LangGraph 记忆存储的三重境界这个题目我琢磨了挺久。做 Agent 应用的朋友应该都有同感底层模型本身不带记忆可用户要的是连贯对话、记住偏好、断点续跑。LangGraph 把记忆这件事拆成了三个层次理解透了Agent 的实际体验能直接上一个台阶。这篇文章就聊聊我在真实项目里对这三重境界的理解以及对应代码怎么写、坑在哪。适合正在用 LangGraph 搭 Agent或者准备从单轮问答往多轮会话应用演进的同学。1. 第一重境界State 里的临时记忆1.1 白板一样的 State字段和 Reducer 到底怎么工作先看最基本的记忆载体State。LangGraph 的整个执行过程可以理解成一群节点围着一张白板干活。这张白板就是 State节点读它、改它改完再传给下一个节点。白板的格式在你定义图的时候就用 TypedDict 写死了不存在的字段不能乱写这跟写普通 Python dataclass 的约束感很像。from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_name: str todo_count: int这里最值得注意的就是Annotated[list, add_messages]。如果不写这个注解多个节点往messages字段写入时后面的会把前面的直接覆盖这在多轮 Agent 里是灾难。add_messages是一个 reducer它定义的是“多个写入值如何合并”。消息列表场景下它会按消息的id去重把新消息追加到列表尾部。也就是说即便两个节点都往messages里塞内容最终也不会出现谁把谁干掉的情况。节点函数本身只负责返回“要改动的增量”不需要把整个 State 原样返回。比如某节点只改了user_name就返回{user_name: A同学}messages和其他字段会自动保留。这套机制跟看板很像你把便利贴贴上去白板上原来的内容不动系统会自动归位。1.2 一轮任务内的“顺手记”结束就消失第一重境界能干什么能支撑单次复杂任务里多步推理的信息流转。举个例子一个调研 Agent先收集资料再整理要点最后写总结。中间每一步的中间结果都可以写进 State下一步节点读出来用。用大白话说这是“超局部记忆”相当于你在做一道复杂的数学题草稿纸上的中间过程。题目做完了草稿纸收走什么都不留。这个特点决定了它的适用范围适合编排类、工具调用类任务不适合多轮会话。因为一旦图执行结束State 就被释放了。你下一次再调用同一个图messages 是空的用户上一轮说“我叫小明”这一轮它就忘了。很多新手第一次跑通 LangGraph 后很开心紧接着就发现“它怎么不记得我说过啥”其实就是被第一重境界的边界卡住了。1.3 这个阶段最容易踩的坑我在实际项目里见过的第一类问题就是 reducer 忘写。某个节点往messages里更新消息另一个节点也更新结果先写的数据被后写的覆盖跑了几轮才发现历史消息越来越少。排查方法很简单把 State 定义里的Annotated[list, add_messages]检查一遍凡是需要追加的列表字段都别漏。还有一个坑是“节点返回了整个 State”。这不会报错但很容易造成字段膨胀。因为 State 是全局的如果你习惯把临时变量塞进去节点多了之后白板会变成杂物间每个节点都被迫拿到一堆无关数据。我的习惯是只在 State 里放必须跨节点流转的字段一次性的中间结果用闭包变量或局部变量处理。2. 第二重境界Checkpointer 与 Thread 持久化2.1 给白板拍快照Checkpointer 让状态活到下一轮State 只活一次图执行那怎么让它在下一轮继续存在LangGraph 的答案是 Checkpointer。它的工作方式非常粗暴有效图的每一步执行完都把当前 State 保存一份快照。下次你用同一个thread_id再来调用图时它会先把上次快照捞出来再往上面叠加新的节点执行。这里的thread_id是核心概念你可以把它理解成“会话 ID”。同一个用户、同一个会话就用同一个thread_id新会话就换一个新的。LangGraph 把同一thread_id下的多次执行串成了一条时间线断点续跑、人工审批后继续执行都是基于这套快照机制实现的。Checkpointer 的实现有几种常见选择Checkpointer存储位置适合场景MemorySaver进程内存本地调试、单机演示重启即丢SqliteSaver本地 SQLite 文件单机持久化重启不丢PostgresSaver数据库生产环境多实例共享支持水平扩展MemorySaver 最省事但它活在内存里进程一重启就全没了。SqliteSaver 适合大多数单机项目本地文件持久化不用额外起服务我个人的很多项目都是从它开始的。2.2 实操做一个能记住上下文的问答 Agent下面是我在项目里一直在用的一个基础结构。假设我们要做一个客服问答 Agent用户会连续提问Agent 要记得上一轮聊到哪了。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.checkpoint.memory import MemorySaver class ChatState(TypedDict): messages: Annotated[list, add_messages] def call_model(state: ChatState): # 这里的 get_llm() 换成你实际使用的模型客户端 llm get_llm() response llm.invoke(state[messages]) return {messages: [response]} graph StateGraph(ChatState) graph.add_node(agent, call_model) graph.add_edge(START, agent) graph.add_edge(agent, END) memory MemorySaver() # 本地调试先用这个 app graph.compile(checkpointermemory)调用方式是在config里带上thread_idconfig {configurable: {thread_id: user-001}} # 第一轮 result app.invoke({messages: [{role: user, content: 我叫小明我有三个订单要查}]}, config) # 第二轮同一个 thread_idAgent 还记得上一轮 result app.invoke({messages: [{role: user, content: 第一个订单到哪了}]}, config)第二轮执行时LangGraph 会把thread_id为user-001的历史消息从 Checkpointer 里恢复出来与当前轮次的新消息合并成一整段上下文再喂给模型。你在代码层面甚至不需要手动拼历史消息只要保证同一个会话用同一个thread_id即可。2.3 Thread ID 的隔离与调试手段有几个细节我觉得值得单独拎出来说。第一thread_id是一个天然隔离边界。用户 A 用user-001用户 B 用user-002两边互不干扰。这既保证了隐私也让并发调用变得安全。但代价是如果你把thread_id写死在代码里那所有用户就共享同一个时间线用户的私有对话会串线。生产环境里一定要动态生成thread_id。第二SqliteSaver 初始化时有个“子路径”问题。如果你传入一个不存在的目录路径它会尝试在路径下创建文件但目录本身不存在时会报Looking for a subpath with prefix之类的错误。解决方法很简单先os.makedirs确保目录存在再传路径。第三调试多轮状态时除了直接跑invokeLangGraph 还提供了两个非常有用的接口app.get_state(config)可以查看当前thread_id下的完整快照app.update_state(config, values)可以手动修改状态。我经常用这两个接口在测试里模拟“用户中途修改信息”的场景不用重新跑完整轮对话就能验证逻辑。第四别忽视 Checkpointer 的恢复粒度。LangGraph 的快照是每个 super-step 结束后自动保存的。也就是说如果某一步模型调用出错你重试时它会从最近一个稳定快照恢复而不是从头跑。这一点在做长时间运行的 Agent 时特别省心。3. 第三重境界跨线程的全局记忆 Store3.1 为什么一个会话的记忆远远不够第二重境界解决的是“一轮对话能记住”但它有个硬边界记忆被锁在thread_id里。用户换个会话或者换台设备历史就断开了。更难受的是有些信息本来就不该跟具体会话绑定比如用户的语言偏好、常用收货地址、会员等级。这些是“属于用户”的数据不是“属于某次会话”的数据。LangGraph 的第三重记忆载体就是 Store。Store 是一个跨线程、跨会话的全局键值存储它跟 Checkpointer 的核心区别是Checkpointer 按thread_id组织Store 按命名空间组织。你可以把它理解成一张按部门分类的档案柜每个柜子是一个命名空间柜子里放着一个个带 key 的档案。from langgraph.store.memory import InMemoryStore store InMemoryStore() # 写入用户偏好namespace 区分不同用户key 标识具体信息 store.put( (user, user-001), preferences, {language: zh, timezone: Asia/Shanghai, theme: dark} ) # 读取 item store.get((user, user-001), preferences) print(item.value) # {language: zh, timezone: Asia/Shanghai, theme: dark}namespace是一个字符串元组LangGraph 官方推荐的做法是把最宽泛的概念放前面、最具体的放后面比如(user, user-001)这样检索和权限控制都比较自然。key是具体档案名value是真正要存的数据可以是任意能被序列化的 Python 对象。3.2 在节点里读写全局记忆Store 的使用不是在调用invoke时传参而是在构建图时通过compile(graph, storestore)注入到执行上下文里。在节点函数内部你可以通过闭包或者把节点方法绑定到某个对象上来访问它。我常用的做法是定义一个类来封装节点这样既有地方放 Store代码也好维护from langgraph.graph import StateGraph, START, END from langgraph.store.memory import InMemoryStore class AgentNode: def __init__(self, store): self.store store def read_user_profile(self, state): # 从 State 里拿 user_id实际项目中可以来自认证信息 user_id state[user_id] profile self.store.get((user, user_id), preferences) if profile: user_prefs profile.value else: user_prefs {language: zh} # 把偏好注入到系统提示词里 system_prompt f用户偏好{user_prefs} messages [{role: system, content: system_prompt}] state[messages] # 这里再调用模型 ... return {messages: [response]} store InMemoryStore() agent AgentNode(store) graph StateGraph(ChatState) graph.add_node(agent, agent.read_user_profile) graph.add_edge(START, agent) graph.add_edge(agent, END) app graph.compile(checkpointermemory, storestore)这种写法的好处是职责清晰Checkpointer 管“这次会话聊到哪”Store 管“这个用户长期有哪些固定信息”。两者互补互不冲突。3.3 把对话历史沉淀成长期记忆的一种务实方案全局 Store 最常见的一个用法是在会话结束时做“摘要沉淀”。比如客服 Agent用户每次会话都会有大量细节不可能全量往 Store 里塞。我的做法是会话结束后用模型生成一段结构化摘要包括用户核心诉求、已经解决的、待跟进的事项然后写入 Store。summary 用户咨询了订单配送问题已解决偏好中文客服希望后续通知通过短信发送。 store.put((user, user-001), summary, {text: summary, updated_at: timestamp})下次用户再来时先读summary把历史摘要拼进 system promptAgent 就能做到“虽然我不记得上轮细节但我记得上轮结论”。这个方法成本低、效果好特别适合客服、导购、医疗咨询这类以“多次复访”为核心场景的应用。更进一步Store 的 value 里其实可以放 embedding 向量。你可以在写入用户行为记录时顺便生成向量查询时先按namespace缩小范围再算相似度实现“语义记忆召回”。严格来说这已经不是 LangGraph 内置的功能而是 Store 加向量检索的组合拳但思路值得留个底。LangGraph 把存储层抽象得足够薄接一个向量数据库做底层是完全可行的。4. 三层记忆架构的配合思路与排查实录4.1 三层应该怎么配合而不是互相替代很多刚接触这三重记忆的人会有一个误区既然 Store 能存全局数据是不是只用 Store 就够了不是的。三层记忆解决的是三个不同时间尺度的问题记忆层级载体生命周期典型用途第一重状态记忆State Reducer单次图执行单轮多步推理、工具调用中间结果第二重会话记忆Checkpointer thread_id多次执行绑定会话多轮对话、断点续跑、人工审批恢复第三重全局记忆Store namespace跨会话长期存在用户偏好、历史结论、业务实体信息我的建议是State 里只放流程需要的东西Checkpointer 承担所有会话级上下文Store 只放“删掉了会亏”的用户级和业务级数据。三层各司其职整个系统的复杂度才是可控的。4.2 实战中遇到过的四个典型问题第一个是 reducer 覆盖问题。表现是多轮对话运行到某个节点后历史消息被截断模型开始“失忆”。除了检查add_messages之外还要注意有些从老代码迁移过来的节点可能直接用state[messages] ...这种覆盖式写法这会绕过 reducer 机制。第二个是 SqliteSaver 的文件锁问题。本地开发时多个进程同时写同一个 SQLite 数据库文件会出现database is locked。LangGraph 的 SqliteSaver 为了跨线程安全内部会做连接管理但多进程场景还是建议用 PostgresSaver或者至少按进程分文件。第三个是thread_id使用姿势不对。最典型的错误是把thread_id设成常量所有用户共享一条时间线会话隐私直接漏穿。另一个常见错误是每次请求都随机生成thread_id导致 Checkpointer 永远无法命中历史快照记忆完全失效。正确做法是从用户身份信息中派生稳定的thread_id比如fuser-{user_id}。第四个是 Store 里存了不可序列化对象。Python 对象五花八门但 Store 在落盘和传输时要求可序列化。我之前试着把模型客户端实例塞进去跑起来直接报错。解决办法是只存 JSON 友好的基础类型或者自己做好序列化转换。4.3 从“能记住”到“记得好”的几个经验记忆不是越多越好。把二十轮对话的全部细节都塞进 Store下次会话不仅响应变慢还可能把模型上下文撑爆。经验法则Store 里存结论不存过程存偏好不存每次点击存可复用的实体信息不存临时生成的内容。时效性问题也值得上心。用户三个月前的地址可能已经失效。我的处理方式是在updated_at字段之外再给重要档案加一个ttl或expires_at读取时做一次过期判断。这不是 LangGraph 内置能力但在 Store 外层包一层数据校验逻辑并不难。还有一个很实用的小技巧把 Store 里的用户偏好定期导出一份精炼版 Profile直接注入 system prompt。这比让模型每次从头推理“用户可能喜欢什么”要可靠得多。我在一个导购 Agent 里试过加了这行 Profile 之后推荐商品的命中率有明显提升而这个改动本身只需要一次 Store 读取。我个人在实操里体会最深的一点是记忆架构要尽早设计不要等 Agent 跑通了再补。thread_id的命名规范、Store 的命名空间结构、摘要的生成策略这些在项目第一天定下来后续所有节点都会受益。等代码满天飞再回头改记忆层牵一发动全身非常痛苦。最后再分享一个细节调试记忆问题时别忘了配合get_state看快照内容很多时候你以为丢了记忆其实只是 View 层的 prompt 组合逻辑漏了字段排查起来比想象中快得多。
