Checkpointer 与 Human-in-the-loop:持久化与人工介入
图的"存档点"与"暂停键":Checkpointer 按 thread 持久化状态,interrupt 实现审批流与断点续跑。
Checkpointer:给图加"存档系统"
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver() # 生产换 SqliteSaver/PostgresSaver
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user-42-thread-1"}}
app.invoke({"messages": [("user", "你好")]}, config)
- 每个thread_id 是一条独立会话,每步执行后的状态自动存档;
- 同一 thread 再次 invoke 自动带上历史 —— 多轮记忆不用手动管理;
- 换 Sqlite/Postgres 存储后,重启服务会话也不丢。
Human-in-the-loop:危险操作前踩刹车
app = graph.compile(
checkpointer=checkpointer,
interrupt_before=["execute_trade"]) # 进该节点前暂停
# 第一次运行:跑到节点前停下
app.invoke(input, config)
# 人工审核后放行(None 表示沿用存档状态继续)
app.invoke(None, config)
典型用途:执行下单/删除/发款等高危工具前暂停等审批;用户输入补充信息后继续;失败现场人工修复后续跑。这是黑盒 AgentExecutor 做不到的事。
Time Travel:时光回溯
因为每步都有存档,你可以取历史某个 checkpoint 的状态、修改后从那里重跑 —— 调试多智能体流程的利器(对应《Harness》课的可观察与恢复职责)。
📝 课后练习
quiz-1 LangGraph 中 interrupt_before + thread_id 组合实现的是?