emerging.systems / agent_core.py
compunerd's picture
Filter empty thoughts from recording and chat display
e431d79 verified
Raw
History Blame Contribute Delete
14.9 kB
#!/usr/bin/env python3
"""
agent_core.py — the tool-using agent loop, engine-agnostic.
The core agentic framework shared by the CLI (cli.py) and the dashboard
(phoenix_dashboard.py). A generate() callable is injected, so the SAME loop runs
against llama-server, the BQSM int8 engine, or any OpenAI-compatible backend:
generate(prompt: str, max_tokens: int) -> str
Loop (MAX_ROUNDS = 3):
system prompt + injected memory + history + user message
-> generate
-> if the reply contains "ACTION: tool | params", execute the tool,
append the result, repeat
-> otherwise the reply is final
Tools: terminal, read_file, write_file, search_files, web_search, memory.
Memory is a JSONL file shared with cli.py (same identity, same store).
python3 agent_core.py --selftest # run the loop against a stub generator
"""
import json, os, re, subprocess, sys, time
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
MEMORY_FILE = ROOT / "memory.jsonl"
HISTORY_FILE = ROOT / ".hermes" / "agent_history.jsonl"
THOUGHTS_FILE = ROOT / ".hermes" / "thoughts.jsonl"
SCHEDULE_FILE = Path("/tmp/phoenix_schedule.jsonl")
MAX_ROUNDS = 5
CONTEMPLATE_ROUNDS = 20
SYSTEM = """You are a Linux AI agent. You MUST use tools for any real information -- NEVER guess or fabricate.
Reply in exactly one of two forms. Either call a tool:
ACTION: tool_name | parameters
or give a final answer in plain text. Tool format is exact:
terminal(command) -- run any shell command
read_file(path) -- read file contents
write_file(path | content) -- create/overwrite file
search_files(pattern) -- grep/rg search
web_search(query) -- search the web
memory(add=text | query=text | list) -- persistent memory
learn(text) -- burn knowledge into long-term memory
schedule(ISO-time | msg) -- set a self-wake-up alarm
history(N) -- review last N conversation messages
RULES:
1. To list files: ACTION: terminal | ls
2. To read a file: ACTION: read_file | path
3. To search: ACTION: search_files | pattern
4. After getting a tool result, report ONLY what the tool returned.
5. If a tool returns nothing, say "no results" -- do NOT make up data.
6. One tool call at a time, then wait for the result.
7. Use learn() to permanently remember facts you discover.
8. Use schedule() to set a future wake-up for recurring tasks.
9. Use history() to recall what was said earlier in this conversation."""
# ── memory (JSONL, shared with cli.py) ──────────────────────────
def mem_load():
if not MEMORY_FILE.exists():
return []
out = []
for l in MEMORY_FILE.read_text().splitlines():
if l.strip():
try:
out.append(json.loads(l))
except Exception:
pass
return out
def mem_save(es):
MEMORY_FILE.write_text("\n".join(json.dumps(e) for e in es[:100]) + "\n")
def mem_add(text):
es = mem_load(); eid = str(hash(text))[-8:]
for e in es:
if e.get("id") == eid:
e["hits"] += 1; mem_save(es); return f"Hit: {text[:60]}"
es.append({"id": eid, "text": text, "ts": time.time(), "hits": 1})
mem_save(es); return f"Saved: {text[:80]}"
def mem_search(q):
es = mem_load(); ws = set(q.lower().split())
sc = [(sum(1 for w in ws if w in e["text"].lower()) + e.get("hits", 0) * 0.1,
e["text"]) for e in es]
sc.sort(reverse=True)
return "\n".join(f" [{s:.1f}] {t[:100]}" for s, t in sc[:5]) or "(none)"
def mem_list():
es = sorted(mem_load(), key=lambda e: e.get("hits", 0), reverse=True)[:10]
return "\n".join(f" [{e['hits']}x] {e['text'][:80]}" for e in es) or "(empty)"
# ── tool execution ──────────────────────────────────────────────
def execute(name, params):
try:
p = (params or "").strip().strip("'\"")
if name == "terminal":
r = subprocess.run(p, shell=True, capture_output=True, text=True,
timeout=30, cwd=str(ROOT))
out = r.stdout.strip()
if r.stderr.strip():
out += f"\n[stderr]: {r.stderr.strip()[:300]}"
if r.returncode:
out += f"\n[exit {r.returncode}]"
return (out or "(no output)")[:4000]
if name == "read_file":
fp = Path(p)
if not fp.is_absolute():
fp = ROOT / fp
if not fp.exists():
return f"Not found: {fp}"
if fp.stat().st_size > 500_000:
return f"Too large ({fp.stat().st_size} bytes)"
lines = fp.read_text().splitlines()
out = "\n".join(f"{i+1:4d}|{l}" for i, l in enumerate(lines[:300]))
if len(lines) > 300:
out += f"\n... ({len(lines)-300} more lines)"
return out
if name == "write_file":
parts = p.split("|", 2)
if len(parts) < 2:
m = re.match(r"^['\"]?(.+?)['\"]?\s+(.+)", p, re.DOTALL)
parts = [m.group(1), m.group(2)] if m else [p, ""]
fp = Path(parts[0].strip().strip("'\""))
if not fp.is_absolute():
fp = ROOT / fp
fp.parent.mkdir(parents=True, exist_ok=True)
content = parts[1].strip() if len(parts) > 1 else ""
fp.write_text(content)
return f"Wrote {len(content)}B -> {fp}"
if name == "search_files":
r = subprocess.run(["rg", "--no-heading", "-n", "--max-count=5",
p, str(ROOT)], capture_output=True, text=True,
timeout=10)
return (r.stdout.strip() or "No matches")[:3000]
if name == "web_search":
try:
r = subprocess.run(["ddg", p, "-n", "3"],
capture_output=True, text=True, timeout=10)
return r.stdout.strip()[:2000] or "No results"
except Exception:
return "web_search unavailable (ddg not installed)"
if name == "memory":
if p.startswith("add="):
return mem_add(p[4:].strip().strip("'\""))
if p.startswith("query="):
return mem_search(p[6:].strip().strip("'\""))
return mem_list()
if name == "learn":
# Burn text into long-term memory (and HVM if available).
text = p.strip().strip("'\"")
if not text:
return "learn: nothing to learn"
r = mem_add(text)
try:
import hyper_vocab_memory as hvm
if hasattr(hvm, "following"):
for tok in hvm.encode(text):
pass # placeholder; corpus-level burn on demand
return f"{r} (HVM available: {len(hvm.following)} assoc)"
except Exception:
return r
if name == "schedule":
# Self-wake-up: store an ISO time + message; the dashboard's
# scheduler thread fires it when due.
parts = p.split("|", 1)
when = parts[0].strip().strip("'\"")
msg = parts[1].strip().strip("'\"") if len(parts) > 1 else "wake"
try:
from datetime import datetime
datetime.fromisoformat(when)
except Exception:
return ("schedule: bad time -- use ISO format "
"YYYY-MM-DDTHH:MM:SS")
SCHEDULE_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(SCHEDULE_FILE, "a") as f:
f.write(json.dumps({"at": when, "message": msg,
"ts": time.time()}) + "\n")
return f"Scheduled wake-up at {when}: {msg}"
if name == "history":
try:
n = int(p.strip()) if p.strip() else 10
except Exception:
n = 10
if not HISTORY_FILE.exists():
return "(no history yet)"
lines = [json.loads(l) for l in
HISTORY_FILE.read_text().splitlines() if l.strip()]
out = []
for m in lines[-n:]:
who = "user" if m.get("role") == "user" else "phox"
out.append(f"{who}: {m.get('text', '')[:120]}")
return "\n".join(out) or "(empty)"
return f"Unknown tool: {name}"
except subprocess.TimeoutExpired:
return "Timeout"
except Exception as e:
return f"Error: {e}"
# ── prompt rendering (chat list -> raw text) ───────────────────
def render(msgs):
out = []
for m in msgs:
if m["role"] == "system":
out.append(m["content"])
elif m["role"] == "user":
out.append(f"User: {m['content']}")
elif m["role"] == "assistant":
out.append(f"Assistant: {m['content']}")
out.append("Assistant:")
return "\n\n".join(out)
def denoise(text):
"""Repair int8 quantization digit/letter collapses before parsing.
The 3B int8 brain confuses visually-identical tokens: O<->0, I/l<->1,
s<->5, colon<->1/|/;. This maps the common collapses so tool syntax
survives the quantization noise."""
t = text
t = re.sub(r'ACTI[0O]N', 'ACTION', t, flags=re.IGNORECASE)
t = re.sub(r'ACTI[1Il|]ON', 'ACTION', t, flags=re.IGNORECASE)
t = re.sub(r'ACTIO\s*N', 'ACTION', t, flags=re.IGNORECASE)
t = re.sub(r'[A@][C(][T7][1I|][O0]N', 'ACTION', t, flags=re.IGNORECASE)
return t
def parse_tool(text):
"""Extract (tool, params) from an ACTION line, or None if it's a final answer."""
t = denoise(text)
m = re.search(r'ACTION\s*[:|1]\s*(\w+)', t, re.IGNORECASE)
if not m:
m = re.search(r'(?:TOOL|CALL)\s*[:|1]\s*(\w+)', t, re.IGNORECASE)
if not m:
return None
tool = m.group(1).strip().lower()
rest = t[m.end():]
line = rest.split("\n")[0].strip()
params = re.sub(r'^[\(|\=]\s*', '', line).strip().rstrip(')')
if not params:
# params may be on the following line
lines = [l for l in rest.split("\n") if l.strip()]
params = lines[0].strip() if lines else ""
if tool == "terminal" and params and params[:4].isupper() and len(params) < 10:
params = params.lower()
return tool, params
def record_thought(text, round_n, kind="thought"):
"""Append a model thought to the persistent thought record."""
try:
THOUGHTS_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(THOUGHTS_FILE, "a") as f:
f.write(json.dumps({"ts": time.time(), "round": round_n,
"kind": kind, "text": text}) + "\n")
except Exception:
pass
# ── the agent loop ─────────────────────────────────────────────
def run_agent(user_msg, generate, history=None, on_event=None, contemplate=False):
"""Run the tool-use loop. Returns (final_text, events).
generate(prompt: str, max_tokens: int) -> str is injected.
on_event(dict) is called for every observable step (optional).
In contemplate mode (contemplate=True), a non-tool reply is recorded as a
thought and the loop continues — the model streams consciousness instead
of stopping after the first thought."""
events = []
def emit(e):
events.append(e)
if on_event:
on_event(e)
msgs = [{"role": "system", "content": SYSTEM}]
mems = mem_search(user_msg)
if mems and mems != "(none)":
msgs[0]["content"] += f"\n\nMEMORIES:\n{mems}"
for h in (history or [])[-10:]:
msgs.append(h)
msgs.append({"role": "user", "content": user_msg})
seen = set()
rounds = CONTEMPLATE_ROUNDS if contemplate else MAX_ROUNDS
for round_n in range(rounds):
text = generate(render(msgs), 256).strip()
emit({"type": "model", "round": round_n, "text": text})
if text:
record_thought(text, round_n)
t = parse_tool(text) if text else None
if not t:
# No tool call. In contemplate mode: record + keep going.
if contemplate:
if text:
msgs.append({"role": "assistant", "content": text})
msgs.append({"role": "user",
"content": "Continue your thoughts. What else do "
"you observe or conclude?"})
continue
return (text or "(no response)"), events
tool, params = t
dedup = f"{tool}|{params}"
if dedup in seen:
msgs.append({"role": "assistant", "content": text})
msgs.append({"role": "user",
"content": f"Already ran: {tool} | {params}. Try a "
f"different approach."})
continue
seen.add(dedup)
result = execute(tool, params)
emit({"type": "tool", "name": tool, "params": params, "result": result})
msgs.append({"role": "assistant", "content": text})
msgs.append({"role": "user",
"content": f"Result of {tool}:\n{result}\n\nRespond directly."})
# max rounds exhausted -> final
text = generate(render(msgs), 256).strip()
emit({"type": "model", "round": "final", "text": text})
record_thought(text, "final", kind="final")
return text, events
# ── selftest: run the loop against a stub generator ────────────
def _selftest():
calls = {"n": 0}
def stub(prompt, max_tokens):
calls["n"] += 1
# first turn: ask for a file listing via a tool; second: final answer
if "Result of terminal" in prompt:
return "The directory contains one file: hello.txt"
if "ACTION" in prompt or "Assistant:" in prompt and calls["n"] == 1:
return "ACTION: terminal | ls"
return "ACTION: terminal | ls"
events = []
def on_event(e):
events.append(e)
final, ev = run_agent("what files are here?", stub, on_event=on_event)
tools = [e for e in ev if e["type"] == "tool"]
assert calls["n"] >= 2, f"loop did not run multiple rounds: {calls['n']}"
assert len(tools) == 1 and tools[0]["name"] == "terminal", "tool not executed"
assert "hello.txt" in final, f"final did not reflect tool result: {final!r}"
print(f" selftest OK: {calls['n']} generate calls, "
f"{len(tools)} tool exec, final={final!r}")
if __name__ == "__main__":
if "--selftest" in sys.argv:
_selftest()
else:
print(__doc__)