Human-in-the-loop 实战:让 AI Agent 在关键节点等人确认

当 AI Agent 能自动调用工具、发消息、改数据库,它离”出事”往往只有一步之遥。Human-in-the-loop(人在回路)就是在这关键一步之前喊停,把最终决定权交还给真正能负责的人。本文用审批门、草稿确认、超时回退三种可落地的设计模式,讲清如何给你的 AI Agent 加上一道人工确认阀,并在自主与可控之间找到平衡。

一、什么是 Human-in-the-loop(人在回路)

一个自主 Agent 的典型闭环是:感知环境 → 大模型推理 → 调用工具 → 观察结果 → 继续推理。这个闭环跑得越快,越容易在某一环做出不可逆的动作。Human-in-the-loop(简称 HITL,中文常译”人在回路”)指的就是:在闭环里故意插入”暂停点”,让真实用户在关键决策上做确认或否决,Agent 拿到许可后才继续执行。

它解决的不是”Agent 不够聪明”,而是”Agent 不够稳”。聪明负责找方案,人在回路负责兜底风险。下面这类动作,几乎都应该进人工确认列表:

# 需要人工确认的高风险动作类型
RISKY_ACTIONS = {
    "payment":      "资金变动:扣款、转账、下单",
    "data_delete":  "数据删除或覆盖:DROP、覆盖记录",
    "external_send":"对外发送:邮件、消息、工单",
    "permission":   "权限变更:授权、角色调整",
}

def is_risky(action_type: str) -> bool:
    return action_type in RISKY_ACTIONS

二、哪些节点必须停下来等人

不是所有动作都该拦。拦截太松会出事故,拦截太紧人会疲劳、最后干脆一路点”同意”。一个实用的判断标准是:动作是否不可逆、是否影响第三方、是否涉及资金或权限。下表给出常见操作的分类参考。

操作类型典型示例是否需人工判断理由
资金变动扣款、转账、下单必须不可逆、直接经济损失
数据删除/覆盖删表、覆盖记录必须恢复成本极高
对外发送发邮件、发消息、发推必须影响第三方、难撤回
权限变更授权、改角色必须安全风险最高
高风险推理结论诊断、法律/医疗建议建议责任边界需人把关
只读查询查数据、搜索不需要无副作用

三、模式一:审批门(Approval Gate)

最直观的做法是”审批门”:任何高风险工具调用先进入 pending 状态,等待人工 approve / reject,只有拿到许可才真正执行。把确认逻辑做成统一拦截层,而不是散落在每个工具里。

from enum import Enum

class Decision(Enum):
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"

def request_approval(action: dict) -> str:
    """返回 ticket_id,状态置为 pending 并通知人工。"""
    ticket_id = create_ticket(action)
    notify_human(ticket_id, action)
    return ticket_id

def execute_if_approved(ticket_id: str, run: callable):
    if get_status(ticket_id) != Decision.APPROVED:
        raise PermissionError("操作未获人工批准,已中止")
    return run()

四、模式二:草稿 + 确认(Draft & Confirm)

有些动作不适合”先做再拦”,而适合”先出草稿,人确认后再落库”。典型如生成一段要发送的邮件、一条要发布的帖子、一份要提交的 SQL。草稿本身无副作用,确认才生效。

def generate_draft(prompt: str) -> dict:
    content = llm_generate(prompt)
    return {"state": "draft", "content": content, "id": new_id()}

def publish_draft(draft_id: str, approved_by: str) -> dict:
    draft = load(draft_id)
    if draft["state"] != "draft":
        raise ValueError("草稿状态异常")
    draft["state"] = "published"
    draft["approved_by"] = approved_by   # 记录责任人
    return persist(draft)

五、模式三:超时与回退(Timeout & Fallback)

人工确认不是永远在线。如果一条审批长时间没人理,Agent 不该无限卡死,也不该偷偷替人做决定。正确做法是:超时自动取消,并把任务降级到人工工单队列,保证”要么被确认,要么被安全搁置”。这跟 AI 代码评审与单元测试生成 里”大模型当副驾、人当主驾”的思路一致——把待决事项显式移交给人。

