第33章 人机协作模式
"最好的AI系统不是替代人类，而是放大人类的能力。" —— Fei-Fei Li
33.1 概述
Agent系统的最终目标不是完全自主运行，而是与人类高效协作。在大多数实际应用中，完全自主的Agent要么不可靠（安全风险），要么不必要（人类参与能提升结果质量）。人机协作模式解决的核心问题是：在什么时机、以什么方式让人类参与Agent的工作流程。
本章探讨六种人机协作模式，它们按照人类参与的深度分为三个层次：
- Human-in-the-Loop：人类深度参与，每次关键决策都需要人类确认
- Human-on-the-Loop：人类监督运行，异常时介入
- Human-over-the-Loop：人类高层控制，Agent自主执行
此外还有三种补充模式：Approval Gate（审批关卡）、Delegation（任务委派）和Escalation（问题升级），它们分别从不同角度丰富了人机协作的工具箱。
选择协作模式的关键考量：
| 考量维度 | Human-in-the-Loop | Human-on-the-Loop | Human-over-the-Loop |
|---|---|---|---|
| 人类参与频率 | 每次关键决策 | 异常时介入 | 仅设定目标 |
| 适用风险等级 | 高 | 中 | 低 |
| 人类时间成本 | 高 | 中 | 低 |
| Agent自主性 | 低 | 中 | 高 |
| 典型场景 | 医疗诊断 | 内容审核 | 数据报表 |
33.2 Human-in-the-Loop 模式
意图
在Agent执行的每个关键决策点引入人类审查和确认，确保Agent的行为始终处于人类的直接控制之下。
动机
在高风险场景中——医疗诊断、法律建议、金融交易、军事决策——Agent的任何错误都可能导致严重后果。Human-in-the-Loop模式确保Agent不是"自作主张"，而是"请示后行动"。
这种模式的本质是信任但验证：Agent负责生成建议或方案，人类负责审核和决策。Agent是"参谋"，人类是"指挥官"。
Microsoft的 Human-in-the-Loop AI 框架和IBM的 AI OpenScale 都将这种模式作为企业级AI部署的必备组件。
结构
┌─────────────────────────────────────────┐
│ Human-in-the-Loop Flow │
│ │
│ Agent ──▶ 生成方案 ──▶ ⏸️ 等待人类审批 │
│ │ │
│ ┌─────┴─────┐ │
│ ▼ ▼ │
│ ✅ 批准 ❌ 拒绝 │
│ │ │ │
│ ▼ ▼ │
│ 继续执行 修改/重试 │
│ │
│ 关键决策点: 数据输入、工具调用、最终输出 │
└─────────────────────────────────────────┘参与者
- Agent：生成建议或执行方案
- Human Reviewer：审核Agent的建议并做出决策
- Approval Interface：人类与Agent交互的界面
- Decision Log：记录人类决策的审计日志
协作
- Agent分析任务，生成建议方案
- Agent将方案提交给人类审核
- 暂停执行，等待人类决策
- 人类审核方案，选择批准、修改或拒绝
- 根据人类决策继续执行或调整
效果
优点：
- 最大化安全性——每次关键决策都有人类把关
- 人类可以纠正Agent的错误
- 逐步建立对Agent的信任
- 完整的审计追踪
缺点：
- 人类参与成本高——每次都需要人工
- 响应延迟大——需要等待人类响应
- 人类疲劳——频繁的审批请求导致注意力下降
- 不适合高吞吐量场景
实现
python
"""
Human-in-the-Loop 模式实现
"""
from typing import Callable, Optional, Any, Dict
from dataclasses import dataclass, field
from enum import Enum
import time
class Decision(Enum):
APPROVE = "approve"
REJECT = "reject"
MODIFY = "modify"
@dataclass
class ApprovalRequest:
"""审批请求"""
request_id: str
agent_name: str
action_type: str
action_description: str
action_details: Dict[str, Any]
risk_level: str = "medium" # low, medium, high, critical
context: str = ""
@dataclass
class ApprovalResponse:
"""审批响应"""
request_id: str
decision: Decision
comment: str = ""
modified_details: Optional[Dict[str, Any]] = None
reviewer: str = "human"
class HumanInLoopAgent:
"""
Human-in-the-Loop Agent
在关键决策点暂停执行，等待人类审批。
支持同步和异步审批模式。
"""
def __init__(
self,
agent_name: str,
approval_callback: Callable[[ApprovalRequest], ApprovalResponse],
auto_approve_low_risk: bool = True,
verbose: bool = False
):
self.agent_name = agent_name
self.approval_callback = approval_callback
self.auto_approve_low_risk = auto_approve_low_risk
self.verbose = verbose
self.approval_history: list = []
def _request_approval(self, request: ApprovalRequest) -> ApprovalResponse:
"""请求人类审批"""
# 低风险自动批准
if self.auto_approve_low_risk and request.risk_level == "low":
if self.verbose:
print(f" ⚡ 低风险自动批准: {request.action_type}")
return ApprovalResponse(
request.request_id, Decision.APPROVE,
"自动批准（低风险）", reviewer="auto"
)
if self.verbose:
print(f" ⏸️ 等待人类审批: {request.action_type} [{request.risk_level}]")
print(f" 详情: {request.action_description}")
response = self.approval_callback(request)
self.approval_history.append({
"request": request, "response": response,
"timestamp": time.time()
})
status = "✅" if response.decision == Decision.APPROVE else "❌"
if self.verbose:
print(f" {status} 审批结果: {response.decision.value} - {response.comment}")
return response
def execute_with_approval(self, action_type: str, action_fn: Callable,
action_args: dict, risk_level: str = "medium",
context: str = "") -> Any:
"""执行需要审批的操作"""
request = ApprovalRequest(
request_id=f"{self.agent_name}_{int(time.time())}",
agent_name=self.agent_name,
action_type=action_type,
action_description=action_fn.__doc__ or action_type,
action_details=action_args,
risk_level=risk_level,
context=context
)
response = self._request_approval(request)
if response.decision == Decision.APPROVE:
return action_fn(**action_args)
elif response.decision == Decision.MODIFY:
modified_args = {**action_args, **(response.modified_details or {})}
return action_fn(**modified_args)
else:
return None
# 使用示例
def mock_approval_ui(request: ApprovalRequest) -> ApprovalResponse:
"""模拟人类审批界面"""
print(f"\n📋 审批请求:")
print(f" Agent: {request.agent_name}")
print(f" 操作: {request.action_type}")
print(f" 风险: {request.risk_level}")
print(f" 详情: {request.action_description}")
# 模拟人类批准
return ApprovalResponse(request.request_id, Decision.APPROVE, "同意执行")
def demo_hitl():
def send_email(to: str, subject: str, body: str):
"""发送邮件"""
return f"邮件已发送至 {to}"
agent = HumanInLoopAgent("email_agent", mock_approval_ui, verbose=True)
result = agent.execute_with_approval(
"send_email", send_email,
{"to": "user@example.com", "subject": "重要通知", "body": "..."},
risk_level="high"
)
print(f"\n执行结果: {result}")适用场景
- 医疗诊断（诊断建议需医生确认）
- 法律文档生成（需律师审核）
- 金融交易（大额操作需人工确认）
- 安全敏感操作（数据删除、权限变更）
相关模式
- Human-on-the-Loop：HitL每次决策需要人类，HotL只需异常时介入
- Approval Gate：HitL的通用化版本，聚焦于审批机制
- Guardrails（第31章）：Guardrails是自动化的规则检查，HitL是人类的检查
33.3 Human-on-the-Loop 模式
意图
让Agent自主执行大部分操作，人类作为监督者在系统出现异常、不确定或超过阈值时介入。
动机
Human-in-the-Loop虽然安全，但人类参与成本太高——想象一下如果每次邮件分类都需要人工审批，那就失去了自动化的意义。Human-on-the-Loop模式找到了一个平衡点：Agent自主处理大部分常规任务，只在"不确定"或"高风险"时才通知人类。
这种模式类似于工厂中的"质量检验员"——不参与每一道工序，但在关键检查点抽检，发现问题时及时介入。
结构
┌─────────────────────────────────────────┐
│ Human-on-the-Loop Flow │
│ │
│ Agent 自主执行 ─────────────────┐ │
│ │ │
│ ┌─────────▼────┐ │
│ │ 异常检测器 │ │
│ └──┬───────┬──┘ │
│ │ │ │
│ 正常运行 检测到异常 │
│ │ │ │
│ ▼ ▼ │
│ 继续 ⚠️ 通知人类 │
│ │ │
│ ┌──────┴────┐│
│ ▼ ▼│
│ 人类介入 忽略 │
│ │ ││
│ ▼ ▼│
│ 修正继续 标记噪声│
└─────────────────────────────────────────┘参与者
- Agent：自主执行任务
- Anomaly Detector：检测执行过程中的异常
- Human Supervisor：人类监督者，接收异常通知并决定是否介入
- Alert System：异常通知系统
效果
优点：
- 人类参与成本低——只在必要时介入
- Agent自主性高——大部分操作自动完成
- 适合高吞吐量场景
缺点：
- 异常检测的准确性影响体验
- 人类可能错过紧急通知
- 需要设计合理的通知策略避免"报警疲劳"
实现
python
"""Human-on-the-Loop 模式实现"""
from typing import Callable, Optional, Dict, Any, List
from dataclasses import dataclass, field
from enum import Enum
import time
class AnomalyType(Enum):
LOW_CONFIDENCE = "low_confidence"
HIGH_RISK = "high_risk"
TIMEOUT = "timeout"
ERROR = "error"
POLICY_VIOLATION = "policy_violation"
@dataclass
class AnomalyEvent:
event_type: AnomalyType
description: str
confidence: float
context: Dict[str, Any] = field(default_factory=dict)
timestamp: float = field(default_factory=time.time)
@dataclass
class HumanIntervention:
event: AnomalyEvent
action: str # "fix", "skip", "escalate"
human_input: Any = None
class HumanOnLoopAgent:
"""Human-on-the-Loop Agent - 自主执行，异常时通知人类。"""
def __init__(self, alert_callback: Callable[[AnomalyEvent], HumanIntervention],
confidence_threshold: float = 0.7, verbose: bool = False):
self.alert_callback = alert_callback
self.confidence_threshold = confidence_threshold
self.verbose = verbose
self.events: List[Dict] = []
def execute(self, task: str, executor: Callable) -> Any:
"""执行任务，自动检测异常"""
try:
result, confidence = executor(task)
if confidence < self.confidence_threshold:
event = AnomalyEvent(
AnomalyType.LOW_CONFIDENCE,
f"置信度 {confidence:.2f} 低于阈值 {self.confidence_threshold}",
confidence
)
intervention = self.alert_callback(event)
self.events.append({"event": event, "intervention": intervention})
if intervention.action == "fix" and intervention.human_input:
return intervention.human_input
return result
except Exception as e:
event = AnomalyEvent(AnomalyType.ERROR, str(e), 0.0)
intervention = self.alert_callback(event)
if intervention.action == "fix":
return intervention.human_input
raise适用场景
- 内容审核（大部分内容自动处理，可疑内容人工审核）
- 数据标注（自动标注 + 人工抽检）
- 客服系统（常见问题自动回复，复杂问题转人工）
相关模式
- Human-in-the-Loop：HotL是HitL的轻量版
- Escalation：HotL的异常检测可以触发Escalation升级流程
33.4 Human-over-the-Loop 模式
意图
人类设定高层目标和约束，Agent完全自主地规划和执行，人类只在任务完成后审查结果或调整方向。
动机
对于低风险、高重复性的任务——如数据报表生成、日志分析、信息检索——人类的精细控制不仅不必要，还可能降低效率。Human-over-the-Loop模式让人类专注于"做什么"和"为什么做"，Agent负责"怎么做"。
这种模式类似于管理者给团队分配任务：管理者说明目标和要求，团队自主决定执行方式。
结构
┌─────────────────────────────────────────┐
│ Human-over-the-Loop Flow │
│ │
│ 人类: 设定目标 + 约束 │
│ │ │
│ ▼ │
│ Agent: 自主规划 + 执行 │
│ │ │
│ ▼ │
│ Agent: 输出结果 + 执行报告 │
│ │ │
│ ▼ │
│ 人类: 审查结果 (可选) │
│ │ │
│ ▼ │
│ 反馈 → 优化下一轮目标设定 │
└─────────────────────────────────────────┘效果
优点： 人类参与成本最低、Agent自主性最高、适合大规模自动化 缺点： 人类对执行过程的控制力弱、错误可能在完成后才被发现
实现
python
"""Human-over-the-Loop 模式实现"""
from dataclasses import dataclass, field
from typing import List, Callable
@dataclass
class AgentGoal:
objective: str
constraints: List[str] = field(default_factory=list)
success_criteria: str = ""
max_duration_seconds: int = 300
class HumanOverLoopAgent:
"""Agent完全自主执行，人类只设定目标。"""
def __init__(self, planner: Callable, executor: Callable, verbose: bool = False):
self.planner = planner
self.executor = executor
self.verbose = verbose
def run(self, goal: AgentGoal) -> dict:
plan = self.planner(goal)
result = self.executor(plan, goal.constraints)
return {
"objective": goal.objective,
"plan": plan,
"result": result
}适用场景
- 数据报表自动生成
- 日志分析和异常检测
- 信息检索和摘要
- 自动化测试执行
相关模式
- Plan-and-Execute（第31章）：HoTL天然支持计划-执行范式
- Delegation：HoTL是Delegation的一种形式
33.5 Approval Gate 模式
意图
在Agent工作流的特定节点设置审批关卡，只有在人类批准后才能继续执行后续步骤。
动机
Human-in-the-Loop在每个决策点都需要人类参与，粒度太细。Approval Gate模式允许你在工作流中精确地选择哪些节点需要审批——比如在"发送邮件"之前设置关卡，但在"搜索信息"时不需要。
这种模式类似于企业审批流程中的"签字盖章"——不是每个步骤都需要签字，但在关键节点（如合同签署、资金拨付）必须有签字。
结构
┌──────────────────────────────────────────┐
│ Workflow with Gates │
│ │
│ Step1 ──▶ Step2 ──▶ 🔒 Gate A ──▶ Step3 │
│ │ │
│ 人类审批 │
│ │ │
│ Step4 ──▶ Step5 ──▶ 🔒 Gate B ──▶ Step6 │
│ │
│ Gate配置: │
│ - 需要审批的操作类型 │
│ - 审批超时策略 │
│ - 审批人/组 │
│ - 自动批准条件 │
└──────────────────────────────────────────┘效果
优点： 精确控制审批粒度、审批策略可配置、支持多级审批 缺点： 需要预先定义审批节点、新增关卡需要修改工作流
实现
python
"""Approval Gate 模式实现"""
from typing import Callable, Dict, Optional, List
from dataclasses import dataclass, field
from enum import Enum
class GateAction(Enum):
APPROVE = "approve"
REJECT = "reject"
REQUEST_CHANGE = "request_change"
@dataclass
class ApprovalGate:
gate_id: str
name: str
description: str
required_action_types: List[str]
approvers: List[str]
timeout_seconds: int = 3600
auto_approve_after_timeout: bool = False
class GatedWorkflow:
"""带审批关卡的工作流"""
def __init__(self, gates: List[ApprovalGate] = None, verbose: bool = False):
self.gates = gates or []
self.verbose = verbose
def check_gate(self, action_type: str, action_details: dict,
approval_fn: Callable) -> bool:
"""检查操作是否需要经过审批关卡"""
for gate in self.gates:
if action_type in gate.required_action_types:
if self.verbose:
print(f"🔒 触发审批关卡: {gate.name}")
result = approval_fn(gate, action_details)
return result == GateAction.APPROVE
return True # 无需审批适用场景
- 企业审批流程（请假、报销、采购）
- 内容发布流程（文章发布前的审核）
- 数据操作（删除、导出敏感数据前的确认）
相关模式
- Human-in-the-Loop：Approval Gate是HitL在工作流中的精确实现
- Escalation：Gate拒绝后可以触发Escalation
33.6 Delegation 模式
意图
人类将特定任务委派给Agent执行，并设定明确的期望、约束和验收标准，Agent独立完成任务后汇报结果。
动机
Delegation模式模拟了人类组织中的任务委派：管理者将任务分配给下属，说明期望和约束，下属自主完成并汇报。在Agent系统中，人类将某些任务委派给Agent，同时保留对其他任务的直接控制。
与Human-over-the-Loop不同的是，Delegation更强调任务边界的明确性——哪些任务可以委派，哪些必须人类亲自处理。
效果
优点： 任务边界清晰、人类保持最终控制权、Agent可以专注于委派的任务 缺点： 委派粒度需要经验、不合适的委派可能导致返工
实现
python
"""Delegation 模式实现"""
from dataclasses import dataclass, field
from typing import List, Optional, Callable
@dataclass
class DelegatedTask:
task_id: str
description: str
constraints: List[str] = field(default_factory=list)
deadline: Optional[str] = None
expected_output: str = ""
priority: int = 0
class DelegationManager:
"""任务委派管理器"""
def __init__(self, agent_executor: Callable, verbose: bool = False):
self.executor = agent_executor
self.verbose = verbose
self.delegated: List[dict] = []
def delegate(self, task: DelegatedTask) -> dict:
if self.verbose:
print(f"📋 委派任务: {task.description}")
print(f" 约束: {', '.join(task.constraints)}")
result = self.executor(task)
self.delegated.append({"task": task, "result": result})
return result
def review(self, task_id: str) -> Optional[dict]:
for item in self.delegated:
if item["task"].task_id == task_id:
return item["result"]
return None适用场景
- 项目管理（将子任务分配给Agent）
- 研究助手（委托Agent进行信息搜集）
- 代码审查（委托Agent进行初步审查）
相关模式
- Human-over-the-Loop：Delegation是HoTL的具体化形式
- Escalation：Agent发现任务超出能力时，可以反向委派给人类
33.7 Escalation 模式
意图
当Agent遇到自身无法处理的情况时，按照预定义的规则将问题升级给更有能力的人类或其他Agent处理。
动机
Agent的能力是有边界的。当遇到超出能力范围的问题——如需要人类判断的主观问题、缺少必要权限的操作、系统异常情况——Agent不应该盲目尝试，而应该及时"求助"。Escalation模式定义了何时升级、升级给谁、升级后如何处理的标准化流程。
结构
┌──────────────────────────────────────────┐
│ Escalation Flow │
│ │
│ Agent 尝试处理 │
│ │ │
│ 遇到以下情况? │
│ ├── 超出能力范围 ──▶ 升级到专家 │
│ ├── 缺少权限 ────▶ 升级到管理员 │
│ ├── 不确定结果 ──▶ 升级到人类审核 │
│ └── 系统错误 ───▶ 升级到运维 │
│ │
│ 升级规则: │
│ - 触发条件 │
│ - 升级目标 │
│ - 升级级别 (L1→L2→L3) │
│ - 超时未处理策略 │
└──────────────────────────────────────────┘效果
优点： 避免Agent做出错误决策、问题能及时得到专业处理、有明确的升级路径 缺点： 升级规则设计复杂、可能过度升级导致人类负担
实现
python
"""Escalation 模式实现"""
from typing import Callable, Dict, List, Optional
from dataclasses import dataclass, field
from enum import Enum
class EscalationLevel(Enum):
L1_AGENT = "l1_agent" # 升级给高级Agent
L2_SPECIALIST = "l2_specialist" # 升级给领域专家
L3_HUMAN = "l3_human" # 升级给人类
@dataclass
class EscalationRule:
rule_id: str
condition: Callable[[dict], bool]
level: EscalationLevel
target: str
description: str
priority: int = 0
@dataclass
class EscalationEvent:
rule: EscalationRule
context: Dict
resolved: bool = False
resolution: str = ""
class EscalationManager:
"""问题升级管理器"""
def __init__(self, handlers: Dict[EscalationLevel, Callable],
verbose: bool = False):
self.rules: List[EscalationRule] = []
self.handlers = handlers
self.verbose = verbose
self.history: List[EscalationEvent] = []
def add_rule(self, rule: EscalationRule) -> None:
self.rules.append(rule)
self.rules.sort(key=lambda r: r.priority, reverse=True)
def check_and_escalate(self, context: dict) -> Optional[dict]:
"""检查是否需要升级"""
for rule in self.rules:
if rule.condition(context):
if self.verbose:
print(f"⚠️ 升级 [{rule.level.value}]: {rule.description}")
print(f" 目标: {rule.target}")
event = EscalationEvent(rule, context)
handler = self.handlers.get(rule.level)
if handler:
result = handler(event)
event.resolved = True
event.resolution = str(result)
self.history.append(event)
return result
return None # 无需升级适用场景
- 客服系统（Agent无法解答时转人工）
- 安全监控（检测到异常时升级给安全团队）
- 运维系统（系统异常时升级给运维人员）
相关模式
- Human-on-the-Loop：Escalation是HotL的具体触发机制
- Router（第31章）：Router做常规分发，Escalation做异常分发
33.8 模式选择指南
本章介绍了六种人机协作模式，它们的核心差异在于人类参与的时机和深度：
人类参与深度
▲
│ ┌─────────────────────┐
│ │ Human-in-the-Loop │ ← 每次关键决策
│ ├─────────────────────┤
│ │ Approval Gate │ ← 特定审批节点
│ ├─────────────────────┤
│ │ Human-on-the-Loop │ ← 异常时介入
│ ├─────────────────────┤
│ │ Delegation │ ← 明确的任务边界
│ ├─────────────────────┤
│ │ Escalation │ ← 能力不足时升级
│ ├─────────────────────┤
│ │ Human-over-the-Loop │ ← 仅设定目标
│ └─────────────────────┘
└──────────────────────────▶ Agent自主性实践原则：
- 从严格到宽松：系统初期使用Human-in-the-Loop，逐步过渡到Human-on-the-Loop
- 风险驱动：高风险操作必须有人类参与，低风险操作可以让Agent自主
- 渐进信任：随着Agent可靠性的提升，减少人类的参与频率
- 可观测性：无论使用哪种模式，都要确保人类可以随时了解Agent的状态
"最好的工具不是让人类变得更少，而是让人类变得更强大。人机协作的未来不是替代，而是增强。"