系列「企业级 AI Agent 实现拆解」E41 篇Part 9 源码深度篇第六章。E39 提到工具可以抛出 interruptE40 提到 ReAct 的SetReturnDirectly。这篇完整拆 Eino 的 HITLHuman-In-The-Loop机制——中断如何产生、如何保存状态、如何恢复。读完这篇你会知道三种中断方式配置级节点拦截 vs 运行时组件中断 vs 复合中断中断信号怎么从工具冒泡到 Graph 层GetInterruptStateGetResumeContext组件怎么感知自己被恢复Resume/ResumeWithData外部怎么恢复一个中断的 Agent地址系统如何唯一定位嵌套结构中的断点一个完整的人工审批流实现三种中断方式方式一配置级节点拦截最粗粒度runnable,err:graph.Compile(ctx,compose.WithInterruptBeforeNodes([]string{PaymentNode}),// 执行前暂停compose.WithInterruptAfterNodes([]string{KYCNode}),// 执行后暂停)编译时告诉执行引擎凡是要执行PaymentNode先暂停等审批。不需要修改节点代码适合全局策略。方式二运行时组件中断细粒度可带状态// 无状态中断只是暂停不需要恢复时传数据returncompose.Interrupt(ctx,map[string]any{reason:需要人工确认,amount:input.Amount,})// 有状态中断暂停恢复时组件能读回自己保存的 statereturncompose.StatefulInterrupt(ctx,map[string]any{reason:需要人工确认},// info给外部看的信息TransferState{Amount:input.Amount,From:input.From},// state恢复时用)方式三复合中断ToolsNode 使用当一批并行工具里只有部分 interrupt// ToolsNode 内部returncompose.CompositeInterrupt(ctx,rerunExtra,// 给外部的信息包含哪些 tool 成功了哪些需要重跑rerunState,// 内部状态已执行成功的 tool 结果errs...,// 每个 interrupt 的 InterruptSignal)// 已成功的 tool 结果存进 rerunState恢复时不重跑中断信号的冒泡路径组件内: return StatefulInterrupt(ctx, info, state) ↓ 产生 *core.InterruptSignal{ID, Address, Info, State} ↓ Graph 执行引擎(graph_run.go): completedTask.err InterruptSignal resolveInterruptCompletedTasks() 收集所有中断任务 ↓ handleInterrupt() 把 state 断点地址 保存进 checkpoint 返回 *interruptError{Info: InterruptInfo{InterruptContexts: [...]}} ↓ 外部调用方: _, err agent.Generate(ctx, messages) info, ok : compose.ExtractInterruptInfo(err) // info.InterruptContexts[0].InterruptID ← 恢复用的唯一 ID // info.InterruptContexts[0].Info ← 组件提供的人工阅读信息地址系统精确定位断点每个断点都有一个层级地址格式是[]AddressSegment{Type, ID}AgentRunnable/ graph_node:ChatModel/ tool_call:transfer:call_abc123 ← 第 3 层精确到具体工具调用这个地址就是InterruptIDResumeWithData用它来路由恢复信号// 中断时 info 里有 InterruptIDinfo.InterruptContexts[0].InterruptID// AgentRunnable/ChatModel/transfer:call_abc123// 恢复时用这个 ID 定向恢复ctxcompose.ResumeWithData(ctx,interruptID,approvalResult)_,erragent.Generate(ctx,messages)// 重新执行只有指定断点会被激活Resume API// 恢复一个断点不带数据ctxcompose.Resume(ctx,interruptID)// 恢复一个断点并传数据人工审批结果ctxcompose.ResumeWithData(ctx,interruptID,ApprovalResult{Approved:true,Approver:alice,Comment:金额合规批准,})// 批量恢复多个断点ctxcompose.BatchResumeWithData(ctx,map[string]any{interruptID1:ApprovalResult{Approved:true},interruptID2:ApprovalResult{Approved:false},})组件内感知恢复组件被恢复后执行的代码func(t*TransferTool)InvokableRun(ctx context.Context,argsstring,opts...tool.Option)(string,error){// Step 1检查自己是否曾经中断过wasInterrupted,hasState,savedState:compose.GetInterruptState[*TransferState](ctx)if!wasInterrupted{// 正常执行路径input:parseArgs(args)ifinput.Amount10000{// 需要审批 → 中断保存状态return,compose.StatefulInterrupt(ctx,map[string]any{reason:超额转账,amount:input.Amount},TransferState{Input:input},)}returndoTransfer(input),nil}// Step 2被恢复了。确认自己是被定向恢复还是顺带恢复isTarget,hasData,approval:compose.GetResumeContext[*ApprovalResult](ctx)if!isTarget{// 不是这次 Resume 的目标 → 重新中断保持自己的状态return,compose.StatefulInterrupt(ctx,map[string]any{reason:超额转账等待其他审批完成},savedState,)}if!hasData||!approval.Approved{return,fmt.Errorf(转账被拒绝%s,approval.Comment)}// 审批通过 → 执行真正的转账returndoTransfer(savedState.Input),nil}两个关键函数函数作用GetInterruptState[T]检查自己是否在上一次中断了返回 (wasInterrupted, hasState, state)GetResumeContext[T]检查自己是否是这次 Resume 的目标返回 (isResumeFlow, hasData, data)两种 Resume 策略策略一隐式 “Resume All”适合只有一个审批点// 不管哪个断点被中断Resume 就认为所有断点都该继续ctxcompose.Resume(ctx,interruptIDs...)// 组件里只检查 wasInterrupted不检查 isResumeFlow策略二显式 “Targeted Resume”适合多个独立审批点// 只恢复指定的断点其他断点继续等待ctxcompose.ResumeWithData(ctx,interruptID1,data1)// 组件里isResumeFlowfalse 的断点必须重新中断自己保持等待if!isTarget{return,compose.StatefulInterrupt(ctx,info,savedState)}DeepFlux 使用策略二多个工具可能同时等待不同审批人每个审批独立进行。完整流程示意用户请求 → Agent.Generate(ctx, messages) ↓ 执行到 TransferTool ↓ return StatefulInterrupt(...) ↓ 执行引擎保存 checkpoint ↓ 返回 interruptError 应用层 info : compose.ExtractInterruptInfo(err) interruptID : info.InterruptContexts[0].InterruptID userInfo : info.InterruptContexts[0].Info // {reason: 超额转账} // 异步等待人工审批可以存 DB、发消息通知 approval : waitForApproval(interruptID, userInfo) // 恢复执行 ctx compose.ResumeWithData(ctx, interruptID, approval) result, err agent.Generate(ctx, messages) // 从 checkpoint 恢复跳过已执行节点 ↓ ↓ TransferTool 重新执行GetInterruptState 拿回 savedState ↓ GetResumeContext 确认自己是目标拿到 approval ↓ 执行真正的转账 ↓ 返回结果小结Eino 的 HITL 机制三个核心设计地址系统每个断点都有层级地址图→节点→工具调用支持嵌套结构中精确定位StatefulInterrupt中断时可以把状态存进 checkpoint恢复时通过GetInterruptState拿回组件不需要外部存储定向恢复ResumeWithData(interruptID, data)只激活指定断点其他断点用GetResumeContext检查isResumeFlow不是目标的重新中断自己三者配合实现了部分恢复一个 batch 里 5 个工具2 个需要审批另外 3 个已经跑完——Resume 只重跑那 2 个3 个已完成的结果直接从 checkpoint 读取。代码来源eino/compose/interrupt.go · eino/compose/resume.go