Agent 与人类协作的交互设计

人机协作模式、审批流程与升级协议的工程化实践

人机协作的设计原则

Agent 不是取代人类，而是与人类协作。关键设计原则：

  1. 透明性（Transparency）：Agent 必须让人类理解它在做什么、为什么这么做
  2. 可控性（Controllability）：人类随时可以介入、修改、中断 Agent 的行为
  3. 适度自主（Calibrated Autonomy）：根据风险和置信度动态调整自主程度
  4. 优雅降级（Graceful Degradation）：当 Agent 不确定时，应主动寻求人类帮助
自主程度光谱：

  完全人工           HITL 协作           完全自主
  ─────────────────────────────────────────────
  │                  │                   │
  每步确认         关键节点确认       全自动执行
  适用：学习期     适用：生产环境     适用：低风险重复任务

一、HITL（Human-in-the-Loop）模式

三种 HITL 粒度

┌─────────────────────────────────────────────────────────────┐
│                   HITL 粒度选择                              │
├──────────────┬──────────────────┬───────────────────────────┤
│ 步骤级 HITL   │ 节点级 HITL      │ 异常级 HITL               │
│ (Step-level)  │ (Checkpoint)     │ (Exception-only)         │
├──────────────┼──────────────────┼───────────────────────────┤
│ 每个工具调用  │ 关键决策点        │ 仅在异常时                │
│ 都需确认      │ 需要确认          │ 请求介入                  │
├──────────────┼──────────────────┼───────────────────────────┤
│ 最安全       │ 平衡安全与效率    │ 最高效                    │
│ 最低效       │                  │ 需要完善的异常检测          │
├──────────────┼──────────────────┼───────────────────────────┤
│ 适用：       │ 适用：            │ 适用：                    │
│ 高风险操作   │ 生产环境标准      │ 低风险重复任务             │
│ 新 Agent 上线│                  │ 成熟 Agent                 │
└──────────────┴──────────────────┴───────────────────────────┘

节点级 HITL 实现

class HITLController:
    """Human-in-the-Loop 控制器"""

    def __init__(self, policy: HITLPolicy):
        self.policy = policy
        self.pending_approvals = {}

    def check_approval_needed(self, action: AgentAction) -> bool:
        """判断当前动作是否需要人工审批"""
        # 1. 强制审批列表
        if action.type in self.policy.always_approve:
            return True

        # 2. 风险评估
        risk = self.assess_risk(action)
        if risk.level in ("high", "critical"):
            return True

        # 3. 置信度检查
        if action.confidence < self.policy.confidence_threshold:
            return True

        # 4. 成本检查
        if action.estimated_cost > self.policy.cost_threshold:
            return True

        return False

    def request_approval(self, action: AgentAction) -> ApprovalRequest:
        """生成审批请求"""
        request = ApprovalRequest(
            id=str(uuid4()),
            action=action,
            context=self._build_context(action),
            risk_assessment=self.assess_risk(action),
            options=[
                ApprovalOption("approve", "批准执行"),
                ApprovalOption("modify", "修改后执行"),
                ApprovalOption("reject", "拒绝"),
                ApprovalOption("delegate", "转交其他人"),
            ],
            timeout_minutes=self.policy.approval_timeout,
            created_at=datetime.now()
        )

        self.pending_approvals[request.id] = request
        self._notify_approver(request)
        return request

    def _build_context(self, action: AgentAction) -> dict:
        """构建审批上下文，帮助人类快速决策"""
        return {
            "what": action.description,
            "why": action.reasoning,
            "impact": self._estimate_impact(action),
            "reversible": action.is_reversible,
            "alternatives": self._suggest_alternatives(action),
            "history": self._get_similar_past_decisions()
        }

HITL 策略配置

class HITLPolicy:
    """HITL 策略定义"""

    # 必须审批的操作
    always_approve = [
        "database_write",
        "database_delete",
        "file_delete",
        "deploy_production",
        "send_email",
        "create_account",
        "modify_permissions",
        "financial_transaction",
    ]

    # 置信度阈值（低于此值需审批）
    confidence_threshold = 0.8

    # 成本阈值（超过此值需审批）
    cost_threshold_usd = 1.0

    # 审批超时（分钟）
    approval_timeout = 30

    # 超时后的默认动作
    timeout_action = "reject"  # reject / approve / escalate

