让 Agent 干活不难,难的是在关键步骤让人类停下来看一眼。LangGraph 把这件事实做成了第一等公民:它每一步都读写一个 checkpoint,把图的执行状态持久化下来。这个持久化层不只是为了容错,更直接支撑了人工介入——图可以无限期暂停,人可以审、可以改,然后再从原地恢复。这篇讲清 checkpoint 与 interrupt 怎么配合。
持久化是基础
LangGraph 的执行引擎每一步都从 checkpoint 读取、再写回图状态。图状态存着 Agent 干活需要的所有东西,所以执行到一半可以暂停、隔很久再恢复,因为 checkpoint 就在那里,从原地继续就行。它甚至可以暂停、让人编辑 checkpoint、再从编辑后的状态继续。你可以把这个持久化层理解成人和 Agent 协作的一块草稿板。实现上,checkpoint 记录了序列化的通道值(默认 MsgPack、可选加密)、版本号、以及每个节点最近见过的通道版本,这让它能在任意机器上恢复,不依赖某个进程还活着。
interrupt:像 input 但能上生产
在 Python 里做人工介入,最直觉的方式是 input(),程序暂停、终端弹个框、输入作为返回值。问题是它同步阻塞、只在命令行和 notebook 里能用。LangGraph 提供了一个对等物 interrupt(),用法几乎一样,但能在生产里跑。它会暂停图执行、把线程标记为 interrupted、把传给 interrupt 的内容放进持久化层。恢复时用 Command 把回答传回去:
from langgraph.graph import StateGraph, MessagesState
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command, interrupt
def approval_node(state: MessagesState):
# 暂停并把问题抛给调用方;恢复时返回值就是这里的 ok
ok = interrupt({"question": "是否批准这次写操作?"})
return {"approved": bool(ok)}
g = StateGraph(MessagesState)
g.add_node("approval", approval_node)
g.set_entry_point("approval")
graph = g.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "task-42"}}
graph.invoke({"messages": []}, config=config) # 命中 interrupt,暂停
graph.invoke(Command(resume={"approved": True}), config=config) # 恢复
这里 thread_id 就是持久化游标:复用同一个值,恢复的是同一个 checkpoint;换一个新值,就开一个空状态的新线程。注意 interrupt 和 input 有一点关键差异——它不保存进程,恢复时会重跑当前节点里 interrupt 之前的部分,但不会重跑之前的节点。这样被中断的线程除了占点存储,不占任何计算资源,可以过几个月、在另一台机器上再恢复。
两种暂停方式
动态中断在节点内部、根据当前状态用 interrupt() 暂停,可以放在代码任意位置、按业务逻辑条件触发。静态中断用 interrupt_before / interrupt_after 在预定义节点前后暂停,适合「到这个节点前一定要拦下来」的场景:
graph = g.compile(
checkpointer=MemorySaver(),
interrupt_before=["approval"], # 执行 approval 前先停
)
动态更灵活,静态更省心,两个可以混用。
四种常见模式
官方把人工介入归纳成四种。批准或拒绝:在关键步骤(如 API 调用)前暂停,审查后批准或拒绝,拒绝就阻止执行并可能走替代路径。审查编辑状态:暂停让人类审或改图状态,纠正错误或补充信息。审查工具调用:在工具执行前审查并编辑 LLM 请求的工具调用,对涉及敏感或需人工监督的工具尤其关键。多轮对话:在多个 agent 之间,某个 agent 需要和人类来回多轮收集信息或反馈。
常见误区
第一个误区是拿 interrupt 当 input 用,以为它会同步阻塞进程——它实际是暂停 + 恢复,要用事件流(stream_events 的 stream.interrupts / stream.interrupted)或 invoke 结果的 interrupt 字段去感知暂停,而不是卡在一个函数调用上等。第二个误区是生产环境用内存 checkpointer(MemorySaver),进程一重启 checkpoint 就没了,生产应该用持久化 checkpointer(比如 Postgres 后端)。第三个误区是忽略 thread_id,它既是恢复指针也是隔离边界,多人多任务要各用各的 id。第四个误区是把所有人工介入都做成静态 breakpoint 而不用 interrupt,导致能暂停的点只有固定的几处,业务一旦要「按状态决定停不停」就捉襟见肘。最后一个误区是以为人工介入免费——中断的线程不占计算资源,但 checkpoint 是占存储的,长期堆积要清理。