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

适用场景： 工作流明确的任务（如代码审查流水线、文档生成管线）

2.2 去中心化编排（Peer-to-Peer Pattern）

没有中央编排器，Agent 之间直接通信、协商和协作。每个 Agent 自主决策何时寻求帮助、向谁请求。

class PeerAgent:
    def __init__(self, name: str, capabilities: list[str]):
        self.name = name
        self.capabilities = capabilities
        self.peers: dict[str, 'PeerAgent'] = {}
        self.mailbox: asyncio.Queue = asyncio.Queue()

    def connect(self, peer: 'PeerAgent'):
        self.peers[peer.name] = peer

    async def send(self, target: str, message: dict):
        await self.peers[target].mailbox.put({
            "from": self.name,
            "payload": message
        })

    async def broadcast(self, message: dict):
        for name, peer in self.peers.items():
            await peer.mailbox.put({
                "from": self.name,
                "payload": message,
                "type": "broadcast"
            })

    async def request_help(self, capability: str, task: str) -> str:
        """通过能力发现找到合适的 peer"""
        for name, peer in self.peers.items():
            if capability in peer.capabilities:
                await self.send(name, {
                    "type": "request",
                    "capability": capability,
                    "task": task
                })
                response = await self.mailbox.get()
                return response["payload"].get("result", "")
        raise RuntimeError(f"No peer has capability: {capability}")

    async def listen(self):
        """持续监听消息"""
        while True:
            msg = await self.mailbox.get()
            await self.handle_message(msg)

    async def handle_message(self, msg: dict):
        # 子类实现具体处理逻辑
        pass

适用场景： 开放式探索任务、多视角辩论、自组织系统

2.3 混合模式（Hierarchical Pattern）

层级结构：顶层编排器管理子编排器，子编排器再管理工作 Agent。兼具中心化的可控性和去中心化的灵活性。

        ┌─────────────┐
        │  Super Orc  │
        └──────┬──────┘
       ┌───────┴───────┐
   ┌───┴───┐       ┌───┴───┐
   │Sub-Orc│       │Sub-Orc│
   ├───┬───┤       ├───┬───┤
   │ A │ B │       │ C │ D │
   └───┴───┘       └───┴───┘

3. 模式对比

维度中心化去中心化混合模式
控制复杂度低高中
扩展性受限于编排器良好良好
容错性单点故障高中等
通信开销O(n)O(n²)O(n log n)
一致性保证强最终一致分层一致
适用规模<20 agents<100 agents<1000 agents
实现难度⭐⭐⭐⭐⭐⭐⭐⭐⭐

4. 通信协议设计

4.1 消息格式

from pydantic import BaseModel, Field
from datetime import datetime
from enum import Enum

class MessageType(str, Enum):
    TASK_ASSIGN = "task_assign"
    RESULT_RETURN = "result_return"
    HELP_REQUEST = "help_request"
    STATUS_UPDATE = "status_update"
    ERROR_REPORT = "error_report"

class AgentMessage(BaseModel):
    msg_id: str = Field(description="唯一消息 ID")
    msg_type: MessageType
    sender: str
    receiver: str  # "*" 表示广播
    content: dict
    timestamp: datetime = Field(default_factory=datetime.now)
    reply_to: Optional[str] = None  # 回复的消息 ID
    ttl: int = Field(default=10, description="消息存活跳数，防环路")
    priority: int = Field(default=0, ge=0, le=10)

4.2 基于发布订阅的事件总线

class AgentEventBus:
    def __init__(self):
        self._subscribers: dict[str, list[callable]] = {}
        self._history: list[AgentMessage] = []

    def subscribe(self, topic: str, handler: callable):
        self._subscribers.setdefault(topic, []).append(handler)

    async def publish(self, topic: str, message: AgentMessage):
        self._history.append(message)
        handlers = self._subscribers.get(topic, [])
        results = await asyncio.gather(
            *[h(message) for h in handlers],
            return_exceptions=True
        )
        return results

    def replay(self, topic: str, since: datetime = None) -> list[AgentMessage]:
        """重放历史消息，用于新 Agent 追赶状态"""
        return [
            m for m in self._history
            if (since is None or m.timestamp > since)
        ]

5. 任务分解与分配策略

5.1 基于能力的任务路由