二、审批流程设计

审批工作流

Agent 提出动作
     │
     v
┌──────────────┐
│ 需要审批？    │──→ No ──→ 直接执行
└──────┬───────┘
       │ Yes
       v
┌──────────────┐
│ 生成审批请求  │
│ - 动作描述    │
│ - 风险评估    │
│ - 影响分析    │
│ - 替代方案    │
└──────┬───────┘
       │
       v
┌──────────────┐
│ 通知审批人    │──→ 邮件 / Slack / 应用内
└──────┬───────┘
       │
       v
┌──────────────┐
│ 等待决策      │──→ 超时 ──→ 执行默认动作
└──────┬───────┘
       │
  ┌────┼────┬─────┐
  v    v    v     v
批准  修改  拒绝  转交
  │    │    │     │
  v    v    v     v
执行  修改  记录  新审批人
     后执行 原因

审批界面信息架构

class ApprovalUI:
    """审批界面的信息组织"""

    def render_approval_card(self, request: ApprovalRequest) -> dict:
        """生成审批卡片的数据结构"""
        return {
            # 第一眼：这是什么
            "headline": request.action.description,
            "risk_badge": self._risk_badge(request.risk_assessment),

            # 第二眼：为什么
            "reasoning": request.action.reasoning,

            # 第三眼：影响是什么
            "impact": {
                "scope": request.context["impact"]["scope"],
                "reversible": request.context["reversible"],
                "affected_resources": request.context["impact"]["resources"]
            },

            # 决策辅助
            "similar_past_decisions": request.context["history"],
            "alternatives": request.context["alternatives"],

            # 操作区
            "actions": [
                {"label": "批准", "style": "primary",
                 "shortcut": "Enter"},
                {"label": "修改", "style": "secondary",
                 "shortcut": "M"},
                {"label": "拒绝", "style": "danger",
                 "shortcut": "R"},
            ],

            # 时间压力
            "deadline": request.created_at + timedelta(
                minutes=request.timeout_minutes
            ),
            "auto_action": request.policy.timeout_action
        }

    def _risk_badge(self, risk) -> dict:
        badges = {
            "low": {"text": "LOW", "color": "green"},
            "medium": {"text": "MED", "color": "orange"},
            "high": {"text": "HIGH", "color": "red"},
            "critical": {"text": "CRITICAL", "color": "darkred"}
        }
        return badges.get(risk.level, badges["medium"])

三、升级协议（Escalation Protocol）

升级触发条件

class EscalationTrigger:
    """升级触发条件定义"""

    TRIGGERS = [
        # 置信度不足
        EscalationRule(
            name="low_confidence",
            condition=lambda ctx: ctx.agent_confidence < 0.5,
            level="L1",
            message="Agent 对当前决策的置信度不足 50%"
        ),

        # 连续失败
        EscalationRule(
            name="repeated_failure",
            condition=lambda ctx: ctx.consecutive_failures >= 3,
            level="L2",
            message="连续 3 次尝试失败"
        ),

        # 超出时间预算
        EscalationRule(
            name="time_budget_exceeded",
            condition=lambda ctx: ctx.elapsed > ctx.time_budget * 0.8,
            level="L1",
            message="已使用 80% 的时间预算"
        ),

        # 超出成本预算
        EscalationRule(
            name="cost_budget_exceeded",
            condition=lambda ctx: ctx.cost > ctx.cost_budget * 0.9,
            level="L2",
            message="已使用 90% 的成本预算"
        ),

        # 安全事件
        EscalationRule(
            name="security_incident",
            condition=lambda ctx: ctx.security_violations > 0,
            level="L3",
            message="检测到安全违规"
        ),

        # 歧义检测
        EscalationRule(
            name="ambiguous_task",
            condition=lambda ctx: ctx.interpretation_count > 2,
            level="L1",
            message="任务存在多种合理解释"
        ),
    ]

升级层级

L0: Agent 自行处理
  │  重试、降级、换工具
  │
L1: 信息补充请求
  │  向用户请求澄清或补充信息
  │  不中断执行流程
  │
