从单兵作战到团队协作

单个 Agent 能力再强，也有上限：上下文有限、单线程思考、无法并行处理子任务。多 Agent 系统像是从"个人英雄"升级到"特种部队"——每个成员专精一项，协作完成复杂任务。

五种架构模式

模式一：中心化编排（Orchestrator）

         ┌─────────────┐
         │ Orchestrator │ ← 规划、分配、汇总
         └──┬──┬──┬──┬─┘
            │  │  │  │
     ┌──────┘  │  │  └──────┐
     ↓         ↓  ↓         ↓
  ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐
  │Agent│ │Agent│ │Agent│ │Agent│
  │  A  │ │  B  │ │  C  │ │  D  │
  └─────┘ └─────┘ └─────┘ └─────┘

class Orchestrator:
    def __init__(self):
        self.agents = {
            "researcher": ResearchAgent(),
            "writer": WriterAgent(),
            "reviewer": ReviewerAgent(),
            "fact_checker": FactCheckAgent(),
        }
    
    async def run(self, task):
        # 1. 分解任务
        subtasks = self.decompose(task)
        
        # 2. 分配给专业 Agent
        results = {}
        for subtask in subtasks:
            agent = self.assign(subtask)
            results[subtask.id] = await agent.run(subtask)
        
        # 3. 汇总结果
        return self.synthesize(results)

# 适用场景：报告撰写、项目管理
# 优点：控制清晰，易于调试
# 缺点：Orchestrator 是瓶颈和单点故障

模式二：流水线（Pipeline）

Input → [Agent A] → [Agent B] → [Agent C] → Output
        搜索        写作        审校

class Pipeline:
    def __init__(self):
        self.stages = [
            ResearchAgent(),    # 搜索资料
            OutlineAgent(),     # 生成大纲
            DraftAgent(),       # 写初稿
            ReviewAgent(),      # 审校修改
            FactCheckAgent(),   # 事实核查
        ]
    
    async def run(self, input):
        data = input
        for stage in self.stages:
            data = await stage.process(data)
            if data.rejected:
                # 回退到上一阶段
                data = await self.stages[stage.id - 1].rework(data)
        return data

# 适用场景：内容生产、数据处理
# 优点：简单直观，每个阶段可独立优化
# 缺点：串行执行，延迟高

模式三：辩论（Debate）

         ┌───────────┐
         │   Judge    │
         └──┬────┬───┘
            │    │
     ┌──────┘    └──────┐
     ↓                  ↓
┌─────────┐        ┌─────────┐
│ Agent A  │ ←→    │ Agent B  │
│ (正方)   │ 辩论  │ (反方)   │
└─────────┘        └─────────┘

class DebateSystem:
    async def run(self, question):
        pro_agent = Agent(role="supporter")
        con_agent = Agent(role="opponent")
        judge = Agent(role="judge")
        
        rounds = []
        for round_num in range(3):  # 3 轮辩论
            pro_arg = await pro_agent.argue(question, rounds, side="pro")
            con_arg = await con_agent.argue(question, rounds, side="con")
            rounds.append({"pro": pro_arg, "con": con_arg})
        
        verdict = await judge.evaluate(question, rounds)
        return verdict

# 适用场景：决策支持、风险评估
# 优点：减少偏见，多角度分析
# 缺点：Token 消耗是单 Agent 的 3-5 倍

模式四：层级委托（Hierarchical）

                ┌──────────┐
                │  Manager  │
                └────┬─────┘
            ┌────────┼────────┐
            ↓        ↓        ↓
        ┌──────┐ ┌──────┐ ┌──────┐
        │Team A │ │Team B │ │Team C │
        │Lead  │ │Lead  │ │Lead  │
        └──┬───┘ └──┬───┘ └──┬───┘
           ↓        ↓        ↓
        Workers  Workers  Workers

class Manager:
    async def handle(self, task):
        if self.can_handle(task):
            return self.do(task)
        
        # 委托给子团队
        team = self.select_team(task)
        result = await team.lead.handle(task)
        return result

class TeamLead:
    async def handle(self, task):
        subtasks = self.split(task)
        results = await asyncio.gather(*[
            worker.run(t) for t, worker in zip(subtasks, self.workers)
        ])
        return self.merge(results)

# 适用场景：复杂项目、大规模任务
# 优点：可扩展性好，模拟人类组织
# 缺点：通信开销大，调试困难

模式五：自由协作（Swarm）

    ┌────────┐
    │ Agent A │←──→┌────────┐
    └────────┘    │ Agent B │
       ↑ ↓        └────────┘
       │ ↑ ↓         ↑
    ┌──┴───┐    ┌───┴──┐
    │Agent C│←──→│Agent D│
    └────────┘    └──────┘