class CapabilityRouter:
    def __init__(self):
        self.registry: dict[str, dict] = {}  # agent_name -> capabilities+load

    def register(self, name: str, capabilities: list[str]):
        self.registry[name] = {
            "capabilities": set(capabilities),
            "load": 0,
            "max_concurrent": 5
        }

    def route(self, task: str, required_capability: str) -> Optional[str]:
        candidates = [
            (name, info) for name, info in self.registry.items()
            if required_capability in info["capabilities"]
            and info["load"] < info["max_concurrent"]
        ]
        if not candidates:
            return None
        # 最少负载优先
        return min(candidates, key=lambda x: x[1]["load"])[0]

    def complete(self, name: str):
        if name in self.registry:
            self.registry[name]["load"] = max(0, self.registry[name]["load"] - 1)

5.2 基于 LLM 的动态任务分解

TASK_DECOMPOSE_PROMPT = """你是一个任务分解专家。给定一个复杂任务，将其分解为可独立执行的子任务。

要求：
1. 每个子任务有明确的输入输出
2. 标注子任务间的依赖关系
3. 指定每个子任务所需的能力标签
4. 子任务数量不超过 7 个

输出 JSON 格式：
{
  "subtasks": [
    {
      "id": "st_1",
      "description": "...",
      "capabilities": ["search", "analysis"],
      "depends_on": [],
      "expected_output": "..."
    }
  ]
}
"""

async def decompose_task(orchestrator_llm, complex_task: str) -> list[dict]:
    response = await orchestrator_llm.chat(
        system=TASK_DECOMPOSE_PROMPT,
        user=complex_task
    )
    plan = parse_json(response)
    return plan["subtasks"]

6. 容错与恢复机制

6.1 超时与重试

class ResilientDispatcher:
    def __init__(self, max_retries: int = 3, timeout: float = 30.0):
        self.max_retries = max_retries
        self.timeout = timeout

    async def execute_with_retry(self, agent: Agent, task: str, context: dict) -> str:
        for attempt in range(self.max_retries):
            try:
                result = await asyncio.wait_for(
                    agent.execute(task, context),
                    timeout=self.timeout * (attempt + 1)
                )
                return result
            except asyncio.TimeoutError:
                if attempt == self.max_retries - 1:
                    raise
                # 指数退避
                await asyncio.sleep(2 ** attempt)
            except Exception as e:
                if attempt == self.max_retries - 1:
                    raise
                # 记录错误并重试
                await asyncio.sleep(1)

class CircuitBreaker:
    """熔断器：连续失败超过阈值时暂停调用"""
    def __init__(self, threshold: int = 5, reset_timeout: float = 60.0):
        self.threshold = threshold
        self.reset_timeout = reset_timeout
        self.failures = 0
        self.last_failure_time = 0
        self.state = "closed"  # closed, open, half-open

    def can_call(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = "half-open"
                return True
            return False
        return True  # half-open

    def record_success(self):
        self.failures = 0
        self.state = "closed"

    def record_failure(self):
        self.failures += 1
        self.last_failure_time = time.time()
        if self.failures >= self.threshold:
            self.state = "open"

7. 实际案例：代码审查多智能体系统

class CodeReviewSystem:
    def __init__(self):
        self.orchestrator = Orchestrator()
        self.orchestrator.register(SecurityAgent("security"))
        self.orchestrator.register(PerformanceAgent("performance"))
        self.orchestrator.register(StyleAgent("style"))
        self.orchestrator.register(SummaryAgent("summary"))

    async def review(self, code: str, language: str) -> dict:
        plan = [
            {"agent": "security", "task": f"审查以下{language}代码的安全漏洞:\n{code}", "depends_on": []},
            {"agent": "performance", "task": f"分析以下{language}代码的性能问题:\n{code}", "depends_on": []},
            {"agent": "style", "task": f"检查以下{language}代码的编码规范:\n{code}", "depends_on": []},
            {"agent": "summary", "task": "汇总所有审查结果，生成报告",
             "depends_on": ["security:审查以下", "performance:分析以下", "style:检查以下"]},
        ]
        return await self.orchestrator.run_pipeline(plan)

8. 总结

多智能体编排架构的选择取决于任务特征：

  • 确定性流程 → 中心化编排，简单可控
  • 开放探索 → 去中心化，激发涌现能力
  • 大规模复杂系统 → 混合模式，分层治理

核心设计原则：高内聚低耦合——每个 Agent 专注自身能力，通过明确的协议通信，由编排策略决定协作方式。在工程实践中，建议从中心化模式起步，随着复杂度增长逐步引入去中心化元素，最终演进到混合架构。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。