llm

人机协作模式（Human-in-the-Loop）：让 Agent 在关键时刻停下来，等待人类确认

深度分析 Agent 系统中人类介入的必要性与设计模式：审批流（Approval）、审核反馈（Review）、接管仲裁（Override）、触发条件与降级策略。涵盖人机协作的 UX 设计、上下文压缩技术、以及监管合规要求（金融/医疗/工业控制的关键决策审计）。附完整的 HITL 审批队列实现、审计日志设计与多层级审批工作流代码。

Leeting Yan 2026-08-08 4 分钟阅读 1863 字

⚠️ 在支付、删除、医疗诊断、法律建议等场景，永远不要把最终决定权交给 Agent。
HITL 不是可选优化——是生产环境的底线安全策略。

1. 为什么必须有人机协作？

据 2024 年 Anthropic 的研究报告，即便在简单任务中，Agent 的自主决策准确率也只有 78%，而在以下场景中，错误率显著上升：

场景	错误类型	典型后果
金融交易	订单量单位理解错误	多买 10 倍
医疗诊断	罕见症状误判	错误用药建议
法律文件	管辖权条款遗漏	合同无效
数据库操作	DROP 语句误触发	数据丢失

HITL 的三种介入时机：

执行前（Pre-approval）：操作实际生效前，必须有人批准
执行中（Inline review）：Agent 自主执行，但关键步骤触发审查
执行后（Post-hoc audit）：操作已完成，留痕供事后审计

2. 人机协作的四大模式

2.1 审批流（Approval）

Agent 生成操作 → 推送给审批人 → 审批人确认/拒绝 → Agent 执行/回滚

适用场景：资金转账、数据删除、合同签署、权限变更

class ApprovalGate:
    REQUIRED_APPROVALS = {
        "transfer_money": "financial_controller",
        "delete_user": "it_admin",
        "deploy_production": "tech_lead",
    }

    def check(self, action: str, context: dict) -> str:
        if action in self.REQUIRED_APPROVALS:
            required_role = self.REQUIRED_APPROVALS[action]
            return f"PENDING_APPROVAL:{required_role}:{uuid4()}"
        return "AUTO_APPROVED"

2.2 审核反馈（Review）

Agent 自主完成输出，但人类可以标记、纠正或拒绝。

class ReviewCycle:
    def submit_output(self, task_id: str, output: dict):
        # 通知人类审核者
        self.notify_reviewer(task_id, output)
        # 设置审核截止日期
        self.set_deadline(task_id, hours=48)
        # 如果超时未审核，走默认策略
        self.set_timeout_fallback(task_id, "approved")

2.3 接管仲裁（Override）

人类可以中断正在执行的 Agent 并接管控制。

class HumanOverride:
    def __init__(self):
        self.interrupt_flags = {}

    def request_interrupt(self, agent_id: str):
        self.interrupt_flags[agent_id] = True

    def check_interrupt(self, agent_id: str):
        return self.interrupt_flags.get(agent_id, False)

    def resume_with_human_input(self, agent_id: str, human_input: str):
        self.interrupt_flags[agent_id] = False
        return {"override": True, "human_input": human_input}

2.4 推荐决策（Suggestion）

Agent 给出推荐，但人类最终拍板。这是最松散的协作模式。

class SuggestionMode:
    def generate_recommendation(self, context):
        prompt = f"""
你是助手，请给出 3 个建议并说明利弊。
不要直接操作，只做推荐。
当前场景: {context}
"""
        return self.llm.invoke(prompt)

3. 触发条件设计（何时介入？）

3.1 基于风险分数

def risk_score(action_type, parameters, user_role):
    score = 0
    # 金额阈值
    if parameters.get("amount", 0) > 10000:
        score += 50
    # 权限等级
    if action_type in ["delete_db", "grant_admin"]:
        score += 80
    # 用户可信度
    if user_role == "intern":
        score += 20
    return score

def should_escalate(score: int) -> bool:
    return score > 60  # 超过阈值必须人工审批

3.2 基于置信度

@dataclass
class AgentOutput:
    answer: str
    confidence: float  # 0-1
    source_references: list[str]

def check_confidence(output: AgentOutput):
    if output.confidence < 0.7:
        return "ESCALATE_TO_HUMAN"
    return "AUTO_APPROVE"

3.3 关键词触发

SENSITIVE_KEYWORDS = [
    "root", "admin", "password", "credit card",
    "DELETE", "DROP TABLE", "TRUNCATE",
]

def contains_sensitive(text: str) -> bool:
    return any(kw in text.lower() for kw in SENSITIVE_KEYWORDS)

4. 完整的 HITL 审批队列实现

import asyncio
from datetime import datetime, timedelta
from enum import Enum

class ApprovalStatus(Enum):
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"
    EXPIRED = "expired"

@dataclass
class ApprovalRequest:
    id: str
    agent_id: str
    action: str
    parameters: dict
    requester:str
    required_role: str
    status: ApprovalStatus
    created_at: datetime
    expires_at: datetime
    reviewer_comment: str = ""

