第 12 篇讲 checkpoint 内部结构时埋过一句话:"第 23 篇的长期记忆走 Store 跨线程共享,与线程内的 checkpoint 互补。"现在兑现。checkpointer 加 thread_id 解决了会话内记忆,但它有一个天然边界:记忆锁死在单个线程里。用户今天在线程 A 里说"我对花生过敏",明天开个新线程问"推荐午餐",Agent 一脸茫然——checkpointer 不会跨线程去翻别的线程状态,这个问题它天生解决不了。
本篇是《AI Agent 开发》系列第 23 篇,也是第五模块"进阶架构"的开篇,代码基于 langgraph 1.2.x。全文按"记忆分类学 → 两大流派 → Store 最小接入 → 写入策略 → 完整闭环 → 踩坑 → 压缩与摘要"推进,示例沿用第 12 篇的聊天图底座。
1、记忆分类学
和认知科学对齐,官方Memory 概念文档把 Agent 记忆先按作用域分两层:会话内的短期记忆、跨会话的长期记忆;长期记忆再按内容形态分三类。全貌一棵树:
Agent 记忆
├── 短期记忆 —— 会话内,checkpointer 托管(第 12 篇)
│ 消息历史与 State 快照,按 thread_id 隔离
└── 长期记忆 —— 跨会话,Store 托管(本篇)
├── 语义记忆:事实与偏好("她对花生过敏")
├── 情景记忆:过往事件("上周一起修过部署脚本")
└── 程序性记忆:操作流程("输出代码前先跑测试")
四类记忆在工程上的对应关系一张表说清:
| 类型 | 存什么 | LangGraph 载体 |
|---|---|---|
| 短期 | 本轮会话的消息历史与 State | checkpointer |
| 语义 | 用户事实与偏好 | Store |
| 情景 | 过往对话与任务的经过 | Store(常为摘要形态) |
| 程序性 | Agent 学到的行为规则 | Store(注入提示词) |
只有短期记忆属于 checkpointer,其余三类都落在 Store——后三类共用同一套 put/search API,靠 namespace 分拣。多数 Agent 先做语义记忆就够用,情景与程序性按需追加;分类学是设计地图,不是起步清单。
2、长期记忆两大流派
让 Agent 拥有跨会话知识,业界就两条路。
检索式:对话中抽取事实存进外部库,来日对话前检索、拼进提示词。记忆与模型分离,store.put 一行写入,下一次对话立即可见。参数式:微调,把知识直接烧进模型权重,问答时无需检索。两条路的成本差距是结构性的:
| 维度 | 检索式 | 参数式(微调) |
|---|---|---|
| 记忆载体 | 外部库 + 提示词注入 | 模型权重 |
| 更新生效 | 写入即可见 | 收集数据 → 训练 → 评估 → 部署全流程 |
| 删除一条记忆 | store.delete 一行 |
做不到,需重训 |
| 上下文占用 | 每轮注入消耗 token | 零 |
| 适合内容 | 用户偏好、动态事实 | 稳定领域知识、输出风格 |
参数式的致命伤在第三行:用户要求"忘掉我提供过的信息",权重里烧进去的东西取不出来。工程默认检索式——记忆的增删改查是运维问题,不该升级成训练问题;参数式留给真正稳定的知识与风格,那已经不是"记忆"而是"能力"。本篇后续全部围绕检索式展开。
3、Store 最小接入
Store 是 LangGraph 的跨线程持久层,统一实现 BaseStore 接口,与 checkpointer 在 compile 处会师。两者的分工一张图看懂:
Agent 图(单例)
/ \
compile(checkpointer=…) compile(store=…)
│ │
会话内:按 thread_id 跨会话:按 namespace
├─ thread-1 ● 快照 ├─ ("alice", "memories")
├─ thread-2 ● 快照 ├─ ("bob", "memories")
└─ thread-3 ● 快照 └─ ("alice", "episodes")
线程之间互相隔离 线程之间按 namespace 共享
解读:同一个图实例同时挂两个持久层,checkpointer 记"这个线程聊到哪了",Store 记"这个用户是谁"——记忆版图由此拼完整。Store 的最小 API 四件套如下:
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
ns = ("alice", "memories") # namespace:字符串元组,可任意分层
store.put(ns, "mem-1", {"text": "用户对花生过敏"}) # value 必须是 dict
item = store.get(ns, "mem-1") # 精确读取一条
print(item.value["text"])
items = store.search(ns, limit=10) # 列出 namespace 内条目
store.delete(ns, "mem-1") # 删除
解读:put 收三样——namespace、key、value。namespace 像目录路径,("alice", "memories") 就是"alice 的 memories 目录",不同用户各写各的目录、互不可见;key 是条目身份证,对同一个 (namespace, key) 再次 put 是覆盖更新而不是新增——第 6 节"记忆冲突"的解药就藏在这。value 必须是 dict,传纯字符串直接报错。开发期用 InMemoryStore(重启即失,定位同 InMemorySaver),生产换数据库后端的 Store 实现,接口不变,详见 Stores 文档。
条目多起来后,全量列出不够用,search 要按语义检索。给 Store 配一个 embedding 索引:
from langchain.embeddings import init_embeddings
from langgraph.store.memory import InMemoryStore
store = InMemoryStore(
index={
"embed": init_embeddings("openai:text-embedding-3-small"),
"dims": 1536, # 维度须与 embedding 模型一致
"fields": ["text"], # 只对 text 字段做嵌入
}
)
# 配了 index 后,带 query 的 search 走语义检索
hits = store.search(ns, query="饮食忌口", limit=3)
解读:index 配置让 Store 在每次 put 时对指定字段自动嵌入,search 带 query 就按相似度返回 top-k。没配 index 时带 query 调用会报错,这是新手高频坑。语义检索与第 14 篇的向量库同源,细节见 InMemoryStore 参考与官方语义检索公告。
4、记忆写入策略
图挂上了 Store,下一个设计决策:谁来决定"记什么"?两条路线。
4.1、模型自主式
把"保存记忆"做成一个工具交给模型,提示词里写明标准,模型在对话中自行判断何时调用。memory-agent 官方仓库就是这个思路:
import uuid
from langchain_core.tools import tool
from langgraph.config import get_store
@tool
def save_memory(memory: str) -> str:
"""记住用户透露的长期事实或偏好,供未来会话使用。"""
store = get_store()
store.put(("alice", "memories"), str(uuid.uuid4()), {"text": memory})
return f"已记住:{memory}"
解读:工具函数不在节点签名里,拿 Store 靠 get_store()——它从运行时上下文取出编译期挂上的那个实例,图内任何代码(节点、工具、task)都能这么取,见 get_store 参考。返回值把记下的内容原样回显,让用户看得见 Agent 记了什么——第 6 节会讲这也是防记忆污染的关键一环。
4.2、规则式抽取
不信模型的自觉,就写死规则:每轮对话后追加一个抽取节点,用便宜的模型从最近消息里抽事实。核心实现:
def extract_memories(state: MessagesState, store: BaseStore):
prompt = (
"从下面的对话里抽取值得长期记住的用户事实,"
"每条一行,没有则输出空行:\n"
+ "\n".join(m.text for m in state["messages"][-2:])
)
result = small_model.invoke(prompt)
for line in result.text.splitlines():
if line.strip():
store.put(("alice", "memories"),
str(uuid.uuid4()), {"text": line.strip()})
return {} # 不改 State,只写记忆
解读:节点签名里声明 store: BaseStore 参数,LangGraph 自动注入——这是节点内访问 Store 的第二种方式,与 get_store() 等价。返回 {} 表示纯副作用节点,用一条边串在模型节点后面即可。两种策略对照:
| 维度 | 工具式(模型自主) | 规则式(每轮后处理) |
|---|---|---|
| 触发时机 | 模型判断,需要时才记 | 每轮必跑 |
| 额外成本 | 零,工具调用走主模型 | 每轮固定多一次抽取调用 |
| 漏记风险 | 有,取决于提示词 | 低 |
| 噪声风险 | 低 | 高,需去重过滤 |
对话型 Agent 优先工具式——用户在语境里的强调程度本身就是"值不值得记"的信号,模型拿得到这个信号;规则式适合无对话循环的流水线,或用户要求全量沉淀的场景。
5、偏好记忆完整闭环
把前两节拼成完整闭环:节点每次调用前检索记忆拼进 system prompt(注入),模型对话中调工具写入(沉淀),checkpointer 与 Store 并存。先看注入节点:
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage
from langgraph.graph import MessagesState
from langgraph.store.base import BaseStore
model = init_chat_model("openai:gpt-5").bind_tools([save_memory])
def call_model(state: MessagesState, store: BaseStore):
items = store.search(("alice", "memories"), limit=10)
facts = "\n".join(f"- {i.value['text']}" for i in items)
system = SystemMessage(
content=f"已知用户信息:\n{facts}\n"
"发现新的长期事实或偏好时,调用 save_memory 记住。"
)
return {"messages": [model.invoke([system] + state["messages"])]}
解读:每次模型调用前现场检索、现场拼接,记忆更新后下一轮立即生效;save_memory 沿用 4.1 的定义。装配与第 9 篇手写 ReAct 同构,只是 compile 多挂一个 store:
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.prebuilt import ToolNode
from langgraph.store.memory import InMemoryStore
def route(state: MessagesState):
return "tools" if state["messages"][-1].tool_calls else END
builder = StateGraph(MessagesState)
builder.add_node("model", call_model)
builder.add_node("tools", ToolNode([save_memory]))
builder.add_edge(START, "model")
builder.add_conditional_edges("model", route)
builder.add_edge("tools", "model")
store = InMemoryStore()
graph = builder.compile(checkpointer=InMemorySaver(), store=store)
解读:checkpointer 管线程内历史,store 管跨线程记忆,互不干扰。验证闭环是否真的跨会话——开两个线程:
c1 = {"configurable": {"thread_id": "chat-1"}}
graph.invoke({"messages": [{"role": "user", "content": "我对花生过敏,帮我记住。"}]}, c1)
c2 = {"configurable": {"thread_id": "chat-2"}}
r = graph.invoke({"messages": [{"role": "user", "content": "推荐一道午餐。"}]}, c2)
print(r["messages"][-1].text)
解读:第一轮模型调 save_memory 把过敏写进 Store;第二轮全新线程,checkpointer 历史为零,但注入节点 search 拿到了那条记忆,推荐会避开花生——跨线程记忆闭环生效。把 alice 换成真实用户 ID、namespace 第一段跟着走,多用户隔离就有了;记忆过百条后把全量列出换成第 3 节的语义检索,取 top-3 即可。
6、记忆系统的坑
闭环跑通只是及格线,长期记忆是要长期运营的数据资产,四个坑提前避开。
**过期与冲突。**用户三月说"我住北京",十月说"我搬到上海了",堆积式写入会让两条并存,检索命中哪条看运气。解法是更新而非堆积:写入前先 search 同主题条目,命中就复用它的 key 再 put——同 key 写入即覆盖;再给 value 加 updated_at 字段,注入时带上时间,让模型知道哪条更新。
**隐私。**记忆是用户数据的长期沉淀,法规层面(如 GDPR 的被遗忘权)要求可查看、可删除。好在 get、search、delete 就是现成的后端——产品上给用户一个"查看我的记忆、删除某条"的入口,成本极低;健康、财务类敏感信息要么单独 namespace 加权限,要么干脆不入库。
**记忆污染。**模型幻觉出的"事实"被工具固化,之后每轮当真话注入,错误自我强化——这是长期记忆独有的失败模式。缓解手段:抽取提示词写死"只记用户明确陈述的内容";工具返回值回显给用户,当场可纠正;重要决策让模型复述记忆来源,而不是直接照办。
**检索漂移。**记忆条数上千后,top-k 里混进无关项,注入的噪声反而稀释提示词。对策三板斧:limit 给小,3~5 条足够;namespace 按主题细分,饮食偏好与工作安排分开存;定期清理长期零命中的条目——记忆库也要做 housekeeping。
7、记忆压缩与摘要
记忆版图还剩最后一块:会话本身变长之后怎么办。两个方向,一个治短期、一个治长期。
短期侧,第 10 篇埋的伏笔正面兑现——SummarizationMiddleware。历史超过阈值时自动把旧消息摘要成一条系统消息,近期消息保留原文:
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
agent = create_agent(
model,
tools=[save_memory],
middleware=[
SummarizationMiddleware(
model="openai:gpt-5-mini", # 摘要交给便宜的小模型
trigger=("tokens", 4000), # 历史超过 4000 token 触发
messages_to_keep=5, # 摘要后保留最近 5 条原文
)
],
checkpointer=InMemorySaver(),
)
解读:上下文占用从随轮次线性增长压到近似常数,且对图结构零侵入。trigger 用元组形式 ("tokens", N),旧的 max_tokens_before_summary 参数已弃用;中间件全家族见中间件文档。
长期侧,把快结束的线程整段摘要成"情景记忆"存进 Store——第 12 篇的 get_state 正好取得到完整历史:
def archive_thread(graph, config, store):
state = graph.get_state(config)
dialog = "\n".join(m.text for m in state.values["messages"])
summary = model.invoke("把这段对话摘要成要点:\n" + dialog)
store.put(("alice", "episodes"),
config["configurable"]["thread_id"],
{"summary": summary.text})
解读:归档进 ("alice", "episodes") 后,未来任何会话检索这个 namespace,都能拿到"上周一起修过部署脚本"这类背景——正是第 1 节说的情景记忆。与 SummarizationMiddleware 的分工:前者压缩当下线程的窗口,后者把经验沉淀给未来的线程,一急一缓。
8、总结
回收全文:记忆分四类,短期归 checkpointer,语义、情景、程序性三类归 Store;实现选检索式而非参数式,更新成本与可删除性差一个量级;Store 的组织轴是 namespace,put、get、search、delete 四件套,同 key 写入即更新;写入策略工具式优先、规则式兜底;闭环 = 启动检索注入 + 工具写入沉淀 + 两个持久层并存;运营上盯住过期冲突、隐私、污染、漂移四个坑;会话变长就摘要——短期用 SummarizationMiddleware,长期把线程归档成情景记忆。
留一个练习:回到第 20 篇的 SQLite 查询 Server,给它加一张可写的 memories 表和 save_memory、search_memory 两个工具(只读连接之外单开一个可写连接),接到 Claude Code 或 Claude Desktop,让任意 Host 拥有跨会话记忆——本质上是用 MCP 工具重做一遍本篇的 Store。加分项:写入前先 SELECT 查重,命中即 UPDATE,把"更新而非堆积"做进 SQL。
对于任何要长期服务用户的 Agent,建议第一天就把 Store 挂上:记忆是用户回头的起点,金鱼 Agent 留不住人。下一篇进入多 Agent 架构:Supervisor、handoff 与层级团队。


