Human-in-the-Loop：人机协作的正确姿势

一只用 AI Agent 搭副业产线的程序员--- 我第一次让 Agent 自动部署代码的时候紧张得手心出汗。 不是因为部署复杂——相反Agent 做得很快拉代码、编译、打包、推到服务器一气呵成。但那 30 秒里我一直想一个问题**如果它搞砸了我有办法在半路拦住它吗** 答案是没有。因为我没设计「半路拦停」的机制。 Human-in-the-LoopHITL就是这道安全阀——在关键时刻Agent 停下来等人按「确定」。 --- ## 不是所有步骤都需要人确认 一个极端是每步都确认——Agent 变成遥控玩具每次行动都弹对话框。这比不用 Agent 还累。 另一个极端是全程不管——Agent 变成脱缰野马。 关键是**在高风险节点插入确认其他步骤自动跑。** --- ## 风险评估模型 go package hitl type RiskLevel int const ( RiskLow RiskLevel iota // 读操作无副作用 RiskMedium // 写操作可回滚 RiskHigh // 写操作不可回滚 RiskCritical // 生产环境变更、财务操作 ) // Action 评估每个步骤的风险等级 func AssessRisk(action string) RiskLevel { switch { case isDestructive(action): return RiskCritical case isExpensive(action): return RiskHigh case isWrite(action): return RiskMedium default: return RiskLow } } func isDestructive(action string) bool { destructive : []string{deploy, drop_table, delete_database, force_push, publish} actionLower : strings.ToLower(action) for _, d : range destructive { if strings.Contains(actionLower, d) { return true } } return false } func isExpensive(action string) bool { expensive : []string{train_model, batch_process, send_sms, charge} actionLower : strings.ToLower(action) for _, e : range expensive { if strings.Contains(actionLower, e) { return true } } return false } func isWrite(action string) bool { write : []string{update, insert, create, write, save, modify, delete} actionLower : strings.ToLower(action) for _, w : range write { if strings.Contains(actionLower, w) { return true } } return false } --- ## 完整实现 go package main import ( bufio fmt os strings time ) // ConfirmationRequest 确认请求 type ConfirmationRequest struct { StepID string json:step_id Action string json:action Risk RiskLevel json:risk Details string json:details Timestamp time.Time json:timestamp } // ConfirmationResponse 人的回复 type ConfirmationResponse struct { Approved bool Comment string } // HumanInTheLoopAgent 支持人机协作的 Agent type HumanInTheLoopAgent struct { confirmFunc func(ConfirmationRequest) ConfirmationResponse autoApprove RiskLevel // 低于此风险等级的自动批准 } func NewHumanInTheLoopAgent(autoApprove RiskLevel) *HumanInTheLoopAgent { return HumanInTheLoopAgent{ confirmFunc: consoleConfirm, // 默认终端输入 autoApprove: autoApprove, } } // SetConfirmFunc 允许自定义确认方式CLI、WebSocket、Slack 消息等 func (a *HumanInTheLoopAgent) SetConfirmFunc(fn func(ConfirmationRequest) ConfirmationResponse) { a.confirmFunc fn } // ExecuteWithConfirmation 带确认的执行 func (a *HumanInTheLoopAgent) ExecuteWithConfirmation(steps []Step) error { for i, step : range steps { risk : AssessRisk(step.Action) fmt.Printf(\n[%d/%d] %s, i1, len(steps), step.Name) fmt.Printf( 风险等级: %s\n, riskLevelName(risk)) // 低风险 自动批准阈值 → 跳过确认 if risk a.autoApprove { fmt.Printf( ✅ 自动批准风险等级 %s ≤ 阈值 %s\n, riskLevelName(risk), riskLevelName(a.autoApprove)) step.Execute() continue } // 高风险 → 请求确认 req : ConfirmationRequest{ StepID: step.ID, Action: step.Action, Risk: risk, Details: step.Details, Timestamp: time.Now(), } resp : a.confirmFunc(req) if !resp.Approved { return fmt.Errorf(步骤 %s 被用户拒绝: %s, step.ID, resp.Comment) } if err : step.Execute(); err ! nil { return fmt.Errorf(步骤 %s 执行失败: %w, step.ID, err) } } return nil } func riskLevelName(r RiskLevel) string { switch r { case RiskLow: return 低 case RiskMedium: return 中 case RiskHigh: return 高 case RiskCritical: return 严重 default: return 未知 } } // ──────────── 确认方式 ──────────── // 方式 1终端交互开发阶段 func consoleConfirm(req ConfirmationRequest) ConfirmationResponse { fmt.Println(strings.Repeat(─, 50)) fmt.Printf(⚠️ 需要人工确认\n) fmt.Printf(步骤: %s\n, req.StepID) fmt.Printf(操作: %s\n, req.Action) fmt.Printf(风险: %s\n, riskLevelName(req.Risk)) fmt.Printf(详情: %s\n, req.Details) fmt.Print(\n确认执行(y/n): ) reader : bufio.NewReader(os.Stdin) input, _ : reader.ReadString(\n) input strings.TrimSpace(strings.ToLower(input)) return ConfirmationResponse{ Approved: input y || input yes, Comment: input, } } // 方式 2Slack 通知团队协作 type SlackNotifier struct { webhookURL string } func (s *SlackNotifier) Confirm(req ConfirmationRequest) ConfirmationResponse { // 发送 Slack 消息等待用户点击按钮 // 实际项目使用 Slack Block Kit Interactive Messages fmt.Printf( 已发送 Slack 确认消息到频道 #agent-approvals\n) fmt.Printf( 步骤: %s | 操作: %s\n, req.StepID, req.Action) // 等待回调... return ConfirmationResponse{Approved: true, Comment: Slack 审批通过} } // 方式 3超时自动拒绝安全默认值 func timeoutConfirm(timeout time.Duration) func(ConfirmationRequest) ConfirmationResponse { return func(req ConfirmationRequest) ConfirmationResponse { resultCh : make(chan ConfirmationResponse, 1) go func() { // 模拟等待用户输入 time.Sleep(2 * time.Second) resultCh - ConfirmationResponse{Approved: true} }() select { case result : -resultCh: return result case -time.After(timeout): fmt.Printf(⏰ 超时 %v自动拒绝步骤 %s\n, timeout, req.StepID) return ConfirmationResponse{ Approved: false, Comment: fmt.Sprintf(超时自动拒绝超过 %v, timeout), } } } } // ──────────── Step 定义 ──────────── type Step struct { ID string Name string Action string Details string Execute func() error } // ──────────── 使用示例 ──────────── func main() { agent : NewHumanInTheLoopAgent(RiskLow) // 低风险自动执行 steps : []Step{ { ID: 1, Name: 读取配置文件, Action: read_config, Details: 读取 config/production.yaml, Execute: func() error { fmt.Println( → 执行读取配置文件...) return nil }, }, { ID: 2, Name: 更新数据库 Schema, Action: update_database, Details: ALTER TABLE users ADD COLUMN tier VARCHAR(20);\n预计影响: 12000 行\n回滚方案: ALTER TABLE users DROP COLUMN tier;, Execute: func() error { fmt.Println( → 执行更新数据库...) return nil }, }, { ID: 3, Name: 部署到生产环境, Action: deploy_production, Details: docker push myapp:v2.3.1 → kubectl rollout restart deployment/myapp\n当前在线用户: 约 200\n预计停机: 0滚动更新, Execute: func() error { fmt.Println( → 执行部署到生产...) return nil }, }, { ID: 4, Name: 发送全站通知, Action: publish_announcement, Details: 内容: 系统已升级到 v2.3.1新增积分功能\n受众: 全站 12000 注册用户, Execute: func() error { fmt.Println( → 执行发送通知...) return nil }, }, } err : agent.ExecuteWithConfirmation(steps) if err ! nil { fmt.Printf(\n❌ 执行中止: %v\n, err) return } fmt.Println(\n✅ 所有步骤完成) } --- ## 确认节点的插入时机 不要每步都确认。只在以下 4 类操作前插入确认 | 类型 | 示例 | 确认策略 | |------|------|---------| | **破坏性操作** | DROP TABLE, rm -rf, force push, deploy | 必须确认 | | **高成本操作** | 批量 API 调用, 模型训练, 发送短信 | 确认 成本估算 | | **生产变更** | 数据库 Schema 变更, DNS 修改 | 确认 回滚方案 | | **对外发布** | 发送公告, 合并 PR, 发布 Release | 确认 预览 | 读操作、代码分析、本地文件生成——这些不需要确认。 --- ## 产品化三种确认通道 **1. CLI开发环境** go func consoleConfirm(req ConfirmationRequest) ConfirmationResponse **2. Slack/钉钉/飞书团队协作** Agent 发一条带按钮的消息人在手机端点「批准」或「拒绝」。异步审批不阻塞开发流程。 **3. Web Dashboard生产运维** 一个简单的审批页面展示所有待确认的 Agent 操作、风险等级、回滚计划。点一下确认Agent 继续执行。 这三种我都在用。个人开发用 CLI团队协作用 Slack运维用 Dashboard。 --- ## 一个真实的坑确认疲劳 我的第一个 HITL Agent 每步都弹确认包括「读取 go.mod」这种零风险操作。结果 10 个步骤我点了 10 次 y。到第 11 步的时候我已经不看了——肌肉记忆在按 y。 **这就是确认疲劳。** 太多确认等于没确认。 解决方案就是上面的 autoApprove 阈值。设定好风险等级低风险的自动过。每天真正需要确认的操作不超过 3-5 次。 阈值推荐个人项目设 RiskMedium只确认高和严重团队项目设 RiskLow确认中及以上。 --- ## 总结 Human-in-the-Loop 不是「每个步骤都问一遍」而是「在关键节点拉手刹」。 人的价值不是点「确定」按钮而是在 Agent 要炸掉生产数据库的时候说「停」。 下一篇是模块四的最后一篇——我们把前面 9 篇的知识整合成一个**通用的 Agent 框架**。你可以直接拿这个框架去写自己的 Agent。 **关注我别错过。** --- 一只用 AI Agent 搭副业产线的程序员 全平台同名虾哥不加班 需要定制 AI 工具来聊聊 → lob_ai 源码[GitHub - lobster-bujiaban](https://github.com/lobster-bujiaban)