20｜人机协作：在关键操作前暂停并等待审批

《Eino 实战：用 Go 构建 AI 应用与智能体》· 第 20 篇
关键词：HITL、Interrupt、Resume、CheckPoint、StatefulInterrupt、审批
前置条件：已完成 16｜从流程到 Agent 与 17｜Runner 与事件流。

第 11 篇强调过：鉴权必须在工具内部完成，模型说“用户有权限”不构成授权。但有一类操作，光有权限还不够——它本身风险高、影响不可逆：

• 删除工单
• 发送对外通知
• 执行上游发布
• 大额退款

即使调用者有权限，也值得让人再确认一次。这就是 human-in-the-loop（HITL，人机协作）。

Eino 的实现方式是中断与恢复：运行到关键点暂停，保存现场，等人给出决定，再从中断处继续。

一、三个核心概念

关键机制是寻址（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)}ifdecision=="approve"{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("context""fmt""log""sync""github.com/cloudwego/eino/adk""github.com/cloudwego/eino/components/model""github.com/cloudwego/eino/components/tool""github.com/cloudwego/eino/compose""github.com/cloudwego/eino/schema")// ---- 审批工具：首次调用时中断，等待人工决定 ----typeapproveStatestruct{TicketIDstring}funcinit(){schema.Register[approveState]()}typecloseTicketToolstruct{}func(t*closeTicketTool)Info(_context.Context)(*schema.ToolInfo,error){return&schema.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)}ifdecision=="approve"{return"已关闭工单 "+state.TicketID,nil}return"已拒绝关闭 "+state.TicketID,nil}// ---- 内存检查点存储：保存中断现场 ----typememoryStorestruct{mu sync.Mutex mmap[string][]byte}funcnewMemoryStore()*memoryStore{return&memoryStore{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.Role==schema.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{ifevent==nil||event.Output==nil{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!=nil&&event.Action.Interrupted!=nil{for_,ic:=rangeevent.Action.Interrupted.InterruptContexts{fmt.Printf("[中断] id=%s info=%v\n",ic.ID,ic.Info)ifinterruptID==""{interruptID=ic.ID}}}// 中断事件里 event.Output 可能为 nil，访问字段前必须判空。ifmv:=messageOutput(event);mv!=nil&&mv.Message!=nil&&mv.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!=nil&&mv.Message!=nil&&mv.Message.Content!=""{fmt.Println("[输出]",mv.Message.Content)}ifevent.Action!=nil&&event.Action.Interrupted!=nil{fmt.Println("[仍在中断]")}}returnnil}

实机输出：

[中断] id=79f0aa61-a2f2-482b-8719-095efece030f info=关闭工单需要审批，请确认 [恢复] 人工决定：approve [输出] 已关闭工单 T-1001 [输出] 处理结果：已关闭工单 T-1001

（interrupt ID 每次运行都不同，这是正常的。）

完整闭环清晰可见：

1. 运行触发中断，输出中断 ID 和提示信息——此时工具没有真正执行。
2. 人工给出决定：approve。
3. 恢复运行，工具读取决定后执行，返回“已关闭工单”。
4. 模型拿到工具结果，给出最终回复。

注意中断时工具并未执行。这正是 HITL 的价值：把“决定”和“执行”分开。

四、两种恢复方式

Eino 提供两个恢复入口：

ResumeWithParams的Targets是“地址 → 数据”的映射：

targets:=map[string]any{interruptID:"approve",// 这个点：批准// otherID: "reject", // 另一个点：拒绝（如果有）}

当有多个中断点时，未被Targets指定的点必须重新中断以保留状态。这不是可选的实现细节，而是框架的约定——官方文档明确指出叶节点组件必须重新中断。上面示例中那段if !isResumeFlow || !hasData分支就是这个约定。

五、CheckPointStore：生产环境的关键

示例用的是内存存储，进程重启就丢。生产环境需要考虑：

如果你是多实例部署，内存存储一定不够用。用户点了“确认”后，请求可能被负载均衡到任意实例——检查点必须在共享存储里。

六、HITL 不等于安全

这是本篇最需要强调的一点。

中断-恢复解决的是“流程控制”问题，不是“数据一致性”问题。

设想这个链路：

1. 工具执行 → 关闭工单成功 2. 保存结果时进程崩溃 / 检查点恢复失败 3. 用户重试 → 工具再次执行

如果“关闭工单”不是幂等的，就会出问题。因此：

恢复时必须重新校验权限和业务状态。审批可能发生在几分钟甚至几小时之后，期间环境已经变化。

还有一点：人工确认不能替代鉴权。用户点了“确认”，服务端仍须验证他是否有权做这件事（第 11 篇）。

七、适合引入 HITL 的场景

判断标准：如果做错了、又很难撤回，就值得让人看一眼。

但也要克制。审批太多会让用户疲劳，最后变成“闭眼点确认”，反而更危险。只对真正关键的操作设置审批，并在界面上说明清楚在审批什么。

八、常见问题

预期中断却没有中断
检查工具是否真的被调用（看事件流），以及CheckPointStore是否配置。没有存储，中断无法保存现场。

恢复时报找不到检查点
ID 不一致，或进程重启导致内存存储丢失。生产环境要用持久化存储。

恢复后状态丢失
状态类型没有注册（schema.Register[T]()），或没有用StatefulInterrupt保存状态。

恢复后一直在中断
GetResumeContext的isResumeFlow为 false，代码重新中断了。确认Targets里的地址与中断 ID 一致。

中断事件处理时 panic
event.Output为nil（本例实测踩到过）。访问前必须逐层判空。

多个中断点只恢复了一个
用ResumeWithParams并在Targets中列出所有需要恢复的地址。

恢复后重复执行了写操作
缺少幂等设计。这是业务层问题，检查点机制不会替你解决。

九、小结

• HITL 通过“中断 → 保存检查点 → 恢复”实现，核心是寻址机制。
• 工具内用StatefulInterrupt中断并保存状态；恢复时用GetInterruptState和GetResumeContext。
• ResumeWithParams支持定向恢复；未被指定为目标的点必须重新中断。
• 状态类型需schema.Register注册才能序列化。
• 生产环境必须用持久化、可跨实例共享的检查点存储。
• HITL 是流程控制，不是安全保证：恢复时要重新校验权限与业务状态，写操作要幂等。
• 只对不可逆或高影响操作设置审批，避免审批疲劳。

下一篇《Agent 与 Graph 结合：自主决策，确定性执行》，我们把已编排好的流程交给 Agent 调用。

参考资料与版本说明

资料核对日期为2026 年 10 月 9 日：

1. Eino human-in-the-loop 框架：技术架构指南
2. Interrupt & CheckPoint 使用手册
3. Eino ADK 概述
4. Eino 主仓库

本文示例按Eino v0.9.21编写，使用模拟模型与内存检查点存储，已实机运行，正文输出（含中断 ID、判空踩坑）为真实结果。