第 10 章 AI AgentsLangChainLangGraph
第 10 章 Agent 记忆：持久化让 Agent 成为演化系统
用 LangGraph 实现 Agent 记忆
关于本 notebook
LangGraph memory：short-term、semantic、episodic、procedural、time travel
本 notebook 涵盖的内容
- 短期：
messages+ checkpointer（thread_id）；可选的/thread summarize与RemoveMessage。 - 长期：
BaseStore—— semantic（事实 + embeddings）、episodic（任务/结果）、procedural（指令），以及用于丰富注入的结构化 profile blob。 - 组织 vs 用户：
(org_id, user_id, kind)vs(org_id, "_shared", kind)。org_id/user_id必须来自认证，绝不能来自模型。 - OpenRouter 用于聊天和 embeddings —— 没有桩实现。
- 组织写入要求
AppContext.is_admin；模型单独的update_global_policy不构成授权。 - 结构化记忆 LLM 按间隔 / 告别 / … 策略（
MEMORY_LLM_*环境变量）运行，而不是每一轮都运行。
概念
用户应该能够查看和编辑 Agent 记住的内容（信任、GDPR）。
服务层 —— UserMemoryService 由图节点和未来的 HTTP API 共享。
卫生管理 —— /thread summarize 和 /memory compact 让历史和事实保持可控。
记忆投毒防护 —— 校验输入；AppContext.is_admin（或来自认证的角色）是组织命名空间写入所必需的，而不是单独的模型标志。
设计加固（架构）
- 规范 profile： profile blob 中的
facts[]是权威数据。Semantic store 行是搜索投影（source_type=profile_projection、projection_of=<fact_id>）。临时捕获使用transient:键和lifecycle=proposed，直到被提升进 profile。 - 两阶段管线（产品）： 这里保留单个 LLM 以便教学；生产环境中把提取与应用分开，以便测试和审计。
- 多 worker： 内存 store/checkpointer 不是并发安全的；这里不演示事务性或幂等的后台同步。
- 旧版
remember:/episode:： 使用LEGACY_REMEMBER_EPISODE_SYNTAX=0禁用；优先使用 tools 或显式 API。
%pip install -q "langgraph>=0.2" "langchain>=0.3" "langchain-openai>=0.3" "langchain-core>=0.3" "python-dotenv>=1.0" "pydantic>=2" "tiktoken>=0.7" "pycryptodome>=3.20"from __future__ import annotations
import os
import re
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any
from dotenv import load_dotenv
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, RemoveMessage, SystemMessage
from langchain_core.runnables import RunnableConfig
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
load_dotenv()
OPENROUTER_API_KEY = os.environ.get("OPENROUTER_API_KEY", "").strip()
OPENROUTER_BASE_URL = os.getenv("OPENROUTER_BASE_URL", "https://openrouter.ai/api/v1").rstrip("/")
OPENROUTER_CHAT_MODEL = os.getenv("OPENROUTER_CHAT_MODEL", "openai/gpt-4o-mini")
OPENROUTER_MEMORY_MODEL = os.getenv("OPENROUTER_MEMORY_MODEL", OPENROUTER_CHAT_MODEL)
OPENROUTER_EMBEDDING_MODEL = os.getenv("OPENROUTER_EMBEDDING_MODEL", "openai/text-embedding-3-small")
EMBEDDING_DIMS = int(os.getenv("OPENROUTER_EMBEDDING_DIMS", "1536"))
OPENROUTER_HTTP_REFERER = os.getenv("OPENROUTER_HTTP_REFERER", "https://localhost")
OPENROUTER_APP_TITLE = os.getenv("OPENROUTER_APP_TITLE", "LangGraph Memory Tutorial")
def _require_openrouter() -> None:
if not OPENROUTER_API_KEY:
raise RuntimeError(
"Set OPENROUTER_API_KEY (e.g. in .env). This notebook uses real models only — no stubs."
)
def openrouter_default_headers() -> dict[str, str]:
return {"HTTP-Referer": OPENROUTER_HTTP_REFERER, "X-Title": OPENROUTER_APP_TITLE}
def build_chat_llm(*, model: str | None = None, temperature: float = 0.0) -> ChatOpenAI:
_require_openrouter()
return ChatOpenAI(
model=model or OPENROUTER_CHAT_MODEL,
api_key=OPENROUTER_API_KEY,
base_url=OPENROUTER_BASE_URL,
default_headers=openrouter_default_headers(),
temperature=temperature,
)
def build_embeddings() -> OpenAIEmbeddings:
_require_openrouter()
return OpenAIEmbeddings(
model=OPENROUTER_EMBEDDING_MODEL,
api_key=OPENROUTER_API_KEY,
base_url=OPENROUTER_BASE_URL,
default_headers=openrouter_default_headers(),
check_embedding_ctx_length=False,
)
_require_openrouter()
llm = build_chat_llm(model=OPENROUTER_CHAT_MODEL)
memory_llm = build_chat_llm(model=OPENROUTER_MEMORY_MODEL)
embeddings = build_embeddings()
def safe_embed_documents(texts: list[str]) -> list[list[float]]:
"""Some providers return no `data` for empty strings; LangGraph may request embeddings for empty index fields."""
if not texts:
return []
placeholder = os.getenv("EMBED_EMPTY_PLACEHOLDER", "[empty]")
cleaned = [(t if isinstance(t, str) and t.strip() else placeholder) for t in texts]
return embeddings.embed_documents(cleaned)
print(
f"OpenRouter OK — chat={OPENROUTER_CHAT_MODEL!r}, memory={OPENROUTER_MEMORY_MODEL!r}, "
f"embeddings={OPENROUTER_EMBEDDING_MODEL!r}, dims={EMBEDDING_DIMS}"
)OpenRouter OK — chat='openai/gpt-4o-mini', memory='openai/gpt-4o-mini', embeddings='openai/text-embedding-3-small', dims=1536import copy
import logging
from datetime import datetime, timezone
log = logging.getLogger("unified_memory")
LEGACY_REMEMBER_EPISODE_SYNTAX = os.getenv("LEGACY_REMEMBER_EPISODE_SYNTAX", "1").strip().lower() in (
"1",
"true",
"yes",
)
MEMORY_CONTENT_MAX_CHARS = int(os.getenv("MEMORY_CONTENT_MAX_CHARS", "4000"))
SEMANTIC_TRANSIENT_PREFIX = "transient:"
ORG_SHARED_ALLOWED_CATEGORIES = frozenset(
x.strip().lower()
for x in os.getenv(
"ORG_SHARED_ALLOWED_CATEGORIES",
"policy,compliance,legal,disclaimer,org_naming,knowledge,context",
).split(",")
if x.strip()
)
CATEGORY_COMBINE_MODE: dict[str, str] = {
"preference": "replace",
"correction": "replace",
"behavior": "replace",
"goal": "merge",
"knowledge": "merge",
"context": "accumulate",
}
FACT_LIFECYCLE_PROPOSED = "proposed"
FACT_LIFECYCLE_CONFIRMED = "confirmed"
FACT_LIFECYCLE_SUPERSEDED = "superseded"
MEMORY_QUARANTINE_CONFIDENCE = float(os.getenv("MEMORY_QUARANTINE_CONFIDENCE", "0.82"))
def validate_durable_memory_text(text: str, *, max_chars: int = MEMORY_CONTENT_MAX_CHARS) -> str:
t = (text or "").strip()
if len(t) > max_chars:
raise ValueError(f"Memory content exceeds max length ({max_chars} chars).")
if re.search(r"\b\d{3}-\d{2}-\d{4}\b", t):
raise ValueError("Refusing possible SSN-like pattern in durable memory.")
if re.search(r"\b(?:\d[ -]*?){13,19}\b", re.sub(r"\s+", " ", t)):
raise ValueError("Refusing possible payment-card-like digit run in durable memory.")
low = t.lower()
if "-----begin" in low and "private key" in low:
raise ValueError("Refusing possible PEM private key material.")
if re.search(r"\bAKIA[0-9A-Z]{16}\b", t):
raise ValueError("Refusing possible AWS access key id in durable memory.")
if re.search(r"\bsk-[A-Za-z0-9]{20,}\b", t):
raise ValueError("Refusing possible API secret pattern in durable memory.")
return t
def _parse_iso_utc(s: str | None) -> datetime | None:
if not s or not isinstance(s, str):
return None
try:
s2 = s.replace("Z", "+00:00")
return datetime.fromisoformat(s2)
except ValueError:
return None
def filter_org_facts_by_policy(memory: dict) -> dict:
mem = copy.deepcopy(memory)
facts = mem.get("facts") or []
kept = []
for f in facts:
if not isinstance(f, dict):
continue
cat = str(f.get("category", "context") or "context").strip().lower() or "context"
if cat in ORG_SHARED_ALLOWED_CATEGORIES:
kept.append(f)
else:
log.info("Dropped org fact (category not allowlisted): %s", cat)
mem["facts"] = kept
return mem
def deterministic_reconcile_profile_facts(memory: dict) -> dict:
mem = copy.deepcopy(memory)
facts = [f for f in (mem.get("facts") or []) if isinstance(f, dict)]
to_remove: set[str] = set()
pairs = [
(
re.compile(r"\b(dark mode|dark theme|dark ui)\b", re.I),
re.compile(r"\b(light mode|light theme|light ui)\b", re.I),
),
]
for i, fa in enumerate(facts):
ca = str(fa.get("content", ""))
cat_a = str(fa.get("category", "") or "").lower()
if cat_a not in ("preference", "behavior", "correction", "context", ""):
continue
for j, fb in enumerate(facts):
if j <= i:
continue
cb = str(fb.get("content", ""))
for pa, pb in pairs:
if pa.search(ca) and pb.search(cb):
ia = _parse_iso_utc(fa.get("updatedAt")) or _parse_iso_utc(fa.get("createdAt"))
ib = _parse_iso_utc(fb.get("updatedAt")) or _parse_iso_utc(fb.get("createdAt"))
if ia and ib and ib < ia:
older = fb
else:
older = fa
oid = older.get("id")
if oid:
to_remove.add(str(oid))
if to_remove:
mem["facts"] = [f for f in facts if str(f.get("id")) not in to_remove]
return memimport copy
import logging
import math
import os
from typing import Any
log = logging.getLogger("unified_memory")
if not logging.getLogger().handlers:
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
NS_SEMANTIC = "semantic"
NS_EPISODIC = "episodic"
NS_PROCEDURAL = "procedural"
NS_PROFILE = "profile"
SHARED_SEGMENT = "_shared"
PROFILE_DOC_KEY = "document"
PROFILE_AGENT_INSTRUCTIONS = "agent_instructions"
CONSOLIDATED_KEY = "consolidated_profile"
@dataclass(frozen=True)
class AppContext:
org_id: str
user_id: str
is_admin: bool = False
def _safe_segment(s: str) -> str:
cleaned = re.sub(r"[^a-zA-Z0-9._-]+", "_", s.strip())[:120]
if not cleaned or cleaned in {".", ".."}:
raise ValueError(f"Invalid id segment: {s!r}")
return cleaned
def ns_user(ctx: AppContext, kind: str) -> tuple[str, str, str]:
return (_safe_segment(ctx.org_id), _safe_segment(ctx.user_id), kind)
def ns_org_shared(org_id: str, kind: str) -> tuple[str, str, str]:
return (_safe_segment(org_id), SHARED_SEGMENT, kind)
def ns_profile_user(ctx: AppContext) -> tuple[str, str, str]:
return ns_user(ctx, NS_PROFILE)
def ns_profile_org(org_id: str) -> tuple[str, str, str]:
return ns_org_shared(org_id, NS_PROFILE)
def _utc_now_z() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S") + "Z"
def empty_memory() -> dict[str, Any]:
return {
"version": "1.0",
"lastUpdated": _utc_now_z(),
"user": {
"workContext": {"summary": "", "updatedAt": ""},
"personalContext": {"summary": "", "updatedAt": ""},
"topOfMind": {"summary": "", "updatedAt": ""},
},
"history": {
"recentMonths": {"summary": "", "updatedAt": ""},
"earlierContext": {"summary": "", "updatedAt": ""},
"longTermBackground": {"summary": "", "updatedAt": ""},
},
"facts": [],
}
def load_profile_blob(store, namespace: tuple[str, ...]) -> dict[str, Any]:
item = store.get(namespace, PROFILE_DOC_KEY)
if item is None:
return empty_memory()
val = item.value
if isinstance(val, dict) and "memory" in val:
return copy.deepcopy(val["memory"])
if isinstance(val, dict):
return copy.deepcopy(val)
return empty_memory()
def save_profile_blob(store, namespace: tuple[str, ...], memory: dict[str, Any]) -> None:
memory = copy.deepcopy(memory)
memory["lastUpdated"] = _utc_now_z()
store.put(namespace, PROFILE_DOC_KEY, {"memory": memory, "updated_at": _utc_now_z()})
try:
import tiktoken
_enc = tiktoken.get_encoding("cl100k_base")
def _count_tokens(text: str) -> int:
return len(_enc.encode(text))
except Exception:
def _count_tokens(text: str) -> int:
return max(1, len(text) // 4)
def _cosine_sim(a: list[float], b: list[float]) -> float:
dot = sum(x * y for x, y in zip(a, b))
na = sum(x * x for x in a) ** 0.5
nb = sum(x * x for x in b) ** 0.5
if na == 0.0 or nb == 0.0:
return 0.0
return dot / (na * nb)
def _coerce_confidence(value: Any, default: float = 0.0) -> float:
try:
x = float(value)
except (TypeError, ValueError):
return max(0.0, min(1.0, default))
if x != x or x in (float("inf"), float("-inf")):
return max(0.0, min(1.0, default))
return max(0.0, min(1.0, x))
def decayed_confidence(fact: dict, *, half_life_days: float = 120.0) -> float:
base = _coerce_confidence(fact.get("confidence"), 0.5)
created = _parse_iso_utc(fact.get("createdAt")) or _parse_iso_utc(fact.get("updatedAt"))
if created is None:
return base
age_days = max(
0.0,
(datetime.now(timezone.utc) - created.astimezone(timezone.utc)).total_seconds() / 86400.0,
)
factor = math.pow(0.5, age_days / max(half_life_days, 1e-6))
return max(0.0, min(1.0, base * factor))
def rank_facts_for_injection(
facts: list[dict[str, Any]],
*,
rerank_query: str,
embed_texts,
top_prefilter: int = 64,
) -> list[dict[str, Any]]:
# Re-order facts by embedding similarity to the query (then confidence).
q = rerank_query.strip()
if not q or not facts:
return facts
take = facts[:top_prefilter]
texts = [q] + [str(f.get("content", "")).strip() for f in take]
vecs = embed_texts(texts)
qv, fv = vecs[0], vecs[1:]
scored: list[tuple[float, float, dict[str, Any]]] = []
for f, v in zip(take, fv):
sim = _cosine_sim(qv, v)
conf = _coerce_confidence(f.get("confidence"), 0.0)
scored.append((sim, conf, f))
scored.sort(key=lambda t: (t[0], t[1]), reverse=True)
return [t[2] for t in scored] + facts[len(take) :]
def format_memory_for_injection(
memory_data: dict[str, Any],
*,
max_tokens: int = 2000,
rerank_query: str | None = None,
embed_texts=None,
) -> str:
if not memory_data:
return ""
sections: list[str] = []
user_data = memory_data.get("user") or {}
user_bits: list[str] = []
for label, key in (
("Work", "workContext"),
("Personal", "personalContext"),
("Current focus", "topOfMind"),
):
block = user_data.get(key) or {}
s = (block.get("summary") or "").strip()
if s:
user_bits.append(f"{label}: {s}")
if user_bits:
nl = chr(10)
sections.append("User context:" + nl + nl.join(f"- {b}" for b in user_bits))
hist = memory_data.get("history") or {}
hist_bits: list[str] = []
for label, key in (
("Recent", "recentMonths"),
("Earlier", "earlierContext"),
("Background", "longTermBackground"),
):
block = hist.get(key) or {}
s = (block.get("summary") or "").strip()
if s:
hist_bits.append(f"{label}: {s}")
if hist_bits:
nl = chr(10)
sections.append("History:" + nl + nl.join(f"- {b}" for b in hist_bits))
facts_raw = memory_data.get("facts") or []
candidates = [f for f in facts_raw if isinstance(f, dict) and str(f.get("content", "")).strip()]
ranked = sorted(
candidates,
key=lambda f: decayed_confidence(
f, half_life_days=float(os.getenv("MEMORY_CONFIDENCE_HALF_LIFE_DAYS", "120"))
),
reverse=True,
)
if rerank_query and embed_texts is not None and len(ranked) > 1:
ranked = rank_facts_for_injection(ranked, rerank_query=rerank_query, embed_texts=embed_texts)
nl2 = chr(10) * 2
base = nl2.join(sections)
base_tokens = _count_tokens(base) if base else 0
header = "Facts:" + chr(10)
sep = nl2 + header if base else header
running = base_tokens + _count_tokens(sep)
fact_lines: list[str] = []
for fact in ranked:
content = str(fact.get("content", "")).strip()
cat = str(fact.get("category", "context") or "context").strip() or "context"
conf = _coerce_confidence(fact.get("confidence"), 0.0)
line = f"- [{cat} | {conf:.2f}] {content}"
add = (chr(10) + line) if fact_lines else line
if running + _count_tokens(add) <= max_tokens:
fact_lines.append(line)
running += _count_tokens(add)
else:
break
if fact_lines:
sections.append("Facts:" + chr(10) + chr(10).join(fact_lines))
out = nl2.join(sections)
if _count_tokens(out) > max_tokens:
ratio = len(out) / max(_count_tokens(out), 1)
out = out[: int(max_tokens * ratio * 0.95)] + chr(10) + "..."
return out
def merge_injection_blocks(
*,
org_memory: dict[str, Any],
user_memory: dict[str, Any],
max_tokens: int = 2400,
rerank_query: str | None = None,
embed_texts=None,
include_org_profile: bool = True,
org_max_tokens: int | None = None,
user_max_tokens: int | None = None,
) -> str:
nl2 = chr(10) * 2
overhead = _count_tokens("### Organization memory (shared)" + nl2 + "### User memory (private)" + nl2)
org_ceiling = org_max_tokens if org_max_tokens is not None else max(120, int(max_tokens * 0.30))
user_ceiling = user_max_tokens if user_max_tokens is not None else max(200, max_tokens - org_ceiling - overhead)
kw = {}
if rerank_query and embed_texts is not None:
kw = {"rerank_query": rerank_query, "embed_texts": embed_texts}
org_block = ""
if include_org_profile:
org_block = format_memory_for_injection(org_memory, max_tokens=org_ceiling, **kw)
org_tokens = _count_tokens(org_block) if org_block.strip() else 0
user_budget = max(200, max_tokens - org_tokens - overhead)
user_budget = min(user_budget, user_ceiling)
user_block = format_memory_for_injection(user_memory, max_tokens=user_budget, **kw)
parts = []
if include_org_profile and org_block.strip():
parts.append("### Organization memory (shared)" + chr(10) + org_block)
if user_block.strip():
parts.append("### User memory (private)" + chr(10) + user_block)
return nl2.join(parts) if parts else "(No structured long-term profile yet.)"from typing import Sequence
from langchain_core.messages import BaseMessage
from pydantic import BaseModel, Field
def format_conversation_for_update(messages: Sequence[BaseMessage]) -> str:
lines: list[str] = []
for msg in messages:
role = getattr(msg, "type", None)
content = getattr(msg, "content", "")
if isinstance(content, list):
text_parts: list[str] = []
for p in content:
if isinstance(p, str):
text_parts.append(p)
elif isinstance(p, dict) and isinstance(p.get("text"), str):
text_parts.append(p["text"])
content = " ".join(text_parts) if text_parts else str(content)
text = str(content).strip()
if len(text) > 1200:
text = text[:1200] + "..."
if role == "human":
lines.append(f"User: {text}")
elif role == "ai":
lines.append(f"Assistant: {text}")
return (chr(10) * 2).join(lines)
class AssistantRouterOutput(BaseModel):
reply: str = Field(..., description="User-facing answer.")
update_global_policy: bool = Field(
False,
description=(
"True only for ORGANIZATION-WIDE information: policies, disclaimers, compliance wording, "
"company-wide naming. False for personal preferences or one-off questions."
),
)
class MemorySectionPatch(BaseModel):
summary: str = ""
shouldUpdate: bool = False
class UserMemoryPatch(BaseModel):
workContext: MemorySectionPatch = Field(default_factory=MemorySectionPatch)
personalContext: MemorySectionPatch = Field(default_factory=MemorySectionPatch)
topOfMind: MemorySectionPatch = Field(default_factory=MemorySectionPatch)
class HistoryMemoryPatch(BaseModel):
recentMonths: MemorySectionPatch = Field(default_factory=MemorySectionPatch)
earlierContext: MemorySectionPatch = Field(default_factory=MemorySectionPatch)
longTermBackground: MemorySectionPatch = Field(default_factory=MemorySectionPatch)
class NewFactItem(BaseModel):
content: str
category: str = "context"
confidence: float = Field(0.8, ge=0.0, le=1.0)
evidence_span: str = Field("", description="Optional short quote from the user turn supporting this fact.")
supersedes_fact_ids: list[str] = Field(
default_factory=list,
description="Optional fact ids this new fact replaces (in addition to factsToRemove).",
)
class MemoryUpdateOutput(BaseModel):
user: UserMemoryPatch = Field(default_factory=UserMemoryPatch)
history: HistoryMemoryPatch = Field(default_factory=HistoryMemoryPatch)
newFacts: list[NewFactItem] = Field(default_factory=list)
factsToRemove: list[str] = Field(default_factory=list)
MEMORY_UPDATE_INSTRUCTIONS = """You are the Memory Manager. Analyze the conversation against the current memory JSON.
**Contradictions**
- When new information overrides an old fact, put outdated fact **id** values in `factsToRemove`, then add replacement in `newFacts`.
**Quality**
- Only set `shouldUpdate` when there is meaningful new information for that section.
- Facts: specific strings; categories: preference, knowledge, context, behavior, goal, correction.
- confidence 0.9+ for explicit statements; 0.7–0.8 for strong implication.
- Optional `evidence_span`: short verbatim snippet from the user message (for audits).
- Optional `supersedes_fact_ids`: ids this fact replaces (redundant with `factsToRemove` but finer-grained).
**Scope** (in human message)
- USER-PRIVATE vs ORG-WIDE as specified.
- ORG-WIDE: prefer category **policy** for customer-facing / pricing / disclaimer rules so allowlisting keeps them.
"""import copy
import json
import uuid
from typing import Any, Sequence
from langchain_core.messages import BaseMessage, HumanMessage
def _fact_key(content: str) -> str | None:
s = content.strip()
return s.casefold() if s else None
def apply_updates(
current_memory: dict[str, Any],
update: MemoryUpdateOutput,
*,
fact_confidence_threshold: float = 0.7,
max_facts: int = 100,
thread_id: str | None = None,
) -> dict[str, Any]:
mem = copy.deepcopy(current_memory)
now = _utc_now_z()
src = thread_id or "unknown"
mem.setdefault("user", empty_memory()["user"])
mem.setdefault("history", empty_memory()["history"])
mem.setdefault("facts", [])
for section in ("workContext", "personalContext", "topOfMind"):
block = getattr(update.user, section)
if block.shouldUpdate and block.summary.strip():
mem["user"][section] = {"summary": block.summary.strip(), "updatedAt": now}
for section in ("recentMonths", "earlierContext", "longTermBackground"):
block = getattr(update.history, section)
if block.shouldUpdate and block.summary.strip():
mem["history"][section] = {"summary": block.summary.strip(), "updatedAt": now}
remove = set(update.factsToRemove)
for nf in update.newFacts:
remove.update(x for x in (nf.supersedes_fact_ids or []) if x)
if remove:
mem["facts"] = [f for f in mem.get("facts", []) if f.get("id") not in remove]
replace_cats: set[str] = set()
for nf in update.newFacts:
c = (nf.category or "context").strip().lower() or "context"
if CATEGORY_COMBINE_MODE.get(c, "accumulate") == "replace":
replace_cats.add(c)
if replace_cats:
mem["facts"] = [
f
for f in mem.get("facts", [])
if str(f.get("category", "context") or "context").strip().lower() not in replace_cats
]
existing_keys = {
k for k in (_fact_key(str(f.get("content", ""))) for f in mem.get("facts", [])) if k
}
for fact in update.newFacts:
conf = _coerce_confidence(fact.confidence, 0.5)
if conf < fact_confidence_threshold:
continue
content = fact.content.strip()
if not content:
continue
fk = _fact_key(content)
if fk and fk in existing_keys:
continue
lifecycle = (
FACT_LIFECYCLE_PROPOSED if conf < MEMORY_QUARANTINE_CONFIDENCE else FACT_LIFECYCLE_CONFIRMED
)
entry = {
"id": f"fact_{uuid.uuid4().hex[:10]}",
"content": content,
"category": (fact.category or "context").strip() or "context",
"confidence": conf,
"createdAt": now,
"updatedAt": now,
"source": src,
"source_type": "llm_structured_update",
"lifecycle": lifecycle,
"needs_review": lifecycle == FACT_LIFECYCLE_PROPOSED,
"last_confirmed_at": now if lifecycle == FACT_LIFECYCLE_CONFIRMED else "",
"source_turn_id": src,
"evidence_span": (fact.evidence_span or "").strip(),
}
mem.setdefault("facts", []).append(entry)
if fk:
existing_keys.add(fk)
facts = mem.get("facts", [])
if len(facts) > max_facts:
mem["facts"] = sorted(
facts, key=lambda f: _coerce_confidence(f.get("confidence"), 0.0), reverse=True
)[:max_facts]
return mem
def extract_memory_update_llm(
memory_llm_structured,
*,
current_memory: dict[str, Any],
messages: Sequence[BaseMessage],
scope: str,
) -> MemoryUpdateOutput | None:
"""Stage 1: LLM proposes a structured patch only (test `apply_updates` separately in production)."""
conv = format_conversation_for_update(messages)
if not conv.strip():
return None
if scope == "org":
scope_line = "Scope: ORG-WIDE memory only — shared across all users in this organization."
scope_line += (
"\nFor newFacts use category **policy** (preferred) or compliance/legal/disclaimer/org_naming/"
"knowledge/context so allowlisting keeps the row."
)
else:
scope_line = "Scope: USER-PRIVATE memory for this individual."
nl = chr(10)
human = (
f"{MEMORY_UPDATE_INSTRUCTIONS}{nl}{nl}{scope_line}{nl}{nl}"
f"Current memory JSON:{nl}{json.dumps(current_memory, indent=2, ensure_ascii=False)}{nl}{nl}"
f"Conversation:{nl}{conv}"
)
try:
update = memory_llm_structured.invoke([HumanMessage(content=human)])
except Exception as e:
log.warning("Memory manager structured invoke failed: %s", e)
return None
if not isinstance(update, MemoryUpdateOutput):
log.warning("Unexpected memory update type: %s", type(update))
return None
return update
def run_memory_update_llm(
memory_llm_structured,
*,
current_memory: dict[str, Any],
messages: Sequence[BaseMessage],
scope: str,
thread_id: str | None,
) -> dict[str, Any] | None:
update = extract_memory_update_llm(
memory_llm_structured,
current_memory=current_memory,
messages=messages,
scope=scope,
)
if update is None:
return None
mem = apply_updates(
current_memory,
update,
fact_confidence_threshold=0.7,
max_facts=100,
thread_id=thread_id,
)
if scope == "org":
mem = filter_org_facts_by_policy(mem)
mem = deterministic_reconcile_profile_facts(mem)
return mem
def sync_profile_facts_semantic(
store,
namespace_semantic: tuple[str, ...],
prev_mem: dict[str, Any],
new_mem: dict[str, Any],
) -> None:
new_facts = [f for f in (new_mem.get("facts") or []) if isinstance(f, dict) and f.get("id")]
canonical_ids = {str(f["id"]) for f in new_facts}
prev_ids = {
str(f.get("id"))
for f in (prev_mem.get("facts") or [])
if isinstance(f, dict) and f.get("id")
}
for rid in prev_ids - canonical_ids:
store.delete(namespace_semantic, rid)
for rid in canonical_ids:
try:
store.delete(namespace_semantic, rid)
except Exception:
pass
now = _utc_now_z()
for f in new_facts:
fid = str(f["id"])
content = str(f.get("content", "")).strip()
if not content:
continue
zw = "\u200c"
store.put(
namespace_semantic,
fid,
{
"fact": content,
"mem_ix": f"{content}{zw}id:{fid}",
"category": f.get("category", "context"),
"confidence": f.get("confidence", 0.8),
"updated_at": now,
"source": "profile",
"source_type": "profile_projection",
"projection_of": fid,
"lifecycle": f.get("lifecycle", "confirmed"),
},
)
for it in list(store.search(namespace_semantic, limit=500)):
v = it.value or {}
if v.get("source_type") == "profile_projection":
pid = str(v.get("projection_of") or it.key)
if pid not in canonical_ids:
store.delete(tuple(it.namespace), it.key)import os
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
def build_checkpointer(*, serde=None):
"""Return a BaseCheckpointSaver. Swap for SqliteSaver / PostgresSaver in production."""
# from langgraph.checkpoint.sqlite import SqliteSaver
# import sqlite3
# conn = sqlite3.connect("langgraph_checkpoints.db", check_same_thread=False)
# return SqliteSaver(conn)
return InMemorySaver(serde=serde) if serde is not None else InMemorySaver()
def build_store() -> InMemoryStore:
"""Semantic index over store values — embeddings via OpenRouter."""
def embed_texts(texts: list[str]) -> list[list[float]]:
return safe_embed_documents(list(texts))
return InMemoryStore(
index={
"embed": embed_texts,
"dims": EMBEDDING_DIMS,
# Single field per row: LangGraph dedupes by string; duplicate `fact` text across keys
# or repeated empty placeholders => len(embeddings) != len(indices). Use unique `mem_ix`.
"fields": ["mem_ix"],
}
)
CHECKPOINTER = build_checkpointer()
STORE = build_store()
print("Checkpointer + store ready (InMemory; embeddings dim=%s)." % EMBEDDING_DIMS)Checkpointer + store ready (InMemory; embeddings dim=1536).Semantic memory —— 隔离操作（无需 graph 即可调用）
from langgraph.store.base import BaseStore
def semantic_list_items(store: BaseStore, ctx: AppContext, *, limit: int = 50):
"""List semantic memories for this org+user. Single concern: semantic namespace only."""
return list(store.search(ns_user(ctx, NS_SEMANTIC), limit=limit))
def semantic_search(
store: BaseStore, ctx: AppContext, query: str, *, limit: int = 6
):
"""Vector search within semantic memories only."""
return list(
store.search(ns_user(ctx, NS_SEMANTIC), query=query, limit=limit)
)
def semantic_delete_key(store: BaseStore, ctx: AppContext, key: str) -> None:
"""Delete one semantic memory key."""
store.delete(ns_user(ctx, NS_SEMANTIC), key)
def semantic_clear_all(store: BaseStore, ctx: AppContext, *, search_limit: int = 500) -> int:
"""Remove every semantic entry for this org+user. Returns number of keys deleted."""
nspath = ns_user(ctx, NS_SEMANTIC)
items = list(store.search(nspath, limit=search_limit))
for it in items:
store.delete(tuple(it.namespace), it.key)
return len(items)
def semantic_upsert_fact(
store: BaseStore,
ctx: AppContext,
key: str,
fact: str,
extra: dict | None = None,
*,
transient: bool = False,
projection_of: str | None = None,
) -> None:
"""Canonical facts live in profile; semantic rows are projections unless transient:."""
text = validate_durable_memory_text(fact)
ex = dict(extra or {})
if projection_of:
ex.setdefault("source_type", "profile_projection")
ex["projection_of"] = projection_of
elif transient or key.startswith(SEMANTIC_TRANSIENT_PREFIX):
ex.setdefault("source_type", "transient_capture")
ex.setdefault("lifecycle", FACT_LIFECYCLE_PROPOSED)
else:
log.warning(
"semantic_upsert_fact: set projection_of= or transient=True / transient: key"
)
zw = "\u200c"
payload = {
"fact": text,
"mem_ix": f"{text}{zw}key:{key}",
"updated_at": _utc_now_z(),
**ex,
}
store.put(ns_user(ctx, NS_SEMANTIC), key, payload)
def semantic_compact_consolidated(
store: BaseStore, ctx: AppContext, llm, *, list_limit: int = 200
) -> tuple[str, int]:
"""Merge semantic atoms into CONSOLIDATED_KEY; returns (summary_text, deleted_count)."""
items = semantic_list_items(store, ctx, limit=list_limit)
if not items:
return "No semantic memories to compact.", 0
nl = chr(10)
facts = nl.join(f"- ({it.key}) {it.value.get('fact', it.value)}" for it in items)
merged = llm.invoke(
[
SystemMessage(
content=(
"Merge these user-specific facts into one structured profile (markdown bullets). "
"De-duplicate aggressively. Drop stale contradictions keeping the newest updated_at if present."
)
),
HumanMessage(content=facts),
]
).content
zw = "\u200c"
m = str(merged)
store.put(
ns_user(ctx, NS_SEMANTIC),
CONSOLIDATED_KEY,
{
"fact": merged,
"mem_ix": f"{m[:6000]}{zw}key:{CONSOLIDATED_KEY}",
"updated_at": _utc_now_z(),
"source": "compact",
},
)
deleted = 0
for it in items:
if it.key == CONSOLIDATED_KEY:
continue
store.delete(tuple(it.namespace), it.key)
deleted += 1
return str(merged), deleted
def semantic_format_lines(store: BaseStore, ctx: AppContext, *, limit: int = 50) -> list[str]:
"""Human-readable lines for UI (semantic only)."""
sem = semantic_list_items(store, ctx, limit=limit)
lines: list[str] = []
lines.append(
f"- Semantic ({len(sem)}): "
+ ", ".join(f"{it.key[:10]}…" for it in sem[:10])
+ (" …" if len(sem) > 10 else "")
)
for it in sem[:5]:
lines.append(f" • [{it.key}] {it.value.get('fact', it.value)}")
return lines Episodic memory —— 隔离操作
def episodic_list_items(store: BaseStore, ctx: AppContext, *, limit: int = 50):
"""List episodic memories. Single concern: episodic namespace only."""
return list(store.search(ns_user(ctx, NS_EPISODIC), limit=limit))
def episodic_search(store: BaseStore, ctx: AppContext, query: str, *, limit: int = 3):
"""Vector search within episodic memories only."""
return list(
store.search(ns_user(ctx, NS_EPISODIC), query=query, limit=limit)
)
def episodic_delete_key(store: BaseStore, ctx: AppContext, key: str) -> None:
store.delete(ns_user(ctx, NS_EPISODIC), key)
def episodic_clear_all(store: BaseStore, ctx: AppContext, *, search_limit: int = 500) -> int:
nspath = ns_user(ctx, NS_EPISODIC)
items = list(store.search(nspath, limit=search_limit))
for it in items:
store.delete(tuple(it.namespace), it.key)
return len(items)
def episodic_record_event(
store: BaseStore, ctx: AppContext, task: str, outcome: str, *, memory_key: str | None = None
) -> str:
"""Store one task/outcome episode; returns the key used."""
key = memory_key or str(uuid.uuid4())
ts, oc = task.strip(), outcome.strip()
zw = "\u200c"
store.put(
ns_user(ctx, NS_EPISODIC),
key,
{
"task": ts,
"outcome": oc,
"mem_ix": f"{ts}\n{oc}{zw}key:{key}",
"created_at": _utc_now_z(),
},
)
return key
def episodic_format_lines(store: BaseStore, ctx: AppContext, *, limit: int = 50) -> list[str]:
epi = episodic_list_items(store, ctx, limit=limit)
lines: list[str] = []
lines.append(f"- Episodic ({len(epi)}): " + ", ".join(it.key[:8] for it in epi[:5]))
for it in epi[:3]:
lines.append(f" • {it.value.get('task')} -> {it.value.get('outcome')}")
return linesProcedural memory —— 隔离操作
def procedural_get_item(store: BaseStore, ctx: AppContext):
"""Return store item for procedural instructions, or None."""
return store.get(ns_user(ctx, NS_PROCEDURAL), PROFILE_AGENT_INSTRUCTIONS)
def procedural_set_instructions(store: BaseStore, ctx: AppContext, instructions: str) -> None:
procedural_put_structured(
store,
ctx,
{"style": instructions.strip(), "workflow": "", "safety": "", "task_policies": ""},
)
def procedural_put_structured(store: BaseStore, ctx: AppContext, fields: dict[str, str]) -> None:
now = _utc_now_z()
cur = procedural_read_structured(store, ctx)
for k in ("style", "workflow", "safety", "task_policies"):
if k in fields and fields[k] is not None:
cur[k] = str(fields[k]).strip()
cur["updated_at"] = now
store.put(ns_user(ctx, NS_PROCEDURAL), PROFILE_AGENT_INSTRUCTIONS, cur)
def procedural_read_structured(store: BaseStore, ctx: AppContext) -> dict[str, str]:
it = procedural_get_item(store, ctx)
if it is None:
return {"style": "", "workflow": "", "safety": "", "task_policies": "", "updated_at": ""}
v = it.value or {}
if isinstance(v, dict) and any(k in v for k in ("style", "workflow", "safety", "task_policies")):
return {
"style": str(v.get("style") or ""),
"workflow": str(v.get("workflow") or ""),
"safety": str(v.get("safety") or ""),
"task_policies": str(v.get("task_policies") or ""),
"updated_at": str(v.get("updated_at") or ""),
}
legacy = str(v.get("instructions", ""))
return {
"style": legacy,
"workflow": "",
"safety": "",
"task_policies": "",
"updated_at": str(v.get("updated_at") or ""),
}
def procedural_format_structured_block(store: BaseStore, ctx: AppContext) -> str:
p = procedural_read_structured(store, ctx)
lines = ["**Procedural (typed slots)**"]
for label, k in (
("Style", "style"),
("Workflow", "workflow"),
("Safety", "safety"),
("Task policies", "task_policies"),
):
if (p.get(k) or "").strip():
lines.append(f"- {label}: {p[k].strip()}")
return chr(10).join(lines) if len(lines) > 1 else ""
def procedural_clear(store: BaseStore, ctx: AppContext) -> None:
store.delete(ns_user(ctx, NS_PROCEDURAL), PROFILE_AGENT_INSTRUCTIONS)
def procedural_ensure_default(store: BaseStore, ctx: AppContext) -> None:
if procedural_get_item(store, ctx) is None:
procedural_set_instructions(
store,
ctx,
"Be concise and helpful. Prefer bullet points for comparisons.",
)
def procedural_format_line(store: BaseStore, ctx: AppContext, *, max_len: int = 200) -> str | None:
block = procedural_format_structured_block(store, ctx)
if not block:
return None
tail = "…" if len(block) > max_len else ""
return block[:max_len] + tailUserMemoryService （轻量门面 —— 委托给上面的隔离函数）
from langgraph.store.base import BaseStore
class UserMemoryService:
"""Stable product API; each method delegates to one isolated concern (see functions above)."""
def __init__(self, store: BaseStore):
self.store = store
def list_semantic(self, ctx: AppContext, limit: int = 50):
return semantic_list_items(self.store, ctx, limit=limit)
def list_episodic(self, ctx: AppContext, limit: int = 50):
return episodic_list_items(self.store, ctx, limit=limit)
def delete_semantic_key(self, ctx: AppContext, key: str) -> None:
semantic_delete_key(self.store, ctx, key)
def delete_episodic_key(self, ctx: AppContext, key: str) -> None:
episodic_delete_key(self.store, ctx, key)
def clear_semantic_all(self, ctx: AppContext) -> int:
return semantic_clear_all(self.store, ctx)
def clear_episodic_all(self, ctx: AppContext) -> int:
return episodic_clear_all(self.store, ctx)
def upsert_semantic(self, ctx: AppContext, key: str, fact: str, extra: dict | None = None) -> None:
semantic_upsert_fact(self.store, ctx, key, fact, extra)
def set_procedural(self, ctx: AppContext, instructions: str) -> None:
procedural_set_instructions(self.store, ctx, instructions)
MEMORY = UserMemoryService(STORE)def _ensure_msg_ids(messages: list[BaseMessage]) -> list[BaseMessage]:
fixed: list[BaseMessage] = []
for m in messages:
mid = getattr(m, "id", None)
if mid is None:
fixed.append(m.model_copy(update={"id": str(uuid.uuid4())}))
else:
fixed.append(m)
return fixed
def _last_human(state) -> HumanMessage | None:
for m in reversed(state["messages"]):
if isinstance(m, HumanMessage):
return m
return None
def summarize_thread_messages(
messages: list[BaseMessage],
*,
drop_command: HumanMessage | None = None,
) -> dict:
"""Short-term hygiene: returns channel update with RemoveMessage + summary HumanMessage."""
msgs = _ensure_msg_ids(list(messages))
body = [m for m in msgs if m is not drop_command]
transcript = (chr(10)).join(
f"{m.type}: {m.content}" for m in body if isinstance(getattr(m, "content", None), str)
)
summary = memory_llm.invoke(
[
SystemMessage(
content=(
"Summarize the conversation for future context in <=8 bullet points. "
"Preserve user preferences and decisions. No preamble."
)
),
HumanMessage(content=transcript),
]
).content
removes = [RemoveMessage(id=m.id) for m in msgs if m.id]
new_human = HumanMessage(
id=str(uuid.uuid4()),
content="Prior conversation (summarized):" + chr(10) + str(summary),
)
return {"messages": removes + [new_human]}Short-term memory & checkpoints（最小 graph）
from operator import add
from typing import Annotated
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict
class PS(TypedDict):
foo: str
bar: Annotated[list[str], add]
def node_a(state: PS):
return {"foo": "a", "bar": ["a"]}
def node_b(state: PS):
return {"foo": "b", "bar": ["b"]}
ps_wf = StateGraph(PS)
ps_wf.add_node(node_a)
ps_wf.add_node(node_b)
ps_wf.add_edge(START, "node_a")
ps_wf.add_edge("node_a", "node_b")
ps_wf.add_edge("node_b", END)
persist_demo = ps_wf.compile(checkpointer=InMemorySaver())
ps_cfg: RunnableConfig = {"configurable": {"thread_id": "persistence-mini-demo"}}
persist_demo.invoke({"foo": "", "bar": []}, ps_cfg)
print("Checkpoint count:", len(list(persist_demo.get_state_history(ps_cfg))))Checkpoint count: 4Encryption（checkpoint serde）
from langgraph.checkpoint.serde.base import CipherProtocol
from langgraph.checkpoint.serde.encrypted import EncryptedSerializer
print("Extensibility:", CipherProtocol, "->", EncryptedSerializer)
key_hex = os.getenv("LANGGRAPH_AES_KEY", "")
if len(key_hex) == 64 and re.fullmatch(r"[0-9a-fA-F]+", key_hex):
try:
serde = EncryptedSerializer.from_pycryptodome_aes()
enc_cp = InMemorySaver(serde=serde)
g = StateGraph(PS)
g.add_node(node_a)
g.add_node(node_b)
g.add_edge(START, "node_a")
g.add_edge("node_a", "node_b")
g.add_edge("node_b", END)
enc_graph = g.compile(checkpointer=enc_cp)
enc_graph.invoke({"foo": "", "bar": []}, {"configurable": {"thread_id": "encrypted-demo"}})
print("Encrypted checkpointer round-trip: OK")
except Exception as exc:
print("Encrypted demo skipped:", exc)
else:
print("Set LANGGRAPH_AES_KEY (64 hex chars) to exercise EncryptedSerializer.from_pycryptodome_aes().")Extensibility: ->
Set LANGGRAPH_AES_KEY (64 hex chars) to exercise EncryptedSerializer.from_pycryptodome_aes(). Memory-LLM 同步：成本控制（"持久化税"）
结构化 MemoryUpdateOutput 调用成本很高（延迟 + tokens）。本 notebook 不会每一轮都运行它们。
MEMORY_LLM_SYNC_STRATEGY：interval（默认）、farewell、both或every_turn（旧版/高成本）。MEMORY_LLM_EVERY_N_HUMANS：自上次成功同步以来，至少累计 N 条非命令人类消息后，才运行 memory manager（默认5）。MEMORY_LLM_FAREWELL_TRIGGERS：以逗号分隔的子串；如果最后一条用户消息包含其中之一，则触发同步（当策略包含 farewell 时）。MEMORY_LLM_MIN_CONV_CHARS：序列化对话文本的最短长度（默认20）。
生产环境： 把同样的 run_memory_update_llm + save_profile_blob 工作放入以 thread_id 为键的后台 worker（Celery/RQ），并做防抖，而不是在 node_persist 中阻塞。
超越 InMemorySaver / InMemoryStore
Notebook 状态在进程退出后会消失。后续步骤：
- Checkpointer：
langgraph.checkpoint.sqlite.SqliteSaver或langgraph.checkpoint.postgres.PostgresSaver（pip install langgraph-checkpoint-sqlite/langgraph-checkpoint-postgres）。Postgres 需要调用一次setup()。 - Store： 使用
langgraph-checkpoint-postgres中的PostgresStore（与生产 checkpoint 相同的包）——它实现了带可选向量搜索的BaseStore。Embedding 维度必须与你的模型匹配（例如text-embedding-3-small→ 1536）；不匹配会导致类似 expected 1536 dimensions, not 768 的错误——请把InMemoryStore(index={"dims": …})/ Postgres 索引配置与OPENROUTER_EMBEDDING_DIMS和OpenAIEmbeddings对齐。 - OpenRouter embeddings 保持兼容：同一个
OpenAIEmbeddings客户端；只有 store 后端和维度会改变。
参见下一个单元格中带注释的替换示例。
多 worker： 这里没有演示并发安全、事务或幂等的后台同步——请在生产中自行设计。
# Commented production pattern (uncomment + install extras when you have a real DB).
# --- Checkpointer (pick one) ---
# from langgraph.checkpoint.sqlite import SqliteSaver
# import sqlite3
# conn = sqlite3.connect("checkpoints.db", check_same_thread=False)
# CHECKPOINTER_PROD = SqliteSaver(conn)
# from langgraph.checkpoint.postgres import PostgresSaver
# CHECKPOINTER_PROD = PostgresSaver.from_conn_string(os.environ["DATABASE_URL"])
# CHECKPOINTER_PROD.setup()
# --- Store with vectors (langgraph-checkpoint-postgres; API names can vary by version) ---
# from langgraph.store.postgres import PostgresStore # or PostgresStore from package docs
# STORE_PROD = PostgresStore.from_conn_string(
# os.environ["DATABASE_URL"],
# index={"dims": EMBEDDING_DIMS, "embed": safe_embed_documents, "fields": ["mem_ix"]},
# )
# agent_app_prod = workflow.compile(checkpointer=CHECKPOINTER_PROD, store=STORE_PROD)
print("In-memory checkpointer + store active; see comments above for Sqlite/Postgres swap.")In-memory checkpointer + store active; see comments above for Sqlite/Postgres swap.斜杠命令：shlex + 分发器（无需正则泥潭）
命令用 shlex.split 解析 / 之后的文本，然后按前缀路由（memory list …、memory delete semantic … 等）。要进一步扩展，可在 FastAPI 路由或一个小的 (关键字元组) -> handler 的 dict 中镜像这张表——思路与 click 相同，却不必把 CLI 依赖拉进 kernel。
统一 Agent graph（加载 → 助手 → 持久化）
import shlex
from typing import Literal, NotRequired
from langgraph.graph.message import add_messages
from langgraph.runtime import Runtime
from langgraph.store.base import BaseStore
MEMORY_LLM_MIN_CONV_CHARS = int(os.getenv("MEMORY_LLM_MIN_CONV_CHARS", "20"))
MEMORY_LLM_SYNC_STRATEGY = os.getenv("MEMORY_LLM_SYNC_STRATEGY", "interval").strip().lower()
MEMORY_LLM_EVERY_N_HUMANS = int(os.getenv("MEMORY_LLM_EVERY_N_HUMANS", "5"))
_MEMORY_FAREWELL_RAW = os.getenv(
"MEMORY_LLM_FAREWELL_TRIGGERS",
"goodbye,that's all for now,talk later,see you,catch you later",
)
MEMORY_LLM_FAREWELL_TRIGGERS = tuple(
x.strip().lower() for x in _MEMORY_FAREWELL_RAW.split(",") if x.strip()
)
INJECTION_USE_EMBEDDING_RERANK = os.getenv("INJECTION_USE_EMBEDDING_RERANK", "1").strip().lower() in (
"1",
"true",
"yes",
)
print(
"Memory-LLM sync:",
f"strategy={MEMORY_LLM_SYNC_STRATEGY!r}, every_n_humans={MEMORY_LLM_EVERY_N_HUMANS}, "
f"injection_embedding_rerank={INJECTION_USE_EMBEDDING_RERANK}",
)
class UnifiedState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
injection_text: NotRequired[str]
update_global_policy: NotRequired[bool]
last_memory_llm_sync_human_count: NotRequired[int]
BASE_ASSISTANT_INSTRUCTIONS = """You are a helpful assistant.
Use long-term memory when relevant; do not invent facts not supported by memory or the conversation.
Set update_global_policy True only for organization-wide policy or shared knowledge that every user should see.
The server persists org-wide memory only when the authenticated session has admin privileges (is_admin on AppContext)—do not assume a policy is saved without admin.
"""
assistant_structured = llm.with_structured_output(AssistantRouterOutput)
memory_structured = memory_llm.with_structured_output(MemoryUpdateOutput)
def _default_procedural(runtime: Runtime[AppContext]) -> None:
procedural_ensure_default(runtime.store, runtime.context)
def route_entry(state: UnifiedState) -> Literal["commands", "load_context"]:
h = _last_human(state)
if h and isinstance(h.content, str) and h.content.strip().startswith("/"):
return "commands"
return "load_context"
def node_load_context(state: UnifiedState, runtime: Runtime[AppContext]) -> dict:
ctx = runtime.context
org_p = load_profile_blob(runtime.store, ns_profile_org(ctx.org_id))
user_p = load_profile_blob(runtime.store, ns_profile_user(ctx))
h = _last_human(state)
q = (h.content if h and isinstance(h.content, str) else "") or ""
ql = q.lower()
org_kw = ("policy", "legal", "compliance", "company", "org", "everyone", "customer-facing", "disclaimer")
include_org = any(k in ql for k in org_kw) or len(ql) < 4
rq = q.strip() if q.strip() else None
embed_fn = safe_embed_documents if INJECTION_USE_EMBEDDING_RERANK else None
inj = merge_injection_blocks(
org_memory=org_p,
user_memory=user_p,
rerank_query=rq if embed_fn else None,
embed_texts=embed_fn,
include_org_profile=include_org,
org_max_tokens=int(os.getenv("INJECTION_ORG_MAX_TOKENS", "600")),
user_max_tokens=int(os.getenv("INJECTION_USER_MAX_TOKENS", "1800")),
)
return {"injection_text": inj}
def _cmd_reply(text: str) -> dict:
return {"messages": [AIMessage(content=text)]}
def _cmd_memory_list_dispatch(store: BaseStore, ctx: AppContext, sub: str) -> dict:
nl = chr(10)
s = sub.lower().strip() or "all"
if s in ("", "all"):
parts = [
"**Memory scope**",
"- **User-private:** semantic + episodic + user profile (this `user_id` only).",
"- **Org-shared:** `_shared` namespaces — can affect **other users** in the org.",
"",
"### User-private listings",
]
parts.extend(semantic_format_lines(store, ctx))
parts.extend(episodic_format_lines(store, ctx))
parts.append("")
parts.append("### Org-shared profile (preview)")
org_prof = load_profile_blob(store, ns_profile_org(ctx.org_id))
parts.append(format_memory_for_injection(org_prof, max_tokens=800) or "(empty)")
pl = procedural_format_line(store, ctx)
if pl:
parts.append("")
parts.append("### Procedural overrides")
parts.append(pl)
return _cmd_reply(nl.join(parts))
if s == "semantic":
return _cmd_reply(nl.join(["**Semantic only**"] + semantic_format_lines(store, ctx)))
if s == "episodic":
return _cmd_reply(nl.join(["**Episodic only**"] + episodic_format_lines(store, ctx)))
if s == "procedural":
pl = procedural_format_line(store, ctx)
return _cmd_reply(pl or "(No procedural instructions.)")
return _cmd_reply(
f"Unknown /memory list subcommand: {sub!r}. Try: semantic | episodic | procedural | all"
)
def dispatch_slash_command(
line: str,
store: BaseStore,
ctx: AppContext,
state: UnifiedState,
h: HumanMessage,
) -> dict | None:
t = line.strip()
low = t.lower()
if low.startswith("/thread summarize"):
msgs = _ensure_msg_ids(list(state["messages"]))
return summarize_thread_messages(msgs, drop_command=h)
if not t.startswith("/"):
return None
try:
parts = shlex.split(t[1:])
except ValueError:
return _cmd_reply("Invalid command quoting (check quotes).")
if not parts:
return None
p0 = parts[0].lower()
p1 = parts[1].lower() if len(parts) > 1 else ""
if p0 == "memory":
if p1 == "list":
sub = parts[2] if len(parts) > 2 else "all"
return _cmd_memory_list_dispatch(store, ctx, sub)
if p1 == "delete" and len(parts) >= 4:
kind, key = parts[2].lower(), parts[3]
if kind == "semantic":
semantic_delete_key(store, ctx, key)
return _cmd_reply(f"Deleted semantic memory `{key}`.")
if kind == "episodic":
episodic_delete_key(store, ctx, key)
return _cmd_reply(f"Deleted episodic memory `{key}`.")
return _cmd_reply("Use: /memory delete semantic or /memory delete episodic ")
if p1 == "clear" and len(parts) >= 3:
scope = parts[2].lower()
if scope == "semantic":
n = semantic_clear_all(store, ctx)
return _cmd_reply(f"Cleared semantic memories ({n} removed).")
if scope == "episodic":
n = episodic_clear_all(store, ctx)
return _cmd_reply(f"Cleared episodic memories ({n} removed).")
if scope == "all":
ns = semantic_clear_all(store, ctx)
ne = episodic_clear_all(store, ctx)
return _cmd_reply(f"Cleared semantic ({ns}) and episodic ({ne}) memories.")
return _cmd_reply("Use: /memory clear semantic | episodic | all")
if p1 == "edit" and len(parts) >= 5 and parts[2].lower() == "semantic":
key = parts[3]
new_fact = " ".join(parts[4:])
semantic_upsert_fact(store, ctx, key, new_fact)
return _cmd_reply(f"Updated semantic memory `{key}`.")
if p1 == "compact":
_summary, deleted = semantic_compact_consolidated(store, ctx, memory_llm)
if deleted == 0 and _summary == "No semantic memories to compact.":
return _cmd_reply("No semantic memories to compact.")
return _cmd_reply(
f"Compacted semantic memories into `{CONSOLIDATED_KEY}` and removed {deleted} prior keys."
)
return None
if p0 == "proc":
if p1 == "reset" and len(parts) == 2:
procedural_clear(store, ctx)
return _cmd_reply(
"Procedural instructions removed; defaults apply on the next normal message."
)
if p1 == "set" and len(parts) >= 3:
text = " ".join(parts[2:])
procedural_set_instructions(store, ctx, text)
return _cmd_reply("Updated procedural / style instructions.")
return None
return None
def node_commands(state: UnifiedState, runtime: Runtime[AppContext]) -> dict:
h = _last_human(state)
if not h or not isinstance(h.content, str):
return {}
line = h.content.strip()
ctx = runtime.context
store = runtime.store
out = dispatch_slash_command(line, store, ctx, state, h)
if out is not None:
return out
return _cmd_reply(f"Unknown command: {line!r}. Try /memory list all")
def _non_command_human_messages(messages: list[BaseMessage]) -> list[HumanMessage]:
out: list[HumanMessage] = []
for m in messages:
if isinstance(m, HumanMessage) and isinstance(m.content, str):
t = m.content.strip()
if t and not t.startswith("/"):
out.append(m)
return out
def should_run_memory_manager_llm(state: UnifiedState) -> bool:
conv = format_conversation_for_update(state["messages"])
if len(conv) < MEMORY_LLM_MIN_CONV_CHARS:
return False
humans = _non_command_human_messages(state["messages"])
if not humans:
return False
last = humans[-1].content.strip().lower()
farewell = any(tok in last for tok in MEMORY_LLM_FAREWELL_TRIGGERS if tok)
n = len(humans)
prev = state.get("last_memory_llm_sync_human_count", 0)
interval_hit = MEMORY_LLM_EVERY_N_HUMANS > 0 and (n - prev) >= MEMORY_LLM_EVERY_N_HUMANS
strat = MEMORY_LLM_SYNC_STRATEGY
if strat in ("every_turn", "always"):
return True
if strat == "farewell":
return farewell
if strat == "interval":
return interval_hit
if strat == "both":
return farewell or interval_hit
return interval_hit
def node_agent(state: UnifiedState, runtime: Runtime[AppContext]) -> dict:
_default_procedural(runtime)
ctx = runtime.context
h = _last_human(state)
q = (h.content if h and isinstance(h.content, str) else "") or ""
sem = semantic_search(runtime.store, ctx, q or "user preferences", limit=6)
epi = episodic_search(runtime.store, ctx, q or "similar task", limit=3)
epi_min = int(os.getenv("EPISODIC_INJECT_MIN_QUERY_CHARS", "14"))
use_epi = len((q or "").strip()) >= epi_min
proc_struct = procedural_read_structured(runtime.store, ctx)
default_style = "Be concise and helpful. Prefer bullet points for comparisons."
has_override = (proc_struct.get("style") or "").strip() not in ("", default_style)
has_override = has_override or bool((proc_struct.get("workflow") or "").strip())
has_override = has_override or bool((proc_struct.get("safety") or "").strip())
has_override = has_override or bool((proc_struct.get("task_policies") or "").strip())
proc_block = (
procedural_format_structured_block(runtime.store, ctx)
if has_override
else "(Default procedural seed only.)"
)
facts = chr(10).join(f"- {s.value.get('fact', s.value)}" for s in sem) or "- (none)"
eps_lines = (
chr(10).join(
f"- task: {e.value.get('task')} => outcome: {e.value.get('outcome')}" for e in epi
)
if use_epi
else "(Episodic block skipped: short query; tune EPISODIC_INJECT_MIN_QUERY_CHARS.)"
)
profile_block = state.get("injection_text") or "(No profile injection.)"
nl = chr(10)
sys = SystemMessage(
content=(
f"{BASE_ASSISTANT_INSTRUCTIONS}{nl}{nl}"
f"## Structured profile (org + user){nl}{profile_block}{nl}{nl}"
f"Procedural:{nl}{proc_block}{nl}{nl}"
f"Semantic facts (retrieved):{nl}{facts}{nl}{nl}"
f"Episodic examples:{nl}{eps_lines}{nl}{nl}"
"Prefer tools/APIs for explicit writes; legacy remember:/episode: behind LEGACY_REMEMBER_EPISODE_SYNTAX."
" Slash commands: `/memory list`, `/thread summarize`, etc."
),
)
conv = [m for m in state["messages"] if isinstance(m, (HumanMessage, AIMessage))]
out = assistant_structured.invoke([sys, *conv])
return {"messages": [AIMessage(content=out.reply)], "update_global_policy": out.update_global_policy}
def node_persist(state: UnifiedState, runtime: Runtime[AppContext]) -> dict:
msgs = [m for m in state["messages"] if not isinstance(m, SystemMessage)]
if len(msgs) < 2:
return {}
human = None
ai = None
for m in reversed(msgs):
if ai is None and isinstance(m, AIMessage):
ai = m
elif human is None and isinstance(m, HumanMessage):
human = m
break
if human is None or ai is None or not isinstance(human.content, str):
return {}
text = human.content.strip()
if text.startswith("/"):
return {}
ctx = runtime.context
out: dict = {"update_global_policy": False}
if LEGACY_REMEMBER_EPISODE_SYNTAX:
m = re.search(r"remember\s*:\s*(.+)", text, flags=re.I | re.S)
if m:
try:
fact_text = validate_durable_memory_text(m.group(1).strip())
semantic_upsert_fact(
runtime.store,
ctx,
SEMANTIC_TRANSIENT_PREFIX + str(uuid.uuid4()),
fact_text,
transient=True,
)
except ValueError as e:
log.warning("remember: rejected: %s", e)
m = re.search(r"episode\s*:\s*(.+?)\s*\|\s*(.+)", text, flags=re.I | re.S)
if m:
try:
episodic_record_event(
runtime.store,
ctx,
validate_durable_memory_text(m.group(1).strip()),
validate_durable_memory_text(m.group(2).strip()),
)
except ValueError as e:
log.warning("episode: rejected: %s", e)
if not should_run_memory_manager_llm(state):
return out
humans = _non_command_human_messages(state["messages"])
tid = ctx.user_id
user_ns = ns_profile_user(ctx)
user_prev = load_profile_blob(runtime.store, user_ns)
updated_user = run_memory_update_llm(
memory_structured,
current_memory=user_prev,
messages=state["messages"],
scope="user",
thread_id=tid,
)
did_sync = False
if updated_user is not None:
sync_profile_facts_semantic(
runtime.store, ns_user(ctx, NS_SEMANTIC), user_prev, updated_user
)
save_profile_blob(runtime.store, user_ns, updated_user)
did_sync = True
org_requested = bool(state.get("update_global_policy"))
org_allowed = org_requested and ctx.is_admin
if org_requested and not ctx.is_admin:
log.warning(
"Blocked org memory write: update_global_policy=True but AppContext.is_admin is False "
"(model output is not trusted for authorization)."
)
if org_allowed:
org_ns = ns_profile_org(ctx.org_id)
org_prev = load_profile_blob(runtime.store, org_ns)
updated_org = run_memory_update_llm(
memory_structured,
current_memory=org_prev,
messages=state["messages"],
scope="org",
thread_id=tid,
)
if updated_org is not None:
sync_profile_facts_semantic(
runtime.store,
ns_org_shared(ctx.org_id, NS_SEMANTIC),
org_prev,
updated_org,
)
save_profile_blob(runtime.store, org_ns, updated_org)
did_sync = True
if did_sync:
out["last_memory_llm_sync_human_count"] = len(humans)
return out
workflow = StateGraph(UnifiedState, context_schema=AppContext)
workflow.add_node("commands", node_commands)
workflow.add_node("load_context", node_load_context)
workflow.add_node("agent", node_agent)
workflow.add_node("persist", node_persist)
workflow.add_conditional_edges(START, route_entry, {"commands": "commands", "load_context": "load_context"})
workflow.add_edge("commands", END)
workflow.add_edge("load_context", "agent")
workflow.add_edge("agent", "persist")
workflow.add_edge("persist", END)
agent_app = workflow.compile(checkpointer=CHECKPOINTER, store=STORE)
print("Compiled agent_app (checkpointer + store + org/user namespaces).") Memory-LLM sync: strategy='interval', every_n_humans=5, injection_embedding_rerank=True
Compiled agent_app (checkpointer + store + org/user namespaces).LLM memory manager —— 结构化更新（用例）
memory manager 是一个独立的结构化调用（MemoryUpdateOutput）：它会修补 profile 的各部分、以置信度添加 facts，并在新的用户信息与旧 facts 矛盾时使用 factsToRemove。
下面，isolated_*_memory_update 运行与 graph 在 node_persist 中相同的 run_memory_update_llm 路径（无需很长的 thread）。node_persist 只在 should_run_memory_manager_llm 为真时运行该 LLM（interval / farewell / 等），所以 use case 5 代码片段临时把 MEMORY_LLM_SYNC_STRATEGY=every_turn，以便两轮对话仍能演示持久化。
代码中面向产品的要点： extract_memory_update_llm（提议）vs apply_updates（应用、可单元测试）、通过 CATEGORY_COMBINE_MODE 的类别 replace/merge/accumulate、通过 MEMORY_QUARANTINE_CONFIDENCE → lifecycle=proposed + needs_review 实现的隔离、组织 allowlist ORG_SHARED_ALLOWED_CATEGORIES、safe_embed_documents 和 mem_ix（每行唯一的字符串；LangGraph 的 store 按精确字符串对 embedding 输入 去重，所以两个键上相同的 fact 文本不能共享一个向量）。
前置要求： 必须已经运行了前面的单元，以便 memory_structured、STORE 和命名空间都存在。
import copy
import json
from pprint import pprint
# --- Isolated helper: same pipeline as node_persist (profile + semantic sync) ---
def isolated_user_memory_update(
store: BaseStore,
ctx: AppContext,
dialogue: list[tuple[str, str]],
*,
thread_id: str | None = None,
) -> tuple[dict, dict]:
"""One memory-manager pass for USER-PRIVATE profile. Returns (before, after) memory dicts."""
ns_prof = ns_profile_user(ctx)
prev = load_profile_blob(store, ns_prof)
msgs: list[BaseMessage] = []
for role, text in dialogue:
if role == "human":
msgs.append(HumanMessage(content=text))
else:
msgs.append(AIMessage(content=text))
updated = run_memory_update_llm(
memory_structured,
current_memory=prev,
messages=msgs,
scope="user",
thread_id=thread_id or ctx.user_id,
)
if updated is None:
return prev, copy.deepcopy(prev)
sync_profile_facts_semantic(store, ns_user(ctx, NS_SEMANTIC), prev, updated)
save_profile_blob(store, ns_prof, updated)
return prev, updated
def isolated_org_memory_update(
store: BaseStore,
ctx: AppContext,
dialogue: list[tuple[str, str]],
*,
thread_id: str | None = None,
) -> tuple[dict, dict]:
"""One memory-manager pass for ORG-WIDE shared profile."""
ns_prof = ns_profile_org(ctx.org_id)
prev = load_profile_blob(store, ns_prof)
msgs: list[BaseMessage] = []
for role, text in dialogue:
if role == "human":
msgs.append(HumanMessage(content=text))
else:
msgs.append(AIMessage(content=text))
updated = run_memory_update_llm(
memory_structured,
current_memory=prev,
messages=msgs,
scope="org",
thread_id=thread_id or ctx.user_id,
)
if updated is None:
return prev, copy.deepcopy(prev)
sync_profile_facts_semantic(
store, ns_org_shared(ctx.org_id, NS_SEMANTIC), prev, updated
)
save_profile_blob(store, ns_prof, updated)
return prev, updated
CTX_LLM = AppContext(org_id="llm-demo-org", user_id="llm-demo-user")
# --- Use case 1: NEW information on an empty profile (sections + facts) ---
_, mem1 = isolated_user_memory_update(
STORE,
CTX_LLM,
[
(
"human",
"I'm Alex. I work on fraud detection and I'm based in Berlin. "
"Always cite regulation articles when you explain compliance.",
),
(
"ai",
"Thanks, Alex — I'll prioritize regulatory citations in compliance explanations.",
),
],
)
print("=== Use case 1: new user context (empty -> filled) ===")
print("workContext summary:", (mem1.get("user") or {}).get("workContext", {}).get("summary", "")[:300])
print("Facts:")
pprint(mem1.get("facts", [])[:5])
# --- Use case 2: NEW information contradicts an existing fact (factsToRemove + replacement) ---
seed = empty_memory()
fid = "fact_manual_dark_001"
seed["facts"] = [
{
"id": fid,
"content": "User wants the product UI in dark mode only.",
"category": "preference",
"confidence": 0.9,
"createdAt": _utc_now_z(),
"source": "seed",
}
]
save_profile_blob(STORE, ns_profile_user(CTX_LLM), seed)
semantic_upsert_fact(STORE, CTX_LLM, fid, seed["facts"][0]["content"], extra={"category": "preference"})
prev2, mem2 = isolated_user_memory_update(
STORE,
CTX_LLM,
[
(
"human",
"Change of plans: I now need light mode everywhere. Dark mode strains my eyes. Forget the dark-only rule.",
),
(
"ai",
"Understood — I'll switch to light mode as your UI preference going forward.",
),
],
)
print("=== Use case 2: contradiction / supersession ===")
print("Old fact ids:", [f.get("id") for f in prev2.get("facts", [])])
print("New fact ids:", [f.get("id") for f in mem2.get("facts", [])])
for f in mem2.get("facts", [])[:6]:
print(" -", f.get("id"), f.get("content", "")[:120])
# --- Use case 3: additive fact without dropping unrelated memory ---
_, mem3 = isolated_user_memory_update(
STORE,
CTX_LLM,
[
(
"human",
"Also: my team's on-call Slack channel is #fraud-ops — ping there for urgent issues.",
),
(
"ai",
"Noted — #fraud-ops for urgent on-call matters.",
),
],
)
print("=== Use case 3: additive fact (unrelated to theme preference) ===")
pprint([f.get("content") for f in mem3.get("facts", [])])
# --- Use case 4: org-wide memory (shared namespace) ---
CTX_ORG = AppContext(org_id="llm-demo-org", user_id="policy-editor")
_, org_mem = isolated_org_memory_update(
STORE,
CTX_ORG,
[
(
"human",
"Org-wide rule: any pricing answer must end with 'Prices may change without notice.'",
),
(
"ai",
"Recorded as organization-wide customer-facing policy.",
),
],
)
print("=== Use case 4: org-wide structured memory ===")
print("Org facts sample:")
pprint(org_mem.get("facts", [])[:4])
print(
"Org shared semantic keys (sample):",
[it.key for it in STORE.search(ns_org_shared(CTX_ORG.org_id, NS_SEMANTIC), limit=5)],
)
# --- Use case 5: full graph — assistant router + persist (same LLM update path) ---
_prev_sync = os.environ.get("MEMORY_LLM_SYNC_STRATEGY")
os.environ["MEMORY_LLM_SYNC_STRATEGY"] = "every_turn"
THREAD_LLM = "thread-llm-manager-demo"
cfg_g: RunnableConfig = {"configurable": {"thread_id": THREAD_LLM}}
agent_app.invoke(
{
"messages": [
HumanMessage(
content=(
"For the record: my preferred meeting window is 09:00–12:00 CET on weekdays only."
),
id=str(uuid.uuid4()),
)
]
},
cfg_g,
context=CTX_LLM,
)
agent_app.invoke(
{
"messages": [
HumanMessage(
content=(
"Actually meetings after 15:00 CET are fine too — update that preference."
),
id=str(uuid.uuid4()),
)
]
},
cfg_g,
context=CTX_LLM,
)
prof_after_graph = load_profile_blob(STORE, ns_profile_user(CTX_LLM))
print("=== Use case 5: two graph turns — profile facts after router + persist ===")
for f in prof_after_graph.get("facts", [])[-6:]:
print(" -", (f.get("content") or "")[:100])
if _prev_sync is None:
os.environ.pop("MEMORY_LLM_SYNC_STRATEGY", None)
else:
os.environ["MEMORY_LLM_SYNC_STRATEGY"] = _prev_syncWARNING:unified_memory:semantic_upsert_fact: set projection_of= or transient=True / transient: key === Use case 1: new user context (empty -> filled) ===
workContext summary: Works on fraud detection
Facts:
[{'category': 'context',
'confidence': 0.9,
'content': 'Works on fraud detection',
'createdAt': '2026-04-06T06:32:51Z',
'evidence_span': "I'm Alex. I work on fraud detection",
'id': 'fact_d6cf98d3c7',
'last_confirmed_at': '2026-04-06T06:32:51Z',
'lifecycle': 'confirmed',
'needs_review': False,
'source': 'llm-demo-user',
'source_turn_id': 'llm-demo-user',
'source_type': 'llm_structured_update',
'updatedAt': '2026-04-06T06:32:51Z'},
{'category': 'context',
'confidence': 0.9,
'content': 'Based in Berlin',
'createdAt': '2026-04-06T06:32:51Z',
'evidence_span': "I'm based in Berlin",
'id': 'fact_807470cae1',
'last_confirmed_at': '2026-04-06T06:32:51Z',
'lifecycle': 'confirmed',
'needs_review': False,
'source': 'llm-demo-user',
'source_turn_id': 'llm-demo-user',
'source_type': 'llm_structured_update',
'updatedAt': '2026-04-06T06:32:51Z'},
{'category': 'goal',
'confidence': 0.9,
'content': 'Always cite regulation articles when explaining compliance',
'createdAt': '2026-04-06T06:32:51Z',
'evidence_span': 'Always cite regulation articles when you explain '
'compliance',
'id': 'fact_a3d66037ca',
'last_confirmed_at': '2026-04-06T06:32:51Z',
'lifecycle': 'confirmed',
'needs_review': False,
'source': 'llm-demo-user',
'source_turn_id': 'llm-demo-user',
'source_type': 'llm_structured_update',
'updatedAt': '2026-04-06T06:32:51Z'}]
=== Use case 2: contradiction / supersession ===
Old fact ids: ['fact_manual_dark_001']
New fact ids: ['fact_0769bcee7b']
- fact_0769bcee7b User wants the product UI in light mode only.
=== Use case 3: additive fact (unrelated to theme preference) ===
['User wants the product UI in light mode only.',
"User's team's on-call Slack channel is #fraud-ops for urgent issues.",
'User prefers to be contacted in the #fraud-ops channel for urgent issues.']
=== Use case 4: org-wide structured memory ===
Org facts sample:
[{'category': 'policy',
'confidence': 0.9,
'content': "Any pricing answer must end with 'Prices may change without "
"notice.'",
'createdAt': '2026-04-06T06:33:06Z',
'evidence_span': "Org-wide rule: any pricing answer must end with 'Prices "
"may change without notice.'",
'id': 'fact_01f4ecf88d',
'last_confirmed_at': '2026-04-06T06:33:06Z',
'lifecycle': 'confirmed',
'needs_review': False,
'source': 'policy-editor',
'source_turn_id': 'policy-editor',
'source_type': 'llm_structured_update',
'updatedAt': '2026-04-06T06:33:06Z'},
{'category': 'policy',
'confidence': 0.9,
'content': 'This is recorded as organization-wide customer-facing policy.',
'createdAt': '2026-04-06T06:33:06Z',
'evidence_span': 'Recorded as organization-wide customer-facing policy.',
'id': 'fact_35846a3c11',
'last_confirmed_at': '2026-04-06T06:33:06Z',
'lifecycle': 'confirmed',
'needs_review': False,
'source': 'policy-editor',
'source_turn_id': 'policy-editor',
'source_type': 'llm_structured_update',
'updatedAt': '2026-04-06T06:33:06Z'}]
Org shared semantic keys (sample): ['fact_01f4ecf88d', 'fact_35846a3c11']
=== Use case 5: two graph turns — profile facts after router + persist ===
- User wants the product UI in light mode only.
- User's team's on-call Slack channel is #fraud-ops for urgent issues.
- User prefers to be contacted in the #fraud-ops channel for urgent issues.Runbook 演示
def run_turn(thread: str, org: str, user: str, text: str, *, is_admin: bool = False):
cfg: RunnableConfig = {"configurable": {"thread_id": thread}}
return agent_app.invoke(
{"messages": [HumanMessage(content=text, id=str(uuid.uuid4()))]},
cfg,
context=AppContext(org_id=org, user_id=user, is_admin=is_admin),
)
ORG = "acme"
USER_A = "alice"
USER_B = "bob"
THREAD = "demo-thread-1"
run_turn(THREAD, ORG, USER_A, "remember: I use dark mode and prefer US English")
run_turn(THREAD, ORG, USER_A, "episode: billing dispute | issued pro-rata credit after tier mismatch")
out = run_turn(THREAD, ORG, USER_A, "What preferences should the UI use?")
print(out["messages"][-1].content[:600])
run_turn(
THREAD,
ORG,
USER_A,
"Company policy: customer-facing answers must include the disclaimer 'Prices may change.'",
is_admin=True,
)
out2 = run_turn("demo-thread-2", ORG, USER_B, "What is our customer-facing disclaimer rule?")
print("--- new thread (Bob) ---")
print(out2["messages"][-1].content[:600])
print(run_turn(THREAD, ORG, USER_A, "/memory list")["messages"][-1].content[:1200])
run_turn(THREAD, ORG, USER_A, "/thread summarize")
run_turn(THREAD, ORG, USER_A, "remember: I want metric units everywhere")
run_turn(THREAD, ORG, USER_A, "/memory compact")
print(run_turn(THREAD, ORG, USER_A, "/memory list")["messages"][-1].content[:1200])The UI should use dark mode and display content in US English, as these are your preferred settings.
--- new thread (Bob) ---
I currently don't have access to specific organizational policies or disclaimers. You may want to check your company's internal documentation or consult with your compliance or legal team for the customer-facing disclaimer rule.
**Memory scope**
- **User-private:** semantic + episodic + user profile (this `user_id` only).
- **Org-shared:** `_shared` namespaces — can affect **other users** in the org.
### User-private listings
- Semantic (1): transient:…
• [transient:7e497291-08d2-4bc6-8f35-a55ff3f1f8b2] I use dark mode and prefer US English
- Episodic (1): 13fb238d
• billing dispute -> issued pro-rata credit after tier mismatch
### Org-shared profile (preview)
(empty)
### Procedural overrides
**Procedural (typed slots)**
- Style: Be concise and helpful. Prefer bullet points for comparisons.
**Memory scope**
- **User-private:** semantic + episodic + user profile (this `user_id` only).
- **Org-shared:** `_shared` namespaces — can affect **other users** in the org.
### User-private listings
- Semantic (1): consolidat…
• [consolidated_profile] - **User Preferences:**
- Dark mode: Enabled
- Language: US English
- Measurement system: Metric units
- Episodic (1): 13fb238d
• billing dispute -> issued pro-rata credit after tier mismatch
### Org-shared profile (preview)
(empty)
### Procedural overrides
**Procedural (typed slots)**
- Style: Be concise and helpful. Prefer bullet points for comparisons.Time travel（统一 thread + 迷你 fork）
cfg_tt: RunnableConfig = {"configurable": {"thread_id": THREAD}}
hist = list(agent_app.get_state_history(cfg_tt))
print("Unified thread snapshots:", len(hist))
if len(hist) >= 3:
mid = hist[len(hist) // 2]
print("Example checkpoint step:", mid.metadata.get("step"), "next:", mid.next)
class TT(TypedDict):
foo: str
bar: Annotated[list[str], add]
def tt_a(state: TT):
return {"foo": "a", "bar": ["a"]}
def tt_b(state: TT):
return {"foo": "b", "bar": ["b"]}
tt_wf = StateGraph(TT)
tt_wf.add_node("node_a", tt_a)
tt_wf.add_node("node_b", tt_b)
tt_wf.add_edge(START, "node_a")
tt_wf.add_edge("node_a", "node_b")
tt_wf.add_edge("node_b", END)
tt_graph = tt_wf.compile(checkpointer=InMemorySaver())
tt_cfg: RunnableConfig = {"configurable": {"thread_id": "time-travel-mini"}}
tt_graph.invoke({"foo": "", "bar": []}, tt_cfg)
before_b = next(s for s in tt_graph.get_state_history(tt_cfg) if s.next == ("node_b",))
forked = tt_graph.update_state(before_b.config, {"foo": "forked"}, as_node="node_a")
continued = tt_graph.invoke(None, forked)
print("Fork continue:", continued)
replay_cfg = {
"configurable": {
"thread_id": "time-travel-mini",
"checkpoint_id": before_b.config["configurable"]["checkpoint_id"],
}
}
print("Replay:", tt_graph.invoke(None, replay_cfg))Unified thread snapshots: 37
Example checkpoint step: 17 next: ('persist',)
Fork continue: {'foo': 'b', 'bar': ['a', 'b']}
Replay: {'foo': 'b', 'bar': ['a', 'b']}生产检查清单
| 替换项 | 位置 |
|---|---|
| 持久性 | build_checkpointer() → SQLite / Postgres |
| 多进程 | 为 checkpoints + store 使用共享数据库 |
| 用户记忆 UI | HTTP handlers 调用 UserMemoryService / 隔离的 semantic_* 等函数 |
| 安全 | 对 org_id / user_id 做 AuthZ，组织写入需要 AppContext.is_admin，静态加密，监控投毒 |
| 记忆 LLM 成本 | 防抖 / MEMORY_LLM_* 环境变量或后台 worker；生产中避免 every_turn |
| 事实注入 | 通过 rank_facts_for_injection + INJECTION_USE_EMBEDDING_RERANK 可选 embedding 重排 |
| 防抖 | 在负载下批量调用 memory manager |
| 组织写入 | 模型 update_global_policy + 管理员 RBAC / 审批 |
在应用边界上，把 store 和 checkpoint payload 视为不可信文本。
有记忆 vs 无记忆
关于本 notebook
本 notebook 演示了在有和没有 checkpoint 记忆的情况下运行 LangGraph 聊天机器人的区别。
该 graph 只包含一个聊天机器人节点，它通过兼容 OpenRouter 的 ChatOpenAI 接口把当前消息历史发送给 LLM。同一个 graph 编译出两个版本：一个不带 checkpointer，一个带 InMemorySaver checkpointer。
第一次运行时，graph 独立接收每条用户消息，因此它不会在多次调用之间记住用户的名字。第二次运行时，两次调用使用同一个 thread_id，使 LangGraph 能够从记忆恢复先前的对话状态。结果，聊天机器人能够根据上一轮提供的信息回答后续问题。
该示例说明了 thread 级 checkpoint 在 Agent 记忆中的作用，并展示了在构建对话式或多轮 Agent 系统时，为什么必须显式配置持久状态。
import os
from dotenv import load_dotenv
load_dotenv()
OPENROUTER_API_KEY = os.getenv('OPENROUTER_API_KEY')!pip install langchain_openaifrom typing import Annotated, TypedDict
import operator
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
llm = ChatOpenAI(model="gpt-5.4-nano", base_url="https://openrouter.ai/api/v1", temperature=0, api_key=OPENROUTER_API_KEY)# pip install -U langgraph langchain openai
# ---------------------------
# Define state
# ---------------------------
class AgentState(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
# ---------------------------
# Define node
# ---------------------------
def chatbot(state: AgentState) -> AgentState:
response = llm.invoke(state["messages"])
return {"messages": [response]}
# ---------------------------
# Build graph
# ---------------------------
builder = StateGraph(AgentState)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)
# Version A: no memory
graph_without_memory = builder.compile()
# Version B: with memory
checkpointer = InMemorySaver()
graph_with_memory = builder.compile(checkpointer=checkpointer)
# ---------------------------
# Helper to print output
# ---------------------------
def show_last_ai_message(result: dict) -> None:
last_message = result["messages"][-1]
print(f"AI: {last_message.content}\n")
# ---------------------------
# Run without memory
# ---------------------------
print("WITHOUT MEMORY")
print("=" * 60)
result = graph_without_memory.invoke(
{
"messages": [
HumanMessage(content="My name is Nicole. I am writing a chapter about memory in AI agents. You don't need to give me tips, just ackknowledge it.")
]
}
)
show_last_ai_message(result)
result = graph_without_memory.invoke(
{
"messages": [
HumanMessage(content="What is my name?")
]
}
)
show_last_ai_message(result)
# ---------------------------
# Run with memory
# ---------------------------
print("WITH MEMORY")
print("=" * 60)
config = {"configurable": {"thread_id": "demo-thread-1"}}
result = graph_with_memory.invoke(
{
"messages": [
HumanMessage(content="My name is Nicole. I am writing a chapter about memory in AI agents.")
]
},
config=config,
)
show_last_ai_message(result)
result = graph_with_memory.invoke(
{
"messages": [
HumanMessage(content="What is my name?")
]
},
config=config,
)
show_last_ai_message(result)WITHOUT MEMORY
============================================================
AI: Hi Nicole — acknowledged. Good luck writing your chapter on memory in AI agents.
AI: I don’t know your name from this chat. If you tell me what you’d like me to call you, I’ll use it.
WITH MEMORY
============================================================
AI: Hi Nicole—nice to meet you! If you’re writing a chapter on memory in AI agents, I can help you shape the content, outline the chapter, or draft sections.
A few quick questions to tailor it:
1. **Audience level:** is this for general readers, undergraduate, or graduate/professional?
2. **Scope:** do you mean *agent memory* broadly (retrieval, long-term/short-term, planning), or mainly *neural memory* (transformers, external memory, RNN/attention)?
3. **Angle:** are you emphasizing **technical mechanisms** (e.g., RAG, vector DBs, episodic memory) or **systems/engineering** (latency, scaling, evaluation), or both?
4. **Length target:** roughly how many pages/words?
Meanwhile, here are a couple of strong chapter section structures you can choose from:
### Option A: Conceptual + taxonomy (great for a “chapter about” framing)
1. What “memory” means for agents
2. Memory requirements (persistence, recall, write/update, prioritization)
3. A taxonomy: short-term vs long-term, parametric vs non-parametric, episodic vs semantic vs working memory
4. Architectures and mechanisms
- Context windows & attention
- Retrieval-augmented memory (RAG)
- External stores (vector DBs, knowledge graphs)
- Episodic/long-term logging
5. Memory management: when to write, what to keep, how to prune
6. Failure modes and risks
- Stale or contradictory memories
- Privacy leakage / data retention
- Hallucinated “remembered” facts
- Temporal inconsistency
7. Evaluation: metrics and benchmarks
8. Future directions
### Option B: Build-up from agents → implementations
1. Agent loop and why memory is needed
2. Minimal baseline: prompting/context
3. Adding retrieval (RAG)
4. Adding updates (write-back memory)
5. Agent-specific memory strategies (summaries, timelines, user profiles)
6. Tool use + memory (how agents act using remembered info)
7. Evaluation and best practices
If you tell me your answers to the 4 questions (especially audience + scope), I can draft:
- a polished **chapter outline**, or
- a **full chapter draft** for one section (e.g., “memory taxonomy” or “write/update strategies”), in your voice.
AI: Your name is **Nicole**.记忆拓扑
关于本 notebook
本 notebook 演示了在 LangGraph 中，当父 graph 调用子 Agent 时，checkpoint 如何影响记忆行为。
示例构建了一个简单的计费 worker，它从用户轮次中提取信息，例如重复扣费、套餐级别和退款意图。父 graph 随后调用该 worker，只传入当前用户消息。三种执行模式并排比较：每次调用新建的 worker、按 thread 持久化的 worker，以及不带 checkpoint 的无状态 worker。
目标是展示子 Agent 的记忆取决于 worker graph 如何编译和 checkpoint。即使父 graph 只传入最新一轮，checkpointer 也能让 worker 在同一 thread 内的多轮之间保留先前的消息。这使得本 notebook 对理解记忆边界、thread 级持久化以及基于 LangGraph 的多 Agent 系统中的子 Agent 状态处理很有用。
from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import START, StateGraph
from langgraph.graph.message import add_messages
# Define worker state and node
class WorkerState(TypedDict):
messages: Annotated[list, add_messages]
def billing_worker_node(state: WorkerState):
human_texts = [
m.content.lower()
for m in state["messages"]
if isinstance(m, HumanMessage)
]
issue = "duplicate" if any("charged twice" in t for t in human_texts) else None
tier = "pro" if any("pro plan" in t for t in human_texts) else None
action = "refund" if any("refund" in t for t in human_texts) else None
msg_count = len(state["messages"])
answer = f"Msgs: {msg_count} | Known: {issue}, {tier}, {action}"
return {"messages": [AIMessage(content=answer)]}
# Build worker graph ---
def build_worker(mode: str):
builder = StateGraph(WorkerState)
builder.add_node("billing_worker", billing_worker_node)
builder.add_edge(START, "billing_worker")
if mode == "per_invocation":
return builder.compile()
if mode == "per_thread":
return builder.compile(checkpointer=True)
if mode == "stateless":
return builder.compile(checkpointer=False)
raise ValueError(f"Unknown mode: {mode}")
# Parent graph passes only the current turn
class ParentState(TypedDict):
current_input: str
answer: str
def make_parent_app(worker_app):
def call_worker(state: ParentState, config: RunnableConfig):
out = worker_app.invoke(
{"messages": [HumanMessage(content=state["current_input"])]},
config=config,
)
return {"answer": out["messages"][-1].content}
builder = StateGraph(ParentState)
builder.add_node("call_worker", call_worker)
builder.add_edge(START, "call_worker")
return builder.compile(checkpointer=InMemorySaver())
# Compare modes side by side
def run_comparison():
apps = {
"Per-invocation (fresh start)": make_parent_app(build_worker("per_invocation")),
"Per-thread (persistent)": make_parent_app(build_worker("per_thread")),
"Stateless (no checkpoint)": make_parent_app(build_worker("stateless")),
}
turns = [
"I was charged twice.",
"I'm on the Pro plan.",
"I want a refund.",
]
results = []
for label, app in apps.items():
cfg = {"configurable": {"thread_id": f"demo-{label}"}}
for i, turn in enumerate(turns, start=1):
out = app.invoke({"current_input": turn}, cfg)
results.append(
{
"Mode": label,
"Turn": i,
"Input": turn,
"Subagent Output": out["answer"],
}
)
print(f"{'MODE':<32} | {'TURN':<5} | {'SUBAGENT OUTPUT'}")
print("-" * 95)
for r in results:
print(f"{r['Mode']:<32} | {r['Turn']:<5} | {r['Subagent Output']}")
run_comparison()MODE | TURN | SUBAGENT OUTPUT
-----------------------------------------------------------------------------------------------
Per-invocation (fresh start) | 1 | Msgs: 1 | Known: duplicate, None, None
Per-invocation (fresh start) | 2 | Msgs: 1 | Known: None, pro, None
Per-invocation (fresh start) | 3 | Msgs: 1 | Known: None, None, refund
Per-thread (persistent) | 1 | Msgs: 1 | Known: duplicate, None, None
Per-thread (persistent) | 2 | Msgs: 3 | Known: duplicate, pro, None
Per-thread (persistent) | 3 | Msgs: 5 | Known: duplicate, pro, refund
Stateless (no checkpoint) | 1 | Msgs: 1 | Known: duplicate, None, None
Stateless (no checkpoint) | 2 | Msgs: 1 | Known: None, pro, None
Stateless (no checkpoint) | 3 | Msgs: 1 | Known: None, None, refund