# 5.4 人机协作

**完全自主的智能体** 并不总是最佳选择。在高风险决策、创意性工作或需要最终确认的场景中，人类的参与是不可或缺的。本节探讨如何在智能体系统中优雅地引入人类监督，实现人机协作的最佳平衡。

## 5.4.1 为什么需要人类介入

人机协作（HITL）的目标不是“让人类替智能体干活”，而是在关键节点提供校验、授权与偏好输入：让系统既保持自动化效率，又能在高风险与高不确定性处收敛到可控边界。

## 5.4.2 智能体自主性的边界

尽管 AI 能力在快速提升，但在以下场景中，人类介入仍然是必要的：

| 场景类型  | 原因       | 示例        |
| ----- | -------- | --------- |
| 高风险决策 | 错误成本过高   | 金融交易、医疗诊断 |
| 价值判断  | 需要主观偏好   | 创意设计、品牌策略 |
| 法律责任  | 需要人类签字确认 | 合同签署、重要审批 |
| 不确定性高 | 模型信心不足   | 模糊需求、边缘案例 |
| 学习反馈  | 持续优化需要   | 错误纠正、偏好学习 |

## 5.4.3 完全自主与人机协作

```
完全自主模式：
用户 → 智能体 → 执行 → 结果
        ↑
      (无监督)

人机协作模式：
用户 → 智能体 → [检查点] → 人类确认 → 执行 → 结果
                    ↓
              (如需修正) → 智能体重试
```

如果把这个过程展开成运行时视角，HITL 更像四条并行泳道之间的持续交接：用户给出目标，智能体提出动作，运行时策略判断是否需要停下，人类则在关键节点批准、拒绝或改写方向。