class SwarmMessageBus:
    """Agent 之间通过消息总线自由通信"""
    def __init__(self):
        self.agents = {}
        self.messages = asyncio.Queue()
    
    async def broadcast(self, sender, content):
        for agent_id, agent in self.agents.items():
            if agent_id != sender:
                await agent.receive({
                    "from": sender,
                    "content": content,
                })
    
    async def run(self, task):
        # 所有 Agent 同时启动，自由协作
        await asyncio.gather(*[
            agent.start(task, self) for agent in self.agents.values()
        ])

# 适用场景：开放式探索、创意协作
# 优点：涌现行为，灵活性极高
# 缺点：不可预测，难以控制

通信机制

消息格式

@dataclass
class AgentMessage:
    sender: str           # 发送者 ID
    receiver: str         # 接收者 ID 或 "broadcast"
    type: str             # request / response / notify
    content: str          # 消息内容
    context: dict         # 上下文（任务ID、对话历史等）
    timestamp: float      # 时间戳
    priority: int = 0     # 优先级

通信协议

class AgentProtocol:
    """简化的 Agent 间通信协议"""
    
    async def request(self, target, action, params):
        """请求-响应模式"""
        msg = AgentMessage(
            sender=self.id,
            receiver=target,
            type="request",
            content=json.dumps({"action": action, "params": params}),
        )
        response = await self.bus.send_and_wait(msg, timeout=30)
        return response
    
    async def notify(self, target, event):
        """通知模式（不需要响应）"""
        msg = AgentMessage(
            sender=self.id,
            receiver=target,
            type="notify",
            content=json.dumps({"event": event}),
        )
        await self.bus.send(msg)
    
    async def broadcast(self, event):
        """广播模式"""
        msg = AgentMessage(
            sender=self.id,
            receiver="broadcast",
            type="notify",
            content=json.dumps({"event": event}),
        )
        await self.bus.send(msg)

冲突处理

当多个 Agent 给出矛盾结果：

class ConflictResolver:
    def resolve(self, results: list[AgentResult]):
        if len(results) == 1:
            return results[0]
        
        # 策略1：投票
        if all(r.type == "decision" for r in results):
            return self.majority_vote(results)
        
        # 策略2：置信度加权
        weighted = sorted(
            results,
            key=lambda r: r.confidence * r.agent_reputation,
            reverse=True
        )
        return weighted[0]
        
        # 策略3：仲裁 Agent
        return self.arbitrator.judge(results)

状态管理

多 Agent 系统需要共享状态：

class SharedState:
    """所有 Agent 共享的状态存储"""
    def __init__(self):
        self.store = {}
        self.locks = {}
    
    async def get(self, key):
        async with self.locks[key]:
            return self.store.get(key)
    
    async def set(self, key, value, agent_id):
        async with self.locks[key]:
            self.store[key] = value
            await self.log_change(key, agent_id, value)
    
    async def get_history(self, key):
        """获取某个状态的变更历史"""
        return self.store.get(f"{key}__history", [])

成本控制

多 Agent 系统的 Token 消耗是单 Agent 的 N 倍以上：

class BudgetManager:
    def __init__(self, total_budget):
        self.budget = total_budget
        self.spent = 0
    
    async def call_agent(self, agent, task):
        estimated_cost = self.estimate(task)
        
        if self.spent + estimated_cost > self.budget:
            # 降级：用更便宜的模型
            agent.downgrade_model()
            estimated_cost = self.estimate(task)
        
        if self.spent + estimated_cost > self.budget:
            raise BudgetExceeded()
        
        result = await agent.run(task)
        self.spent += result.tokens_used * result.price_per_token
        return result

实际案例：研究报告生成

async def generate_research_report(topic):
    # 5 个 Agent 协作
    system = MultiAgentSystem()
    
    system.add_agent("researcher", ResearchAgent(model="gpt-4o"))
    system.add_agent("analyst", AnalysisAgent(model="gpt-4o"))
    system.add_agent("writer", WritingAgent(model="claude-4"))
    system.add_agent("reviewer", ReviewAgent(model="gpt-4o-mini"))
    system.add_agent("fact_checker", FactCheckAgent(model="gpt-4o"))
    
    # 流水线 + 反馈循环
    research = await system["researcher"].run(topic)
    analysis = await system["analyst"].run(research)
    draft = await system["writer"].run(analysis)
    
    # 并行审校
    review, fact_check = await asyncio.gather(
        system["reviewer"].run(draft),
        system["fact_checker"].run(draft),
    )
    
    # 根据反馈修改
    if review.needs_revision or fact_check.has_issues:
        feedback = merge_feedback(review, fact_check)
        draft = await system["writer"].revise(draft, feedback)
    
    return draft

# 成本：~$0.50/报告
# 耗时：~2 分钟
# 质量：远超单 Agent 生成

结论

多 Agent 不是银弹。在以下情况才值得用：

  • 任务可分解：子任务之间有清晰边界
  • 专业分工明确：不同 Agent 擅长不同领域
  • 质量要求高：值得付出 N 倍成本换取更高质量
  • 并行收益大：子任务可并行执行

否则，一个强模型 + 好的 Prompt 往往比多 Agent 更经济。多 Agent 是手段，不是目的。

加入讨论

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

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