import time

def wait_for_approval(ticket_id: str, timeout: int = 1800):
    deadline = time.time() + timeout
    while time.time() < deadline:
        if get_status(ticket_id) in (Decision.APPROVED, Decision.REJECTED):
            return get_status(ticket_id)
        time.sleep(10)
    # 超时:不自动执行,转为人工工单并通知
    escalate_to_human_queue(ticket_id)
    return Decision.REJECTED  # 安全默认:超时即中止

六、用 Function Calling 串起"调用 → 申请确认"

Function Calling 工具调用 体系里,可以让每个工具声明自己是否需要确认。运行时统一拦截:凡是 confirmation_required 为 true 的调用,先走审批门再执行,模型本身无需关心确认细节。

tools = [{
    "type": "function",
    "function": {
        "name": "send_email",
        "description": "向用户发送邮件",
        "parameters": {"type": "object", "properties": {
            "to": {"type": "string"}, "body": {"type": "string"}
        }},
        "confirmation_required": True   # 自定义字段,标记需人工确认
    }
}]

def run_tool_call(call):
    if call.function.get("confirmation_required"):
        tid = request_approval(call)          # 进审批门
        if get_status(tid) != Decision.APPROVED:
            return {"error": "rejected_by_human"}
    return dispatch(call)

七、和多智能体、记忆、可观测的协同

HITL 不是孤立的一环。在 多智能体协作 架构里,建议把所有"需确认动作"集中到一个主管 Agent 统一申请,避免每个子 Agent 各自弹窗;在 Agent 记忆系统 中记录"哪些操作曾被人工否决",下次直接走人工,不再反复申请;同时把每次确认的耗时、通过率、驳回原因埋点到 LLM 应用可观测 体系,用数据判断拦截策略是否过紧或过松。

八、生产落地清单与常见坑

下面是实践中最容易踩的五个坑,以及对应解法:

现象解法
确认疲劳每次都弹窗,人麻木乱点低风险自动过,只拦高风险
确认绕过前端漏掉确认直接执行后端强制校验 pending 状态
超时丢操作超时后无人知晓转工单 + 主动通知
责任不清出事不知谁确认记录 operator_id + 审计日志
重放攻击同一确认被复用一次性 token,用完即废

九、串起来:一个最小可运行骨架

把前面的模式收拢成一个骨架:所有工具调用先过一道高风险拦截,命中的进入草稿或审批门,超时未决则安全中止并转人工。下面是一段示意性的主循环,可以直接作为你自己 Agent 的脚手架来填充。

def agent_loop(goal: str):
    plan = llm_plan(goal)
    for step in plan:
        call = llm_pick_tool(step)
        if needs_human(call):                      # 命中高风险清单
            tid = request_approval(call)           # 审批门 + 草稿
            if wait_for_approval(tid) != Decision.APPROVED:
                log("已安全中止,转人工工单"); continue
        result = dispatch(call)                    # 拿到许可才执行
        plan = llm_reflect(plan, result)           # 观察结果后继续

这个骨架的关键在于:dispatch 永远只在审批通过之后发生。即便大模型"幻觉"出一个危险步骤,它也会卡在审批门前,把最终决定权留给人。等这条链路跑顺了,再把低风险动作从确认清单里移除,让人只处理真正重要的那几步。

十、小结

Human-in-the-loop 的目标不是让 Agent 变慢,而是把"不可逆风险"交给最合适做判断的人。审批门、草稿确认、超时回退三套模式可以组合使用:能自动的一律自动,必须等人的坚决等人。即便在 Dify + Ollama 私有化部署 里,也可以借工作流的"人工节点"原生实现这套机制。把确认做成显式的、可审计的、带超时的流程,你的 Agent 才真正敢放上生产。

上一篇 Java 内存模型实战:volatile、happen-before 与可见性陷阱
下一篇 Nginx 413 文件上传失败排查与根治