Agent人工介入Human‑In‑The‑Loop流程

在自动化脚本、AI Agent 或 CI/CD 流水线里，删除文件、重启服务、释放防火墙端口这类高危操作一旦误执行，代价可能很大。
人工介入（Human-In-The-Loop）就是在 Agent 真正执行这些操作前，强制插入一道“人工审批”关卡。
本文用可落地的思路，讲清楚怎么设计这个流程、写最小实现、避开常见坑，并完成验证。

先把环境准备明白。
你需要一个能运行脚本的机器，比如 Linux 服务器或本地开发机，同时确认 Agent 有明确的“操作分类”能力——至少能区分哪些动作属于高危。
还要输入高管的审批渠道：企业微信群机器人、钉钉群、邮件或简单的 Web 审批页面都可以，目的是让审批人能及时收到请求并回复“同意/拒绝”。
最轻量的方案是用一个 JSON 文件当审批单存储，再加一个监听函数轮询状态，适合学习和小团队。

核心实现分四步。
第一步，定义高危操作名单和审批规则，比如“执行 rm -rf、systemctl stop、iptables -F 前必须审批”。
第二步，在 Agent 执行前截获操作，生成带唯一 ID 的审批单，通过 webhook 发到审批群。
第三步，审批人点击同意或拒绝后，回调脚本更新审批单状态。
第四步，Agent 只在状态为 approved 时继续执行，否则终止。
下面用一个 Python 伪代码展示关键逻辑：

# 伪代码：人工审批流程核心
class HighRiskAction:
    def execute(self, action):
        if action in HIGH_RISK_LIST:
            ticket = create_ticket(action)          # 生成审批单
            send_webhook("群消息", f"{ticket.id} 请求执行 {action}")
            while True:
                status = check_ticket(ticket.id)    # 轮询审批结果
                if status == "approved":
                    break
                if status == "rejected" or timeout:
                    print("审批未通过，已中止")
                    return False
                sleep(30)
        return action.run()  # 真正执行操作

实际对接企业微信或钉钉时，把 send_webhook 换成对应机器人接口，审批回调用它们的卡片回调即可，逻辑完全一致。
重点提醒：人工审批必须能阻塞 Agent 执行，不能在审批未完成时让 Agent 继续跑，否则这个流程就形同虚设。

这一步也是最容易踩坑的地方。
先看审批超时：审批人可能开会、休假，所以轮询不能无限等，超时要设置默认拒绝或转交给备胎审批人。
再看误操作风险：审批人点错怎么办？
建议在审批单里写清楚操作对象、影响范围和回滚方案，要求审批人填备注，这样所有人都知道这个动作是自己批的。
还有消息漏发问题，webhook 推送失败会导致审批单一直挂着，所以每次提交审批单都要失败重试，并支持在后台手动补发。
最后，所有审批记录都要留日志，包括谁在什么时间批了多少次，方便事后审计，这通常是人工审批流里最容易忽略但最重要的要求。

怎么验证你的流程是真正可用的？
最简单的方式是准备一台测试机器，故意触发一条高危操作。
确认 Agent 在执行前停下来，群消息里出现审批请求；
点“拒绝”后观察任务被中止，点“同意”后操作才执行。
再测超时场景，故意不回复，确认按预设策略失败。
验证一个关键结论：审批不是通知，必须能被阻塞等待，否则就没有达到“高危险需要人工把关”的目的。 把以上测试结果固化到你的发布检查单里，以后每次调整 Agent 流程都重新跑一遍，这套人工介入机制才能真正落地。
如果你还想了解审批回调怎么和现有自动化平台结合，可以先从留审批日志、接群机器人开始，逐步扩展成完整的风控系统。