L2: 决策审批请求
  │  暂停执行，等待人类做关键决策
  │  提供选项和建议
  │
L3: 紧急中断
  │  立即停止所有操作
  │  通知所有相关人员
  │  保存现场状态用于事后分析

class EscalationManager:
    """升级管理器"""

    def escalate(self, level: str, context: dict) -> EscalationResponse:
        if level == "L1":
            return self._request_info(context)
        elif level == "L2":
            return self._request_decision(context)
        elif level == "L3":
            return self._emergency_halt(context)

    def _request_info(self, context: dict) -> EscalationResponse:
        """L1：请求补充信息"""
        question = self._generate_clarification_question(context)
        return EscalationResponse(
            level="L1",
            action="ask_user",
            message=question,
            blocking=False,  # 不阻塞，继续执行其他不依赖此信息的任务
            fallback_action=context.get("default_assumption")
        )

    def _request_decision(self, context: dict) -> EscalationResponse:
        """L2：请求决策"""
        options = self._generate_decision_options(context)
        return EscalationResponse(
            level="L2",
            action="request_approval",
            message=f"需要您的决策：{context['description']}",
            options=options,
            blocking=True,   # 阻塞，等待决策
            timeout_minutes=30,
            timeout_action="pause"
        )

    def _emergency_halt(self, context: dict) -> EscalationResponse:
        """L3：紧急中断"""
        # 保存现场
        self._save_state(context)
        # 停止所有执行
        self._halt_all_agents()
        # 通知
        self._send_alert(
            severity="critical",
            message=f"Agent 紧急中断: {context['reason']}",
            channels=["slack", "email", "sms"]
        )
        return EscalationResponse(
            level="L3",
            action="halted",
            state_snapshot=context["state"]
        )

四、反馈循环设计

显式反馈

class FeedbackCollector:
    """用户反馈收集器"""

    def collect_after_action(self, action: AgentAction,
                              result: ActionResult) -> None:
        """在关键动作完成后收集反馈"""
        feedback = {
            "action_id": action.id,
            "timestamp": datetime.now(),
            "result_summary": result.summary,
        }

        # 根据场景选择反馈方式
        if action.type in ("code_write", "document_generate"):
            # 内容生成类：请求质量评分
            feedback["type"] = "quality_rating"
            feedback["prompt"] = (
                "生成的内容质量如何？"
                "\n1=很差 2=较差 3=一般 4=较好 5=很好"
            )

        elif action.type in ("search", "retrieval"):
            # 检索类：请求相关性评价
            feedback["type"] = "relevance_check"
            feedback["prompt"] = "检索结果是否满足需求？"

        elif action.type in ("decision", "recommendation"):
            # 决策类：请求决策认可
            feedback["type"] = "decision_approval"
            feedback["prompt"] = (
                "您认同这个建议吗？"
                "\n可选：采纳 / 修改后采纳 / 不采纳"
            )

        self._store_feedback(feedback)

隐式反馈

class ImplicitFeedbackTracker:
    """隐式反馈追踪"""

    def track(self, event: UserEvent):
        """通过用户行为推断反馈"""
        signals = []

        # 用户是否修改了 Agent 的输出
        if event.type == "edit_agent_output":
            signals.append({
                "signal": "output_edited",
                "interpretation": "输出未完全满足需求",
                "severity": self._calc_edit_severity(
                    event.original, event.edited
                )
            })

        # 用户是否重新描述了同一个任务
        if event.type == "task_rephrase":
            signals.append({
                "signal": "task_rephrased",
                "interpretation": "Agent 可能误解了任务",
                "original": event.original_task,
                "rephrased": event.new_task
            })

        # 用户是否回退了 Agent 的操作
        if event.type == "undo":
            signals.append({
                "signal": "action_undone",
                "interpretation": "Agent 的操作不符合预期",
                "undone_action": event.action_id
            })

        # 用户沉默时间过长
        if event.type == "long_pause" and event.duration_s > 30:
            signals.append({
                "signal": "long_pause",
                "interpretation": "用户可能困惑或在思考",
                "duration_s": event.duration_s
            })

        return signals

五、交互设计模式

模式一：渐进式披露（Progressive Disclosure）

用户请求："帮我分析这份报告"