{% @mermaid/diagram content="flowchart LR
classDef user fill:#e0f2fe,stroke:#0284c7,stroke-width:2px,color:#0f172a;
classDef agent fill:#ecfccb,stroke:#65a30d,stroke-width:2px,color:#0f172a;
classDef policy fill:#fef3c7,stroke:#d97706,stroke-width:2px,color:#0f172a;
classDef human fill:#fce7f3,stroke:#db2777,stroke-width:2px,color:#0f172a;

```
subgraph LaneUser["用户"]
    direction TB
    U1["提交目标"] --> U2["查看进度<br/>补充约束"]
end

subgraph LaneAgent["智能体"]
    direction TB
    A1["解析任务"] --> A2["生成动作方案"] --> A3["执行或等待"] --> A4["根据反馈重试"] --> A5["交付结果"]
end

subgraph LanePolicy["策略与运行时"]
    direction TB
    P1["风险评估"] --> P2{"命中审批<br/>或熔断规则?"} --> P3["记录 trace<br/>并暂停任务"] --> P4["恢复同一任务线程"]
end

subgraph LaneHuman["人工审阅者"]
    direction TB
    H1["接收审批包"] --> H2["批准 / 拒绝 / 修改"] --> H3["写回决策"]
end

U1 --> A1
A2 --> P1
P2 -- "否" --> A3
P2 -- "是" --> H1
H3 --> P4
P4 --> A4
A5 --> U2
U2 -.追加信息.-> A4

style LaneUser fill:#f8fafc,stroke:#cbd5e1,stroke-width:1.5px,color:#0f172a;
style LaneAgent fill:#f8fafc,stroke:#cbd5e1,stroke-width:1.5px,color:#0f172a;
style LanePolicy fill:#f8fafc,stroke:#cbd5e1,stroke-width:1.5px,color:#0f172a;
style LaneHuman fill:#f8fafc,stroke:#cbd5e1,stroke-width:1.5px,color:#0f172a;
class U1,U2 user;
class A1,A2,A3,A4,A5 agent;
class P1,P2,P3,P4 policy;
class H1,H2,H3 human;" %}
```

图 5-3：HITL 的四泳道协作流程

## 5.4.4 人机协作模式设计

在关键操作前设置人类审批节点：

```python
from typing import Dict

class ApprovalResult:
    def __init__(self, approved: bool, reason: str = ""):
        self.approved = approved
        self.reason = reason


class ApprovalGate:
    def __init__(self, approval_type: str, timeout_hours: int = 24):
        self.approval_type = approval_type
        self.timeout_hours = timeout_hours

    async def request_approval(
        self,
        action: str,
        context: Dict
    ) -> ApprovalResult:
        # 发送审批请求到人类

        request_id = await send_approval_request(
            action=action,
            context=context,
            approval_type=self.approval_type
        )

        # 等待人类响应

        result = await wait_for_human_response(
            request_id,
            timeout=self.timeout_hours * 3600
        )

        return result

# 在工作流中使用

async def execute_high_risk_action(action: str):
    gate = ApprovalGate("financial_operation")

    approval = await gate.request_approval(
        action=f"执行转账: {action}",
        context={"amount": 10000, "recipient": "vendor_001"}
    )

    if approval.approved:
        return await perform_action(action)
    else:
        raise ActionRejected(approval.reason)
```

### 置信度阈值

当智能体对决策信心不足时自动请求人类帮助：

```python
from typing import Tuple

class ConfidenceBasedHITL:
    def __init__(self, confidence_threshold: float = 0.8):
        self.threshold = confidence_threshold

    async def decide_with_human_backup(
        self,
        question: str
    ) -> Tuple[str, str]:
        # 智能体生成回答和置信度
        response, confidence = await agent.generate_with_confidence(question)

        if confidence >= self.threshold:
            return response, "agent"
        else:
            # 置信度不足，请求人类帮助
            human_response = await request_human_input(
                question=question,
                agent_suggestion=response,
                agent_confidence=confidence,
                message="智能体不确定此回答，请人工确认或修正"
            )
            return human_response, "human"
```

### 迭代反馈循环

人类持续提供反馈，智能体不断改进：

```python
class IterativeFeedbackLoop:
    def __init__(self, max_iterations: int = 3):
        self.max_iterations = max_iterations

    async def refine_with_feedback(
        self,
        task: str,
        initial_output: str
    ) -> str:
        current_output = initial_output

        for i in range(self.max_iterations):
            # 展示当前结果给人类
            feedback = await get_human_feedback(
                iteration=i + 1,
                current_output=current_output,
                prompt="请提供修改建议，或输入 'approve' 接受当前结果"
            )

            if feedback.lower() == "approve":
                return current_output

            # 根据反馈改进
            current_output = await agent.refine(
                original=current_output,
                feedback=feedback
            )

        return current_output
```

## 5.4.5 以 LangGraph 为代表的框架实践

像 LangGraph 这类强调状态图执行与持久化恢复的编排框架，通常会提供一组便于实现 HITL 的运行时机制。这里常见的 **checkpoint/checkpointer**、**interrupt**、**resume** 更适合被理解为**框架 API 语汇**，而不是行业统一标准术语。

以 LangGraph 的做法为例：

1. **Checkpointer / Checkpoint**：在图执行过程中持久化状态，使同一个 `thread_id` 后续可以继续运行。
2. **Interrupt**：在节点内部显式暂停执行，把“待审批内容”返回给外部调用方。
3. **Resume**：外部拿到人类决策后，再用 `Command(resume=...)` 之类的恢复接口继续同一条执行线程。
4. **持久化与幂等**：这是更普遍的工程要求，不专属于 LangGraph，但在这类框架中通常和 checkpoint 机制一起实现。

示意伪代码（不同 LangGraph 版本的返回结构略有区别，这里只表达控制流）：

```python
config = {"configurable": {"thread_id": "approval-1"}}

# 首次调用：图在 interrupt(...) 处暂停，并返回待审批内容
result = graph.invoke({"input": task}, config=config)
approval_payload = result["__interrupt__"]

# ... later ...

decision = get_human_decision(approval_payload)
resumed = graph.invoke(Command(resume=decision), config=config)
```

不同框架也可能提供 approval queue、workflow token、人工任务节点等能力，但它们的命名方式、恢复粒度和重放语义并不完全一致。

### 明确介入时机

定义清晰、**可前置验证** 且 **可量化** 的介入规则，是防止系统走向两个极端（过度打扰或监督不足）的关键原则：

1. **资源与预算边界 (Resource-based)**：
   * 消耗的预测 Token 成本或单次任务流程费用超过特定额度（如 `> $5.00`）。
   * 请求的计算周期或系统存留时间超过预警窗口。
2. **状态变更边界 (State-based)**：
   * 尝试写库、修改外部状态或执行不可逆操作（如“删除用户数据”、“发送对外群发邮件”）。
   * 触发了基座框架设定的硬编码拦截器（如检测到 PII 出境请求）。
3. **置信度与错误阈值 (Confidence/Error-based)**：
   * 智能体内在置信度（大模型自我评分或 Logprobs 衍生）低于设定的及格线（如 `0.80`）。
   * 单个工具连续发生错误返回超过阈值（如 `> 3次 Retry Exhaustion`），或陷入规划死循环（Looping）。

```python
HITL_TRIGGERS = {
    # 状态与边界 (State & Boundaries)
    "always": ["delete_production_data", "financial_transaction > 10000", "send_external_email"],
    # 置信度与错判 (Confidence & Errors)
    "conditional": ["confidence < 0.8", "tool_retry_count > 3", "cost_estimate > $5.0"],
    # 自由放行 (Safe)
    "never": ["search_query", "read_document", "local_sandbox_test"]
}
```

### 提供足够上下文

人类做决策时需要完整信息：

```python
def format_approval_request(action: str, context: Dict) -> str:
    return f"""
📋 需要您的审批

操作类型：{action}

背景信息：
- 智能体的推理过程：{context["reasoning"]}
- 置信度：{context["confidence"]}
- 潜在风险：{context["risks"]}

预期结果：{context["expected_outcome"]}

请选择：[✅ 批准] [❌ 拒绝] [✏️ 修改后批准]
"""
```

### 支持异步审批

不要让人类一直等待，支持稍后处理：

```python
class AsyncApprovalSystem:
    async def submit_for_approval(self, request):
        # 存储请求
        await self.db.save(request)

        # 发送通知（邮件、IM、工单等）
        await self.notify(request.reviewer, request)

        # 返回请求ID，不阻塞
        return request.id

    async def check_approval_status(self, request_id: str):
        request = await self.db.get(request_id)
        return request.status  # pending | approved | rejected
```

## 5.4.6 实时可观测与行动透明

前面讨论的审批门控和置信度阈值属于**被动式**人机协作——人类只在特定节点被叫停审批。但对于长时间运行的复杂任务，用户需要**实时看到** Agent 正在做什么，而不是等到执行完毕才获得一份事后报告。

### 为什么需要实时可观测？

传统的可观测性（observability）侧重于事后分析：收集日志、构建 trace、生成仪表板。但在 Agent 场景下，任务可能持续数分钟甚至数小时，用户需要：

* **即时感知**：当前 Agent 在思考什么？正在调用哪个工具？
* **信任建立**：看到 Agent 的推理过程，用户才能判断它是否走在正确方向上
* **及时干预**：发现偏差时立即介入，而不是等到错误累积后再回滚
* **成本感知**：实时了解 token 消耗和 API 调用次数

### Action Streaming 模式

Agent 在执行过程中，应将每一步操作以结构化事件的形式实时推送给外部系统：

```python
from dataclasses import dataclass
from enum import Enum
from typing import Any, Callable
import asyncio

class EventType(Enum):
    THINKING = "thinking"       # Agent 正在推理
    TOOL_CALL = "tool_call"     # 准备调用工具
    TOOL_RESULT = "tool_result" # 工具返回结果
    DECISION = "decision"       # 做出决策
    ERROR = "error"             # 遇到错误

@dataclass
class AgentEvent:
    event_type: EventType
    content: Any
    step_index: int
    timestamp: float

class StreamingAgent:
    """支持实时事件流的 Agent"""

    def __init__(self):
        self._subscribers: list[Callable] = []

    def subscribe(self, callback: Callable):
        self._subscribers.append(callback)

    async def _emit(self, event: AgentEvent):
        for cb in self._subscribers:
            await cb(event)

    async def run(self, task: str):
        step = 0
        while not self.is_complete():
            # 1. 推理阶段 — 推送思考过程
            thought = await self.think(task)
            await self._emit(AgentEvent(
                EventType.THINKING, thought, step, time.time()
            ))

            # 2. 工具调用 — 推送调用详情
            tool, args = self.select_tool(thought)
            await self._emit(AgentEvent(
                EventType.TOOL_CALL,
                {"tool": tool.name, "args": args},
                step, time.time()
            ))

            # 3. 执行并推送结果
            result = await tool.execute(**args)
            await self._emit(AgentEvent(
                EventType.TOOL_RESULT, result, step, time.time()
            ))
            step += 1
```

### UI 展示模式

实际产品中，常见的实时展示模式包括：

* **步骤进度流**：类似 Claude Code 的终端输出，逐步显示 “Thinking → Calling search\_api → Got 5 results → Analyzing...”
* **思维链可视化**：将 Agent 的推理过程以树状结构展示，便于追踪决策分叉
* **工具调用预览**：在执行前展示即将调用的工具和参数（类似 Cursor 的 diff 预览），用户可确认或修改后再执行

### 实时流 vs 事后追踪

| 维度   | 事后追踪（传统）           | 实时流（Agent 场景）      |
| ---- | ------------------ | ------------------ |
| 时机   | 任务结束后分析            | 执行过程中同步推送          |
| 用途   | 调试、审计、优化           | 监控、干预、信任建立         |
| 数据格式 | OpenTelemetry Span | SSE / WebSocket 事件 |
| 延迟容忍 | 分钟级                | 秒级甚至毫秒级            |
| 交互性  | 只读                 | 可触发暂停、修改、终止        |

## 5.4.7 运行时控制：暂停、恢复与指令注入

审批门控是一种**预设的**控制点，但现实中用户往往需要在**任意时刻**介入 Agent 执行。这就需要运行时控制机制。

### 控制粒度

运行时控制可以在不同层级生效：

```
┌─────────────────────────────────────────┐
│ 任务级控制                                │
│  "停止当前任务" / "切换到另一个目标"        │
│  ┌───────────────────────────────────┐  │
│  │ 步骤级控制                         │  │
│  │  "跳过这一步" / "换个方法试试"      │  │
│  │  ┌─────────────────────────────┐  │  │
│  │  │ 工具调用级控制               │  │  │
│  │  │  "别调这个 API" / "改下参数"  │  │  │
│  │  └─────────────────────────────┘  │  │
│  └───────────────────────────────────┘  │
└─────────────────────────────────────────┘
```

### 暂停与恢复

Agent 需要在**安全点**（如工具调用前后、步骤边界）检查控制信号：

```python
import asyncio
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class ControlSignal:
    action: str  # "pause", "resume", "stop", "inject"
    payload: Optional[str] = None

class ControllableAgent:
    """支持暂停/恢复/指令注入的 Agent"""

    def __init__(self):
        self._control_queue: asyncio.Queue = asyncio.Queue()
        self._paused = False
        self._state_snapshot = None

    async def send_control(self, signal: ControlSignal):
        """外部系统发送控制信号"""
        await self._control_queue.put(signal)

    async def _check_control(self):
        """在安全点检查是否有控制信号"""
        while not self._control_queue.empty():
            signal = await self._control_queue.get()
            if signal.action == "pause":
                self._paused = True
                self._save_state()
            elif signal.action == "stop":
                raise StopIteration("User requested stop")
            elif signal.action == "inject":
                return signal.payload  # 返回注入的新指令

        # 如果处于暂停状态，阻塞等待恢复
        while self._paused:
            signal = await self._control_queue.get()
            if signal.action == "resume":
                self._paused = False
            elif signal.action == "stop":
                raise StopIteration("User requested stop")

        return None

    async def run(self, task: str):
        steps = self.decompose(task)
        for i, step in enumerate(steps):
            # 每个步骤前检查控制信号
            injected = await self._check_control()
            if injected:
                steps = self.replan(injected, remaining=steps[i:])

            await self.execute_step(step)
```

### 指令注入的典型场景

运行中注入新指令是 Agent 从“自动化脚本”升级为“协作伙伴”的关键能力：

* **修正方向**：“你搜索的方向不对，应该关注 X 而不是 Y”
* **补充约束**：“记得结果要用中文输出”
* **跳过步骤**：“这一步不用做了，我已经有结果了”
* **追加任务**：“顺便也检查一下 Z”
* **调整策略**：“别用暴力搜索，试试先缩小范围”

这些在传统批处理任务中不可能实现，但在 Agent 的 Think-Act-Observe 循环中，每一轮循环都是天然的干预点。

### 安全考量

运行时控制带来灵活性的同时也引入风险：

* **状态一致性**：暂停时 Agent 可能处于工具调用中间态，恢复时需确保状态正确
* **指令冲突**：注入的指令可能与 Agent 当前上下文矛盾，需要验证和协调机制
* **权限分级**：不同用户应有不同的控制权限（查看者只能观看，操作者可暂停，管理员可注入指令）
* **成本控制**：暂停/恢复不应产生额外 token 消耗，应通过检查点机制实现零成本恢复

## 5.4.8 小结

人机协作不是对智能体能力的否定，而是对其能力边界的务实认知。从最基础的审批门控，到实时可观测，再到运行时控制，人机协作机制形成了一个由浅入深的层次：

| 层次       | 机制            | 人类角色  |
| -------- | ------------- | ----- |
| L1 审批门控  | 预设检查点，人工批准/拒绝 | 审批者   |
| L2 置信度阈值 | 低置信度自动升级给人类   | 后备决策者 |
| L3 实时可观测 | 全程可见 Agent 动态 | 监督者   |
| L4 运行时控制 | 随时暂停、恢复、注入指令  | 协作者   |

关键在于找到 **自动化与人工介入的最佳平衡点**——既不能让智能体完全失控，也不能让人类成为瓶颈。随着 Agent 能力的增强，人类的角色会从“审批者”逐步演变为“监督者”和“协作者”。

***

**下一节**: [本章小结](/agentic_ai_guide/di-er-bu-fen-qun-ti-zhi-neng-yu-jin-hua/05_collaboration/summary.md)


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://yeasy.gitbook.io/agentic_ai_guide/di-er-bu-fen-qun-ti-zhi-neng-yu-jin-hua/05_collaboration/5.4_hitl.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.