28. LangGraph 中的持久化与 Checkpointer 是什么？

难度 P1 高频 · 岗位 应用 · 频率 ★★★ · 预计阅读 5 min

本题阅读地图

1. 💡 简要回答
2. 📝 详细解析
3. 🎯 面试总结

👔面试官：LangGraph 中的持久化与 Checkpointer 是什么？

🙋‍♂️我：就是保存状态，可以恢复。

👔面试官：对，但还有更丰富的能力：支持断点续跑、人机协同、时间旅行（回溯到某状态）。你能讲清楚 checkpointer 的工作机制和适用场景吗？

🙋‍♂️我：人机协同就是暂停等用户输入？

👔面试官：对，还有：每个 step 后自动保存、可以查看历史轨迹、出错后从检查点恢复。这些能力让 LangGraph 适合生产级 Agent。

💡 简要回答

Checkpointer 是 LangGraph 的状态持久化机制。

核心能力：

1. 自动保存：每个 step 后自动 checkpoint
2. 断点续跑：中断后从最后 checkpoint 恢复
3. 人机协同：在特定节点暂停，等待人工输入
4. 时间旅行：回溯到任意历史状态重放

适用场景：

• 长任务需要容错
• 需要人工审核/干预
• 需要调试和回溯

📝 详细解析

Checkpointer 机制

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
workflow = StateGraph(State)
# ... 定义节点和边
app = workflow.compile(checkpointer=checkpointer)

# 运行，自动 checkpoint
result = app.invoke(initial_state, config={"thread_id": "conversation_123"})

自动保存：

• 每个 Node 执行后，State 自动保存
• 用 thread_id 区分不同会话

断点续跑

# 中断后，用相同 thread_id 恢复
result = app.invoke(None, config={"thread_id": "conversation_123"})
# 自动从最后 checkpoint 继续

人机协同

# 在特定节点设置断点
workflow.add_node("human_review", human_review_node, interrupt_before=True)

# 运行到这里会暂停，等待人工输入
result = app.invoke(state, config={"thread_id": "xxx"})
# 暂停，人工通过接口更新 state
# 然后继续
result = app.invoke(None, config={"thread_id": "xxx"})

时间旅行

# 查看历史
history = list(app.get_state_history(config))

# 回溯到某个 checkpoint
old_state = app.get_state(config, checkpoint_id="checkpoint_5")

# 从这个状态重新运行
result = app.invoke(old_state.values, config)

🎯 面试总结

Checkpointer 实现自动保存、断点续跑、人机协同、时间旅行。让 LangGraph 适合生产级复杂 Agent。

---

章节首页 · ← Q27 · Q29 →