多智能体编排架构：从中心化到去中心化的设计模式
1. 引言：为什么需要多智能体编排 单个 LLM Agent 在复杂任务中面临上下文窗口限制、角色混淆、推理链断裂等问题。多智能体架构通过任务分解、角色专精和协作机制，将复杂问题分配给多个专业化 Agent 协同完成。然而，如何编排这些 Agent——谁来调度、如何通信、何时同步——是工程落地的核心挑战。 2. 三种核心编排模式 2.1 中心化编排（Orchestrator Pattern） 一个中央编排器（Orchestrator）负责任务分配、状态管理和结果汇总。所有 Agent 只与编排器通信，互不直接交互。 ┌──────────────────────────────────┐ │ Orchestrator │ │ ┌──────┐ ┌──────┐ ┌──────┐ │ │ │Worker│ │Worker│ │Worker│ │ │ │ A │ │ B │ │ C │ │ │ └──────┘ └──────┘ └──────┘ │ └──────────────────────────────────┘ 核心代码实现： from abc import ABC, abstractmethod from typing import Any, Optional import asyncio class Agent(ABC): def __init__(self, name: str, system_prompt: str): self.name = name self.system_prompt = system_prompt self.message_history: list[dict] = [] @abstractmethod async def execute(self, task: str, context: dict) -> str: pass class Orchestrator: def __init__(self): self.agents: dict[str, Agent] = {} self.task_queue: list[dict] = [] self.results: dict[str, Any] = {} def register(self, agent: Agent): self.agents[agent.name] = agent async def dispatch(self, agent_name: str, task: str, context: dict = None) -> str: agent = self.agents[agent_name] result = await agent.execute(task, context or {}) self.results[f"{agent_name}:{task[:20]}"] = result return result async def run_pipeline(self, plan: list[dict]) -> dict: """按计划顺序执行任务，支持依赖传递""" for step in plan: agent_name = step["agent"] task = step["task"] deps = step.get("depends_on", []) merged_context = {d: self.results.get(d) for d in deps} await self.dispatch(agent_name, task, merged_context) return self.results 适用场景： 工作流明确的任务（如代码审查流水线、文档生成管线） ...
