一、为什么需要短期记忆?
我们先想一个最朴素的问题:你用 LangGraph 写了一个最简单的问答 Graph,用户第一句说"我叫小明",第二句问"我叫什么"——Agent 答得出来吗?
答不出来。
默认情况下,LangGraph 的每一次invoke都是无状态的:节点函数只看这一轮传进来的state,跑完就扔。想要让它在同一场对话里记住前面说过什么,就必须引入 LangGraph 的短期记忆机制。
短期记忆的核心只有两个东西:
| 概念 | 作用 | 类比 |
|---|---|---|
| Thread | 给一场对话起个"会话 ID" | 微信里一个聊天窗口 |
| Checkpoint | 把某一时刻的完整 state 存成快照 | 游戏存档点 |
| Checkpointer | 负责存快照、读快照的组件 | 存档系统 |
一句话:Thread 是身份证,Checkpoint 是存档,Checkpointer 是存档管理员。
二、整体架构图
短期记忆在 LangGraph 里的位置,可以用下面这张图概括:
关键点:
- 每个 super-step(一次节点执行完)都会自动落一次盘,不是你手动存;
- 下次同一个
thread_id再来请求,Checkpointer 自动把最近一次的 state 捞出来喂给图; - 存储后端可以换(内存 / SQLite / PostgreSQL),业务代码完全不动。
三、Thread:给对话一个身份
Thread 本身不是一个"数据结构",它就是一个字符串 ID(比如"user-001-session-01")。你在调用图的时候通过config传进去:
config={"configurable":{"thread_id":"thread-1"}}LangGraph 看到这个thread_id,就会去找这个会话对应的所有历史快照。不同的thread_id之间完全隔离,A 用户的对话 B 用户看不到。
小提示:
thread_id你自己起名字就行,生产环境一般用「用户ID + 会话ID」拼出来。
四、Checkpoint:一次执行的完整快照
每次图执行一步(或者一轮完整跑完),LangGraph 会把当前 state 完整序列化成一份Checkpoint。它里面通常包括:
- 所有节点写到 state 里的数据(消息列表、中间变量……);
- 当前执行到哪一步了(下一步该走哪个节点);
- 自定义的一些元信息。
你可以把它理解成"游戏存档":不仅存了你的等级装备(数据),还存了你现在在哪个地图哪个存档点(执行位置)。这也是为什么 LangGraph 能做"中断后恢复"和"人工介入"——本质就是从某个 Checkpoint 接着跑。
五、Checkpointer:真正干活的组件
光有 Thread 和 Checkpoint 的概念还不够,你得在编译图的时候显式挂一个 Checkpointer,LangGraph 才会自动帮你存快照。
最常用的是SqliteSaver(持久化到本地文件)和MemorySaver(存在内存里,重启就没)。
5.1 最小可跑示例
fromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.checkpoint.memoryimportMemorySaverfromtypingimportTypedDictfromlangchain_core.messagesimportHumanMessage,AIMessage# 1. 定义 stateclassChatState(TypedDict):messages:list# 2. 假装是个 LLM 节点defchat_node(state:ChatState):last=state["messages"][-1].contentreturn{"messages":[AIMessage(content=f"你刚才说的是:{last}")]}# 3. 建图builder=StateGraph(ChatState)builder.add_node("chat",chat_node)builder.add_edge(START,"chat")builder.add_edge("chat",END)# 4. 关键:挂 Checkpointercheckpointer=MemorySaver()graph=builder.compile(checkpointer=checkpointer)# 5. 用同一个 thread_id 连续对话config={"configurable":{"thread_id":"t-1"}}graph.invoke({"messages":[HumanMessage("我叫小明")]},config)resp=graph.invoke({"messages":[HumanMessage("我叫什么?")]},config)print(resp["messages"][-1].content)# 输出:你刚才说的是:我叫什么?注意第 4 步和第 5 步:
builder.compile(checkpointer=checkpointer)—— 这一行加上,短期记忆就生效了;- 两次
invoke用的是同一个thread_id,第二次进来时 Checkpointer 会自动把第一次的 messages 塞进 state。
5.2 换成 SQLite,重启不丢
fromlanggraph.checkpoint.sqliteimportSqliteSaverwithSqliteSaver.from_conn_string("chat.db")ascheckpointer:graph=builder.compile(checkpointer=checkpointer)# ... 跑对话生产环境推荐SqliteSaver(单机)或langgraph-checkpoint-postgres(多实例部署)。换后端的时候,业务代码一行都不用改。
六、查看 Thread 状态:人是怎么"看见"记忆的?
挂了 Checkpointer 之后,你随时可以把某个 thread 的历史捞出来:
# 获取某个 thread 的最新 statestate=graph.get_state(config)print(state.values)# 当前 state 里有啥print(state.next)# 下一步要走哪个节点(如果是中断状态会很有用)# 遍历这个 thread 的所有历史快照forsnapingraph.get_state_history(config):print(snap.config,snap.metadata)get_state_history特别有用:它能列出这个会话每一次存档点。配合"人工介入"功能,你可以让用户回退到任意一个历史存档点重新跑——相当于"撤销对话"。
七、几个新手必踩的坑
- 忘了传
thread_id:invoke时 config 里没带thread_id,Checkpointer 不知道存到哪个会话里,等于白挂。 - 每次都新建
MemorySaver:在函数里compile图、每次请求新建 checkpointer,快照存在新对象里,下次又读不到了。Checkpointer 要复用。 - state 里塞了不可序列化的对象:比如数据库连接、文件句柄。Checkpoint 要序列化落盘,这种字段会直接报错。
- 以为短期记忆能跨会话:换了
thread_id就是新会话,老对话"看不见"。想要跨 Thread 记忆,得用 Store(下篇讲)。
八、小结
- 短期记忆 =Thread(会话ID)+ Checkpoint(快照)+ Checkpointer(存档器);
- 编译图时
compile(checkpointer=...)一行接入,每个 super-step 自动存档; - 同一个
thread_id连续调用,Agent 就能记住本场对话的上下文; - 存储后端可插拔:Memory(调试用)→ SQLite(单机生产)→ Postgres(分布式);
get_state/get_state_history是调试和做"撤销/人工介入"的核心 API。
下一篇我们聊长期记忆 Store——怎么让 Agent 跨会话记住"这个用户喜欢喝美式不加糖"这种跟具体对话无关的偏好。