第 12 章 持久化：Checkpointer 与线程

本章目标：让图"记住"执行历史。掌握 Checkpointer 的挂载方式与线程（thread）模型——它是会话恢复、断点重跑、时间旅行、中断恢复这四大能力的共同地基。第 1 章立下的"状态不可恢复"痛点，本章兑现。

12.1 一场断电引发的需求

批次评估跑到第 17 份简历时，进程崩了。重启后你面对的问题是：30 份里前 16 份的结果找不回来——状态活在内存里，进程死了状态就没了。

更日常的版本：HR 上午启动了一轮评估，下午回来想继续这个会话；或者你调试时想让图从第 3 个节点重跑而不要从头再来。

这些需求的共同解法：在每个超步结束时，把状态快照存下来。这就是 Checkpointer 的工作——也是 LangGraph 区别于"玩具编排库"的分水岭特性。

12.2 挂载 Checkpointer

三步：选实现、编译时传入、运行时给线程 ID。

from typing import Annotated, TypedDict
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

class ChatState(TypedDict):
    messages: Annotated[list, operator.add]      # 演示用，正式场景用 add_messages

def echo(state: ChatState) -> dict:
    return {"messages": [f"收到：{state['messages'][-1]}"]}

builder = StateGraph(ChatState)
builder.add_node("echo", echo)
builder.add_edge(START, "echo")
builder.add_edge("echo", END)

# 关键 1：编译时挂载 checkpointer
memory = InMemorySaver()
graph = builder.compile(checkpointer=memory)

# 关键 2：运行时通过 config 指定线程
config = {"configurable": {"thread_id": "hr-session-001"}}

r1 = graph.invoke({"messages": ["你好"]}, config)
print(r1["messages"])       # ['你好', '收到：你好']

# 同一线程再次 invoke：新输入会追加到历史状态之上
r2 = graph.invoke({"messages": ["在吗"]}, config)
print(r2["messages"])       # ['你好', '收到：你好', '在吗', '收到：在吗']

第二次 invoke 只传了 ["在吗"]，但图看到的状态里带着第一次的完整历史——同一线程的连续 invoke，天然构成多轮会话。这就是对话记忆的底层机制（第 15 章在此基础上做裁剪）。

InMemorySaver 把检查点存在当前进程的字典里：零配置、适合开发调试，但进程一死全没。生产可换 SqliteSaver（单机文件）或 PostgresSaver（多实例共享），接口一致：

# from langgraph.checkpoint.sqlite import SqliteSaver
# with SqliteSaver.from_conn_string("checkpoints.db") as saver:
#     graph = builder.compile(checkpointer=saver)

实现一个 Checkpointer 只需满足基类协议（get/put/list 等），需要接入自建存储时可自行实现——第 13 章的 API 都建立在协议之上，与具体后端无关。

12.3 线程：检查点的命名空间

线程是理解持久化的钥匙，值得精确定义：线程 = 一次逻辑会话的容器，以 thread_id 标识，内部按超步顺序存放检查点。

三条设计规则决定了它的行为：

1. invoke 必须带 thread_id（挂了 checkpointer 时），否则不知道读写哪个容器。config = {"configurable": {"thread_id": ...}} 是固定写法。
2. 不同线程互相不可见。hr-session-002 看不到 001 的任何状态——两场独立会话。
3. 同一线程的多次 invoke 是"追加"语义：每次从最新检查点恢复，接着跑。

这组规则立刻派生出一个常见误区：线程不是并发安全的写入单位。两个请求带着同一个 thread_id 同时 invoke，后写者会基于同一快照推进，状态合并结果不可预期。规矩是：一个 thread_id 同一时刻只跑一个 invoke；要并发就开新线程。

线程 ID 怎么生成？按业务语义定：Web 应用里每场会话一个（session_id）、每个用户每个任务一个（user-42-task-7）。它的本质是"你想让哪些运行共享历史"的答案。

12.4 检查点的实用 API

挂上 checkpointer 后，编译产物多了一组检查点操作能力（全部通过 config 定位线程）：

# 当前快照：读线程最新状态
snap = graph.get_state(config)
print(snap.values)         # 当前状态值，如 {'messages': [...]}
print(snap.next)           # 下一步待执行的节点元组，如 ('echo',)——空元组表示已跑完

# 历史检查点列表：本线程的全部超步快照
for state in graph.get_state_history(config):
    print(state.config, state.next)
# 每条记录带自己的 checkpoint_id，可据此定位任意历史时刻（第 13 章时间旅行的入口）

两个高频用途：

