AI Agent 人机协作实战：让智能体知道"什么时候该停下来问人"

作者：yumking | 2026 年 9 月 2 日 | 技术标签：AI Agent / Human-in-the-loop / 置信度路由 / 人机协同

摘要

前七篇我们把 Agent 从"单轮对话"武装到"会记忆、会协作、会用工具、会查知识、会防攻击、可评估、会规划"的智能体。但有一个事实被刻意回避：Agent 永远不可能 100% 自动。强行追求全自动，要么在低风险任务上反复打扰人工、把效率红利还回去，要么在高风险操作上一脚踩空、酿成不可逆事故。本文剖析 Agent 自动化率的天花板与错误代价的不对称性，提出一套"置信度路由 + 四种确认门 + 异步超时降级"的 HITL 架构：让 Agent 在不确定时主动停下、在越界时强制停下、在人缺席时优雅降级而非硬闯。实测将高风险操作的误执行率从 5.2% 压降至 0.3%，同时把人工介入率控制在 8%——既不放过风险，也不淹没人工。

一、为什么 Agent 必须人机协作？

1.1 自动化率的天花板

业界常把"全自动"当作 Agent 的终极形态，但真实数据告诉我们另一回事：

结论：自动化率不是越高越好，而是要贴着"风险曲线"走。一个把删库也自动执行的 Agent，不是智能体，是定时炸弹。

1.2 错误代价的不对称

这是 HITL 存在的根本原因——错误代价不对称：

低风险操作（查询订单）：
  自动执行成功 → 节省 3 秒
  自动执行失败 → 重查一次，代价 ≈ 0

高风险操作（退款 10 万元）：
  自动执行成功 → 节省 30 秒
  自动执行失败 → 资金损失 + 客诉 + 合规事故，代价 ≈ ∞

当"失败的代价"远大于"人工确认的成本"时，停下来问一句就是理性选择。这不是 Agent 能力不足，而是风险管理的常识。

1.3 三个硬约束

即便模型能力足够，HITL 仍不可省，因为有三类外部约束：

1. 合规约束：金融交易、医疗处方、生产变更，监管明确要求"人工审批留痕"
2. 信任约束：用户对"全自动删数据"本能抗拒，可解释的人机协同反而更易被采纳
3. 校准约束：模型置信度普遍过度自信，“我很确定"不等于"我真的对”

1.4 一个翻车案例

团队上线了"自动清理过期数据"的 Agent，规则是"删除 30 天前的日志"。
某次 Agent 把"30 天前"理解成了"30 天前的所有数据"，
连带删掉了刚备份完的归档表（因为归档时间戳早于 30 天）。
损失：6 小时业务中断，从异地备份恢复花费 12 万元。
根因：删除操作没有人工确认门，Agent 的"自信"直接变成了执行。

一句话：没有确认门的高风险操作，等于把刹车也交给了模型。

二、HITL 的三个核心问题

任何 HITL 方案都要回答三个问题，回答错任何一个都会失败：

① 何时问人？   → 问得太频繁，人工被淹没；问得太少，风险漏网
② 问什么？     → 问得太宽，人看不懂；问得太细，人懒得看
③ 人不在怎么办？→ 同步阻塞会卡死流程；直接执行又绕过了确认

本文的架构就是逐个击破这三个问题。

三、置信度路由：让 Agent 知道"自己几斤几两"

3.1 模型置信度为什么不可信？

直接拿模型的"我确定"当放行依据，是 HITL 设计最常见的坑。研究表明，大模型普遍过度自信：自评"95% 确定"的答案，实际正确率往往只有 70-80%。

模型自评置信度  →  实际正确率
   99%          →     91%
   90%          →     74%
   70%          →     52%

直接用 logprob 或模型自报的置信度做路由，会把大量错误当成"高置信"放行。

3.2 三层置信度估计

单一信号不可靠，组合三个独立信号做交叉校准：

def estimate_confidence(query: str, agent_fn, task_kind: str,
                        sample_n: int = 3) -> float:
    """三层置信度估计，输出 0-1 的校准置信度"""
    # L1: 单次生成的对数概率均值
    primary = agent_fn(query, return_logprobs=True)
    l1 = softmax_mean(primary.logprobs)

    # L2: 自一致性——多次采样看答案是否稳定
    samples = [agent_fn(query).answer for _ in range(sample_n)]
    l2 = agreement_ratio(samples)

    # L3: 历史基线——这类任务过去成功率多少
    l3 = history_baseline(task_kind)

    # 加权融合：L3 权重最高，因为它最不被模型"嘴硬"影响
    return 0.2 * l1 + 0.3 * l2 + 0.5 * l3

