持久化与 Checkpoint
LangGraph 内置了持久化层（Persistence Layer），在每次执行步骤时自动保存图状态的快照，称为 Checkpoint（检查点）。这是实现人机协同、对话记忆、时间旅行调试和容错执行的基石。
为什么需要持久化
持久化层带来以下关键能力：
| 能力 | 说明 |
|---|---|
| 人机协同（HITL） | 中断执行、人工审查和修改状态后恢复 |
| 对话记忆 | 跨多次调用保持上下文 |
| 时间旅行 | 回放历史执行步骤进行调试 |
| 容错 | 节点失败后从最近的成功 checkpoint 恢复 |
基本用法
添加 Checkpointer
python
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.runnables import RunnableConfig
# 1. 定义图
builder = StateGraph(State)
builder.add_node("process", process_node)
builder.add_edge(START, "process")
builder.add_edge("process", END)
# 2. 创建 checkpointer
checkpointer = InMemorySaver()
# 3. 编译时传入 checkpointer
graph = builder.compile(checkpointer=checkpointer)
# 4. 执行时必须指定 thread_id
config = {"configurable": {"thread_id": "my_thread_1"}}
result = graph.invoke({"input": "hello"}, config)Thread（线程）
Thread 是 checkpoint 的容器。每次执行时必须指定 thread_id：
python
# 同一个 thread_id 会继承之前的状态
config = {"configurable": {"thread_id": "conversation_1"}}
result1 = graph.invoke({"messages": [{"role": "user", "content": "你好"}]}, config)
result2 = graph.invoke({"messages": [{"role": "user", "content": "还记得刚刚的对话吗?"}]}, config)
# result2 能看到 result1 的消息历史Checkpoint 存储后端
1. InMemorySaver（内存，开发用）
python
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
# 进程重启后数据丢失2. SQLite（本地持久化）
python
from langgraph.checkpoint.sqlite import SqliteSaver
# 文件模式
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
# 内存模式
checkpointer = SqliteSaver.from_conn_string(":memory:")3. Postgres（生产环境）
python
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string(
"postgresql://user:pass@localhost:5432/langgraph"
)4. 自定义 Checkpointer
实现 BaseCheckpointSaver 接口：
python
from langgraph.checkpoint.base import BaseCheckpointSaver
class MyCheckpointer(BaseCheckpointSaver):
async def aget(self, config):
# 获取 checkpoint
pass
async def aput(self, config, checkpoint, metadata):
# 保存 checkpoint
pass
async def alist(self, config, *, before=None, limit=None):
# 列出历史 checkpoint
pass操作 State
获取当前状态
python
config = {"configurable": {"thread_id": "1"}}
current_state = graph.get_state(config)
print(current_state.values) # 当前 state 值
print(current_state.next) # 下一步要执行的节点获取历史状态
python
history = list(graph.get_state_history(config))
for snapshot in history:
print(f"Step {snapshot.metadata.get('step')}: {snapshot.values}")恢复历史状态
python
# 获取特定 checkpoint
config = {
"configurable": {
"thread_id": "1",
"checkpoint_id": "1ef663ba-28fe-6528-8002-5a559208592c"
}
}
state = graph.get_state(config)手动更新状态
python
# 在指定 checkpoint 分支，修改状态
graph.update_state(
config,
{"messages": [{"role": "user", "content": "修正后的输入"}]},
)Checkpoint 生命周期
每次图执行时，checkpoint 的创建流程：
新建线程 → 空 checkpoint
↓
用户输入 → 输入 checkpoint（source: "input"）
↓
Step 1 执行 → step 1 checkpoint（source: "loop"）
↓
Step 2 执行 → step 2 checkpoint（source: "loop"）
↓
... → 最终 checkpoint（next: ()）生产建议
- 开发用 InMemorySaver，集成测试用 SQLite，生产用 Postgres
- 设置合理的
checkpoint_id管理策略，避免存储膨胀 - 使用 Agent Server 部署时无需手动配置 checkpointer，由平台自动管理
- 定期清理过旧的 checkpoint 以节约存储