RSS

第 9 章 / 共 14 章

人在回路:让流程停在人身上

约 10 分钟 更新于

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章:人在回路:让流程停在人身上
图 9.1:中断不是异常,是一次有记录的暂停。注意中间那两步——落盘和唤醒之间可以隔任意长时间,进程死掉也不影响,因为「原地」被存在了检查点里,不在内存里。

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(),调试用静态断点。

9.6 练习与检查点