多Agent协作架构模式实战：从分层规划到动态编排的完整指南

单Agent能力再强，也有天花板。2026年，多Agent协作已从学术概念走向生产实践。本文从4种核心架构模式到代码实现，带你掌握多Agent系统的设计精髓。

一、为什么需要多Agent协作？

1.1 单Agent的极限

单个AI Agent就像一个全栈工程师——什么都能做，但什么都不精。当任务复杂度超过一定阈值，单Agent面临：

• 上下文窗口溢出：复杂任务的上下文轻松超过100K tokens
• 角色混淆：同时扮演分析师、开发者、测试者，prompt越来越长，效果越来越差
• 错误累积：单链推理中一步出错，后续全部偏离
• 无法并行：串行执行，效率低下

多Agent协作的核心思想：专业分工 + 协调机制 = 超越单体的涌现能力。

1.2 2026年的产业现状

多Agent不再是Demo。Salesforce的Agentforce、Microsoft的AutoGen生产版、Google的Vertex AI Agent Builder都已在企业中部署多Agent系统。Gartner预测，到2027年将有30%的企业AI应用采用多Agent架构。

二、四种核心协作模式

模式一：Supervisor（监督者模式）

核心思想：一个"老板"Agent负责任务分解和分配，多个"员工"Agent各司其职。

适用场景：复杂的多步骤任务，如研究报告生成、代码审查流程。

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import create_react_agent

# 定义专业Agent
researcher = create_react_agent(
    model, tools=[web_search, arxiv_search],
    prompt="你是一个资深研究员，专注于收集和分析技术资料。"
)

writer = create_react_agent(
    model, tools=[write_document, format_markdown],
    prompt="你是一个技术写作专家，擅长将复杂概念转化为清晰文章。"
)

reviewer = create_react_agent(
    model, tools=[check_facts, grammar_check],
    prompt="你是一个严格的审稿人，确保内容的准确性和质量。"
)

# Supervisor逻辑
def supervisor(state):
    """监督者决定下一步由谁执行"""
    messages = state["messages"]

    response = model.invoke([
        SystemMessage(content="""你是团队leader。根据当前进度决定：
        - 需要更多研究 → 'researcher'
        - 可以开始写作 → 'writer'  
        - 需要审核 → 'reviewer'
        - 任务完成 → 'FINISH'"""),
        *messages
    ])

    return {"next": response.content}

# 构建图
graph = StateGraph(AgentState)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_node("reviewer", reviewer)

graph.add_conditional_edges("supervisor", 
    lambda state: state["next"],
    {"researcher": "researcher", "writer": "writer", 
     "reviewer": "reviewer", "FINISH": END})

# 所有Agent完成后都回到Supervisor
for agent in ["researcher", "writer", "reviewer"]:
    graph.add_edge(agent, "supervisor")

graph.set_entry_point("supervisor")
app = graph.compile()

优势：

• 集中控制，流程清晰
• 易于调试（所有决策经过Supervisor）
• 天然支持任务优先级

劣势：

• Supervisor是单点瓶颈
• Supervisor的LLM调用次数 = 所有轮次之和
• 不适合需要Agent之间直接对话的场景

模式二：Handoff（交接模式）

核心思想：Agent之间可以直接"交接"控制权，无需中间人。OpenAI Agents SDK的核心设计模式。

from openai import agents

# 定义Agent及其交接规则
triage_agent = agents.Agent(
    name="分诊Agent",
    instructions="根据用户需求，交接给合适的专业Agent。",
    handoffs=[
        agents.Handoff(target="tech_agent", 
                      description="技术问题"),
        agents.Handoff(target="sales_agent",
                      description="销售咨询"),
        agents.Handoff(target="support_agent",
                      description="售后支持"),
    ]
)

tech_agent = agents.Agent(
    name="技术Agent",
    instructions="你是技术专家。解决技术问题，必要时交回分诊。",
    tools=[search_docs, run_code],
    handoffs=[
        agents.Handoff(target="triage_agent",
                      description="非技术问题，交回分诊")
    ]
)

sales_agent = agents.Agent(
    name="销售Agent",
    instructions="你是销售顾问。处理定价、方案咨询。",
    tools=[get_pricing, create_quote],
    handoffs=[
        agents.Handoff(target="triage_agent",
                      description="非销售问题，交回分诊"),
        agents.Handoff(target="tech_agent",
                      description="需要技术确认")
    ]
)

# 运行
result = agents.run(triage_agent, "我想了解你们的API性能指标")

优势：

• 去中心化，Agent之间直接通信
• 延迟低（不经过中间层）
• 自然模拟人类团队协作

劣势：

