从单兵作战到团队协作 单个 Agent 能力再强，也有上限：上下文有限、单线程思考、无法并行处理子任务。多 Agent 系统像是从"个人英雄"升级到"特种部队"——每个成员专精一项，协作完成复杂任务。
五种架构模式 模式一：中心化编排（Orchestrator） ┌─────────────┐ │ Orchestrator │ ← 规划、分配、汇总 └──┬──┬──┬──┬─┘ │ │ │ │ ┌──────┘ │ │ └──────┐ ↓ ↓ ↓ ↓ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ │Agent│ │Agent│ │Agent│ │Agent│ │ A │ │ B │ │ C │ │ D │ └─────┘ └─────┘ └─────┘ └─────┘ class Orchestrator: def __init__(self): self.agents = { "researcher": ResearchAgent(), "writer": WriterAgent(), "reviewer": ReviewerAgent(), "fact_checker": FactCheckAgent(), } async def run(self, task): # 1. 分解任务 subtasks = self.decompose(task) # 2. 分配给专业 Agent results = {} for subtask in subtasks: agent = self.assign(subtask) results[subtask.id] = await agent.run(subtask) # 3. 汇总结果 return self.synthesize(results) # 适用场景：报告撰写、项目管理 # 优点：控制清晰，易于调试 # 缺点：Orchestrator 是瓶颈和单点故障 模式二：流水线（Pipeline） Input → [Agent A] → [Agent B] → [Agent C] → Output 搜索 写作 审校 class Pipeline: def __init__(self): self.stages = [ ResearchAgent(), # 搜索资料 OutlineAgent(), # 生成大纲 DraftAgent(), # 写初稿 ReviewAgent(), # 审校修改 FactCheckAgent(), # 事实核查 ] async def run(self, input): data = input for stage in self.stages: data = await stage.process(data) if data.rejected: # 回退到上一阶段 data = await self.stages[stage.id - 1].rework(data) return data # 适用场景：内容生产、数据处理 # 优点：简单直观，每个阶段可独立优化 # 缺点：串行执行，延迟高 模式三：辩论（Debate） ┌───────────┐ │ Judge │ └──┬────┬───┘ │ │ ┌──────┘ └──────┐ ↓ ↓ ┌─────────┐ ┌─────────┐ │ Agent A │ ←→ │ Agent B │ │ (正方) │ 辩论 │ (反方) │ └─────────┘ └─────────┘ class DebateSystem: async def run(self, question): pro_agent = Agent(role="supporter") con_agent = Agent(role="opponent") judge = Agent(role="judge") rounds = [] for round_num in range(3): # 3 轮辩论 pro_arg = await pro_agent.argue(question, rounds, side="pro") con_arg = await con_agent.argue(question, rounds, side="con") rounds.append({"pro": pro_arg, "con": con_arg}) verdict = await judge.evaluate(question, rounds) return verdict # 适用场景：决策支持、风险评估 # 优点：减少偏见，多角度分析 # 缺点：Token 消耗是单 Agent 的 3-5 倍 模式四：层级委托（Hierarchical） ┌──────────┐ │ Manager │ └────┬─────┘ ┌────────┼────────┐ ↓ ↓ ↓ ┌──────┐ ┌──────┐ ┌──────┐ │Team A │ │Team B │ │Team C │ │Lead │ │Lead │ │Lead │ └──┬───┘ └──┬───┘ └──┬───┘ ↓ ↓ ↓ Workers Workers Workers class Manager: async def handle(self, task): if self.can_handle(task): return self.do(task) # 委托给子团队 team = self.select_team(task) result = await team.lead.handle(task) return result class TeamLead: async def handle(self, task): subtasks = self.split(task) results = await asyncio.gather(*[ worker.run(t) for t, worker in zip(subtasks, self.workers) ]) return self.merge(results) # 适用场景：复杂项目、大规模任务 # 优点：可扩展性好，模拟人类组织 # 缺点：通信开销大，调试困难 模式五：自由协作（Swarm） ┌────────┐ │ Agent A │←──→┌────────┐ └────────┘ │ Agent B │ ↑ ↓ └────────┘ │ ↑ ↓ ↑ ┌──┴───┐ ┌───┴──┐ │Agent C│←──→│Agent D│ └────────┘ └──────┘ class SwarmMessageBus: """Agent 之间通过消息总线自由通信""" def __init__(self): self.agents = {} self.messages = asyncio.Queue() async def broadcast(self, sender, content): for agent_id, agent in self.agents.items(): if agent_id != sender: await agent.receive({ "from": sender, "content": content, }) async def run(self, task): # 所有 Agent 同时启动，自由协作 await asyncio.gather(*[ agent.start(task, self) for agent in self.agents.values() ]) # 适用场景：开放式探索、创意协作 # 优点：涌现行为，灵活性极高 # 缺点：不可预测，难以控制 通信机制 消息格式 @dataclass class AgentMessage: sender: str # 发送者 ID receiver: str # 接收者 ID 或 "broadcast" type: str # request / response / notify content: str # 消息内容 context: dict # 上下文（任务ID、对话历史等） timestamp: float # 时间戳 priority: int = 0 # 优先级 通信协议 class AgentProtocol: """简化的 Agent 间通信协议""" async def request(self, target, action, params): """请求-响应模式""" msg = AgentMessage( sender=self.id, receiver=target, type="request", content=json.dumps({"action": action, "params": params}), ) response = await self.bus.send_and_wait(msg, timeout=30) return response async def notify(self, target, event): """通知模式（不需要响应）""" msg = AgentMessage( sender=self.id, receiver=target, type="notify", content=json.dumps({"event": event}), ) await self.bus.send(msg) async def broadcast(self, event): """广播模式""" msg = AgentMessage( sender=self.id, receiver="broadcast", type="notify", content=json.dumps({"event": event}), ) await self.bus.send(msg) 冲突处理 当多个 Agent 给出矛盾结果：
...
