多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自主决定何时、与谁、如何协作。

[Agent A] ←→ [Agent B]
    ↕           ↕
[Agent C] ←→ [Agent D]

黑板模式实现

import asyncio
from dataclasses import dataclass
from datetime import datetime

@dataclass
class BlackboardEntry:
    agent_id: str
    content: Any
    timestamp: str
    topic: str
    confidence: float = 0.0

class Blackboard:
    """共享黑板：Agent间异步通信媒介"""
    def __init__(self):
        self._entries: dict[str, list[BlackboardEntry]] = {}
        self._subscribers: dict[str, list[asyncio.Queue]] = {}

    async def post(self, topic: str, entry: BlackboardEntry):
        self._entries.setdefault(topic, []).append(entry)
        # 通知订阅者
        for queue in self._subscribers.get(topic, []):
            await queue.put(entry)

    def subscribe(self, topic: str) -> asyncio.Queue:
        queue = asyncio.Queue()
        self._subscribers.setdefault(topic, []).append(queue)
        return queue

class DecentralizedAgent:
    def __init__(self, agent_id: str, role: str, 
                 blackboard: Blackboard, llm):
        self.id = agent_id
        self.role = role
        self.bb = blackboard
        self.llm = llm

    async def run(self, initial_topic: str):
        queue = self.bb.subscribe(initial_topic)
        while True:
            entry = await queue.get()
            if entry is None:  # 终止信号
                break

            # 自主决策：是否处理这条消息
            should_handle = await self._decide(entry)
            if not should_handle:
                continue

            # 处理并发布结果
            result = await self._process(entry)
            output_topic = self._determine_output_topic(result)
            await self.bb.post(output_topic, BlackboardEntry(
                agent_id=self.id,
                content=result,
                timestamp=datetime.now().isoformat(),
                topic=output_topic,
                confidence=result.get("confidence", 0.8)
            ))

    async def _decide(self, entry: BlackboardEntry) -> bool:
        """Agent自主决策是否响应"""
        decision = await self.llm(
            f"你的角色: {self.role}\n"
            f"收到消息: {entry.content}\n"
            f"你是否应该处理？只回答YES或NO。"
        )
        return "YES" in decision.upper()

优缺点

优点 缺点
高可扩展性 全局协调困难
无单点故障 调试复杂
天然并行 可能死锁/活锁
灵活适应变化 一致性难以保证

混合架构：分层编排

设计理念

混合架构在不同层级使用不同模式——宏观用中心化保证一致性，微观用去中心化保证灵活性。

[Top Orchestrator]  ← 中心化：全局规划
    ├── [Team A]  ← 去中心化：团队内协作
    │     ├── Agent A1 ↔ Agent A2
    │     └── Agent A3 ↔ Agent A2
    ├── [Team B]  ← 去中心化：团队内协作
    │     ├── Agent B1 ↔ Agent B2
    │     └── ...
    └── [Team C]

实现

class HybridOrchestrator:
    def __init__(self, llm):
        self.llm = llm
        self.teams: dict[str, TeamCoordinator] = {}

    async def execute(self, task: str):
        # 第一层：中心化分解任务
        subtasks = await self._decompose(task)
        
        # 分配给各团队
        team_tasks = {}
        for st in subtasks:
            team_id = self._assign_team(st)
            team_tasks.setdefault(team_id, []).append(st)

        # 第二层：各团队去中心化执行
        team_results = {}
        results = await asyncio.gather(*[
            self.teams[tid].execute_cooperative(tasks)
            for tid, tasks in team_tasks.items()
        ])
        
        for tid, result in zip(team_tasks.keys(), results):
            team_results[tid] = result

        # 第三层：中心化综合
        final = await self._synthesize(team_results)
        return final

class TeamCoordinator:
    """团队级去中心化协调器"""
    def __init__(self, team_id: str, agents: list, blackboard: Blackboard):
        self.team_id = team_id
        self.agents = agents
        self.bb = blackboard

    async def execute_cooperative(self, subtasks: list[dict]) -> dict:
        # 发布所有子任务到黑板
        for st in subtasks:
            await self.bb.post(
                f"team_{self.team_id}_tasks",
                BlackboardEntry(
                    agent_id="coordinator",
                    content=st,
                    timestamp=datetime.now().isoformat(),
                    topic=f"team_{self.team_id}_tasks"
                )
            )
        
        # 启动所有Agent，等待完成信号
        runners = [agent.run(f"team_{self.team_id}_tasks") 
                   for agent in self.agents]
        
        # 设置完成条件
        results = await self._wait_for_completion(runners)
        return results

三种架构对比

维度 中心化 去中心化 混合架构
决策效率 高（单点决策） 低（需协商） 中（分层决策）
扩展性 差 好 中
容错性 差 好 中
调试难度 易 难 中
一致性 强 弱 可调
适用规模 ≤5 Agent 10-100 Agent 5-50 Agent
典型场景 流水线任务 开放探索 复杂项目

选择决策树

任务是否可分解为独立子任务?
├─ 是 → 子任务间是否有严格依赖?
│   ├─ 是 → 中心化编排（流水线）
│   └─ 否 → 是否需要全局最优?
│       ├─ 是 → 中心化编排（规划型）
│       └─ 否 → 去中心化编排
└─ 否 → 单Agent足够，无需编排

Agent数量 > 10?
├─ 是 → 混合架构（分组+组内去中心化）
└─ 否 → 根据上述决策选择

总结

多Agent编排没有银弹。中心化适合确定性流水线，去中心化适合开放式探索，混合架构适合复杂项目分解。核心设计原则：

  • 匹配任务结构：任务依赖关系决定编排模式
  • 控制Agent数量：5个以下中心化，10个以上考虑混合
  • 定义清晰接口：无论哪种模式，Agent间通信协议要明确
  • 设置超时机制：防止某个Agent卡住整个系统

在实际项目中，建议从中心化起步，随复杂度增长逐步演进到混合架构。不要一上来就设计复杂的去中心化网络——简单系统的可靠性永远高于复杂系统。