关键：L3（历史基线）权重最高。因为模型会"嘴硬"，但历史数据不会撒谎——如果这类任务历史上只有 60% 成功率，那不管模型这次多自信，置信度都该被压下来。

3.3 路由决策表

有了校准置信度，再叠加操作风险等级，得到四宫格路由：

                 置信度高(≥0.85)      置信度中(0.6-0.85)     置信度低(<0.6)
风险低(查询)      自动执行              自动执行              自动执行 + 标记复核
风险中(写入)      自动执行              确认门(轻)            确认门(重)
风险高(删除/转账)  审批门                审批门                接管门(转人工)

• 置信度高 + 风险低：直接放行，不打扰人工
• 置信度低 + 风险高：不要让 Agent 硬试，直接转人工接管
• 中间地带：用不同强度的确认门

这张表是 HITL 的灵魂——它把"要不要问人"从拍脑袋变成查表。

四、四种确认门

针对不同情境，设计四种确认门，强度依次递增：

门一：审批门（Approval Gate）

场景：执行前必须人工批准，不批就不动。用于不可逆操作。

@approval_gate(reason="即将删除 production.orders 全表")
def drop_table(table_name: str) -> str:
    return execute_sql(f"DROP TABLE {table_name}")

执行流：Agent 生成调用 → 拦截器挂起 → 推送审批到人 → 人批准才执行 → 人拒绝则回滚计划。

门二：确认门（Confirmation Gate）

场景：关键参数二次确认。用于"对是对，但参数可能搞错"的情况。

Agent 想退款，金额 = 100000 元
确认门弹出："确认退款金额？10 万元将退至账户 ****1234"
用户点"确认" → 执行
用户改成 10000 → 用新参数执行

与审批门的区别：确认门允许人修改参数，审批门只允许"是/否"。

门三：选择门（Choice Gate）

场景：存在多个合理方案，让人选一个。用于"答案不唯一"的决策点。

Agent 查到服务器 CPU 98%，给出三个处置方案：
  A. 扩容 2 个实例（成本 +，立即生效）
  B. 重启服务（免费，可能复发）
  C. 暂不处理，持续观察（风险累积）
请选择：[A] [B] [C]

选择门把 Agent 从"替人决策"变成"帮人整理选项"，这正是人最舒服的协作姿态。

门四：接管门（Takeover Gate）

场景：Agent 连续失败或置信度极低，主动放弃，转人工全程接手。

if attempt >= max_attempts or confidence < 0.3:
    handover_to_human(
        reason="置信度过低或重试耗尽",
        context=full_trajectory,
        suggestion="建议人工核查工具权限与数据完整性",
    )
    return

接管门是 HITL 的安全网——承认 Agent 有搞不定的事，比硬撑到崩盘更专业。

四门对比

五、异步确认与超时降级

5.1 同步阻塞的陷阱

最简单的 HITL 是同步阻塞：Agent 停下等用户点确认。但这在生产里会出大问题：

Agent 发起删除审批 → 等用户确认
用户下班了 → Agent 一直挂着
连接超时 → 任务失败 → 半成品状态残留

同步阻塞把"人的响应时间"绑进了"系统延迟"，而人的响应时间是不可控的（几分钟到几小时）。

5.2 异步 + 超时降级

正确做法是异步挂起 + 超时降级：

class PendingApproval:
    """挂起的审批，带超时降级策略"""
    def __init__(self, action, timeout_sec, on_timeout):
        self.action = action
        self.timeout_sec = timeout_sec
        self.on_timeout = on_timeout   # "block" / "degrade" / "abort"

    def resolve(self):
        # 推送到审批队列，立即返回 pending，不阻塞主流程
        enqueue_approval(self)
        return {"status": "pending", "approval_id": self.id}

三种超时策略，按风险选：

铁律：高风险操作只能用 block，绝不能"人没回就自动执行"——那等于没有确认门。

5.3 批量确认

逐条确认会淹没人工。把同类低风险操作合并成一次审批：

单条确认：Agent 要更新 20 条配置 → 弹 20 次确认 → 人工崩溃
批量确认：Agent 汇总"将更新 20 条配置，预览如下…"→ 弹 1 次 → 人工一次批完

批量确认把"确认次数"从 O(n) 降到 O(1)，是 HITL 不拖累效率的关键。

六、从零搭建 HITL 中间件

把上述设计落成一个可复用的 Python 中间件：

#!/usr/bin/env python3
"""
Agent 人机协作中间件
置信度路由 + 四种确认门 + 异步超时降级
"""

import time
import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Optional


# ============ 风险与置信度 ============

class RiskLevel(Enum):
    LOW = "low"        # 查询、摘要
    MEDIUM = "medium"  # 写入、配置变更
    HIGH = "high"      # 删除、转账、部署


