Multi-Agent 系统实战：多个 AI Agent 怎么协作

2026-01-22 约 942 字预计阅读 2 分钟

先说什么时候需要 Multi-Agent

不是所有场景都需要多个 Agent。

单一 Agent 能搞定：
  - 任务边界清晰
  - 需要集中决策
  - 工具调用链不长

Multi-Agent 值得考虑：
  - 任务需要不同专业能力
  - 需要并行处理独立子任务
  - 决策需要多方校验
  - 系统规模大，需要职责分离

常见架构

1. Supervisor + Workers

Supervisor Agent
    ├── Worker Agent A（负责某领域）
    ├── Worker Agent B（负责另一领域）
    └── Worker Agent C（负责第三领域）

        
        
        
    
# Supervisor 负责任务分解和协调
class Supervisor:
    def handle(self, task):
        # 分析任务，分配给合适的 Worker
        subtasks = self.decompose(task)
        
        results = []
        for subtask in subtasks:
            worker = self.select_worker(subtask)
            result = worker.execute(subtask)
            results.append(result)
        
        # 汇总结果
        return self.synthesize(results)

适用场景：任务可分解，每个子任务相对独立。

2. Sequential Pipeline

Agent A → Agent B → Agent C → Agent D

        
        
        
    
# 流水线式，每一步依赖前一步输出
pipeline = [
    ("research", ResearchAgent()),
    ("analyze", AnalyzeAgent()),
    ("write", WriteAgent()),
    ("review", ReviewAgent()),
]

def run_pipeline(initial_input):
    output = initial_input
    for name, agent in pipeline:
        output = agent.process(output)
        if agent.needs_retry(output):
            output = agent.retry(output)
    return output

适用场景：有序步骤，每步有明确输入输出。

3. Parallel + Merge

        → Agent A (并行)
Task → → Agent B (并行) → Merge Agent → Result
        → Agent C (并行)

        
        
        
    
# 同一任务多角度处理，结果合并
async def parallel_analyze(task):
    agents = [ResearchAgent(), AnalyzeAgent(), CritiqueAgent()]
    
    # 并行执行
    results = await asyncio.gather(*[
        agent.process(task) for agent in agents
    ])
    
    # 合并结果
    return merge_agent.synthesize(results)

适用场景：需要多角度分析、综合不同观点。

通信协议

Multi-Agent 需要通信。有两种方式：

1. 共享消息队列

        
        
        
    
# 通过共享队列通信
class Agent:
    def __init__(self, name, inbox, outbox):
        self.name = name
        self.inbox = inbox
        self.outbox = outbox
    
    async def run(self):
        while True:
            msg = await self.inbox.get()
            result = await self.process(msg)
            await self.outbox.put(Message(
                from_agent=self.name,
                to="merge",
                content=result
            ))

2. 显式消息传递

        
        
        
    
# Agent 之间直接发消息
class Agent:
    async def send_to(self, target, message):
        await self.messenger.send(
            to=target,
            message=message,
            metadata={"priority": "normal"}
        )
    
    async def receive(self):
        return await self.messenger.receive(from_agent=self.context)

踩过的坑

坑 1：Agent 之间信息丢失

        
        
        
    
# 问题：Agent A 的输出信息，Agent B 没法理解

# 解决：标准化输出格式
class ResearchAgent:
    def output_format(self):
        return {
            "findings": [...],       # 关键发现
            "sources": [...],        # 来源
            "confidence": 0.8,       # 置信度
            "raw_data": {...}       # 原始数据
        }

坑 2：循环依赖

Agent A → Agent B → Agent C
  ↑                     ↓
  └─────────────────────┘

# Agent C 依赖 A 的输出
# Agent A 也间接依赖 C

        
        
        
    
# 解决：用有向无环图（DAG）组织
# 或者引入仲裁机制
class Orchestrator:
    def resolve_conflict(self, agent_a_result, agent_c_result):
        # 冲突时用投票或优先级
        if agent_a_result.confidence > agent_c_result.confidence:
            return agent_a_result
        return agent_c_result

坑 3：调试困难

        
        
        
    
# 问题：多 Agent 出问题时，很难定位是哪个 Agent 的问题

# 解决：每个 Agent 有完整日志，消息队列可回放
class Agent:
    def process(self, msg):
        # 记录输入、输出、时间
        logger.info(f"{self.name} received: {msg}")
        result = self._process_impl(msg)
        logger.info(f"{self.name} output: {result}")
        return result

框架选择

框架	适用场景	特点
LangGraph	复杂流程编排	状态管理强
AutoGen	多 Agent 对话	协作模式丰富
CrewAI	角色分工明确	上手简单
自建	高度定制	灵活但工作量大

结论

Multi-Agent 的价值：让专业的人做专业的事。

多个 Agent 各有所长，组合起来能处理比单一 Agent 更复杂的任务。

但 Multi-Agent 也带来复杂度：通信、调试、一致性都是额外成本。

建议：先用单一 Agent，确定不够用了再拆。不要过度设计。

目录