checkpoint与thread_id

用户第二轮提问时，Agent 必须还记得第一轮说了什么——仅靠单次 invoke 传入的 state 不够。需要 checkpointer 在每次 superstep 后落盘，并用 thread_id 区分会话。

段末注释：checkpoint（检查点）= 某一 superstep 的 State 快照 + 元数据；thread_id = 会话主键第一维。

社区方案：开发用官方 InMemorySaver（BaseCheckpointSaver），不要自写 dict 当记忆。风险：进程内三本账，重启/多 worker 不共享；MemorySaver 只是同名别名。

---

1. 定位

维度	内容
角色	跨 invoke 的短时记忆：Saver 存、图调
输入 → 输出	configurable.thread_id → 加载/写入该会话最新快照
核心 API	InMemorySaver、compile(checkpointer=...)、get_tuple / put
依赖 LangChain	messages 常用 add_messages

出现背景：无 Saver 时每次 invoke 从本次 input 起跑，多轮对话断裂。

---

2. 图拓扑

节点表

节点名	职责	读 State	写 State
chat	追加回复	messages	messages

边表

源	目标	类型
START	chat	固定
chat	END	固定

---

3. invoke 生命周期（两次 invoke）

compile(checkpointer=memory) 把 InMemorySaver 挂到 CompiledStateGraph

第一次 invoke(thread_id=T):
  1. memory.get_tuple({thread_id:T}) → None
  2. 用本次 input 灌 channels
  3. chat 经 add_messages 追加 AI
  4. 每 superstep：memory.put → storage[T][""][新id]

第二次 invoke(同一 T):
  1. get_tuple(T) → 最新 checkpoint，channels 含上轮 messages
  2. 新 HumanMessage 与旧列表合并
  3. chat 再追加 AI → put 新 id（parent 指向上一拍）

图从不直接读 memory.storage；只调用 Saver 接口。

---

4. 原理

4.1 compile 如何绑上 Saver

InMemorySaver() 在进程里建三本账（皆为 dict / defaultdict）：

账本	键	存什么
storage	[thread_id][checkpoint_ns][checkpoint_id]	序列化后的 checkpoint 本体 + metadata + 父 id
blobs	(thread_id, ns, channel, version)	各 State 字段（如 messages）的序列化值
writes	(thread_id, ns, checkpoint_id)	本拍各节点的 pending writes（容错用）

builder.compile(checkpointer=memory) 把该实例赋给 Pregel 的 self.checkpointer。之后：

图方法	实际打到 Saver
invoke / stream 调度前	get_tuple(config)
每个 superstep 结束	put(...)；节点完成时 put_writes(...)
get_state(config)	get_tuple → 包成 StateSnapshot
get_state_history(config)	list(config)，新→旧
update_state(config, values)	再 put 一条 source=update 的新 checkpoint

未传 checkpointer 时 invoke 仍能跑，但第 2 次不会看到第 1 次的 messages。

4.2 一次 invoke 的存取链路

cfg = {"configurable": {"thread_id": "user-42"}}。省略的 checkpoint_ns 运行时补 ""。

1. invoke(input, cfg) 取出 thread_id="user-42"。无 thread_id 且绑了 Saver → 报错（无法定位行）。
2. 读：memory.get_tuple(cfg)。无 checkpoint_id 时取 storage["user-42"][""] 里 id 最大 的一条（ULID 字典序即最新）；有 id 则精确命中。命中后用 serde.loads_typed 解 checkpoint，再按 channel_versions 从 blobs 拼回 channel_values。未命中返回 None，channels 用本次 input。
3. 节点 chat 返回 {"messages": [AIMessage(...)]}；add_messages 追加，不覆盖。
4. 写：put 把 channel_values 拆进 blobs，本体写入 storage[thread_id][ns][checkpoint["id"]]，并把 config 里旧的 checkpoint_id 记为 parent。返回的 config 带上新 id。put_writes 把本拍节点输出挂到同一三维键。
5. 到达 END，invoke 返回合并后的 state。
6. get_state(cfg) 再走第 2 步，不跑节点。

少了第 2 步 → 每次都从空 messages 起。
第 2 步换了 thread_id → storage 另一棵空树，看起来像「忘了」。
第 4 步只改内存、进程退出 → 三本账全没，这就是「仅开发可用」。

