多 Agent 协作与编排
·
多 Agent 协作与编排
目录
1. 什么时候需要多 Agent
1.1 核心判断标准
不是所有场景都需要多 Agent。引入多 Agent 会带来通信开销、一致性问题、调试复杂度。
问题是否可以被单一 Agent 有效处理？
+-- 是 -> 单 Agent 足够，不要过度设计
+-- 否 -> 检查是否满足以下条件之一：
+-- 条件A：任务需要多种异构能力
+-- 条件B：任务天然可并行分解
+-- 条件C：需要多视角交叉验证
+-- 条件D：需要领域知识隔离
+-- 条件E：需要角色扮演与对抗
1.2 五大典型场景
场景 A：异构能力组合
单一任务需要多种不同性质的能力，单一 Agent 难以全部胜任。
用户需求："分析这份财报，评估投资价值，并生成PPT"
多 Agent 方案：
Orchestrator
+-- 财务分析Agent
+-- 行业研究Agent
+-- 估值建模Agent
+-- PPT生成Agent
场景 B：任务天然可并行分解
任务可拆分为多个独立子任务，各子任务之间无强依赖。
用户需求："调研A、B、C三家云服务商的GPU实例价格"
单Agent串行：调研A -> 调研B -> 调研C -> 汇总 (耗时 = T_A+T_B+T_C)
多Agent并行：
Agent-A 调研A --+
Agent-B 调研B --+-> 汇总Agent -> 输出
Agent-C 调研C --+ (耗时 = max(T_A,T_B,T_C))
场景 C：多视角交叉验证
需要从不同角度审视同一问题，提高结论可靠性。
用户需求："这个架构方案有什么潜在风险？"
多 Agent 方案：
架构师Agent -> 提出方案
安全Agent -> 从安全角度审查
成本Agent -> 从成本角度审查
运维Agent -> 从可运维性角度审查
|
汇总Agent -> 合并各视角发现，去重排序
场景 D：领域知识隔离
不同子任务需要不同的专业知识库和工具集，混在一起会导致 Prompt 臃肿和工具调用混乱。
用户需求："我身体不舒服，帮我分析可能的原因并推荐就医科室"
多 Agent 方案：
分诊Agent（通用医学知识）
|
专科Agent-A（内科知识库） 专科Agent-B（外科知识库）
| |
预约Agent（医院HIS系统对接）
场景 E：角色扮演与对抗
需要模拟多方互动、博弈或对抗场景。
辩论模拟： 正方Agent <-> 反方Agent <-> 裁判Agent
谈判模拟： 买方Agent <-> 卖方Agent <-> 调解Agent
红蓝对抗： 红队Agent -> 目标系统 <- 蓝队Agent
代码审查： 开发者Agent -> 提交代码 -> 审查者Agent -> 反馈 -> 开发者Agent -> 修改
1.3 单 Agent vs 多 Agent 决策树
任务是否需要多种异构专业能力？
/ \
是 否
| |
多 Agent 推荐 任务是否可并行分解且耗时敏感？
/ \
是 否
| |
多 Agent 推荐 是否需要多视角交叉验证？
/ \
是 否
| |
多 Agent 推荐 单 Agent 足够
2. 消息传递与状态同步
2.1 五种消息传递模式
模式 A：点对点（Direct）
Agent-A -> Agent-B
适用：两个 Agent 之间有明确的上下游关系。
# 直接调用
result = agent_b.process(agent_a.output)
# 或通过消息队列
message_queue.send(sender="agent_a", receiver="agent_b", payload={...})
优点：简单，延迟低
缺点：耦合度高，Agent 需要知道对方的存在
模式 B：广播（Broadcast）
Agent-A -> Agent-B, Agent-C, Agent-D（同时）
适用：一个 Agent 的输出需要被多个 Agent 同时消费。
# 并行调用
results = await asyncio.gather(
agent_b.process(message),
agent_c.process(message),
agent_d.process(message),
)
优点：解耦，易于扩展
缺点：无法保证所有消费者都成功处理
模式 C：发布-订阅（Pub/Sub）
Agent-A -> Topic “task.completed” -> Agent-B（订阅）, Agent-C（订阅）
适用：基于事件驱动的松耦合架构。
class EventBus:
def __init__(self):
self.subscribers = {}
def subscribe(self, event_type, agent):
self.subscribers.setdefault(event_type, []).append(agent)
def publish(self, event):
for agent in self.subscribers.get(event.type, []):
agent.on_event(event)
bus.subscribe("document.parsed", embedding_agent)
bus.subscribe("document.parsed", indexing_agent)
bus.publish(Event(type="document.parsed", data=doc))
优点：高度解耦，Agent 可动态加入/离开
缺点：调试困难，事件顺序难以保证
模式 D：请求-响应（Request-Reply）
Agent-A --请求–> Agent-B
Agent-A <–响应-- Agent-B
适用：需要同步获取结果的场景。
# 异步带超时
try:
result = await asyncio.wait_for(
agent_b.ask_async("请分析这段代码的安全性"),
timeout=30
)
except asyncio.TimeoutError:
result = "分析超时，请稍后重试"
优点：语义清晰，调用方明确知道在等结果
缺点：同步阻塞，被调用方故障时调用方会卡住
模式 E：黑板模式（Blackboard）
多个 Agent 读写同一个共享数据结构，各自贡献部分解。
class Blackboard:
def __init__(self):
self.data = {}
self.partial_results = []
def write(self, key, value, contributor):
self.data[key] = {"value": value, "contributor": contributor}
def read(self, key):
return self.data.get(key)
# 使用示例：多 Agent 协同解题
blackboard = Blackboard()
blackboard.write("problem_understanding", "...", "agent_a")
understanding = blackboard.read("problem_understanding")
blackboard.write("solution_approach", "...", "agent_b")
优点：灵活，支持增量式问题求解
缺点：需要设计良好的数据结构，否则容易混乱
2.2 状态同步策略
策略对比
| 策略 | 原理 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|---|
| 无状态 | 每个请求独立 | 简单问答、单次计算 | 最简单，易扩展 | 无法多轮交互 |
| 会话状态 | 每会话独立状态 | 多轮对话 | 会话隔离 | 跨Agent共享需额外设计 |
| 共享内存 | 多Agent共享状态空间 | 频繁交换中间结果 | 低延迟，即时可见 | 并发控制复杂 |
| 事件溯源 | 记录所有事件，状态由重放得出 | 需完整审计 | 可回溯，可调试 | 存储开销大 |
2.3 消息传递模式选型
| 场景 | 推荐模式 | 原因 |
|---|---|---|
| 简单顺序流水线 | 点对点 | 上下游明确，无需解耦 |
| 一对多通知 | 广播 | 一个结果多个消费者 |
| 松耦合事件驱动 | Pub/Sub | Agent 可独立演进 |
| 需要返回结果 | 请求-响应 | 同步语义清晰 |
| 多Agent协同解题 | 黑板模式 | 增量贡献部分解 |
| 生产级高可靠 | Pub/Sub + 消息队列 | 持久化 + 重试 + 死信 |
3. 任务分配策略
3.1 策略分类
任务分配策略
+-- 静态分配（预定义规则）
| +-- 基于角色
| +-- 基于能力标签
| +-- 基于优先级
+-- 动态分配（运行时决策）
| +-- 基于负载
| +-- 基于能力匹配（语义路由）
| +-- 基于竞价/拍卖
+-- 混合分配
+-- 静态路由 + 动态负载均衡
+-- 规则引擎 + LLM 决策
3.2 策略详解
策略 A：基于角色的静态分配
class RoleBasedRouter:
def __init__(self):
self.role_registry = {
"code_analysis": ["security_agent", "review_agent"],
"data_query": ["sql_agent", "analytics_agent"],
"content_creation": ["writer_agent", "editor_agent"],
}
def route(self, task):
role = task.metadata.get("role")
candidates = self.role_registry.get(role, [])
return candidates[0] if candidates else "general_agent"
优点：简单，可预测
缺点：无法处理角色边界模糊的任务
策略 B：基于能力标签的语义路由
class SemanticRouter:
def __init__(self):
self.agent_capabilities = {
"security_agent": {
"skills": ["code_review", "vulnerability_scan"],
"tools": ["bandit", "snyk"],
"knowledge": ["owasp", "cve"],
},
"sql_agent": {
"skills": ["sql_generation", "query_optimization"],
"tools": ["postgres", "mysql"],
"knowledge": ["database_schema"],
},
}
def find_best_agent(self, task):
task_embedding = self.embed(task.description)
best_agent = None
best_score = -1
for name, caps in self.agent_capabilities.items():
cap_text = " ".join(caps["skills"] + caps["knowledge"])
cap_embedding = self.embed(cap_text)
score = cosine_similarity(task_embedding, cap_embedding)
if score > best_score:
best_score = score
best_agent = name
return best_agent if best_score > 0.6 else "general_agent"
优点：自动匹配，适应模糊任务
缺点：依赖 embedding 质量，边界任务可能误路由
策略 C：基于负载的动态分配
class LoadBalancedRouter:
def __init__(self):
self.agent_loads = {} # {agent_name: current_tasks}
self.agent_pools = {} # {role: [agent_names]}
def route(self, task):
role = task.metadata.get("role")
pool = self.agent_pools.get(role, [])
# 选择当前负载最低的 Agent
best = min(pool, key=lambda a: len(self.agent_loads.get(a, [])))
self.agent_loads.setdefault(best, []).append(task.id)
return best
def on_task_complete(self, agent_name, task_id):
self.agent_loads[agent_name].remove(task_id)
优点：资源利用率高，避免单点过载
缺点：需要实时监控负载状态
策略 D：LLM 驱动的智能分配
class LLMBasedRouter:
async def route(self, task, available_agents):
prompt = f"""
任务：{task.description}
可用Agent及其能力：
{self._format_agents(available_agents)}
请选择最合适的Agent来处理此任务。
只返回Agent名称，不要解释。
"""
agent_name = await llm.generate(prompt)
return agent_name.strip()
def _format_agents(self, agents):
return "\n".join([
f"- {name}: {caps['description']}"
for name, caps in agents.items()
])
优点：最灵活，能处理复杂决策
缺点：增加延迟和成本，LLM 可能做出错误选择
策略 E：拍卖/竞价模式
class AuctionRouter:
async def route(self, task, agents):
bids = []
for agent in agents:
bid = await agent.evaluate_task(task)
bids.append({
"agent": agent.name,
"confidence": bid.confidence,
"estimated_time": bid.estimated_time,
"cost": bid.cost,
})
# 综合评分：confidence * 0.5 + (1/time) * 0.3 + (1/cost) * 0.2
best = max(bids, key=lambda b:
b["confidence"] * 0.5 +
(1 / max(b["estimated_time"], 1)) * 0.3 +
(1 / max(b["cost"], 0.01)) * 0.2
)
return best["agent"]
优点：Agent 自主决策，适合异构Agent池
缺点：增加一轮通信开销
3.3 任务分配策略对比
| 策略 | 灵活性 | 延迟 | 实现复杂度 | 适用场景 |
|---|---|---|---|---|
| 基于角色 | 低 | 极低 | 低 | 任务类型明确固定 |
| 语义路由 | 中 | 低 | 中 | 任务类型多样 |
| 负载均衡 | 中 | 低 | 中 | 高并发同质任务 |
| LLM 驱动 | 高 | 高 | 中 | 复杂非标任务 |
| 拍卖竞价 | 高 | 高 | 高 | 异构Agent竞争 |
推荐组合
：语义路由（初筛）+ 负载均衡（精选）+ LLM 驱动（兜底复杂任务）
4. 冲突解决机制
4.1 冲突类型
| 冲突类型 | 描述 | 示例 |
|---|---|---|
| 事实冲突 | 不同Agent对同一事实给出矛盾结论 | Agent-A说GDP增长5%，Agent-B说3% |
| 方案冲突 | 对同一问题提出互斥的解决方案 | 一个建议微服务，一个建议单体 |
| 资源冲突 | 多个Agent竞争同一资源 | 两个Agent同时要修改同一配置 |
| 目标冲突 | Agent的优化目标互相矛盾 | 成本Agent要省钱，性能Agent要扩容 |
| 时序冲突 | 执行顺序不一致导致状态错误 | Agent-B在Agent-A写之前读了旧数据 |
4.2 解决策略
策略 A：多数投票（Majority Voting）
适用：分类、判断、有标准答案的任务
优点：简单，可解释
缺点：多数不一定正确（群体盲从）
策略 B：加权投票（Weighted Voting）
适用：Agent能力不均衡的场景
优点：更信任历史表现好的Agent
缺点：权重需要持续更新
策略 C：仲裁者模式（Arbiter）
适用：复杂争议，需要深度分析
优点：能理解争议的本质
缺点：仲裁者本身也可能出错
策略 D：置信度比较
适用：Agent能提供置信度评分的场景
优点：简单直接
缺点：Agent可能自信地犯错
策略 E：协商一致（Consensus Building）
适用：需要多方达成一致的决策场景
优点：充分讨论，结果更可靠
缺点：耗时长，可能无法收敛
策略 F：升级人工（Human Escalation）
适用：高风险、高不确定性场景
优点：最终兜底，安全可靠
缺点：增加延迟，依赖人工可用性
4.3 冲突解决策略对比
| 策略 | 准确性 | 延迟 | 成本 | 适用场景 |
|---|---|---|---|---|
| 多数投票 | 中 | 低 | 低 | 有标准答案的分类任务 |
| 加权投票 | 中高 | 低 | 低 | Agent能力不均衡 |
| 仲裁者 | 高 | 中 | 中 | 复杂争议 |
| 置信度比较 | 中 | 极低 | 极低 | Agent有可靠置信度 |
| 协商一致 | 高 | 高 | 高 | 需要多方共识的决策 |
| 升级人工 | 极高 | 极高 | 极高 | 高风险场景兜底 |
推荐组合
：置信度比较（快速筛选）+ 仲裁者（中等争议）+ 升级人工（高风险兜底）
5. 多 Agent 工作流编排
5.1 编排模式
模式 A：顺序流水线（Sequential Pipeline）
Agent-A -> Agent-B -> Agent-C -> 输出
每个 Agent 的输出是下一个 Agent 的输入。
适用：数据处理流水线、ETL、文档处理
优点：简单，易理解，易调试
缺点：串行执行，无法利用并行性
模式 B：并行分发（Parallel Fan-out）
+-- Agent-A --+
Orchestrator -+-- Agent-B --+-> Aggregator -> 输出
+-- Agent-C --+
适用：并行搜索、多源数据采集、多模型推理
优点：大幅降低延迟
缺点：需要聚合逻辑，结果可能冗余
模式 C：路由分发（Router）
+-- Agent-A（处理类型A的任务）
Router ------ +
+-- Agent-B（处理类型B的任务）
适用：多领域客服、智能路由
优点：专业化处理，资源高效
缺点：路由错误会导致错误处理
模式 D：Map-Reduce
+-- Agent-A --+
Input -- +-- Agent-B --+-> Reducer -> 输出
+-- Agent-C --+
适用：大规模数据分析、报告生成、代码审查
优点：可处理大规模输入
缺点：拆分和汇总逻辑需要精心设计
模式 E：辩论/对抗（Debate）
Agent-A（正方）<-> Agent-B（反方）
|
Judge（裁判）-> 最终结论
适用：决策分析、风险评估、方案评审
优点：多角度审视，减少盲区
缺点：耗时长，可能陷入无限争论
模式 F：层次化（Hierarchical）
Supervisor
/ | \
Manager-A Manager-B Manager-C
/ \ | / \
Worker Worker Worker Worker Worker
适用：大型复杂项目、企业级工作流
优点：可扩展，职责清晰
缺点：层级过多增加延迟和复杂度
模式 G：动态规划（Dynamic Planning）
Orchestrator 根据中间结果动态决定下一步
适用：开放式任务、探索性问题
优点：灵活，能应对不确定性
缺点：不可预测，可能无限循环
5.2 编排模式对比
| 模式 | 并行度 | 灵活性 | 复杂度 | 典型场景 |
|---|---|---|---|---|
| 顺序流水线 | 无 | 低 | 低 | 文档处理、ETL |
| 并行分发 | 高 | 低 | 中 | 多源搜索、多模型推理 |
| 路由分发 | 中 | 中 | 中 | 多领域客服 |
| Map-Reduce | 高 | 中 | 中 | 大规模分析 |
| 辩论/对抗 | 中 | 中 | 中 | 决策分析、风险评估 |
| 层次化 | 高 | 中 | 高 | 大型复杂项目 |
| 动态规划 | 中 | 极高 | 高 | 开放式探索任务 |
5.3 工作流定义 DSL（5.3与5.4内容由AI提供）
# 声明式工作流定义
workflow:
name: investment_report
description: 生成投资分析报告
steps:
- id: fetch_data
agent: data_fetcher
inputs:
ticker: "{{ user_input.ticker }}"
outputs: [financial_data, market_data]
- id: analyze_financials
agent: financial_analyst
inputs:
data: "{{ steps.fetch_data.financial_data }}"
outputs: [financial_analysis]
depends_on: [fetch_data]
- id: analyze_market
agent: market_analyst
inputs:
data: "{{ steps.fetch_data.market_data }}"
outputs: [market_analysis]
depends_on: [fetch_data]
- id: risk_assessment
agent: risk_analyst
inputs:
financial: "{{ steps.analyze_financials.financial_analysis }}"
market: "{{ steps.analyze_market.market_analysis }}"
outputs: [risk_report]
depends_on: [analyze_financials, analyze_market]
- id: generate_report
agent: report_writer
inputs:
financial: "{{ steps.analyze_financials.financial_analysis }}"
market: "{{ steps.analyze_market.market_analysis }}"
risk: "{{ steps.risk_assessment.risk_report }}"
outputs: [final_report]
depends_on: [risk_assessment]
error_handling:
on_step_failure: retry
max_retries: 2
fallback: "generate_partial_report"
5.4 工作流引擎核心实现
class WorkflowEngine:
def __init__(self, workflow_def, agent_registry):
self.defn = workflow_def
self.agents = agent_registry
self.step_results = {}
self.step_status = {}
async def execute(self, user_input):
# 构建依赖图
dag = self._build_dag()
# 拓扑排序 + 并行执行
for level in self._topological_levels(dag):
# 同一层级的步骤可以并行执行
tasks = []
for step in level:
if self._dependencies_met(step):
tasks.append(self._execute_step(step, user_input))
await asyncio.gather(*tasks)
return self._get_final_output()
async def _execute_step(self, step, user_input):
self.step_status[step.id] = "running"
try:
# 解析输入（替换模板变量）
inputs = self._resolve_inputs(step, user_input)
# 获取 Agent 并执行
agent = self.agents[step.agent]
result = await agent.process(inputs)
self.step_results[step.id] = result
self.step_status[step.id] = "completed"
except Exception as e:
self.step_status[step.id] = "failed"
await self._handle_error(step, e)
def _build_dag(self):
dag = {}
for step in self.defn["steps"]:
dag[step["id"]] = step.get("depends_on", [])
return dag
def _topological_levels(self, dag):
"""返回拓扑排序的层级列表，每层可并行"""
in_degree = {node: len(deps) for node, deps in dag.items()}
levels = []
while in_degree:
current_level = [
node for node, deg in in_degree.items() if deg == 0
]
if not current_level:
raise ValueError("Workflow has circular dependency")
levels.append(current_level)
for node in current_level:
del in_degree[node]
for other, deps in dag.items():
if node in deps:
in_degree[other] -= 1
return levels
6. 框架选型对比
6.1 主流框架
| 框架 | 语言 | 编排模式 | 特点 | 适用场景 |
|---|---|---|---|---|
| LangGraph | Python | 有状态图 | 精确控制流，检查点/回滚 | 复杂Agent工作流 |
| CrewAI | Python | 角色分工 | 开箱即用，角色定义简单 | 快速原型，团队协作模拟 |
| AutoGen | Python | 对话驱动 | 微软出品，多Agent对话 | 研究探索，对话式协作 |
| OpenAI Swarm | Python | 路由+交接 | 轻量级，Agent间handoff | 客服路由，简单多Agent |
| Dify | 低代码 | 可视化编排 | 拖拽式，非开发人员可用 | 业务人员自建工作流 |
| n8n | 低代码 | 可视化编排 | 开源，丰富的集成节点 | 自动化工作流 |
| TaskWeaver | Python | 代码优先 | 将自然语言转为代码执行 | 数据分析，代码生成 |
6.2 选型决策矩阵
| 维度 | LangGraph | CrewAI | AutoGen | Swarm | Dify |
|---|---|---|---|---|---|
| 学习曲线 | 陡峭 | 平缓 | 中等 | 平缓 | 极平缓 |
| 控制粒度 | 极细 | 粗 | 中 | 中 | 粗 |
| 灵活性 | 极高 | 中 | 高 | 中 | 低 |
| 生产就绪 | 是 | 部分 | 部分 | 实验性 | 是 |
| 可视化 | 否 | 否 | 否 | 否 | 是 |
| 调试能力 | 强 | 弱 | 中 | 弱 | 中 |
| 社区活跃度 | 高 | 中 | 高 | 低 | 高 |
6.3 推荐选型路径
你的需求是什么？
|
+-- 快速验证想法，不需要精确控制 -> CrewAI
|
+-- 业务人员自建工作流，低代码 -> Dify / n8n
|
+-- 需要精确控制每一步，生产级 -> LangGraph
|
+-- 多Agent对话协作研究 -> AutoGen
|
+-- 简单的Agent路由和交接 -> Swarm
|
+-- 代码生成和数据分析 -> TaskWeaver
7. 总结与最佳实践
7.1 核心原则
- 从简单开始 ：能用单 Agent 就不用多 Agent，能用顺序就不用并行
- 明确边界 ：每个 Agent 的职责、输入、输出必须清晰定义
- 松耦合通信 ：优先使用 Pub/Sub 或消息队列，避免 Agent 间直接依赖
- 失败隔离 ：一个 Agent 的失败不应导致整个工作流崩溃
- 可观测性 ：每个 Agent 的输入、输出、耗时、状态可追踪
- 渐进式复杂度 ：先顺序流水线，再并行，再动态规划
7.2 常见反模式
| 反模式 | 问题 | 正确做法 |
|---|---|---|
| Agent 过多 | 通信开销超过并行收益 | 合并职责相近的Agent |
| 循环依赖 | Agent-A等Agent-B，Agent-B等Agent-A | 使用DAG，禁止循环 |
| 上帝编排器 | 编排器逻辑过于复杂 | 让Agent自主决策，编排器只做协调 |
| 同步阻塞 | 一个慢Agent拖慢整个流程 | 设置超时，异步化，降级策略 |
| 状态混乱 | 多个Agent同时修改共享状态 | 使用乐观锁或事件溯源 |
7.3 与 RICE 框架的关系
多 Agent 协作可以嵌入到 RICE 框架的编排层中：
RICE 编排层
|
+-- 单 Agent 模式（简单任务）
|
+-- 多 Agent 模式（复杂任务）
+-- 消息传递：Pub/Sub + 消息队列
+-- 状态同步：会话状态 + 事件溯源
+-- 任务分配：语义路由 + 负载均衡
+-- 冲突解决：置信度比较 + 仲裁者 + 人工兜底
+-- 工作流编排：DAG + 拓扑排序并行执行
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
8
0- 0
已为社区贡献2条内容
所有评论(0)