持久化与检查点

前面的例子里，图一执行完状态就没了。加上 Checkpointer（检查点）后，每一步的状态都会被保存下来——这是实现多轮记忆、断点续跑、人机交互、状态回溯的基础。

一、Checkpoint 是什么

Checkpoint = 图执行每一步之后的状态快照。LangGraph 在每个节点执行完，自动把当时的完整状态序列化存起来。

flowchart LR
    N1[节点1] -->|存快照1| CP[(Checkpointer)]
    CP --> N2[节点2]
    N2 -->|存快照2| CP
    CP --> N3[节点3]
    N3 -->|存快照3| CP

为什么需要它：

场景	没有 Checkpointer	有 Checkpointer
多轮记忆	每次 invoke 都是全新开始，AI 失忆	同 thread_id 自动加载历史，AI 记得上下文
断点续跑	进程崩了就得从头跑	用相同 thread_id 重启，从断点继续
人机交互 HITL	无法暂停等人工操作	暂停在指定节点，人工处理后再继续
状态回溯	出错了只能重来	回滚到任意历史快照重跑

二、Checkpointer 接口与几种实现

LangGraph 提供几种开箱即用的 Checkpointer：

实现	存储	适用	安装
MemorySaver	进程内存	开发/测试，重启即丢	内置 langgraph
SqliteSaver	本地 SQLite 文件	单机持久化、小项目	pip install langgraph-checkpoint-sqlite
PostgresSaver	PostgreSQL	生产环境、多实例共享	pip install langgraph-checkpoint-postgres

安装（国内源）：

pip install langgraph-checkpoint-sqlite -i https://pypi.tuna.tsinghua.edu.cn/simple

Windows 用户直接在 PowerShell 执行。macOS/Linux 同命令。

三、compile 时传入 checkpointer

把 checkpointer 传给 compile，图就具备了持久化能力：

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END

# ...定义状态、节点、边...
graph = StateGraph(State)
# graph.add_node(...)

# 关键一步：compile 传入 checkpointer
app = graph.compile(checkpointer=MemorySaver())

之后每次 invoke 时传 config={"configurable": {"thread_id": "xxx"}}，框架就会按这个 id 保存/恢复状态。

四、thread_id：会话的身份证

一个 checkpointer 可以同时服务多个会话，靠 thread_id 区分。同一个 thread_id 的多次调用，状态是连续的；不同 thread_id 互相隔离。

app = graph.compile(checkpointer=MemorySaver())

# 会话 A
cfg_a = {"configurable": {"thread_id": "user-A"}}
app.invoke({"user_input": "我叫阿宝"}, config=cfg_a)
app.invoke({"user_input": "我叫什么"}, config=cfg_a)  # 记得"阿宝"

# 会话 B（独立，不记得 A 的内容）
cfg_b = {"configurable": {"thread_id": "user-B"}}
app.invoke({"user_input": "我叫什么"}, config=cfg_b)  # 不记得

thread_id 是你自定义的字符串，比如用户 ID、会话 ID。同一会话从头到尾用同一个。

五、get_state：查看当前状态

随时可以用 get_state 看某个会话现在停在哪、状态是什么：

state = app.get_state(cfg_a)
print(state.values)        # 当前状态字典
print(state.next)          # 接下来要执行的节点（tuple）
print(state.config)        # 这个状态的 config

state.next 在人机交互场景特别有用——它告诉你图"暂停在"哪个节点等着继续。

六、状态历史与回溯

get_state_history 返回这个会话的所有检查点，从新到旧。可以拿到任意一个历史点，然后用它作为"起点"继续跑，实现回滚。

for snap in app.get_state_history(cfg_a):
    print(snap.config["configurable"]["checkpoint_id"],
          "next:", snap.next,
          "values keys:", list(snap.values.keys()))

回溯到某个历史点继续：