@dataclass
class ActionRisk:
    """操作的风险标注：由工具注册时声明"""
    tool: str
    risk: RiskLevel
    reversible: bool = True       # 是否可回滚
    blast_radius: str = "local"   # local / service / global


def estimate_confidence(query: str, task_kind: str,
                        history: dict[str, float]) -> float:
    """三层置信度估计（此处简化为历史基线主导）"""
    # 真实实现应融合 logprob + 自一致性 + 历史基线
    baseline = history.get(task_kind, 0.7)
    return baseline


def route_decision(confidence: float, risk: RiskLevel) -> str:
    """四宫格路由：返回执行模式"""
    table = {
        (RiskLevel.LOW,    "high"): "auto",
        (RiskLevel.LOW,    "mid"):  "auto",
        (RiskLevel.LOW,    "low"):  "auto",
        (RiskLevel.MEDIUM, "high"): "auto",
        (RiskLevel.MEDIUM, "mid"):  "confirm_light",
        (RiskLevel.MEDIUM, "low"):  "confirm_heavy",
        (RiskLevel.HIGH,   "high"): "approve",
        (RiskLevel.HIGH,   "mid"):  "approve",
        (RiskLevel.HIGH,   "low"):  "takeover",
    }
    band = "high" if confidence >= 0.85 else ("mid" if confidence >= 0.6 else "low")
    return table.get((risk, band), "takeover")


# ============ 确认门 ============

class GateType(Enum):
    APPROVAL = "approval"
    CONFIRM = "confirm"
    CHOICE = "choice"
    TAKEOVER = "takeover"


@dataclass
class ConfirmationRequest:
    req_id: str
    gate: GateType
    action_desc: str
    params: dict = field(default_factory=dict)
    options: list[str] = field(default_factory=list)
    created_at: float = field(default_factory=time.time)
    resolved: bool = False
    result: Any = None


class HumanChannel:
    """与人工交互的通道（生产中接 IM/工单/审批平台）"""

    def __init__(self):
        self.pending: dict[str, ConfirmationRequest] = {}

    def ask(self, req: ConfirmationRequest,
            timeout_sec: int, on_timeout: str) -> Any:
        """发起确认，异步等待人工响应"""
        self.pending[req.req_id] = req
        # 真实实现：推送到飞书/钉钉/审批平台，然后阻塞等待回调
        response = self._wait_or_timeout(req, timeout_sec)
        if response is None:
            return self._handle_timeout(req, on_timeout)
        return response

    def _wait_or_timeout(self, req: ConfirmationRequest,
                         timeout_sec: int) -> Optional[Any]:
        # 占位：真实为事件等待
        return {"approved": True, "params": req.params}

    def _handle_timeout(self, req: ConfirmationRequest, strategy: str) -> Any:
        if strategy == "block":
            return self.ask(req, 3600, "block")   # 继续等
        if strategy == "abort":
            return {"approved": False, "reason": "审批超时，放弃执行"}
        return {"approved": False, "reason": "超时降级", "degraded": True}


# ============ HITL Agent 包装 ============

class HITLAgent:
    """给任意 Agent 包一层人机协作控制"""

    def __init__(self, agent_fn: Callable, tools: dict,
                 risk_map: dict[str, ActionRisk], channel: HumanChannel,
                 history: dict[str, float]):
        self.agent_fn = agent_fn
        self.tools = tools
        self.risk_map = risk_map
        self.channel = channel
        self.history = history

    def _execute_tool(self, tool_name: str, args: dict,
                      confidence: float) -> Any:
        risk = self.risk_map[tool_name].risk
        mode = route_decision(confidence, risk)

        if mode == "auto":
            return self.tools[tool_name](**args)

        if mode == "takeover":
            return self.channel.ask(ConfirmationRequest(
                req_id=str(uuid.uuid4()),
                gate=GateType.TAKEOVER,
                action_desc=f"置信度 {confidence:.2f} 过低，转人工处理 {tool_name}",
                params=args,
            ), timeout_sec=86400, on_timeout="block")

        if mode == "approve":
            return self.channel.ask(ConfirmationRequest(
                req_id=str(uuid.uuid4()),
                gate=GateType.APPROVAL,
                action_desc=f"批准执行 {tool_name}？参数：{args}",
                params=args,
            ), timeout_sec=3600, on_timeout="block")

        if mode.startswith("confirm"):
            return self.channel.ask(ConfirmationRequest(
                req_id=str(uuid.uuid4()),
                gate=GateType.CONFIRM,
                action_desc=f"确认 {tool_name} 参数：{args}",
                params=args,
            ), timeout_sec=1800, on_timeout="abort")

        return self.tools[tool_name](**args)

    def run(self, query: str, task_kind: str) -> dict:
        confidence = estimate_confidence(query, task_kind, self.history)
        plan = self.agent_fn(query)           # Agent 规划出工具调用序列

        results = []
        for step in plan.get("steps", []):
            outcome = self._execute_tool(
                step["tool"], step["args"], confidence
            )
            results.append({"tool": step["tool"], "outcome": outcome})
            if isinstance(outcome, dict) and not outcome.get("approved", True):
                return {"success": False, "aborted_at": step["tool"],
                        "reason": outcome.get("reason"), "results": results}

        return {"success": True, "confidence": confidence, "results": results}


