第 11 篇做审批流时，interrupt 停得下来、Command(resume=...) 续得上去，靠的都是当时一笔带过的那个参数——compile(checkpointer=...)。这篇把它正面讲透：checkpointer 是 LangGraph"有状态"的全部基础设施，会话记忆、崩溃恢复、人机协同、时间旅行，全都长在它上面。
本文按"金鱼困境 → 最小接入 → thread_id 会话模型 → 生产后端切换 → 状态检视 → 时间旅行 → 内部结构"组织，代码基于 langgraph 1.2.x。全文用同一张单节点 chat 图做贯穿示例，后端从内存一路换到 Postgres。
1、Agent 的金鱼困境
不接 checkpointer 的 Agent 是一条金鱼，记忆只有一次 invoke 的长度。具体会在三个场景翻车：
- 进程退出：重启后历史清零，用户隔天问"接着昨天聊"，Agent 一脸茫然
- 多轮对话：默认每次
invoke都是独立运行，第二轮不带第一轮的历史 - 崩溃恢复：跑到一半 OOM 只能从头重跑，工具副作用可能重放
第三条最危险，展开说说：Agent 节点常带副作用——下单、发邮件、写库。没有 checkpoint，崩溃后重跑意味着副作用可能再来一遍；有了它，恢复时从最后一个 checkpoint 继续，已完成的节点不会重放。
第二条第 10 篇见过结论：多轮记忆不是 Agent 自带的，是 checkpointer 给的。第一条和第三条更隐蔽——传统 Web 应用里你会拿数据库存会话、拿日志做恢复，LangGraph 则把这件事做进了图运行时：每执行完一个超步（superstep），框架自动写一份 checkpoint。节点函数里不出现任何"保存"代码，存档时机由运行时统一掌控。
一份 checkpoint 是某一时刻的完整快照：State 全部通道的值、下一个要执行的节点、元信息。负责存取快照的组件叫 checkpointer，统一实现 BaseCheckpointSaver 接口，官方维护三档实现：内存、SQLite、Postgres。第 10 篇 create_agent 的 checkpointer 参数最终也传给底层 compile，两篇讲的是同一套机制。概念全景见 Persistence 文档。
2、Checkpointer 最小接入
贯穿全文的示例是一张单节点 chat 图：State 用内置的 MessagesState（自带 add_messages reducer），一个节点调模型。接入持久化只动两处——compile 多传一个参数，invoke 多传一份 config。完整代码如下：
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import MessagesState, StateGraph, START, END
model = init_chat_model("openai:gpt-5")
def chat(state: MessagesState):
return {"messages": [model.invoke(state["messages"])]}
builder = StateGraph(MessagesState)
builder.add_node("chat", chat)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "demo-1"}}
graph.invoke({"messages": [{"role": "user", "content": "你好"}]}, config)
解读：InMemorySaver 从 langgraph.checkpoint.memory 导入，langgraph 主包自带，不用额外安装。挂上之后，每次 invoke 都会先按 thread_id 加载该线程的既有状态，每个节点跑完再写回新快照。thread_id 是你随手起的字符串，一张图加一个 thread_id 唯一定位一份会话，它是下一节的主角。这也是接入的全部成本：图结构、节点逻辑一行不改，持久化是编译期的一次性决策。
还有一处写法变化值得点破：输入里只有一条新消息。历史由 checkpointer 托管，你只负责递增量——同样的写法放在不接 checkpointer 的图上跑，第二轮就会把第一轮丢掉。
3、thread_id 即会话
把第 2 节的行为钉死成规则：同一个 thread_id 连续 invoke，后一轮自动带上全部历史；换一个 thread_id，就是全新会话。继续用第 2 节的图：
config_a = {"configurable": {"thread_id": "user-42"}}
graph.invoke({"messages": [{"role": "user", "content": "我叫李雷，工号 888。"}]}, config_a)
r2 = graph.invoke({"messages": [{"role": "user", "content": "我的工号是多少？"}]}, config_a)
print(r2["messages"][-1].text)
config_b = {"configurable": {"thread_id": "user-43"}}
r3 = graph.invoke({"messages": [{"role": "user", "content": "我的工号是多少？"}]}, config_b)
print(r3["messages"][-1].text)
两次输出如下：
你的工号是 888。
抱歉，我这里没有你的工号信息，方便告诉我吗？
解读：第二轮输入只有一个问题，模型却答得出工号——上下文来自 checkpoint 里第一轮的两条消息。换成 user-43，同一张图、同一个模型，历史归零，只能反问。thread_id 就是会话的身份证。
这也是"多用户多会话"的钥匙。Web 服务里给每个会话分配一个 ID（用户 ID 拼会话 ID 即可），把 thread_id 存进业务库；用户切换会话就是切换 thread_id。图保持单例，会话靠 ID 隔离，不需要为每个用户克隆任何东西。
常见的手工替代是每次把全部历史拼进输入：也能跑，但截断、隔离、崩溃恢复全得自己管，等于重造一个残缺的 checkpointer。
4、生产后端切换
InMemorySaver 把 checkpoint 存在进程内存里，重启即失，只配出现在开发调试与单元测试。上生产前先换后端，装对应的包：
pip install langgraph-checkpoint-sqlite
pip install langgraph-checkpoint-postgres
SQLite 与 Postgres 的 Saver 不在 langgraph 主包里，各自拆成独立的 langgraph-checkpoint-* 包，导入路径也各回各家。三档后端一张表说清：
| 后端 | 导入路径 | 定位 |
|---|---|---|
InMemorySaver |
langgraph.checkpoint.memory |
开发调试、单测 |
SqliteSaver |
langgraph.checkpoint.sqlite |
单机演示、小工具 |
PostgresSaver |
langgraph.checkpoint.postgres |
生产、多进程并发 |
SQLite 版的用法有个反直觉的点：from_conn_string 返回的是上下文管理器而不是实例，得用 with 接：
from langgraph.checkpoint.sqlite import SqliteSaver
with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
checkpointer.setup() # 首次运行建表，之后可省
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user-42"}}
graph.invoke({"messages": [{"role": "user", "content": "我叫李雷"}]}, config)
解读：with 块退出时自动关连接，所以图的全部调用都得写在块内。嫌 with 碍事，可以拿现成连接直接构造：SqliteSaver(sqlite3.connect("checkpoints.db", check_same_thread=False))，连接生命周期自己管。重启进程后 checkpoints.db 还在，user-42 的会话原样恢复——第 3 节的记忆不再是内存里的幻影。
注意：
SqliteSaver是单连接实现，多线程并发写同一线程会出错，官方定位就是轻量同步场景。
Postgres 版面向生产，重点是连接管理。Web 服务不能一条连接打天下，用 psycopg_pool 连接池：
from psycopg_pool import ConnectionPool
from langgraph.checkpoint.postgres import PostgresSaver
DB_URI = "postgresql://app:pass@localhost:5432/agents"
pool = ConnectionPool(
DB_URI,
max_size=20,
kwargs={"autocommit": True, "prepare_threshold": 0}, # 官方要求的两项
)
checkpointer = PostgresSaver(pool)
checkpointer.setup() # 建表，幂等，启动时调一次
graph = builder.compile(checkpointer=checkpointer)
解读：psycopg_pool 随 langgraph-checkpoint-postgres 一起装上。两个 kwargs 是官方参考明确要求的：autocommit 走自动提交，prepare_threshold=0 关掉 prepared statement 缓存，避免池化连接被服务端重置后复用报错。多实例部署时，所有 worker 共享同一张 checkpoint 表，会话天然跨进程存活。进程退出前记得 pool.close()。
换后端只是换 compile 那一行：invoke、config、thread_id、后面要讲的 get_state 全部原样——这正是 BaseCheckpointSaver 接口的意义。细节见 Checkpointers 文档 与 Postgres Saver 参考。
5、状态检视与修改
接上 checkpointer 后，图对象多出一组"上帝视角" API，沿用第 3 节的 user-42 线程。get_state 拿当前快照：
snapshot = graph.get_state(config_a)
print(len(snapshot.values["messages"])) # 4：两轮对话四条消息
print(snapshot.next) # ()：图已跑完，无待执行节点
print(snapshot.config["configurable"]["checkpoint_id"]) # 最新快照的 id
返回的 StateSnapshot 常用字段五个：
| 字段 | 含义 |
|---|---|
values |
State 全部通道的当前值 |
next |
待执行节点元组，() 表示已跑完 |
config |
本快照的 config，含 checkpoint_id |
metadata |
步数、来源节点等元信息 |
tasks |
待执行任务细节，含中断节点的写入 |
next 最值得盯：跑完的线程是空元组；第 11 篇审批流 interrupt 之后，next 里就是等待批准的那个节点——判断线程卡在哪，看 next 一眼便知。
get_state_history 把线程全部快照从新到旧吐给你：
for snap in graph.get_state_history(config_a):
cid = snap.config["configurable"]["checkpoint_id"]
print(cid, len(snap.values["messages"]), snap.next)
输出（id 有删节）：
1ef4f7e2-2c1b-6f6c-8002-9d0a7b6c5e21 4 ()
1ef4f7e2-2c1b-6f6c-8001-4a8f3d2c1b09 3 ('chat',)
1ef4f7e2-2c1b-6f6c-8000-7e5a9c4d2f88 2 ()
1ef4f7e1-9a3d-6f6c-8003-1b6c8e5f7a44 1 ('chat',)
解读：两轮对话留下四个 checkpoint，一个超步一个——invoke 写入输入是一个，chat 跑完是一个。中间那两个 ('chat',) 快照就是"历史时点"：状态齐了、节点排队中。这张清单就是下一节时间旅行的车站表。排障时这两个 API 配合成对：先 get_state 看 next 停在哪，再翻 get_state_history 看从哪一轮开始跑偏，全程不用加日志重启进程。
状态还能直接改。update_state 不跑图，往线程里注入内容：
from langchain_core.messages import SystemMessage
graph.update_state(
config_a,
{"messages": [SystemMessage(content="从现在起，回答控制在两句话以内。")]},
as_node="chat",
)
解读：更新立即按 reducer 生效——这条 System 消息此刻已进入 user-42 的历史，下一轮 invoke 模型自然看得到。as_node 声明这次更新"算谁写的"，影响图后续从哪继续；单纯注入历史按示例写即可，改完用 get_state 复核。
6、时间旅行
第 5 节清单里 next 非空的快照，就是可以"回到"的时点。从历史 checkpoint 恢复重跑，官方分两种：replay 原样重跑，fork 先改状态再跑。先看 replay：
history = list(graph.get_state_history(config_a))
to_replay = history[-1] # 最早的一个：第一轮输入刚落库、模型还没回话
result = graph.invoke(None, to_replay.config)
print(result["messages"][-1].text)
解读：to_replay.config 自带那个时点的 checkpoint_id；invoke 输入传 None，意为"不给新输入，从该快照继续执行排队中的节点"。于是 chat 重跑第一轮，产出一条全新回复，作为新 checkpoint 写回同一线程，旧快照原封不动。
时间旅行永远在"长新分支"，不覆盖既有历史——checkpoint 只追加，不修改。
另一个高频用法是复现线上问题：把出问题的线程历史拉到本地，从出错前的 checkpoint 反复重放，稳定复现再动手修——比对着日志猜要硬得多。
fork 是加强版：先用 update_state 改旧快照的状态（换提示词、改参数、注入消息），再 invoke(None, 该快照的 config) 续跑，第 11 篇的 Command(resume=...) 内部走的正是这条路。调试价值就在这里："如果那步换个参数会怎样"从思想实验变成可反复执行的对照实验——挑时点、改状态、重跑、对比两条分支。LangGraph Studio 的重放按钮、LangSmith 的时间旅行视图，底层就是这组 API，见时间旅行文档。
7、checkpoint 内部结构
最后掀盖看一眼。每个 checkpoint 落库存四样东西：
- State 值：全部通道的快照，消息历史就在其中，不单独存对话
- 待执行任务：排队的节点与已产生的部分写入
- 元信息：
checkpoint_id、父 checkpoint、步数、时间戳 - 序列化字节：以上内容按序列化器编码后的形态
两个实用推论。其一，State 里的对象必须可序列化：默认的 JsonPlusSerializer 认得消息与 Pydantic 模型，自定义类要么落成基本类型、要么自己处理序列化，否则要到运行时才报错。其二，checkpoint_id 内嵌时间戳、可排序，线程的"当前状态"永远取最大的 id——第 6 节重跑后新分支成为"当前"，原因就在这里。
这套结构也预告了后文：第 13 篇的流式输出，事件节奏与节点写 checkpoint 的节奏一一对应；第 23 篇的长期记忆走 Store 跨线程共享，与线程内的 checkpoint 互补。把 checkpoint 想成"每个超步一张存档加一份待办"，后面的机制都能顺着推出来。
8、总结
回收全文：checkpointer 接入只是 compile 一个参数加 invoke 一份 config；thread_id 是会话身份证，同号续聊、换号新生；后端三档按场景选，生产认 Postgres 连接池；get_state、get_state_history、update_state 是检视与修改三件套；时间旅行拿 checkpoint_id 回到任意时点，replay 重跑、fork 改造。第 11 篇审批流的"停与续"，到这不再是魔法。
留一个练习：把第 11 篇审批示例的 InMemorySaver 换成 SqliteSaver，跑出一次 interrupt 后直接杀掉进程；重启进程，用 get_state 确认线程仍停在审批节点（next 非空），再 Command(resume="通过") 续跑——会话在进程死亡后无损恢复，才算真正理解了持久化。加分项：用 get_state_history 数一数一轮带工具调用的对话写出几个 checkpoint，对齐"一个超步一份"的节奏。
对于要上生产的 Agent，建议第一天就接 Postgres 后端：会话即资产，内存后端等于裸奔。下一篇讲流式输出，让 Agent 的思考与工具调用实时吐到前端。