class ApprovalService:
    def __init__(self, db):
        self.db = db
        self.notifications = NotificationService()

    async def submit(self, request: ApprovalRequest) -> str:
        # 保存到数据库
        await self.db.approvals.insert_one(request.to_dict())
        # 发送通知给对应角色
        await self.notifications.send(
            to_role=request.required_role,
            message=f"新的审批请求 #{request.id}: {request.action}"
        )
        return request.id

    async def approve(self, approval_id: str, reviewer: str, comment=""):
        await self.db.approvals.update_one(
            {"id": approval_id},
            {"$set": {
                "status": ApprovalStatus.APPROVED.value,
                "reviewed_by": reviewer,
                "reviewed_at": datetime.utcnow(),
                "reviewer_comment": comment,
            }}
        )
        # 通知 Agent 继续执行
        await self.notifications.send(
            to_agent=requester_id,
            message={"status": "approved", "approval_id": approval_id}
        )

    async def reject(self, approval_id: str, reviewer: str, reason=""):
        await self.db.approvals.update_one(
            {"id": approval_id},
            {"$set": {
                "status": ApprovalStatus.REJECTED.value,
                "reviewed_by": reviewer,
                "reviewed_at": datetime.utcnow(),
                "rejection_reason": reason,
            }}
        )

5. 上下文压缩：给人类看的摘要

Agent 的运行日志通常很长，人类不可能逐条审阅。

class ContextCompressor:
    def compress(self, agent_history: list) -> str:
        # 结构化摘要 Prompt
        summary_prompt = f"""
请将以下 Agent 执行历史压缩为审批摘要，包含：
1. 做了什么（一句话）
2. 为什么做（目标）
3. 涉及的数据/金额/权限
4. 可能的风险点
5. 建议的决策（批准/拒绝/需要更多信息）

历史: {agent_history[-20:]}  （最近 20 步）
"""
        return self.llm.invoke(summary_prompt)

# 示例输出
"""
📋 审批摘要 #REQ-8823
━━━━━━━━━━━━━━━━━━━━━━
做了什么：尝试将用户 Alice 的数据迁移到生产数据库。
为什么做：用户要求从测试环境迁移到生产环境。
涉及权限：生产数据库写权限（root 级操作）。
风险点：⚠️ 可能覆盖现有生产数据。建议先备份。
建议：拒绝，要求先提供数据备份计划。
"""

6. 审计日志设计（不可篡改）

import hashlib
from dataclasses import asdict

class AuditLog:
    def __init__(self, storage):
        self.storage = storage
        self.last_hash = "0" * 64

    async def append(self, event: dict):
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "event": event,
            "prev_hash": self.last_hash,
            "event_hash": "",  # will be computed
        }
        # 计算链式哈希
        data = json.dumps(entry, sort_keys=True)
        entry["event_hash"] = hashlib.sha384(
            (self.last_hash + data).encode()
        ).hexdigest()

        await self.storage.insert(entry)
        self.last_hash = entry["event_hash"]

    async def verify_chain(self) -> bool:
        """验证审计日志链是否完整（未被篡改）"""
        entries = await self.storage.find_all().sort("timestamp")
        for i, entry in enumerate(entries):
            data = json.dumps({k: v for k, v in entry.items() if k != "event_hash"}, sort_keys=True)
            expected_hash = hashlib.sha384(
                (entry["prev_hash"] + data).encode()
            ).hexdigest()
            if expected_hash != entry["event_hash"]:
                return False
        return True

7. 按行业倍增阈值设计

行业	必须 HITL 的场景	特殊合规要求
金融	转账>¥1万、开立新账户、权限变更	完整审计链 7 年、操作留痕
医疗	诊断建议、开药、手术排程	HIPAA / GDPR、医生签字确认
法律	合同条款修改、诉讼建议	律师执业资格验证、决策归因
工业	设备停机、高温高压操作	SCADA 系统集成、人工冗余开关
教育	成绩修改、减免学费	多级审批、家长知情同意

8. 总结：HITL 架构原则

┌─────────────────────────────────────┐
│ 1. 默认不信任（Defense in Depth）   │
│ 2. 人类只做决策，不做信息收集       │
│ 3. 上下文压缩，30 秒内能看懂        │
│ 4. 设置明确的 Timeout 和降级策略    │
│ 5. 所有操作不可逆之前必须有人确认   │
│ 6. 审计日志必须完整、不可篡改       │
└─────────────────────────────────────┘

📂 相关专题：
AI 智能体架构设计 — 记忆、工具、执行组件详解
Agent 工作流编排设计 — 暂停/恢复机制实现
LLM 安全与评估 — 幻觉检测、红队测试、对齐

← 上一篇 多智能体协作（Multi-Agent）：从团队协作到自主组织的技术实现 下一篇 → MCP 服务端开发实战：从 SQLite 查询服务到文件系统与 GitHub API 完整实现

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页