Agent 与人类协作的交互设计
原创
灵阙教研团队
S 精选 进阶 |
约 10 分钟阅读
更新于 2026-02-28 AI 导读
Agent 与人类协作的交互设计 人机协作模式、审批流程与升级协议的工程化实践 人机协作的设计原则 Agent 不是取代人类，而是与人类协作。关键设计原则： 透明性（Transparency）：Agent 必须让人类理解它在做什么、为什么这么做 可控性（Controllability）：人类随时可以介入、修改、中断 Agent 的行为 适度自主（Calibrated...
Agent 与人类协作的交互设计
人机协作模式、审批流程与升级协议的工程化实践
人机协作的设计原则
Agent 不是取代人类，而是与人类协作。关键设计原则：
- 透明性（Transparency）：Agent 必须让人类理解它在做什么、为什么这么做
- 可控性（Controllability）：人类随时可以介入、修改、中断 Agent 的行为
- 适度自主（Calibrated Autonomy）：根据风险和置信度动态调整自主程度
- 优雅降级（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 准备执行：发送邮件给 team@company.com
预览界面：
┌──────────────────────────────────────┐
│ 收件人: team@company.com │
│ 主题: 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 | maurice_wen@proton.me