企业官网建设流程全解析

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 的实现有几种常见选择：

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(checkpointer=memory)

调用方式是在config里带上thread_id：

config = {"configurable": {"thread_id": "user-001"}} # 第一轮 result = app.invoke({"messages": [{"role": "user", "content": "我叫小明，我有三个订单要查"}]}, config) # 第二轮，同一个 thread_id，Agent 还记得上一轮 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确保目录存在，再传路径。

第三，调试多轮状态时，除了直接跑invoke，LangGraph 还提供了两个非常有用的接口：app.get_state(config)可以查看当前thread_id下的完整快照；app.update_state(config, values)可以手动修改状态。我经常用这两个接口在测试里模拟“用户中途修改信息”的场景，不用重新跑完整轮对话就能验证逻辑。

第四，别忽视 Checkpointer 的恢复粒度。LangGraph 的快照是每个 super-step 结束后自动保存的。也就是说，如果某一步模型调用出错，你重试时它会从最近一个稳定快照恢复，而不是从头跑。这一点在做长时间运行的 Agent 时特别省心。

3. 第三重境界：跨线程的全局记忆 Store

3.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, store=store)注入到执行上下文里。在节点函数内部，你可以通过闭包或者把节点方法绑定到某个对象上来访问它。

我常用的做法是定义一个类来封装节点，这样既有地方放 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(checkpointer=memory, store=store)

这种写法的好处是职责清晰：Checkpointer 管“这次会话聊到哪”，Store 管“这个用户长期有哪些固定信息”。两者互补，互不冲突。

3.3 把对话历史沉淀成长期记忆的一种务实方案

全局 Store 最常见的一个用法，是在会话结束时做“摘要沉淀”。比如客服 Agent，用户每次会话都会有大量细节，不可能全量往 Store 里塞。我的做法是：会话结束后，用模型生成一段结构化摘要，包括用户核心诉求、已经解决的、待跟进的事项，然后写入 Store。

summary = "用户咨询了订单配送问题，已解决；偏好中文客服，希望后续通知通过短信发送。" store.put(("user", "user-001"), "summary", {"text": summary, "updated_at": timestamp})

下次用户再来时，先读summary，把历史摘要拼进 system prompt，Agent 就能做到“虽然我不记得上轮细节，但我记得上轮结论”。这个方法成本低、效果好，特别适合客服、导购、医疗咨询这类以“多次复访”为核心场景的应用。

更进一步，Store 的 value 里其实可以放 embedding 向量。你可以在写入用户行为记录时顺便生成向量，查询时先按namespace缩小范围，再算相似度，实现“语义记忆召回”。严格来说这已经不是 LangGraph 内置的功能，而是 Store 加向量检索的组合拳，但思路值得留个底。LangGraph 把存储层抽象得足够薄，接一个向量数据库做底层是完全可行的。

4. 三层记忆架构的配合思路与排查实录

4.1 三层应该怎么配合，而不是互相替代

很多刚接触这三重记忆的人会有一个误区：既然 Store 能存全局数据，是不是只用 Store 就够了？不是的。三层记忆解决的是三个不同时间尺度的问题：

我的建议是：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，比如f"user-{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 组合逻辑漏了字段，排查起来比想象中快得多。