LLM 是 stateless 纯函数;所有「记忆」都是你 token 预算里硬塞进去的状态。会塞和不会塞,差一个量级。
问 100 个做 AI 产品的人「你怎么管理 memory」,90 个会说「我塞 vector DB 里 retrieve」。这就是大多数 AI agent 用三轮就开始胡言乱语的根本原因——把所有 state 当成一种东西,用一种工具(embedding + RAG)处理。资深玩家的认知是:context window 是稀缺资源,state 至少有四类,每类的写入/读取/失效逻辑完全不同。这一期不讲怎么调 chroma 参数(那是 101),讲四件事:怎么把混在一起的 state 拆成 conversation / scratchpad / profile / knowledge 四层并分别管理;短期记忆为什么 truncate > summarize > hierarchical 三档要分场景选;长期记忆为什么 vector retrieval 只能解决三分之一问题,剩下两类要 structured KV + episodic event log;以及 MemGPT / Letta 路线的 self-maintained profile 怎么做到「越用越懂你」。读完你应该能在白板上画出自己 agent 的 state 拓扑,并知道每条线该走哪条路。
OS 教科书第一章告诉你内存有 register / cache / RAM / disk 的分层。LLM 的「内存」也有同样的分层,只是大多数人没意识到。Anthropic 在 Building Effective Agents 里反复强调一句话——「context is a budget」——你必须主动决定每个 token 用在哪。下面四类 state 必须显式分开:
四种 state 用一种机制(往 vector DB 塞)会同时踩三个坑:(1)对话历史按 embedding 检索丢失时序——「上一句」可能 retrieve 不到;(2)user profile 被淹在 1M 条 chunk 里,cosine similarity 选不到关键事实;(3)scratchpad 永远留存,下一次任务被无关中间结果污染。正确做法是给每层独立的存储 + 独立的 read/write policy。
把四层显式建模成 Python class,让 prompt 组装时强制走分层接口:
# memory_layers.py — 强制分层,禁止「往一个 dict 里乱塞」
from dataclasses import dataclass, field
from typing import Protocol
class MemoryLayer(Protocol):
def read(self, query: str) -> str: ...
def write(self, content: str, meta: dict) -> None: ...
@dataclass
class Conversation: # 短期,append-only
messages: list = field(default_factory=list)
def read(self, _): return self.messages[-20:] # 默认滑窗
@dataclass
class Scratchpad: # 单任务,task_id scoped
by_task: dict = field(default_factory=dict)
def clear(self, task_id): self.by_task.pop(task_id, None)
@dataclass
class Profile: # 跨会话稳定事实
facts: dict = field(default_factory=dict) # {"name":"BigCat", ...}
def read(self, _): return self.facts # 全量
@dataclass
class Knowledge: # 文档/代码,RAG 入口
index: object # 你的 vector store
def read(self, query): return self.index.search(query, k=5)
# —— 组装 prompt 时强制走四层 ——
def build_context(query, conv, scratch, profile, kb):
return {
"system": f"User profile:\n{profile.read(None)}",
"messages": conv.read(None),
"working": scratch.by_task.get(current_task_id, []),
"retrieved": kb.read(query), # 仅与 query 相关
}
关键不是这段代码本身,是它强制你回答四个问题:profile 哪些字段全量进 system?scratch 什么时候 clear?conv 滑窗多少?kb retrieve top-k 多少?想不清楚四个答案的 agent,上线必出 memory 事故。
Context 超了怎么办?三档处理方式,复杂度和损耗都不同:
三档的选择标准很简单——「我以后会不会查回早期对话的具体细节?」。客服一般不会(truncate);写小说会但不要求精确(summarize);做投研复盘要精确引用(hierarchical)。
三档可以放在同一个 manager 里按 message 数自动升级:
# conversation_manager.py — 三档自动切换
class ConversationManager:
def __init__(self, mode="truncate", window=20, summary_threshold=40):
self.messages, self.summary = [], None
self.mode, self.window, self.threshold = mode, window, summary_threshold
def add(self, msg): self.messages.append(msg)
def build(self):
if self.mode == "truncate":
return self.messages[-self.window:]
if self.mode == "summarize":
if len(self.messages) > self.threshold:
old = self.messages[:-self.window]
self.summary = compact(self.summary, old) # LLM 增量摘要
self.messages = self.messages[-self.window:]
head = [{"role":"system","content":f"Prior summary:\n{self.summary}"}] \
if self.summary else []
return head + self.messages
if self.mode == "hierarchical":
recent = self.messages[-self.window:]
old_chunks = self.vector.search(latest_user_msg(recent), k=5)
return [recall_block(old_chunks)] + recent
# —— compaction prompt(关键细节别丢)——
COMPACT = """Summarize the prior conversation. PRESERVE:
- All factual decisions made
- All open questions / TODOs
- Any explicit user preferences mentioned
DISCARD: greetings, clarifications already resolved, model's apologies.
Existing summary: {old_summary}
New messages: {new_msgs}
Output (≤300 tokens):"""
compaction prompt 的 PRESERVE / DISCARD 列表是决定摘要质量的关键。Anthropic 文档里给的版本特别强调「保留 decisions 和 open TODOs」——这正是研究/编码场景最容易丢的两类信息。
把 vector DB 当万能长期记忆是 2023 年的迷思。资深做法是按查询模式分三种存储,各司其职:
这三种不是互斥而是互补。一个成熟 agent 会同时有:vector store 装文档 + KV 装 profile + event log 装行为历史。Letta(MemGPT 的产品版本)官方文档把这套叫做「memory hierarchy」,并显式区分 core_memory(KV)、archival_memory(vector)、recall_memory(event log)。Microsoft GraphRAG 走了第四条路(知识图谱),适合实体关系密集的场景,但门槛高,普通团队优先建好前三种。
用三种 store 实现一个能「记住你」的 personal agent,关键是写入路由——一条信息进来,决定它进哪个 store:
# memory_router.py — 一条信息进来,路由到正确的 store
ROUTE_PROMPT = """Classify this user utterance into ONE memory type:
- FACT : a stable fact about the user (name, role, preference)
- EVENT : an action/decision/experience that happened
- DOCUMENT : reference content (article, code, doc)
- NONE : transient (greeting, ack, clarification)
Return JSON: {"type": "...", "extract": "..."}"""
def ingest(utterance, user_id):
r = llm(ROUTE_PROMPT, utterance)
if r["type"] == "FACT":
profile_kv.upsert(user_id, r["extract"]) # structured KV
elif r["type"] == "EVENT":
event_log.append(user_id, ts=now(), text=r["extract"])
elif r["type"] == "DOCUMENT":
vector_store.add(embed(r["extract"]), meta={"user":user_id})
# NONE → drop
# —— 读取时按 query 类型路由 ——
def recall(query, user_id):
intent = classify(query) # fact / temporal / semantic
if intent == "fact":
return profile_kv.get_all(user_id) # 全量 profile
if intent == "temporal":
return event_log.range(user_id, last="7d") # 按时间
return vector_store.search(query, k=5, filter={"user":user_id})
这套路由的精髓——问题决定存储。「我叫什么?」走 KV;「上周我们聊了什么?」走 event log;「找一段我之前提过的关于禅修的内容」走 vector。一种存储解所有问题是新手错觉。
第 3 节的 structured KV 解决了「存什么、怎么读」,但还有个工程问题——谁来决定写什么。三种方案:
update_profile tool,让它在每轮对话后自主决定要不要更新 profile。MemGPT / Letta / ChatGPT memory / Claude memory 走的都是这条路。self-maintained 的难点不是「让它写」,是「让它写对」。Profile 是要全量注入下一次会话 system prompt 的,里面塞了垃圾就永远污染。要解决四个问题:
给 agent 一个 profile 工具,每轮对话后跑一遍「需不需要更新」:
# profile_tools.py — 给 LLM 的 self-maintained profile 工具
PROFILE_UPDATE_PROMPT = """Review the latest user message and decide if the user
profile should be updated. Only write facts that:
1. Are explicitly stated by the user (cite message)
2. Are likely to be true beyond this session
3. Are not already in the profile (or contradict it)
Current profile: {profile_json}
Latest message: {message}
Output JSON:
{
"action": "add" | "update" | "delete" | "none",
"key": "...",
"value": "...",
"evidence_msg_id": "...", // 必填,无 evidence 直接 "none"
"reason": "..."
}"""
def maybe_update_profile(user_id, latest_msg, msg_id):
current = profile_kv.get_all(user_id)
decision = llm(PROFILE_UPDATE_PROMPT.format(
profile_json=json.dumps(current), message=latest_msg))
if decision["action"] == "none": return
if not decision.get("evidence_msg_id"): return # 无 evidence 拒绝
profile_kv.apply(user_id, decision)
changelog.append(user_id, decision, ts=now()) # 可追溯
# —— 每个 session 起手前注入 profile ——
def build_system_prompt(user_id):
p = profile_kv.get_all(user_id)
return f"""You are a personal assistant for the following user.
Stable facts about them (use to personalize, but verify before acting):
{json.dumps(p, indent=2, ensure_ascii=False)}
Important: if a fact seems outdated, ASK the user to confirm rather than
silently override the profile."""
三个细节决定品质:必填 evidence_msg_id(挡幻觉)、changelog(可追溯,用户问「你怎么知道我素食」能答出来)、系统 prompt 提示模型主动验证(避免基于过时 profile 自信犯错)。这三条都做到了,profile 才从「玩具」变「工程产物」。
挑你正在做的或常用的一个 agent(个人 research bot / coding agent / 客服 / 写作助手),按以下 6 步画清 state 拓扑:
30 分钟后你应该有:一张 state 拓扑图、四层各自的 read/write policy、写入路由的判定逻辑、遗忘机制。这就是把 agent 从 demo 升级到「越用越懂你」的产品的关键文档。下一次同事问「我们的 agent 怎么记忆的」,你递这张图,不再支支吾吾说「塞 chroma 里」。