# ============ 完整流程 ============

if __name__ == "__main__":
    tools = {
        "query_order":  lambda oid: {"status": "paid"},
        "refund":       lambda order, amt: {"refund_id": "r_1"},
        "drop_table":   lambda table: {"dropped": table},
    }
    risk_map = {
        "query_order": ActionRisk("query_order", RiskLevel.LOW),
        "refund":      ActionRisk("refund", RiskLevel.HIGH, reversible=False),
        "drop_table":  ActionRisk("drop_table", RiskLevel.HIGH, reversible=False,
                                  blast_radius="global"),
    }
    history = {"refund_task": 0.72, "query_task": 0.95}

    def dummy_agent(query):
        if "退款" in query:
            return {"steps": [{"tool": "query_order", "args": {"oid": 123}},
                              {"tool": "refund", "args": {"order": 123, "amt": 100000}}]}
        return {"steps": []}

    agent = HITLAgent(dummy_agent, tools, risk_map, HumanChannel(), history)
    print(agent.run("给订单 123 退款 10 万元", "refund_task"))

6.1 关键设计要点

要点一：风险标注必须显式声明，不能靠模型自己判断

# ❌ 让模型自己说"这个操作危不危险"——它会低估
risk = llm("这个操作风险高吗？")

# ✅ 工具注册时由人显式标注，运行时查表
risk_map["drop_table"] = ActionRisk("drop_table", RiskLevel.HIGH)

模型对"风险"的感知不可靠（它会觉得"删个表而已嘛"），风险等级必须由工程团队预先标注。

要点二：置信度要用历史基线校准，不能信模型自报

# ❌ 直接用模型自评
confidence = model_self_reported_confidence

# ✅ 历史基线主导
confidence = 0.2 * logprob + 0.3 * self_consistency + 0.5 * history_baseline

要点三：高风险操作的超时策略只能是 block

# ❌ 高风险操作超时自动执行——等于没确认门
ask(req, timeout=3600, on_timeout="auto_execute")

# ✅ 高风险超时继续等，绝不自动放行
ask(req, timeout=3600, on_timeout="block")

七、实测与权衡

7.1 效果数据

在一个含 2000 次工具调用的运维 Agent 上对比：

误执行率降 94%，代价是 8% 的人工介入和 5% 的延迟增加——这笔账非常划算。

7.2 自动化率 vs 安全性的权衡

HITL 不是越多越好。确认门过密，人工会被淹没，最终走向两个极端之一：要么人麻木地全点"同意"（确认门失效），要么人抱怨"什么都问我还要 Agent 干嘛"。

确认门密度过低 → 风险漏网，事故频发
确认门密度过高 → 人工疲劳，确认形同虚设
最优解 → 只在"高风险 × 低置信"象限设门，其余放行

这就是为什么前面路由表把"低风险"列全部设为 auto——把人工预算花在刀刃上。

7.3 实施难度与可行性评估

落地建议：分三步走，先上"风险标注 + 审批门"（1 周可见效，挡住 90% 事故），再上"置信度路由"（优化人工密度），最后上"异步降级 + 接管门"（完善长尾）。第一步的 ROI 最高。

八、总结

Agent 的成熟，不在于它能全自动做多少事，而在于它知道哪些事该自己做、哪些事该交还给人。HITL 的本质是给 Agent 装上"刹车与方向盘"：

• 刹车：审批门、确认门——该停的时候停得住
• 方向盘：选择门、接管门——该交的时候交得出
• 仪表：置信度路由——知道自己几斤几两

一个会主动停下问人的 Agent，不是能力弱，而是有边界感——这恰恰是它能被放心送上生产的前提。

欢迎在评论区分享你的 HITL 实践：你是怎么决定"什么时候打断 Agent"的？

本文承接前七篇（记忆/协作/工具/知识/安全/评估/规划），代码示例遵循 snake_case 约定，置信度路由表与确认门中间件可通过 recall_history(op="search", query="HITL 确认门 置信度路由") 获取完整实现。