# 找到想回到的那个 checkpoint
history = list(app.get_state_history(cfg_a))
target = history[3]  # 比如第 4 个检查点

# 用它的 config 继续运行（会把状态恢复到那一刻，再往后跑）
app.invoke(None, config=target.config)

传 None 表示"不输入新内容，从当前状态继续"。详见 中断与恢复。

七、完整对比：无 checkpointer vs 有 checkpointer

无 checkpointer：每次都是新的

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import HumanMessage

def echo(state):
    last = state["messages"][-1].content
    return {"messages": [{"role": "assistant", "content": f"你说：{last}"}]}

g = StateGraph(MessagesState)
g.add_node("echo", echo)
g.add_edge(START, "echo")
g.add_edge("echo", END)
app_no_cp = g.compile()  # 没传 checkpointer

# 第二次调用不记得第一次
r1 = app_no_cp.invoke({"messages": [HumanMessage(content="我叫阿宝")]})
r2 = app_no_cp.invoke({"messages": [HumanMessage(content="我叫什么")]})
print(r2["messages"][-1].content)  # 你说：我叫什么（不记得阿宝）

有 checkpointer：跨调用记忆

from langgraph.checkpoint.memory import MemorySaver

app_cp = g.compile(checkpointer=MemorySaver())
cfg = {"configurable": {"thread_id": "t1"}}

r1 = app_cp.invoke({"messages": [HumanMessage(content="我叫阿宝")]}, config=cfg)
r2 = app_cp.invoke({"messages": [HumanMessage(content="我叫什么")]}, config=cfg)

# 看历史
print(len(r2["messages"]))  # 4 条：u1, ai1, u2, ai2 —— 历史被保留了

第二次 invoke 时，框架先从检查点加载了第一次的 messages（u1, ai1），再追加 u2 执行，所以 r2["messages"] 有 4 条。这就是记忆。

八、用 SqliteSaver 持久化到文件

MemorySaver 进程退出就没了。要持久到磁盘，用 SqliteSaver：

from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

# Windows/Mac/Linux 都用本地文件路径
conn = sqlite3.connect("checkpoints.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)

app = graph.compile(checkpointer=checkpointer)

下次进程重启，用同一个 thread_id 调用，状态还在。生产环境换 PostgresSaver 即可，API 完全一致。

九、常见踩坑

踩坑 1：MemorySaver 重启丢失

开发用 MemorySaver 很方便，但部署到生产没换持久化实现，一重启所有会话失忆。生产必须用 SqliteSaver/PostgresSaver。

踩坑 2：thread_id 管理混乱

• 不同用户用同一个 thread_id → 串话，A 看到 B 的历史。
• 同一会话中途换了 thread_id → 状态从头开始。

建议 thread_id 用 <用户ID>-<会话ID> 格式，全局唯一。

踩坑 3：状态里塞了不可序列化对象

Checkpointer 要把状态序列化存起来。塞了文件句柄、连接、lambda 等会报序列化错误。详见 定义与组织状态 的踩坑一节。

踩坑 4：以为传了 checkpointer 就自动记忆

还要记得每次 invoke 传相同的 thread_id。光 compile(checkpointer=...) 但 invoke 不带 config，仍然每次新会话。

踩坑 5：SqliteSaver 多线程报错

SQLite 默认不允许跨线程使用连接。要么 check_same_thread=False，要么每个线程建自己的连接。生产用 Postgres 没这问题。

十、小结

• Checkpoint = 每步状态快照，是多轮记忆、断点续跑、HITL、回溯的基础。
• 三种实现：MemorySaver（开发）、SqliteSaver（单机持久化）、PostgresSaver（生产）。
• compile(checkpointer=...) 开启，thread_id 区分会话。
• get_state 看当前状态，get_state_history 看历史、做回溯。
• 生产必须用持久化实现，且 thread_id 要全局唯一管理。

下一篇 人机交互 HITL 讲怎么在执行中间暂停等人工操作。