• get_state 查进度：长任务跑着，轮询 snap.next 就知道进行到哪个节点（配合 (('evaluate',),) 这类值判断阶段）。
• get_state_history 做审计：每个超步的完整快照链，谁在什么时候改了什么，全有据可查。

12.5 恢复：从检查点继续跑

"断点续跑"有两个层次。

层次一：从最新检查点继续（默认行为）。进程崩溃重启后，拿同一个 thread_id 再 invoke——引擎自动从最新检查点恢复状态，接着往下跑。前面 12.2 的多轮对话就是这个机制。对"第 17 份崩了想从第 17 份继续"的场景，只要你的图是按超步推进的，恢复粒度就是超步级。

层次二：从任意历史检查点重跑。指定历史 checkpoint_id 作为起点：

history = list(graph.get_state_history(config))
target = history[-2]                        # 选一个历史时刻

# 以该检查点为新起点继续执行
for chunk in graph.stream(
    None,                                   # None：不注入新输入，用检查点原状态
    target.config,                          # 带历史 checkpoint_id 的 config
):
    print(chunk)

target.config 里携带了那个历史时刻的 checkpoint_id，stream/invoke 会从那里"复活"。这就是时间旅行的雏形——完整展开在第 13 章。

注意一个微妙点：从历史检查点继续跑，会产生新的检查点分支（历史不是被改写，而是长出新枝）。这保证任何时候回退都不会毁掉原有记录。

12.6 checkpointer 与子图：各记各的账

第 9 章埋的伏笔在此兑现：父图与子图各自的检查点如何组织？

规则很简洁：每个图（无论层级）一个检查点命名空间，子图的记录挂在自己这条线上。checkpointer 在编译时挂在父图上，子图作为节点运行时，引擎在父图检查点里记录"子图节点这一步"，子图内部各超步的细节记录在子图自己的命名空间里（通过 stream(subgraphs=True) 或带命名空间的 config 可访问）。

对日常开发的含义：父图状态是恢复的最小单位，子图中间态不是。模式二（黑盒子图）下恢复时，子图整个重跑——这再次印证第 9 章的设计纪律：父子边界放最终产物，别指望从子图中间续跑。

12.7 中断恢复预告

有一种"暂停"不是故障而是设计：高风险简历要等 HR 确认。图运行到审核节点前主动挂起，人确认后带着决定恢复执行。这件事不需要自己造轮子——interrupt 机制就是"官方的暂停/恢复"，底层正是本章的检查点：挂起时存检查点，恢复时从检查点续跑。第 14 章专门展开。

常见坑与排错

1. 挂了 checkpointer 却忘传 thread_id。报错类似 LangGraphError: Checkpointer requires thread_id。固定疗法：invoke/stream 的第二个参数永远带上 config。

2. InMemorySaver 上线生产。多进程部署下各进程各一份内存，检查点互相看不见；重启即失忆。InMemorySaver 只用于开发与测试，生产换 Sqlite/Postgres 或自研实现。

3. 同一线程并发 invoke。12.3 的重点，再钉一遍：同一 thread_id 的并发写入结果不可预期。Web 服务里"每个请求开新线程"或加会话级锁。

4. 把 thread_id 当全局变量。图 A 的线程和图 B 的线程即便同名也互不相干（线程挂在"图 + checkpointer"的组合上）。跨图共享数据用第 15 章的 Store，不要蹭线程。

5. 状态里塞不可序列化对象。检查点要序列化落盘，状态字段里放了数据库连接、函数对象、大自定义类实例，写入时炸。状态保持"纯数据"（dict/基础类型/Pydantic 模型），连接和句柄放节点局部变量。

6. reducer 不纯导致恢复后状态错乱。第 3 章的纪律在这里显出分量：恢复/重放会重新执行 reducer，如果 reducer 里改了入参或依赖外部状态，历史快照被污染，时间旅行（第 13 章）结果不可信。

7. 误以为 invoke 后线程就"结束"了。线程不随单次 invoke 生灭——它持续存在到存储被清空。下次同 thread_id 的 invoke 会接着上次的尾巴跑。想要"全新会话"，换新 thread_id。

动手练习

1. 把 12.2 的 demo 换成 add_messages + 真实 LLM，做三轮对话。中途用 get_state 查看 snap.next 与 messages 长度，体会"线程即会话"。

2. 三线程实验：同一份输入分别用 t-1、t-2、t-1 invoke 三次。打印每次结果，验证：t-2 不受 t-1 影响；t-1 的第三次 invoke 接着前两次的历史。

3. 用 get_state_history 找到第 12.5 demo 的第一个检查点，从那里重新 stream 一遍（输入传 None）。对比两次执行产生的检查点历史，验证"回退产生新分支而非改写历史"。

本章速查