4.3 thread_id 与 ns、id 的分工

复合主键 (thread_id, checkpoint_ns, checkpoint_id)：

维	谁填	作用
thread_id	调用方	会话 / 租户隔离
checkpoint_ns	默认 ""；子图由框架改写	同一会话内主图 vs 子图
checkpoint_id	Saver 每次 put 生成	该抽屉里的历史版本

多租户只换 thread_id（如 org:user）。get_tuple 先用 thread_id 进 storage 第一层，再进 ns。

---

5. 最小可运行示例

from typing import Annotated, TypedDict

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages


class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]


def chat(state: State) -> dict:
    n = len(state["messages"])
    return {"messages": [AIMessage(content=f"reply#{n}")]}  # 拓扑 §2 chat


builder = StateGraph(State)
builder.add_node("chat", chat)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)

memory = InMemorySaver()
graph = builder.compile(checkpointer=memory)

cfg = {"configurable": {"thread_id": "user-42"}}
graph.invoke({"messages": [HumanMessage(content="r1")]}, cfg)
out = graph.invoke({"messages": [HumanMessage(content="r2")]}, cfg)
print(len(out["messages"]))  # 4

snap = graph.get_state(cfg)
print(snap.values["messages"][-1].content)  # reply#3
print(snap.config["configurable"]["thread_id"])  # user-42

# 换 thread = 另一棵 storage 树
other = graph.invoke(
    {"messages": [HumanMessage(content="r1")]},
    {"configurable": {"thread_id": "user-99"}},
)
print(len(other["messages"]))  # 2

# 直接看 Saver（调试用；业务代码走 graph.get_state）
tup = memory.get_tuple(cfg)
print(tup is not None)

---

6. 执行追踪

调用	get_tuple	put 后 messages	最后 AI
第1次 invoke user-42	None	2	reply#1
第2次 invoke user-42	上一拍 2 条	4	reply#3
get_state(user-42)	最新 4 条	不写	reply#3
invoke user-99	None	2	reply#1

线性图 START → chat → END 一次 invoke 会落多枚 checkpoint（input / 节点后）；get_state 只看最新。

---

重要配置参数

参数（API 名）	类型 / 默认值	功能说明	作用与影响	参考起点 / 常用范围	配置指导
compile(checkpointer=...)	BaseCheckpointSaver / None	把 Saver 挂到 Pregel	不传则无跨 invoke 记忆	InMemorySaver()	生产换 Sqlite/Postgres
InMemorySaver()	进程内三本账	实现 get_tuple / put / list	重启即空	开发、单测	勿多 worker 共享
configurable.thread_id	str，有 Saver 时必填	storage 第一层键	换 ID = 新会话	UUID / org:user	多租户只改这一维
configurable.checkpoint_ns	str，默认 ""	storage 第二层	主图空串	子图自动	勿当租户键
configurable.checkpoint_id	str，可选	定点 get_tuple	不传取 id 最大	时间旅行	与 thread/ns 同时匹配
invoke(input, config)	先读后写	调度前 get_tuple，每拍 put	只传增量 messages	多轮	靠 add_messages
get_state(config)	→ StateSnapshot	再调 get_tuple，不跑节点	看 values / next	调试	读 .config 不是 .configurable
memory.get_tuple(config)	→ CheckpointTuple / None	Saver 底层读	图内部用这个	排障	业务优先 get_state

---

7. 易踩坑

1. compile 未传 checkpointer：invoke 能跑，第二次没有第一次的 messages。
2. 换 thread_id 却期待旧对话：get_tuple 走进另一棵空树。
3. 生产用 InMemorySaver：扩缩容 / 重启 / 多副本各持一份内存。
4. 业务代码翻 memory.storage：键和序列化格式会变；用 get_state / get_tuple。

---

小结

• 图只调用 Saver：读走 get_tuple，走 put / put_writes。
• thread_id 是第一层键；同一 ID 多次 invoke 自动续跑。
• InMemorySaver 三本账在进程内；上线换 Sqlite/Postgres，接口不变。

---

参考链接

• Checkpointers
• InMemorySaver