多 Agent 协作系统的设计模式

为什么需要多 Agent？

单个 Agent 的能力是有限的。当任务变得复杂时，单个 Agent 面临以下挑战：

上下文窗口限制：太多工具和信息塞不进一次 LLM 调用

专业化不足：一个 Agent 很难同时精通数据分析、写作、编程

可靠性下降：任务链越长，单 Agent 越容易在某一步出错

多 Agent 协作通过"分工"来解决这些问题——就像一个团队里有项目经理、工程师、设计师一样。

四种核心设计模式

模式一：主管模式（Supervisor Pattern）

最经典的架构。一个"主管 Agent"负责任务分配和结果汇总，多个"专家 Agent"负责具体执行。

         ┌─────────────┐
         │  主管 Agent  │
         │  (Orchestrator) │
         └──────┬──────┘
                │
    ┌───────────┼───────────┐
    │           │           │
┌───┴───┐ ┌───┴───┐ ┌───┴───┐
│ 编码   │ │ 写作   │ │ 搜索   │
│ Agent  │ │ Agent  │ │ Agent  │
└───────┘ └───────┘ └───────┘

优点：逻辑清晰，主管统一控制流程 缺点：主管成为瓶颈，token 消耗大 适用：任务步骤明确、需要严格质量控制的场景

class SupervisorAgent:
    def __init__(self):
        self.agents = {
            "coder": CoderAgent(),
            "writer": WriterAgent(),
            "researcher": ResearchAgent(),
        }

    def run(self, task: str) -> str:
        # 主管分析任务，决定分工
        plan = self._plan(task)

        results = []
        for step in plan:
            agent = self.agents[step["agent"]]
            result = agent.execute(step["instruction"])
            results.append(result)

        # 主管汇总结果
        return self._synthesize(results)

模式二：群聊模式（Group Chat Pattern）

所有 Agent 在同一个"群"里，通过对话协调工作。谁发言、说什么，由一个调度器决定。

┌──────────────────────────────────┐
│          群聊协调器               │
│  ┌─────┐ ┌─────┐ ┌─────┐       │
│  │Agent│ │Agent│ │Agent│       │
│  │  A  │ │  B  │ │  C  │       │
│  └──┬──┘ └──┬──┘ └──┬──┘       │
│     └───────┼───────┘           │
│          消息总线               │
└──────────────────────────────────┘

优点：灵活，Agent 之间可以互相启发 缺点：容易跑题，token 消耗大 适用：开放式讨论、头脑风暴、创意生成

模式三：流水线模式（Pipeline Pattern）

任务按固定顺序依次经过多个 Agent，每个 Agent 处理完传给下一个。

需求 → Agent A → Agent B → Agent C → 最终结果

优点：简单、可预测、易于调试 缺点：缺乏灵活性，不能跳过或重做 适用：数据处理流水线、文档生成、内容审核

class PipelineAgent:
    def __init__(self):
        self.pipeline = [
            ResearchAgent(),    # 先调研
            OutlineAgent(),     # 再列提纲
            WritingAgent(),     # 然后写作
            ReviewAgent(),      # 最后审查
        ]

    def run(self, topic: str) -> str:
        output = topic
        for agent in self.pipeline:
            output = agent.execute(output)
        return output

模式四：竞争模式（Tournament Pattern）

多个 Agent 同时处理同一个任务，由评审 Agent 选出最佳结果。

         ┌──────────┐
         │ 评审 Agent │
         └────┬─────┘
              │ 选择最佳
    ┌─────────┼─────────┐
    │         │         │
┌───┴──┐ ┌───┴──┐ ┌───┴──┐
│Agent │ │Agent │ │Agent │
│  A   │ │  B   │ │  C   │
└──────┘ └──────┘ └──────┘
    同时处理同一任务

优点：质量最高，鲁棒性好 缺点：成本最高（3倍 token） 适用：质量要求极高的场景（如法律文书、重要报告）

通信协议

多 Agent 之间如何传递信息？主要有三种方式：

1. 结构化消息（推荐）

@dataclass
class AgentMessage:
    sender: str          # 发送者 Agent ID
    receiver: str        # 接收者 Agent ID
    msg_type: str        # 消息类型: task, result, question, answer
    content: str         # 消息内容
    metadata: dict       # 元数据
    timestamp: datetime  # 时间戳

2. 共享状态

所有 Agent 读写同一个状态对象：

shared_state = {
    "task": "分析 Q1 销售数据",
    "data": None,
    "analysis": None,
    "report": None,
    "status": "in_progress"
}

3. 事件驱动

Agent 发布事件，其他 Agent 订阅感兴趣的事件：

event_bus.publish("data.ready", {"path": "sales_q1.csv"})
DataAnalysisAgent 订阅了 "data.ready" 事件，自动触发分析

实战案例：AI 内容团队

下面是一个由 4 个 Agent 组成的内容创作团队：

用户需求: "写一篇关于 AI Agent 的技术博客"
        │
        ▼
┌─────────────────┐
│  主编 Agent      │ ← 接收需求，制定计划
└────────┬────────┘
         │
    ┌────┴────┐
    ▼         ▼
┌────────┐ ┌────────┐
│ 调研   │ │ 大纲   │ ← 同时执行
│ Agent  │ │ Agent  │
└───┬────┘ └───┬────┘
    │          │
    └────┬─────┘
         ▼
┌─────────────────┐
│ 写作 Agent      │ ← 基于调研和大纲撰写
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│ 编辑 Agent      │ ← 审查、润色、优化
└────────┬────────┘
         │
         ▼
    最终文章

最佳实践

保持 Agent 单一职责：每个 Agent 只做一件事，做到极致

定义清晰的接口：输入输出格式要标准化

设置超时和重试：防止某个 Agent 卡住拖垮整个系统

记录完整日志：每个 Agent 的输入输出都要记录，便于调试

渐进式复杂度：先用最简单的流水线，确认可行再升级架构

总结

选择哪种模式取决于你的具体需求。建议从流水线模式开始，随着需求复杂化再逐步升级。

---

想了解更多多 Agent 系统的设计细节？欢迎[联系我们](/zh/contact)获取专业方案。