第一层回复（摘要）：
  "报告核心结论：收入同比增长 15%，成本增长 8%。"
  [查看详细分析] [查看数据来源]

第二层（展开详情）：
  "详细分析：
   1. 收入结构变化...
   2. 成本驱动因素..."
  [查看原始数据] [查看计算过程]

第三层（原始数据）：
  完整的数据表格和计算过程

模式二：确认前预览（Preview before Commit）

Agent 准备执行：发送邮件给 [email protected]

预览界面：
┌──────────────────────────────────────┐
│ 收件人: [email protected]             │
│ 主题: Q4 项目进展报告                │
│ 正文:                                │
│   各位好，                           │
│   以下是 Q4 的项目进展...            │
│                                      │
│ 附件: progress_report.pdf            │
│                                      │
│ [发送]  [修改]  [取消]               │
└──────────────────────────────────────┘

模式三：置信度可视化

Agent 的分析结果：

 问题原因判断：
 ┌─────────────────────────────────────┐
 │ 数据库连接超时   ████████████░░ 85%  │  <-- 高置信度，可以直接处理
 │ 网络波动         ████░░░░░░░░░ 35%  │  <-- 低置信度，需要更多证据
 │ 服务器过载       ██░░░░░░░░░░░ 15%  │  <-- 可能性低
 └─────────────────────────────────────┘

 建议：优先排查数据库连接超时（置信度 85%）。
 如果您认为实际原因不同，请告诉我，我会调整分析方向。

模式四：检查点恢复（Checkpoint Recovery）

长任务执行中，设置检查点供用户审查：

任务：重构用户认证模块

 [v] Step 1: 分析现有代码结构              -- 已完成
 [v] Step 2: 设计新的认证架构              -- 已完成
 [ ] Step 3: 实现新的 JWT 认证             -- 等待确认
     --> [查看架构设计] [批准继续] [修改方案]
 [ ] Step 4: 迁移数据
 [ ] Step 5: 回归测试
 [ ] Step 6: 部署

当前状态：等待 Step 3 审批
预计剩余时间：2 小时
已消耗成本：$3.50

六、通知与状态同步

通知策略

class NotificationManager:
    """Agent 状态通知管理"""

    NOTIFICATION_RULES = {
        # 高优先级：立即通知
        "task_blocked": {
            "priority": "high",
            "channels": ["push", "slack"],
            "sound": True
        },
        "security_alert": {
            "priority": "critical",
            "channels": ["push", "slack", "sms"],
            "sound": True
        },
        "approval_needed": {
            "priority": "high",
            "channels": ["push", "slack"],
            "sound": False
        },

        # 中优先级：批量通知
        "task_completed": {
            "priority": "medium",
            "channels": ["slack"],
            "batch": True,
            "batch_interval_minutes": 5
        },

        # 低优先级：静默记录
        "progress_update": {
            "priority": "low",
            "channels": ["dashboard"],
            "silent": True
        }
    }

    def notify(self, event_type: str, context: dict):
        """根据事件类型选择通知方式"""
        rule = self.NOTIFICATION_RULES.get(event_type)
        if not rule:
            return

        message = self._format_message(event_type, context)

        if rule.get("batch"):
            self._add_to_batch(event_type, message)
        else:
            for channel in rule["channels"]:
                self._send(channel, message, rule["priority"])

交互设计检查清单

Agent 交互设计审查：
- [ ] Agent 的每个动作都有清晰的意图说明
- [ ] 高风险操作有确认流程
- [ ] 用户可以随时中断 Agent
- [ ] 用户可以回退 Agent 的操作
- [ ] 低置信度的决策向用户透明展示
- [ ] 长任务有进度反馈和检查点
- [ ] 升级路径明确（何时自动、何时求助）
- [ ] 通知不过度打扰（分级、可静音）
- [ ] 审批界面信息充分（影响、替代方案、时限）
- [ ] 隐式反馈被收集并用于改进

参考资料

  • Google PAIR (People + AI Research) 人机协作指南
  • Apple Human Interface Guidelines - AI 交互设计
  • Microsoft HAX (Human-AI Experience) Toolkit
  • Nielsen Norman Group: AI UX 研究报告
  • Anthropic HITL 模式文档

Maurice | [email protected]