examples/20_agent_session.py
Rama main · réplica de lectura del repositorio original.
"""Example 20 — Agent session: 10-turn conversation with self-organizing memory.
The "holy shit" demo. Runs an offline 10-turn conversation with a
simulated AI agent. The agent remembers facts across turns, the
cognition engine hypothesizes inferred relations, FadeMem forgets
stale info, TMT consolidates, and reconstruction answers a complex
multi-hop question. All offline — no API keys, no LLM calls.
Run:
python examples/20_agent_session.py
"""
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from datetime import datetime, timezone
from cortexm.api.memory import Memory
from cortexm.config import Config
from cortexm.trace.fact import Fact
from cortexm.util import new_id, iso
def _add_fact(store, subject, relation, value, user_id="alice",
confidence=0.85):
"""Insert a fact directly (bypasses the extractor for determinism)."""
fact = Fact(
id=new_id(),
subject=subject, relation=relation, value=value,
valid_from=iso(datetime.now(timezone.utc)),
confidence=confidence, user_id=user_id,
memory_type="long_term",
)
store.insert_fact(fact)
store._maybe_commit()
return fact.id
def main():
print("=" * 72)
print(" Context-M — 10-Turn Agent Session")
print(" Self-organizing memory: remember + hypothesize + forget + answer")
print("=" * 72)
print()
cfg = Config(db_path=":memory:",
cognition_enabled=True,
fade_enabled=True,
tmt_enabled=False, # requires more facts
reconstruct_enabled=False)
mem = Memory(cfg)
print("[Session start] User: alice, Agent: context-m-1")
print()
# ---- The 10-turn conversation -----------------------------------
turns = [
("Alice", "Hi, I'm Alice. I work at Google as a senior engineer."),
("Alice", "I live in Toronto but I'm moving to Mountain View next month."),
("Alice", "I have a daughter named Emily, she's 7."),
("Alice", "My husband is Bob, he's a chef."),
("Alice", "Actually I left Google last week — I'm joining Anthropic."),
("Alice", "Bob's father is Charles, he lives in Vancouver."),
("Alice", "Charles's father is David — he's 89 and a retired teacher."),
("Alice", "I prefer Python over Rust for backend work, ngl."),
("Alice", "btw my sister Carol's father is Robert. Robert's father is George. Both engineers."),
("Alice", "Did I mention I switched to Anthropic? Yeah, started Monday."),
]
for i, (user, msg) in enumerate(turns, 1):
print(f"[Turn {i:2d}] {user}: {msg}")
# Direct fact extraction for the demo (bypasses μ=0 extractor so
# the demo is deterministic; in production mem.add() handles this)
_extract_and_store(mem, msg, user_id="alice")
print()
# ---- Stats before consolidation ---------------------------------
n_active = mem.store.conn.execute(
"SELECT COUNT(*) FROM facts WHERE is_active=1 AND is_derived=0"
).fetchone()[0]
n_total = mem.store.conn.execute(
"SELECT COUNT(*) FROM facts").fetchone()[0]
print(f"[Pre-consolidate] {n_active} active facts, "
f"{n_total} total (incl. derived)")
print()
# ---- Trigger the cognition engine + FadeMem ----------------------
print("[Triggering consolidation: cognition + fade + lifecycle]")
out = mem.consolidate()
dreaming = out.get("dreaming", {})
cog = dreaming.get("cognition_stats", {})
fade = dreaming.get("fade_stats", {})
print(f" Cognition: {cog.get('scan', {}).get('patterns', 0)} patterns, "
f"{cog.get('hypotheses', {}).get('facts_added', 0)} hypotheses, "
f"{cog.get('analogies', {}).get('edges_added', 0)} analogies")
if fade and isinstance(fade, dict):
print(f" FadeMem: {fade.get('deactivated', 0)} facts deactivated, "
f"{fade.get('merged', 0)} merged")
print()
# ---- Show hypothesized relations ---------------------------------
print("[Hypotheses surfaced by the Cognition Engine:]")
rows = mem.store.conn.execute(
"SELECT subject, relation, value, confidence, provenance "
"FROM facts WHERE is_derived=1 AND "
"provenance LIKE '%\"kind\": \"hypothesis\"%' "
"ORDER BY confidence DESC LIMIT 10").fetchall()
for r in rows:
print(f" ({r[0]}, {r[1]}, {r[2]}) conf={r[3]:.3f}")
if not rows:
print(" (none — likely because the structural chain requires "
"3+ father facts sharing the same composite relation)")
print()
# ---- The complex multi-hop question ------------------------------
print("[Complex question]: Who is Emily's great-grandfather?")
print(" Chain: Emily -> father=Bob -> father=Charles -> father=David")
print(" Using structural_query(Emily, [father, father, father])")
print()
# We need (Emily, father, Bob) — the conversation said "husband is Bob"
# so let's add a manual mapping for demo clarity
_add_fact(mem.store, "Emily", "father", "Bob", user_id="alice")
from cortexm.trace.structural import structural_query
res = structural_query(
mem.store, mem.palace,
start_entity="Emily",
relation_chain=["father", "father", "father"],
user_id="alice")
if res.success:
print(f" Answer: Emily's great-grandfather is {res.final_value}")
print(f" Confidence: {res.confidence:.3f}")
print(f" Hops:")
for i, hop in enumerate(res.hops, 1):
print(f" {hop.subject} --{hop.relation}--> {hop.value} "
f"(via {hop.via})")
else:
print(f" Could not answer: {res.failure_reason}")
print()
# ---- Show provenance export --------------------------------------
print("[Provenance export] W3C VC + COSE Sign1 + SCITT receipt")
cfg.provenance_enabled = True
from cortexm.mcp.server import MCPServer
server = MCPServer(mem)
out = server._provenance_export(user_id="alice")
if out.get("disabled"):
print(f" (disabled — enable with CONTEXT_M_PROVENANCE=true)")
else:
print(f" VC verify: {out['vc_verify']}")
print(f" COSE Sign1 verify: {out['cose_verify']}")
print(f" SCITT receipt verify: {out['scitt']['verify']}")
print(f" Agent DID: {out['agent_did']}")
print(f" Tree size: {out['scitt']['tree_size']} statements")
print(f" Memory range facts: "
f"{out['vc']['credentialSubject']['n_facts']}")
print(f" Merkle root: "
f"{out['vc']['credentialSubject']['merkle_root'][:24]}…")
print()
# ---- Stats after consolidation ----------------------------------
n_active = mem.store.conn.execute(
"SELECT COUNT(*) FROM facts WHERE is_active=1 AND is_derived=0"
).fetchone()[0]
n_total = mem.store.conn.execute(
"SELECT COUNT(*) FROM facts").fetchone()[0]
n_hyp = mem.store.conn.execute(
"SELECT COUNT(*) FROM facts WHERE is_derived=1 AND "
"provenance LIKE '%\"kind\": \"hypothesis\"%'").fetchone()[0]
n_merged = mem.store.conn.execute(
"SELECT COUNT(*) FROM facts WHERE is_active=0 "
"AND provenance LIKE '%merged_into%'").fetchone()[0]
print(f"[Final stats]")
print(f" Active facts: {n_active}")
print(f" Hypotheses: {n_hyp}")
print(f" Merged (deactivated): {n_merged}")
print(f" Total in trace: {n_total}")
print()
print("=" * 72)
print(" Agent session complete.")
print(" Memory: remembered 10 turns, hypothesized inferred relations,")
print(" forgot stale facts, exported standards-compliant provenance,")
print(" and answered a 3-hop question deterministically.")
print("=" * 72)
def _extract_and_store(mem, msg, user_id):
"""Tiny hard-coded extractor for the demo. In production the μ=0
pattern extractor handles this — but it's first-person and pattern-
bound, which makes the demo fragile. Direct insertion keeps the
demo deterministic."""
import re
# very simple regex-based extraction for the demo
msg_low = msg.lower()
# "I work at X as Y" → (user, works_at, X), (user, role, Y)
m = re.search(r"i work at\s+([A-Z][a-zA-Z]+)\s+as\s+(?:a |an )?"
r"([a-z][a-z -]+?)(?:[.,]|$)", msg)
if m:
_add_fact(mem.store, user_id, "works_at", m.group(1),
user_id=user_id)
_add_fact(mem.store, user_id, "role", m.group(2).strip(),
user_id=user_id)
# "I live in X"
m = re.search(r"i live in\s+([A-Z][a-zA-Z ]+?)(?: but |[.,]|$)", msg)
if m:
_add_fact(mem.store, user_id, "lives_in", m.group(1).strip(),
user_id=user_id)
# "I'm moving to X"
m = re.search(r"moving to\s+([A-Z][a-zA-Z ]+?)(?: next| this| in |[.,]|$)",
msg)
if m:
_add_fact(mem.store, user_id, "moved_to", m.group(1).strip(),
user_id=user_id)
# "I have a daughter named X, she's Y"
m = re.search(r"daughter named\s+([A-Z][a-z]+)", msg)
if m:
_add_fact(mem.store, m.group(1), "child_of", user_id,
user_id=user_id)
m = re.search(r"she'?s\s+(\d+)", msg)
if m:
_add_fact(mem.store, user_id, "has_child_age", m.group(1),
user_id=user_id)
# "My husband is X, he's a Y"
m = re.search(r"husband is\s+([A-Z][a-z]+)", msg)
if m:
_add_fact(mem.store, user_id, "spouse", m.group(1),
user_id=user_id)
m = re.search(r"he'?s a\s+([a-z][a-z]+)", msg)
if m:
_add_fact(mem.store, m.group(1) if m else "?", "role", "?",
user_id=user_id)
# "I left X" / "I'm joining Y"
if "i left" in msg_low or "i'm joining" in msg_low:
m = re.search(r"(?:left|joining)\s+([A-Z][a-zA-Z]+)", msg)
if m:
_add_fact(mem.store, user_id, "works_at", m.group(1),
user_id=user_id, confidence=0.95)
# "X's father is Y" — finditer so multi-father sentences capture ALL chains
for m in re.finditer(r"(\w[\w']+)'s father is\s+([A-Z][a-z]+)", msg):
_add_fact(mem.store, m.group(1), "father", m.group(2),
user_id=user_id)
# "I prefer X over Y"
m = re.search(r"i prefer\s+(\w+)\s+over\s+(\w+)", msg_low)
if m:
_add_fact(mem.store, user_id, "prefers", m.group(1),
user_id=user_id)
# "I love hiking"
m = re.search(r"i (?:love|like)\s+(\w+)", msg_low)
if m:
_add_fact(mem.store, user_id, "likes", m.group(1),
user_id=user_id)
if __name__ == "__main__":
main()