多智能体协作系统：五大编排架构深度解析（2026）

from ag2 import AssistantAgent, UserProxyAgent

# 编排器
supervisor = AssistantAgent(
    name="supervisor",
    system_message="""你是任务编排器。
    1. 将用户请求分解为子任务
    2. 分配给合适的专家Agent
    3. 聚合所有结果输出最终答案""",
    llm_config={"model": "gpt-4o"}
)

researcher = AssistantAgent(
    name="researcher",
    system_message="你是研究专家，负责信息检索和数据分析。",
    llm_config={"model": "gpt-4o"}
)

writer = AssistantAgent(
    name="writer",
    system_message="你是写作专家，负责内容创作和润色。",
    llm_config={"model": "gpt-4o"}
)

# 通过GroupChat实现编排
from ag2 import GroupChat, GroupChatManager
group_chat = GroupChat(
    agents=[supervisor, researcher, writer],
    messages=[], max_round=10
)
manager = GroupChatManager(groupchat=group_chat)

class PipelineStage:
    def __init__(self, name: str, model: str = "gpt-4o-mini"):
        self.name = name
        self.model = model

    async def execute(self, input_data: dict) -> dict:
        if not self._quality_gate(input_data):
            raise ValueError(f"[{self.name}] 质量检查失败")
        return await self._process(input_data)

class ContentPipeline:
    """研究 -> 大纲 -> 写作 -> 审核 -> 发布"""
    def __init__(self):
        self.stages = [
            PipelineStage("researcher"),
            PipelineStage("outliner"),
            PipelineStage("writer"),
            PipelineStage("editor"),
            PipelineStage("publisher"),
        ]

    async def run(self, topic: str) -> str:
        data = {"topic": topic}
        for stage in self.stages:
            data = await stage.execute(data)
            await shared_storage.save(f"pipeline/{stage.name}.json", data)
        return data["final_output"]

@dataclass
class BlackboardEntry:
    agent_id: str
    hypothesis: str
    evidence: str
    confidence: float

class SharedBlackboard:
    """共享黑板 - 蜂群Agent的协调核心"""
    def __init__(self):
        self.entries: List[BlackboardEntry] = []
        self.convergence_threshold = 0.85

    def write(self, entry: BlackboardEntry):
        self.entries.append(entry)

    def read_recent(self, limit: int = 10) -> List[BlackboardEntry]:
        return self.entries[-limit:]

    def is_converged(self) -> bool:
        if len(self.entries) < 3:
            return False
        recent = self.entries[-5:]
        avg_conf = sum(e.confidence for e in recent) / len(recent)
        return avg_conf >= self.convergence_threshold

async def research_swarm(query: str, num_agents: int = 20):
    bb = SharedBlackboard()
    tasks = [swarm_agent(f"agent-{i}", query, bb) for i in range(num_agents)]
    await asyncio.gather(*tasks)
    return synthesize(bb.entries)

class MeshAgent:
    """网格Agent - 显式对等通信"""
    def __init__(self, name: str):
        self.name = name
        self.peers: Dict[str, asyncio.Queue] = {}
        self.handlers: Dict[str, Callable] = {}

    def connect(self, peer: 'MeshAgent'):
        q1, q2 = asyncio.Queue(), asyncio.Queue()
        self.peers[peer.name] = q1
        peer.peers[self.name] = q2
        self.inbox, peer.inbox = q2, q1

    async def send(self, peer_name: str, msg: AgentMessage):
        await self.peers[peer_name].put(msg)

# 构建编码网格：规划-编码-测试 三角迭代环
planner, coder, tester = MeshAgent("planner"), MeshAgent("coder"), MeshAgent("tester")
planner.connect(coder)
coder.connect(tester)
tester.connect(planner)

from crewai import Agent, Task, Crew, Process

ceo = Agent(role="项目总监", goal="将高层目标分解为部门任务",
            allow_delegation=True)
eng_manager = Agent(role="工程经理", goal="将需求转化为开发任务",
                    allow_delegation=True)
backend_dev = Agent(role="后端开发者", goal="实现API和数据层",
                   tools=[code_search_tool, file_write_tool])
qa_engineer = Agent(role="QA工程师", goal="编写和执行测试",
                    tools=[test_runner_tool])

crew = Crew(
    agents=[ceo, eng_manager, backend_dev, qa_engineer],
    tasks=[Task(description="设计和实现用户认证系统", agent=ceo)],
    process=Process.hierarchical,
    manager_llm="gpt-4o"
)

+-------------------------------------------------------------+
|              多智能体协作系统架构全景                          |
+-------------------------------------------------------------+
|  1. 编排器-工作者        2. 管道模式                          |
|  +---------+           +---+ +---+ +---+ +---+              |
|  | 编排器   |           |研 |>|写 |>|审 |>|发 |              |
|  | Agent   |           |究 | |作 | |核 | |布 |              |
|  +--/-+-\--+           +---+ +---+ +---+ +---+              |
|   A   B   C            阶段隔离|质量门控|线性流转             |
|                                                             |
|  3. 蜂群+黑板           4. 网格模式                           |
|  +-----------+         +---+ <--> +---+                      |
|  | Blackboard|         | A |      | B |                      |
|  +^--^--^--^-+         +-^-+ <--> +-^-+                      |
|  A1 A2 A3 AN           | C | <--> | D |                      |
|  涌现行为 收敛判断       显式连接 对等通信                     |
|                                                             |
|  5. 层级模式                                                  |
|  +------------+                                              |
|  |  CEO Agent | <-- 战略层                                    |
|  +--+-----+---+                                              |
|  |工程M| |内容M|  <-- 战术层                                   |
|  BF FE   WR ED   <-- 执行层                                  |
+-------------------------------------------------------------+
|  +----------+  +----------+  +----------+  +----------+      |
|  |共享存储   |  |状态管理   |  |A2A协议   |  |监控/追踪  |      |
|  +----------+  +----------+  +----------+  +----------+      |
+-------------------------------------------------------------+