多Agent系统的编排挑战 单个Agent的能力边界是有限的——一个Agent很难同时擅长写代码、做数据分析和生成报告。多Agent系统通过分工协作突破这个限制，但"如何编排"成了新的核心问题。
编排架构决定了系统的效率、可扩展性和容错能力。选错架构，轻则效率低下，重则系统不可用。
中心化编排：指挥官模式 架构设计 中心化架构中，一个"编排Agent"（Orchestrator）负责全局决策：分配任务给子Agent、收集结果、决定下一步。
[Orchestrator] / | \ [Agent A] [Agent B] [Agent C] ↓ ↓ ↓ [结果A] [结果B] [结果C] \ | / → [综合输出] ← 工程实现 from typing import Any, Protocol from dataclasses import dataclass, field from enum import Enum class AgentRole(Enum): ANALYZER = "analyzer" CODER = "coder" REVIEWER = "reviewer" WRITER = "writer" @dataclass class AgentCapability: role: AgentRole description: str input_schema: dict output_schema: dict @dataclass class TaskResult: agent_id: str success: bool output: Any confidence: float = 0.0 class CentralizedOrchestrator: def __init__(self, agents: dict[str, AgentCapability], llm): self.agents = agents self.llm = llm self.execution_history: list[dict] = [] async def plan(self, task: str) -> list[dict]: """根据任务生成分配计划""" agent_descriptions = "\n".join( f"- {aid}: {cap.description}" for aid, cap in self.agents.items() ) plan = await self.llm( f"任务: {task}\n可用Agent:\n{agent_descriptions}\n" f"请生成分步执行计划，指定每步使用的Agent。" ) return self._parse_plan(plan) async def execute(self, task: str) -> Any: plan = await self.plan(task) context = {"original_task": task, "results": {}} for step in plan: agent_id = step["agent_id"] step_input = self._prepare_input(step, context) result = await self._invoke_agent(agent_id, step_input) context["results"][step["step_id"]] = result # 编排者决策：是否需要调整后续计划 if not result.success: adjustment = await self._handle_failure(step, result, context) if adjustment.get("abort"): return self._partial_result(context) plan = await self._replan(adjustment, context) self.execution_history.append({ "step": step, "result": result }) return await self._synthesize(context) async def _handle_failure(self, step, result, context): """故障处理策略""" if result.confidence < 0.3: return {"abort": True, "reason": "置信度过低"} # 重试或切换Agent alternative = self._find_alternative_agent(step["agent_id"]) return {"retry_with": alternative} if alternative else {"abort": True} async def _synthesize(self, context: dict) -> Any: """汇总所有Agent的输出""" all_results = context["results"] summary = await self.llm( f"原始任务: {context['original_task']}\n" f"各Agent结果: {all_results}\n" f"请综合所有结果，给出最终输出。" ) return summary 优缺点 优点 缺点 全局最优决策 编排者是性能瓶颈 执行顺序清晰 单点故障风险 易于调试追踪 难以并行化 上下文一致性高 扩展性受限 去中心化编排：协作网络模式 架构设计 没有中心编排者，Agent之间直接通信。每个Agent自主决定何时、与谁、如何协作。
...