20人机协作在关键操作前暂停并等待审批《Eino 实战用 Go 构建 AI 应用与智能体》· 第 20 篇关键词HITL、Interrupt、Resume、CheckPoint、StatefulInterrupt、审批前置条件已完成 16从流程到 Agent 与 17Runner 与事件流。第 11 篇强调过鉴权必须在工具内部完成模型说“用户有权限”不构成授权。但有一类操作光有权限还不够——它本身风险高、影响不可逆删除工单发送对外通知执行上游发布大额退款即使调用者有权限也值得让人再确认一次。这就是 human-in-the-loopHITL人机协作。Eino 的实现方式是中断与恢复运行到关键点暂停保存现场等人给出决定再从中断处继续。一、三个核心概念概念作用Interrupt中断在工具或节点内暂停执行等待外部输入CheckPoint检查点保存中断时的运行现场Resume恢复带着外部输入从中断处继续执行关键机制是寻址Addressing每个可中断点有一个地址interrupt ID恢复时用地址指定“我要恢复哪一个点”。这解决了一个现实问题一次运行可能有多个中断点比如并行调用了两个需要审批的工具。恢复时可以精确指定也可以全部放行。二、在工具中实现审批最典型的做法是在工具内部中断。看这段代码的结构func(t*closeTicketTool)InvokableRun(ctx context.Context,argsstring,_...tool.Option)(string,error){// 1. 是否处于中断状态wasInterrupted,_,state:tool.GetInterruptState[approveState](ctx)if!wasInterrupted{// 首次调用中断并保存状态等待人工审批。return,tool.StatefulInterrupt(ctx,关闭工单需要审批请确认,approveState{TicketID:T-1001})}// 2. 恢复时读取人工给出的决定。isResumeFlow,hasData,decision:tool.GetResumeContext[string](ctx)if!isResumeFlow||!hasData{// 不是本次恢复的目标保持中断状态。return,tool.StatefulInterrupt(ctx,等待审批结果,state)}ifdecisionapprove{return已关闭工单 state.TicketID,nil}return已拒绝关闭 state.TicketID,nil}这段代码包含三个关键点1.StatefulInterrupt带状态。它不仅中断还把approveState保存下来。恢复时通过GetInterruptState取回这样工具就知道“我要审批的是哪个工单”。如果用不保存状态的Interrupt恢复后就丢失了上下文。2.isResumeFlow决定是否消费数据。GetResumeContext返回三个值isResumeFlow是否被本次恢复指定为目标、hasData是否带了数据、data数据本身。当isResumeFlow为 false 时说明恢复的目标不是这个中断点——此时必须重新中断否则会丢掉状态。官方文档把这条写得很明确叶节点组件必须重新中断以保留状态。3. 用类型注册支持序列化。状态要被序列化进检查点需要注册类型funcinit(){schema.Register[approveState]()}忘了注册检查点序列化会失败。三、完整示例下面这段程序演示“中断 → 保存现场 → 带审批结果恢复 → 执行”的完整闭环不访问网络packagemainimport(contextfmtlogsyncgithub.com/cloudwego/eino/adkgithub.com/cloudwego/eino/components/modelgithub.com/cloudwego/eino/components/toolgithub.com/cloudwego/eino/composegithub.com/cloudwego/eino/schema)// ---- 审批工具首次调用时中断等待人工决定 ----typeapproveStatestruct{TicketIDstring}funcinit(){schema.Register[approveState]()}typecloseTicketToolstruct{}func(t*closeTicketTool)Info(_context.Context)(*schema.ToolInfo,error){returnschema.ToolInfo{Name:close_ticket,Desc:关闭工单需要人工审批},nil}func(t*closeTicketTool)InvokableRun(ctx context.Context,argsstring,_...tool.Option)(string,error){// 1. 是否处于中断状态wasInterrupted,_,state:tool.GetInterruptState[approveState](ctx)if!wasInterrupted{// 首次调用中断并保存状态等待人工审批。return,tool.StatefulInterrupt(ctx,关闭工单需要审批请确认,approveState{TicketID:T-1001})}// 2. 恢复时读取人工给出的决定。isResumeFlow,hasData,decision:tool.GetResumeContext[string](ctx)if!isResumeFlow||!hasData{// 不是本次恢复的目标保持中断状态。return,tool.StatefulInterrupt(ctx,等待审批结果,state)}ifdecisionapprove{return已关闭工单 state.TicketID,nil}return已拒绝关闭 state.TicketID,nil}// ---- 内存检查点存储保存中断现场 ----typememoryStorestruct{mu sync.Mutex mmap[string][]byte}funcnewMemoryStore()*memoryStore{returnmemoryStore{m:map[string][]byte{}}}func(s*memoryStore)Get(_context.Context,keystring)([]byte,bool,error){s.mu.Lock()defers.mu.Unlock()v,ok:s.m[key]returnv,ok,nil}func(s*memoryStore)Set(_context.Context,keystring,value[]byte)error{s.mu.Lock()defers.mu.Unlock()cp:make([]byte,len(value))copy(cp,value)s.m[key]cpreturnnil}// ---- 模拟模型先请求调用审批工具看到结果后总结 ----typemockModelstruct{}func(m*mockModel)Generate(_context.Context,in[]*schema.Message,_...model.Option)(*schema.Message,error){for_,msg:rangein{ifmsg.Roleschema.Tool{returnschema.AssistantMessage(处理结果msg.Content,nil),nil}}returnschema.AssistantMessage(,[]schema.ToolCall{{ID:call_1,Type:function,Function:schema.FunctionCall{Name:close_ticket,Arguments:{}},}}),nil}func(m*mockModel)Stream(ctx context.Context,in[]*schema.Message,opts...model.Option)(*schema.StreamReader[*schema.Message],error){msg,err:m.Generate(ctx,in,opts...)iferr!nil{returnnil,err}sr,sw:schema.Pipe[*schema.Message](1)gofunc(){defersw.Close();sw.Send(msg,nil)}()returnsr,nil}func(m*mockModel)WithTools([]*schema.ToolInfo)(model.ToolCallingChatModel,error){returnm,nil}// messageOutput 安全地取出消息输出避免空指针。funcmessageOutput(event*adk.AgentEvent)*adk.MessageVariant{ifeventnil||event.Outputnil{returnnil}returnevent.Output.MessageOutput}funcmain(){iferr:run();err!nil{log.Fatal(err)}}funcrun()error{ctx:context.Background()agent,err:adk.NewChatModelAgent(ctx,adk.ChatModelAgentConfig{Name:assistant,Description:工单助手,Instruction:需要时调用工具。,Model:mockModel{},ToolsConfig:adk.ToolsConfig{ToolsNodeConfig:compose.ToolsNodeConfig{Tools:[]tool.BaseTool{closeTicketTool{}},}},})iferr!nil{returnerr}runner:adk.NewRunner(ctx,adk.RunnerConfig{Agent:agent,CheckPointStore:newMemoryStore()})// 第一次运行应在工具处中断。// 显式指定检查点 ID中断时据此保存现场。iter:runner.Query(ctx,请关闭工单 T-1001,adk.WithCheckPointID(checkpoint-1))varinterruptIDstringfor{event,ok:iter.Next()if!ok{break}ifevent.Err!nil{returnevent.Err}ifevent.Action!nilevent.Action.Interrupted!nil{for_,ic:rangeevent.Action.Interrupted.InterruptContexts{fmt.Printf([中断] id%s info%v\n,ic.ID,ic.Info)ifinterruptID{interruptIDic.ID}}}// 中断事件里 event.Output 可能为 nil访问字段前必须判空。ifmv:messageOutput(event);mv!nilmv.Message!nilmv.Message.Content!{fmt.Println([输出],mv.Message.Content)}}ifinterruptID{returnfmt.Errorf(预期发生中断但没有拿到 interrupt ID)}// 第二次带审批数据恢复。fmt.Println([恢复] 人工决定approve)iter2,err:runner.ResumeWithParams(ctx,checkpoint-1,adk.ResumeParams{Targets:map[string]any{interruptID:approve},})iferr!nil{returnfmt.Errorf(恢复失败: %w,err)}for{event,ok:iter2.Next()if!ok{break}ifevent.Err!nil{returnevent.Err}ifmv:messageOutput(event);mv!nilmv.Message!nilmv.Message.Content!{fmt.Println([输出],mv.Message.Content)}ifevent.Action!nilevent.Action.Interrupted!nil{fmt.Println([仍在中断])}}returnnil}实机输出[中断] id79f0aa61-a2f2-482b-8719-095efece030f info关闭工单需要审批请确认 [恢复] 人工决定approve [输出] 已关闭工单 T-1001 [输出] 处理结果已关闭工单 T-1001interrupt ID 每次运行都不同这是正常的。完整闭环清晰可见运行触发中断输出中断 ID 和提示信息——此时工具没有真正执行。人工给出决定approve。恢复运行工具读取决定后执行返回“已关闭工单”。模型拿到工具结果给出最终回复。注意中断时工具并未执行。这正是 HITL 的价值把“决定”和“执行”分开。四、两种恢复方式Eino 提供两个恢复入口方法特点适用Resume(ctx, checkPointID)隐式恢复全部所有中断点都继续单个“继续”按钮的简单场景ResumeWithParams(ctx, id, params)显式定向精确指定恢复哪个点、带什么数据多中断点、需要传数据推荐ResumeWithParams的Targets是“地址 → 数据”的映射targets:map[string]any{interruptID:approve,// 这个点批准// otherID: reject, // 另一个点拒绝如果有}当有多个中断点时未被Targets指定的点必须重新中断以保留状态。这不是可选的实现细节而是框架的约定——官方文档明确指出叶节点组件必须重新中断。上面示例中那段if !isResumeFlow || !hasData分支就是这个约定。五、CheckPointStore生产环境的关键示例用的是内存存储进程重启就丢。生产环境需要考虑要求说明持久化用 Redis、数据库等外部存储分布式多实例部署时恢复请求可能落在另一台机器上过期清理检查点不能永久堆积要有 TTL信息最小化检查点是序列化的运行现场可能包含敏感内容需评估存储安全幂等性恢复不等于数据一致见下如果你是多实例部署内存存储一定不够用。用户点了“确认”后请求可能被负载均衡到任意实例——检查点必须在共享存储里。六、HITL 不等于安全这是本篇最需要强调的一点。中断-恢复解决的是“流程控制”问题不是“数据一致性”问题。设想这个链路1. 工具执行 → 关闭工单成功 2. 保存结果时进程崩溃 / 检查点恢复失败 3. 用户重试 → 工具再次执行如果“关闭工单”不是幂等的就会出问题。因此风险对策恢复导致重复执行写操作要幂等幂等键、状态判断恢复前数据已被他人修改恢复时重新校验业务状态审批期间权限被撤销恢复时重新校验权限不要信任中断前的判断检查点内容泄露存储加密、访问控制、及时清理恢复时必须重新校验权限和业务状态。审批可能发生在几分钟甚至几小时之后期间环境已经变化。还有一点人工确认不能替代鉴权。用户点了“确认”服务端仍须验证他是否有权做这件事第 11 篇。七、适合引入 HITL 的场景场景是否需要 HITL查询类操作通常不需要可逆的写操作视风险可加审计而非审批不可逆操作删除、发布、退款建议需要涉及对外发送建议需要高金额或高影响建议需要判断标准如果做错了、又很难撤回就值得让人看一眼。但也要克制。审批太多会让用户疲劳最后变成“闭眼点确认”反而更危险。只对真正关键的操作设置审批并在界面上说明清楚在审批什么。八、常见问题预期中断却没有中断检查工具是否真的被调用看事件流以及CheckPointStore是否配置。没有存储中断无法保存现场。恢复时报找不到检查点ID 不一致或进程重启导致内存存储丢失。生产环境要用持久化存储。恢复后状态丢失状态类型没有注册schema.Register[T]()或没有用StatefulInterrupt保存状态。恢复后一直在中断GetResumeContext的isResumeFlow为 false代码重新中断了。确认Targets里的地址与中断 ID 一致。中断事件处理时 panicevent.Output为nil本例实测踩到过。访问前必须逐层判空。多个中断点只恢复了一个用ResumeWithParams并在Targets中列出所有需要恢复的地址。恢复后重复执行了写操作缺少幂等设计。这是业务层问题检查点机制不会替你解决。九、小结HITL 通过“中断 → 保存检查点 → 恢复”实现核心是寻址机制。工具内用StatefulInterrupt中断并保存状态恢复时用GetInterruptState和GetResumeContext。ResumeWithParams支持定向恢复未被指定为目标的点必须重新中断。状态类型需schema.Register注册才能序列化。生产环境必须用持久化、可跨实例共享的检查点存储。HITL 是流程控制不是安全保证恢复时要重新校验权限与业务状态写操作要幂等。只对不可逆或高影响操作设置审批避免审批疲劳。下一篇《Agent 与 Graph 结合自主决策确定性执行》我们把已编排好的流程交给 Agent 调用。参考资料与版本说明资料核对日期为2026 年 10 月 9 日Eino human-in-the-loop 框架技术架构指南Interrupt CheckPoint 使用手册Eino ADK 概述Eino 主仓库本文示例按Eino v0.9.21编写使用模拟模型与内存检查点存储已实机运行正文输出含中断 ID、判空踩坑为真实结果。