• 交接链可能形成环路
• 调试困难（难以追踪控制流）
• 需要每个Agent都理解何时交接

模式三：Router（路由器模式）

核心思想：一个轻量级路由层根据输入特征快速分发到专业Agent，不参与后续处理。

# 高效路由器实现——不用LLM做路由决策
import re
from typing import Dict, Callable

class AgentRouter:
    def __init__(self):
        self.routes: list[tuple[Callable, str]] = []
        self.agents: Dict[str, Agent] = {}

    def register(self, agent_name: str, agent: Agent, 
                 matcher: Callable[[str], bool]):
        self.agents[agent_name] = agent
        self.routes.append((matcher, agent_name))

    def route(self, query: str) -> str:
        # 规则优先（快速、可预测、零成本）
        for matcher, agent_name in self.routes:
            if matcher(query):
                return agent_name
        # fallback到LLM路由
        return self._llm_route(query)

    def _llm_route(self, query: str) -> str:
        """仅在规则匹配失败时使用LLM"""
        response = model.invoke(
            f"将以下查询分类到: {list(self.agents.keys())}\n查询: {query}"
        )
        return response.content.strip()

# 注册Agent和路由规则
router = AgentRouter()
router.register("code_agent", code_agent, 
    lambda q: any(kw in q for kw in ["代码", "bug", "编程", "函数"]))
router.register("data_agent", data_agent,
    lambda q: any(kw in q for kw in ["数据", "SQL", "查询", "报表"]))
router.register("infra_agent", infra_agent,
    lambda q: any(kw in q for kw in ["部署", "服务器", "K8s", "Docker"]))

优势：

• 路由延迟极低（规则匹配<1ms）
• 可预测性强
• 易于监控和计费

劣势：

• 规则维护成本
• 不支持多Agent协作
• 一次只能选一个Agent

模式四：Swarm（群体智能模式）

核心思想：多个Agent同时工作，通过共享状态或消息总线协调，无中心控制。

import asyncio
from dataclasses import dataclass, field

@dataclass
class SharedBlackboard:
    """共享黑板——所有Agent都可以读写"""
    findings: list = field(default_factory=list)
    tasks: list = field(default_factory=list)
    status: dict = field(default_factory=dict)
    lock: asyncio.Lock = field(default_factory=asyncio.Lock)

    async def post(self, agent_name: str, content: dict):
        async with self.lock:
            self.findings.append({
                "agent": agent_name,
                "content": content,
                "timestamp": time.time()
            })

    async def get_recent(self, n: int = 10) -> list:
        return self.findings[-n:]

# Swarm中的每个Agent独立运行
async def research_swarm(topic: str, num_agents: int = 5):
    blackboard = SharedBlackboard()

    agents = [
        ("学术Agent", search_arxiv),
        ("新闻Agent", search_news),
        ("GitHub Agent", search_github),
        ("专利Agent", search_patents),
        ("社交Agent", search_twitter),
    ]

    async def agent_worker(name, search_fn):
        # 独立搜索
        results = await search_fn(topic)
        await blackboard.post(name, {"raw": results})

        # 查看其他Agent的发现，补充研究
        others = await blackboard.get_recent()
        gaps = identify_gaps(results, others)
        if gaps:
            supplementary = await search_fn(gaps)
            await blackboard.post(name, {"supplementary": supplementary})

    # 所有Agent并行执行
    await asyncio.gather(*[
        agent_worker(name, fn) for name, fn in agents
    ])

    # 汇总Agent合成最终结果
    return await synthesize(blackboard.findings)

优势：

• 真正并行，效率最高
• 去中心化，无单点故障
• Agent之间可以互相启发

劣势：

• 协调复杂
• 可能重复工作
• 结果合成是难点

三、模式对比：如何选择？

维度	Supervisor	Handoff	Router	Swarm
控制方式	集中式	分布式	入口集中	去中心化
延迟	高（多轮决策）	中	低	低（并行）
复杂度	中	高	低	最高
可调试性	⭐⭐⭐⭐⭐	⭐⭐	⭐⭐⭐⭐	⭐⭐
可扩展性	中	高	高	最高
LLM成本	高	中	低	高
适用场景	复杂流程	客服/对话	分类分发	研究/创意

选型决策树

你的场景是什么？
├── 需要严格流程控制 → Supervisor
├── 用户对话/客服 → Handoff  
├── 简单分类分发 → Router
├── 需要并行探索 → Swarm
└── 不确定 → 先Router，再升级Supervisor

四、通信协议：MCP与A2A

多Agent系统需要标准化通信协议：

4.1 MCP（AI ↔ 工具）

# Agent通过MCP调用外部工具
async def agent_with_mcp_tools():
    async with MCPClient("http://tools-server:8000/mcp") as client:
        tools = await client.list_tools()
        result = await client.call_tool("search", {"query": "AI trends"})

