文章目录第 6 章Human-in-the-Loop 人工干预6.1 本章目标6.2 核心概念什么是 Human-in-the-Loop中断-恢复时序图interrupt() vs interrupt_before/after6.3 实战实战 1敏感操作审批流程实战 2AI 输出人工审核interrupt_after实战 3动态修改状态6.4 API 速查6.5 错误与避坑指南坑 1在 interrupt() 外用 try/except 包裹坑 2interrupt() 返回复杂对象坑 3中断前副作用不可幂等坑 4忘记配置 checkpointer6.6 最佳实践总结第 6 章Human-in-the-Loop 人工干预6.1 本章目标学完本章你将能够理解 Human-in-the-Loop 的设计理念和典型应用场景掌握interrupt()动态中断和interrupt_before/after静态中断学会Command(resume)恢复执行实现审批流程和状态编辑6.2 核心概念什么是 Human-in-the-LoopHuman-in-the-Loop人机协同是指 AI 在执行过程中暂停等待人类确认、修改或拒绝后再继续执行。典型场景包括审批流程敏感操作如退款、删除数据需要人工确认内容审核AI 生成的内容需要人工检查纠错干预AI 推理出错时人类介入修正数据标注AI 不确定时请人类帮助判断中断-恢复时序图人类审核者节点函数LangGraph应用程序人类审核者节点函数LangGraph应用程序invoke(input, config)执行节点执行业务逻辑interrupt(请确认此操作)抛出 GraphInterrupt 异常检测到中断展示中断信息等待决策确认/拒绝/修改invoke(Command(resume批准), config)从开头重新执行节点interrupt() 返回 批准返回结果返回最终结果interrupt() vs interrupt_before/after方式设置时机粒度使用场景interrupt()运行时节点内节点内特定位置需要动态决定是否中断、展示不同信息interrupt_before编译时compile 参数整个节点前固定的审批点如付款前必须审批interrupt_after编译时compile 参数整个节点后审核节点输出如检查 AI 生成的内容6.3 实战实战 1敏感操作审批流程fromtypingimportTypedDict,Annotatedfromlanggraph.graphimportStateGraph,START,END,add_messagesfromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.typesimportinterrupt,Commandfromlangchain_core.messagesimportBaseMessage,HumanMessage,AIMessage# # 定义 State# classState(TypedDict):messages:Annotated[list[BaseMessage],add_messages]action:str# 待审批的操作approved:bool# 是否已批准# # 定义节点# defrequest_refund(state:State)-dict: 退款请求节点提交退款申请然后暂停等待审批。 注意interrupt() 之后的代码只有在恢复后才会执行。 print( [退款节点] 收到退款请求)# 中断并等待人类审批# interrupt() 的参数会返回给调用方作为中断信息decisioninterrupt({question:确认退款 ¥999 给用户吗,options:[批准,拒绝],order_id:ORDER-12345,})print(f [退款节点] 审批结果:{decision})ifdecision批准:# 执行退款逻辑return{approved:True,messages:[AIMessage(content退款已处理¥999 将退回原支付账户。)],}else:return{approved:False,messages:[AIMessage(content退款已拒绝。)],}# # 构建图# checkpointerMemorySaver()builderStateGraph(State)builder.add_node(refund,request_refund)builder.add_edge(START,refund)builder.add_edge(refund,END)graphbuilder.compile(checkpointercheckpointer)# # 模拟审批流程# config{configurable:{thread_id:refund-001}}print( 第1步发起退款请求 )try:resultgraph.invoke({messages:[HumanMessage(content我要退款)],action:refund,approved:False},configconfig,)exceptExceptionase:print(f ⏸️ 图已暂停等待审批:{e})# 获取中断信息snapshotgraph.get_state(config)# 中断信息通常通过 stream_events 获取这里用 invoke 直接抛异常print(\n 第2步人类审批批准 )resultgraph.invoke(Command(resume批准),# 恢复执行传入审批结果configconfig,)print(f 审批结果: approved{result[approved]})print(f AI 回复:{result[messages][-1].content})实战 2AI 输出人工审核interrupt_afterfromtypingimportTypedDict,Annotatedfromlanggraph.graphimportStateGraph,START,END,add_messagesfromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.typesimportCommandfromlangchain_core.messagesimportBaseMessage,HumanMessage,AIMessageclassState(TypedDict):messages:Annotated[list[BaseMessage],add_messages]defgenerate_response(state:State)-dict:AI 生成回复# 模拟 AI 生成return{messages:[AIMessage(content尊敬的客户根据我们的记录您的订单已发货。预计 3-5 个工作日内送达。如有疑问请联系客服400-123-4567。)]}deffinalize(state:State)-dict:最终处理节点return{messages:[AIMessage(content[已审核通过] state[messages][-1].content)]}checkpointerMemorySaver()builderStateGraph(State)builder.add_node(generate,generate_response)builder.add_node(finalize,finalize)builder.add_edge(START,generate)builder.add_edge(generate,finalize)builder.add_edge(finalize,END)# 关键在 generate 节点后中断让人类审核 AI 输出graphbuilder.compile(checkpointercheckpointer,interrupt_after[generate],# 在 generate 节点执行后暂停)config{configurable:{thread_id:review-001}}print( 第1步AI 生成回复 )# 图会在 generate 执行后暂停try:resultgraph.invoke({messages:[HumanMessage(content我的订单状态是什么)]},configconfig,)exceptException:print( ⏸️ 图已暂停等待人工审核)# 查看 AI 生成的内容snapshotgraph.get_state(config)print(f AI 生成的内容:{snapshot.values[messages][-1].content})print(\n 第2步人工审核通过 )# 直接恢复执行不需要修改resultgraph.invoke(None,configconfig)print(f 最终输出:{result[messages][-1].content})实战 3动态修改状态# 在上面的场景中如果人类审核者想修改 AI 的输出print(\n 场景人工修改 AI 输出 )# 假设 AI 生成了错误信息人类手动修改graph.update_state(configconfig,values{messages:[AIMessage(content【人工修正】尊敬的客户您的订单预计 1-2 个工作日内送达比之前估计的更快。)]},as_nodegenerate,)# 继续执行resultgraph.invoke(None,configconfig)print(f 修正后输出:{result[messages][-1].content})6.4 API 速查API完整签名入参说明返回值说明interrupt(value)interrupt(value: Any)value: 中断时返回给调用方的信息应 JSON 可序列化恢复时传入的值在节点内暂停执行Command(resumevalue)Command(resume: Any)resume: 恢复执行时传给 interrupt() 的值Command对象恢复中断执行compile(interrupt_before[...])compile(interrupt_before: list[str])interrupt_before: 在这些节点前暂停CompiledGraph编译时指定预中断点compile(interrupt_after[...])compile(interrupt_after: list[str])interrupt_after: 在这些节点后暂停CompiledGraph编译时指定后中断点stream.interrupts属性无中断信息列表v3 事件流式中获取中断详情6.5 错误与避坑指南坑 1在 interrupt() 外用 try/except 包裹# ❌ 错误写法defbad_node(state:State):try:answerinterrupt(请确认)# 你的逻辑exceptException:# 捕获了 GraphInterrupt 异常pass# 中断机制被破坏无法恢复# ✅ 正确写法defgood_node(state:State):answerinterrupt(请确认)# 不要包裹 interrupt()# 这里的代码只有在恢复后才会执行return{result:answer}坑 2interrupt() 返回复杂对象# ❌ 错误写法interrupt(MyComplexObject())# 不可序列化# ✅ 正确写法interrupt({question:确认执行此操作,context:{order_id:123,amount:999},options:[批准,拒绝],})坑 3中断前副作用不可幂等# ❌ 错误写法defbad_node(state:State):send_notification()# 发送通知副作用answerinterrupt(请确认)# 恢复后节点重新执行send_notification() 会再次被调用process_order()# ✅ 正确写法defgood_node(state:State):answerinterrupt(请确认)# 把所有副作用放在 interrupt() 之后send_notification()process_order()坑 4忘记配置 checkpointer# ❌ 错误写法graphbuilder.compile(interrupt_before[approval],# 没有 checkpointer)# → 中断后无法恢复因为状态没有被保存# ✅ 正确写法graphbuilder.compile(checkpointerMemorySaver(),# 必须配置 checkpointerinterrupt_before[approval],)6.6 最佳实践总结审批类用interrupt()调试类用interrupt_before/after动态 vs 静态按需选择中断信息使用结构化数据dict 格式方便调用方解析和展示中断前的操作保持幂等所有副作用放在interrupt()之后必须配置 checkpointer没有 checkpointer 无法保存中断状态使用 v3 事件流式 API 检测中断stream.interrupts比 try/except 更优雅