第 9 章 / 共 14 章
人在回路:让流程停在人身上
9.1 停下来这件事,难在哪
前面几章已经把地基铺好了:图有状态(第 5、6 章),状态能落盘(第 8 章)。现在只差最后一块——在节点中间停住,把问题抛给外面的人,等答案回来再从原地继续。
难点在于「从原地继续」。传统的做法是把节点拆成两半:前半段跑完写个状态叫「等待审批」,后半段做成另一个入口。流程越复杂,这种拆分越难维护。
LangGraph 的做法是让你在节点函数中间直接停:
from langgraph.types import interrupt
def review(state):
decision = interrupt({"draft": state["draft"], "question": "这笔退款批准吗?"})
return {"approved": decision == "yes"}
interrupt() 这一行会让整个图停下来,把它的参数抛到外面;恢复时,这一行会返回你传进来的答案,函数从这里往下继续。
前提是必须有 checkpointer——没有落盘就没有「原地」可言。
9.2 完整的一次停与续(本机实测)
第一次运行:
config = {"configurable": {"thread_id": "ticket-4821"}}
for chunk in graph.stream(initial_state, config, stream_mode="updates"):
print(chunk)
真实输出:
{'write': {'steps': ['write'], 'draft': '第一版草稿'}}
{'__interrupt__': (Interrupt(value={'draft': '第一版草稿', 'question': '通过吗?'},
id='d92d1fd6e6764ce12ce4fdd310c268b7'),)}
注意看第二行的键是 __interrupt__。这是你在前端判断「要不要弹审批框」的信号。
此时查状态:
next: ('review',)
interrupts: (Interrupt(value={...}, id='d92d1fd6...'),)
next 停在 review,说明这个节点还没跑完。进程现在可以关掉。
明天恢复:
from langgraph.types import Command
for chunk in graph.stream(Command(resume="yes"), config, stream_mode="updates"):
print(chunk)
{'review': {'steps': ['review'], 'approved': True}}
{'publish': {'steps': ['publish']}}
最终状态:
{'steps': ['write', 'review', 'publish'], 'draft': '第一版草稿', 'approved': True}

9.3 用中间件做工具级审批
上面是自己画图时的写法。如果你还在用 create_agent,第 4 章那个 HumanInTheLoopMiddleware 能做到同样的事,粒度是工具:
agent = create_agent(
model=model, tools=[execute_refund],
middleware=[HumanInTheLoopMiddleware(interrupt_on={"execute_refund": True})],
checkpointer=InMemorySaver(),
)
它抛出的中断结构(本机实测):
Interrupt(value={
'action_requests': [{'name': 'execute_refund',
'args': {'order_id': 'ORD-2026-000123', 'amount': 89.0},
'description': "Tool execution requires approval\n\nTool: execute_refund\nArgs: {...}"}],
'review_configs': [{'action_name': 'execute_refund',
'allowed_decisions': ['approve', 'edit', 'reject', 'respond']}]
}, id='8c48bc54b88ac2938530126412dd1959')
恢复时的载荷形状和上一节不一样,必须是:
Command(resume={"decisions": [{"type": "approve"}]})
形状写错的真实报错(我第一次就写错了):
File ".../langchain/agents/middleware/human_in_the_loop.py", line 456, in after_model
decisions = interrupt(hitl_request)["decisions"]
TypeError: list indices must be integers or slices, not str
During task with name 'HumanInTheLoopMiddleware.after_model'
记住这条:interrupt() 抛什么、resume 收什么,是写抛出那一方定的契约。自己写的 interrupt 想收什么就传什么;用中间件时必须按它的形状来。
9.4 四种决策,四种结果
allowed_decisions 里的四个值分别做什么(前两个为本机实测):
| 决策 | 实测消息序列 | 工具真的执行了吗 |
|---|---|---|
{"type": "approve"} | Human → AI(tool_calls) → Tool 'deleted note-a.md' → AI '已删除。' | 是(自定义的 wrap_tool_call 中间件被触发) |
{"type": "reject", "message": "这条不能删"} | Human → AI(tool_calls) → Tool 'User rejected the tool call for ...with reason: ...' → AI | 否(wrap_tool_call 完全没被调用) |
{"type": "edit", ...} | 用修改后的参数执行 | 是(参数被改过) |
{"type": "respond", ...} | 用人写的内容当作工具结果 | 否 |
这个设计很值得琢磨:在 agent 里,「失败」的标准表达方式是一条消息,不是一个异常。第 4 章的错误处理、第 11 章的重试,用的都是同一个思路。
9.5 静态断点
除了在代码里写 interrupt(),还可以在编译时指定「在某个节点前后停一下」:
graph = builder.compile(checkpointer=saver, interrupt_before=["execute_refund"])
它适合调试(跑到这里停下让我看看状态),不适合做业务审批——因为它停的位置是节点边界,拿不到「要审批什么」的结构化信息,前端没法据此渲染一个审批界面。业务审批用 interrupt(),调试用静态断点。