4.2 A2A（Agent ↔ Agent）

Google的A2A协议专门解决Agent间通信：

# A2A协议核心概念
class A2AMessage:
    sender: str          # 发送者Agent ID
    receiver: str        # 接收者Agent ID  
    task_id: str         # 任务标识
    content: dict        # 消息内容
    artifacts: list      # 附带的工件（文件、数据等）

# Agent注册和发现
class A2ARegistry:
    async def register(self, agent_card: AgentCard):
        """注册Agent的能力描述"""
        # AgentCard包含：名称、能力列表、输入输出Schema
        pass

    async def discover(self, capability: str) -> list[AgentCard]:
        """发现具有特定能力的Agent"""
        pass

4.3 MCP + A2A 协同

用户请求
  ↓
Supervisor Agent
  ├── A2A → Research Agent ── MCP → Web Search Tool
  ├── A2A → Code Agent ── MCP → IDE Tool  
  └── A2A → Review Agent ── MCP → Linting Tool
  ↓
聚合结果 → 用户

五、框架对比：谁更适合多Agent？

框架	核心模式	状态管理	人工介入	生产就绪
LangGraph	Supervisor+自定义	检查点持久化	⭐⭐⭐⭐⭐	⭐⭐⭐⭐⭐
CrewAI	角色扮演+流程	内置记忆	⭐⭐⭐	⭐⭐⭐⭐
AG2	对话驱动	GroupChat	⭐⭐⭐⭐	⭐⭐⭐⭐
OpenAI Agents	Handoff	Runner	⭐⭐⭐	⭐⭐⭐
Bedrock Agents	托管编排	会话状态	⭐⭐⭐⭐	⭐⭐⭐⭐⭐

推荐：

• 需要最大灵活性 → LangGraph
• 快速原型+角色定义 → CrewAI
• 对话式协作 → AG2
• 简洁的交接流程 → OpenAI Agents SDK
• AWS生态+托管 → Bedrock Agents

六、生产实践：五个关键教训

教训1：从2个Agent开始，不要一步到位

# ❌ 错误：一开始就搞10个Agent
agents = [research, write, review, format, translate, 
          illustrate, seo, publish, notify, archive]

# ✅ 正确：从最小可用集开始
agents = [research, write]  # 先验证核心流程
# 然后逐步增加：review → format → publish

教训2：Agent之间的Schema必须严格定义

from pydantic import BaseModel

class ResearchOutput(BaseModel):
    """研究Agent的输出格式——下游Agent依赖这个结构"""
    topic: str
    key_findings: list[str]
    sources: list[dict]  # {"url": str, "credibility": float}
    confidence: float  # 0-1

class WriteInput(BaseModel):
    """写作Agent的输入——必须匹配Research的输出"""
    research: ResearchOutput
    style: str = "technical"
    max_words: int = 3000

教训3：设置明确的终止条件

MAX_ROUNDS = 10  # 防止无限循环
MAX_TOKENS = 50000  # 防止token爆炸

def should_terminate(state) -> bool:
    return (
        state["rounds"] >= MAX_ROUNDS or
        state["total_tokens"] >= MAX_TOKENS or
        state["task_complete"] == True
    )

教训4：日志和可观测性是救命稻草

import structlog

logger = structlog.get_logger()

def agent_middleware(agent_name):
    def wrapper(func):
        async def inner(state):
            logger.info("agent.start", 
                       agent=agent_name, 
                       input_tokens=count_tokens(state))
            result = await func(state)
            logger.info("agent.complete",
                       agent=agent_name,
                       output_tokens=count_tokens(result),
                       duration_ms=elapsed)
            return result
        return inner
    return wrapper

教训5：人工介入不是失败，是安全网

# 关键决策点插入人工审核
def human_checkpoint(state):
    if state["risk_level"] > 0.8:
        # 高风险操作，等待人工确认
        return "wait_for_human"
    return "continue"

七、总结

多Agent协作在2026年已经从"有趣的实验"变成"生产必备"。选择合适的架构模式是成功的关键：

1. Supervisor适合需要严格控制的流程
2. Handoff适合对话式交互
3. Router适合简单高效的分发
4. Swarm适合需要并行探索的创意任务

记住：先简单，后复杂。先2个Agent，后N个Agent。先规则，后AI。

多Agent的未来不是越多越好，而是恰到好处的专业分工。

---

作者：JiaDe Wu | AWS Solutions Architect | sample-OpenClaw-on-AWS-with-Bedrock Owner | GitHub: github.com/JiaDe-Wu
关注我获取更多AI Agent技术深度文章