多智能体协作实战：4 种架构模式选择指南

凌晨三点，我的单 Agent 系统崩了。不是崩溃的那种崩，而是输出了整整三页废话——明明只是让它”检查一下代码风格”。我盯着屏幕，心想：这玩意儿怎么比我那个话痨同事还能扯？

这不是个例。过去一年，Agent 论文数量从 820 篇激增到 2500 多篇，所有人都在问同一个问题：单个 Agent 真的够用吗？

Anthropic 的研究给了我当头一棒：多 Agent 系统比单 Agent 性能高出 90.2%。差点一倍。我突然意识到，我那”一个大 Agent 搞定所有事”的想法，跟”一个人干完一个团队的工作”一样荒唐。

这篇文章，我想聊聊多智能体协作系统的设计——从四种核心架构模式的选择逻辑，到生产环境里踩过的坑，再到可直接运行的代码实现。如果你也在纠结”用 Subagents 还是 Skills”、“怎么管理 Agent 间的状态”，这篇可能会帮你省下不少头发。

为什么需要多智能体系统

说实话，单 Agent 最大的问题不是”能不能用”，而是”能用多久”。

你有没有遇到过这种情况：一个 Agent 本来写得挺好的代码，突然开始输出乱七八糟的东西？或者你明明只问了 A，它偏要扯到 B 再扯到 C，最后连 Z 都给你搬出来？这不是 Agent”笨”，是它的上下文窗口被撑爆了。

单 Agent 有三个致命伤。

上下文限制。一个 Agent 再怎么强大，200k token 的上下文也就那么多。你让它同时处理代码审查、安全分析、性能优化，它能记住一半就不错了。我见过一个 Agent 在同一个对话里，前 20 轮还在聊 Python，第 21 轮就开始输出 JavaScript——它彻底忘了自己该干什么。

能力分散。你塞给一个 Agent 越多的技能，它就越像个”万金油”——什么都会一点，什么都不精。让它做个代码审查，它可能给你写个单元测试；让它写文档，它可能顺便重构了代码。方向感？不存在的。

调试困难。一个 Agent 挂了，你根本不知道是哪个环节出的问题。提示词太长？工具调用失败？还是上下文污染？排查起来像大海捞针。

多智能体系统，说白了就是 AI 界的”微服务架构”。每个 Agent 只做一件事，做好一件事。它们之间通过清晰的消息传递协作，而不是一股脑全塞进一个”超级 Agent”里。

Google、LangChain、Anthropic 都在推这套思路。O’Reilly 的报告显示，2025 年 Agent 相关论文已经从年初的 820 篇涨到 2500 多篇，翻了三倍。为什么？因为大家发现：单 Agent 打天下的时代，结束了。

四种核心架构模式

LangChain 和 Google 总结出了几种主流的多智能体架构。我跑了几个项目后，发现它们各有各的适用场景，选错了就是”用大炮打蚊子”或者”拿筷子喝汤”。

Subagents（子代理）- 中央编排模式

这是最直观的模式：一个”主 Agent”当指挥官，手下带一群”子 Agent”。子 Agent 就是主 Agent 的工具，主 Agent 决定什么时候调用谁。

用户请求 → 主 Agent（协调器）→ 分发给子 Agent A/B/C → 汇总结果 → 返回用户

什么时候用？ 你的任务涉及多个独立领域。比如一个客服系统：一个子 Agent 处理订单查询，一个处理退款，一个处理投诉。每个领域有自己的知识库和工具，主 Agent 只负责”分流”。

代码示例（LangGraph）：

from langgraph.prebuilt import create_react_agent

# 定义子 Agent
order_agent = create_react_agent(
    model="claude-3-5-sonnet-20241022",
    tools=[query_order, update_order],
    prompt="你是订单专家，只处理订单相关问题。"
)

refund_agent = create_react_agent(
    model="claude-3-5-sonnet-20241022",
    tools=[check_refund_policy, process_refund],
    prompt="你是退款专家，只处理退款相关问题。"
)

# 主 Agent 持有子 Agent 作为工具
main_agent = create_react_agent(
    model="claude-3-5-sonnet-20241022",
    tools=[order_agent, refund_agent],  # 子 Agent 就是工具
    prompt="你是客服总管，根据用户问题分发给合适的专家。"
)

优点：上下文隔离干净，每个子 Agent 只看自己该看的。并行执行效率高。

缺点：每个子 Agent 都是独立的 LLM 调用，token 消耗大。如果子 Agent 之间要共享状态，得绕一圈。

Skills（技能）- 按需加载模式

一个 Agent，多套”人格”。Skills 本质上是动态加载的提示词模板。Agent 根据任务切换”身份”，但始终是同一个 Agent。

用户请求 → 单一 Agent → 加载"代码审查"Skill → 执行 → 加载"文档生成"Skill → 执行

什么时候用？ 你的任务需要”单线程”处理，但不同阶段需要不同的专业知识。比如一个编程助手：写代码时用”开发者”模式，写文档时用”技术写手”模式。

代码示例：

# Skills 目录结构
skills/
├── code_review.md      # 代码审查提示词
├── doc_writer.md       # 文档生成提示词
└── security_audit.md   # 安全审计提示词

# 动态加载 Skill
def load_skill(skill_name: str) -> str:
    with open(f"skills/{skill_name}.md") as f:
        return f.read()

# 使用示例
agent = create_react_agent(
    model="claude-3-5-sonnet-20241022",
    tools=[...],
    prompt=load_skill("code_review")  # 运行时切换
)

优点：轻量，不需要额外的 Agent 协调开销。token 消耗比 Subagents 低。

缺点：上下文会累积。你切换了 10 次 Skill，之前 9 次 Skill 的内容还在上下文里，越堆越乱。

Handoffs（移交）- 状态驱动模式

Agent 之间像传球一样交接任务。Agent A 干完活，把状态”扔”给 Agent B，Agent B 继续。有点像接力赛。

用户请求 → Agent A（收集信息）→ 移交 → Agent B（分析问题）→ 移交 → Agent C（给出方案）

什么时候用？ 多阶段对话场景。比如一个技术支持流程：先收集问题 → 诊断问题 → 给方案 → 确认解决。每个阶段可能需要不同的专业知识。

代码示例：

from langchain_core.tools import tool

# 定义移交工具
@tool
def handoff_to_diagnosis(issue_summary: str) -> str:
    """将问题移交给诊断专家。"""
    return f"已接收问题：{issue_summary}，开始诊断..."

@tool
def handoff_to_solution(diagnosis_result: str) -> str:
    """将诊断结果移交给方案专家。"""
    return f"根据诊断：{diagnosis_result}，制定方案中..."

# Agent 链
triage_agent = create_react_agent(
    tools=[handoff_to_diagnosis],
    prompt="你是问题分拣员，收集用户问题并移交给诊断专家。"
)

diagnosis_agent = create_react_agent(
    tools=[handoff_to_solution],
    prompt="你是诊断专家，分析问题根因并移交给方案专家。"
)

优点：对话流自然，符合人类协作直觉。每个 Agent 只关注当前阶段。

缺点：状态管理复杂。你得保证 Agent A 传给 Agent B 的数据格式是对的，否则链条就断了。

Router（路由）- 并行分发模式

一个”路由 Agent”分析请求，然后并行调用多个专业 Agent，最后把结果综合起来。

用户请求 → Router（分类）→ 并行调用 Agent A/B/C → 结果综合 → 返回用户

什么时候用？ 一个请求需要查询多个数据源。比如一个企业知识库问答：Router 判断问题类型，然后并行查询内部文档、外部 API、数据库，最后综合答案。

代码示例：

from langgraph.graph import StateGraph

# 定义并行执行节点
async def query_internal_docs(state):
    # 查询内部文档
    return {"internal_results": [...]}

async def query_external_api(state):
    # 查询外部 API
    return {"external_results": [...]}

async def query_database(state):
    # 查询数据库
    return {"db_results": [...]}

async def synthesize(state):
    # 综合所有结果
    all_results = state["internal_results"] + state["external_results"] + state["db_results"]
    return {"final_answer": summarize(all_results)}

# 构建并行图
graph = StateGraph(State)
graph.add_node("internal", query_internal_docs)
graph.add_node("external", query_external_api)
graph.add_node("database", query_database)
graph.add_node("synthesize", synthesize)

# 并行执行
graph.add_edge("router", ["internal", "external", "database"])
graph.add_edge(["internal", "external", "database"], "synthesize")

优点：并行执行，速度最快。无状态，每个查询独立。

缺点：不适合多轮对话。每个请求都是新的，Agent 不记得上一轮聊了什么。

架构选择决策框架

说了这么多，到底该选哪个？我画了个简单的决策流程：

你的需求是什么？
         │
         ├─→ 多个独立域需要并行处理？
         │         │
         │         └─→ Subagents（中央编排）
         │
         ├─→ 单 Agent 多阶段切换技能？
         │         │
         │         └─→ Skills（按需加载）
         │
         ├─→ 顺序工作流，一棒接一棒？
         │         │
         │         └─→ Handoffs（状态驱动）
         │
         └─→ 多数据源查询需要综合？
                   │
                   └─→ Router（并行分发）

光看流程可能还不够直观，我整理了一个对比表：

模式	分布式开发	并行化	多跳对话	直接用户交互	Token 消耗
Subagents	高	高	高	低	高
Skills	高	中	高	高	低
Handoffs	无	无	高	高	中
Router	中	高	无	中	高

这张表怎么读？

• 分布式开发：你的团队是不是分头开发不同模块？如果是，Subagents 和 Skills 都适合，每个成员负责一个子 Agent 或 Skill。
• 并行化：你追求速度吗？Router 和 Subagents 能并行跑多个 Agent，效率最高。
• 多跳对话：用户需要多轮交互吗？Handoffs 和 Skills 天然支持对话流。
• 直接用户交互：用户是不是直接跟子 Agent 说话？Skills 和 Handoffs 支持，Router 不支持。
• Token 消耗：成本敏感的话，Skills 最省，Router 和 Subagents 最费。

我的经验是：从简单开始。先用 Skills 或 Handoffs 跑通一个 MVP，发现瓶颈了再升级到 Subagents 或 Router。别一上来就搞分布式架构，过度设计的痛苦我懂。

生产级实现要点

从 Demo 到生产，中间隔着一个太平洋。这几个坑我全踩过。

状态管理

多 Agent 共享状态，最容易出问题的就是”竞态条件”——两个 Agent 同时写同一个变量，最后谁覆盖谁？

LangGraph 的解决方案是 output_key：每个 Agent 只能写入自己专属的键。

from langgraph.graph import StateGraph, MessagesState

class GraphState(MessagesState):
    security_result: str = ""   # 安全 Agent 专属
    style_result: str = ""      # 风格 Agent 专属
    perf_result: str = ""       # 性能 Agent 专属

# 安全 Agent 只写 security_result
async def security_agent(state: GraphState):
    result = await analyze_security(state["messages"])
    return {"security_result": result}  # 只写这一个键

# 风格 Agent 只写 style_result
async def style_agent(state: GraphState):
    result = await analyze_style(state["messages"])
    return {"style_result": result}

这样，无论并行还是串行，每个 Agent 只动自己那一亩三分地，互不干扰。

另一个常见问题是”上下文污染”。Agent A 的输出被 Agent B 读到，但 B 根本不需要这些信息。我的做法是：在状态里加一个 relevant_keys 字段，每个 Agent 只读自己需要的键。

性能优化

多 Agent 系统的 token 消耗是个无底洞。几个省 token 的技巧：

1. Subagents 比 Skills 省 67% token（多域场景）

LangChain 的测试数据：如果一个任务涉及 3 个独立领域，Subagents 的 token 消耗是 Skills 的三分之一。为什么？因为 Subagents 的上下文隔离，每个子 Agent 只看自己领域的内容。Skills 的话，所有 Skill 的上下文累积在一起，越堆越大。

2. 有状态模式节省 40-50% 重复调用

如果你的任务有大量重复查询（比如同一个问题问 10 遍），用有状态的 Handoffs 模式，Agent 能记住之前的答案。LangChain 的数据：有状态比无状态节省接近一半的 LLM 调用。

3. 反思模式限制迭代次数

很多人喜欢给 Agent 加”反思”能力——让它自己检查输出、发现问题、重新生成。这东西好是好，但容易陷入无限循环。我一般限制 max_iterations=2 或 3，超过就强制退出。

from langgraph.checkpoint.memory import MemorySaver

# 设置迭代上限
graph = create_react_agent(
    model="claude-3-5-sonnet-20241022",
    tools=[...],
    checkpointer=MemorySaver(),
    config={"configurable": {"max_iterations": 3}}  # 最多反思 3 次
)

常见踩坑

无限循环：Agent 调用自己，自己调自己，自己调自己…没完没了。解决方案：设置 max_iterations 和明确的退出条件。

def should_continue(state):
    if state["iteration_count"] >= 3:
        return "end"
    if "done" in state["messages"][-1].content:
        return "end"
    return "continue"

上下文膨胀：Agent 越来越”笨”，输出越来越短。多半是上下文塞太多东西了。解决方案：用 Blackboard 模式（共享黑板），只保留必要的上下文，定期清理。

协调税：Agent 数量增加，通信开销指数级增长。我测试过一个系统：从 3 个 Agent 加到 10 个，响应时间从 2 秒变成 15 秒。解决方案：合并职责相近的 Agent，控制 Agent 数量在 5 个以内。

完整实现示例

说了这么多理论，来点实际的。我搭了一个代码审查多智能体系统，用的是 Router + ParallelAgent 模式。

架构：Router 判断代码语言和类型 → 并行调用安全审计、风格检查、性能分析三个 Agent → 综合结果输出报告。

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic

# 定义状态
class CodeReviewState(TypedDict):
    code: str
    language: str
    security_issues: list
    style_issues: list
    perf_issues: list
    final_report: str

# 初始化 LLM
llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")

# Router: 判断语言
async def route_code(state: CodeReviewState) -> dict:
    code = state["code"]
    # 简单判断，实际可以用 LLM 分类
    if "def " in code or "import " in code:
        language = "python"
    elif "function" in code or "const " in code:
        language = "javascript"
    else:
        language = "unknown"
    return {"language": language}

# 安全审计 Agent
async def security_audit(state: CodeReviewState) -> dict:
    code = state["code"]
    prompt = f"""你是安全审计专家。检查以下代码的安全问题：
- SQL 注入风险
- XSS 漏洞
- 敏感信息泄露
- 不安全的依赖

代码：
{code}

以 JSON 列表形式输出问题，每项包含：line（行号）、severity（严重程度）、description（描述）。
"""
    response = await llm.ainvoke(prompt)
    # 解析结果...
    return {"security_issues": []}

# 风格检查 Agent
async def style_check(state: CodeReviewState) -> dict:
    code = state["code"]
    language = state["language"]
    prompt = f"""你是代码风格专家。检查以下 {language} 代码的风格问题：
- 命名规范
- 代码格式
- 注释完整性

代码：
{code}

以 JSON 列表形式输出问题。
"""
    response = await llm.ainvoke(prompt)
    return {"style_issues": []}

# 性能分析 Agent
async def perf_analysis(state: CodeReviewState) -> dict:
    code = state["code"]
    prompt = f"""你是性能分析专家。检查以下代码的性能问题：
- 时间复杂度过高
- 不必要的循环
- 内存泄漏风险

代码：
{code}

以 JSON 列表形式输出问题。
"""
    response = await llm.ainvoke(prompt)
    return {"perf_issues": []}

# 综合报告
async def generate_report(state: CodeReviewState) -> dict:
    security = state.get("security_issues", [])
    style = state.get("style_issues", [])
    perf = state.get("perf_issues", [])

    total_issues = len(security) + len(style) + len(perf)

    report = f"""# 代码审查报告

## 概览
- 语言：{state['language']}
- 总问题数：{total_issues}

## 安全问题 ({len(security)} 个)
{format_issues(security)}

## 风格问题 ({len(style)} 个)
{format_issues(style)}

## 性能问题 ({len(perf)} 个)
{format_issues(perf)}

## 建议
根据以上分析，建议优先修复安全问题...
"""
    return {"final_report": report}

# 构建图
graph = StateGraph(CodeReviewState)
graph.add_node("router", route_code)
graph.add_node("security", security_audit)
graph.add_node("style", style_check)
graph.add_node("perf", perf_analysis)
graph.add_node("report", generate_report)

# 流程：Router → 并行执行三个检查 → 生成报告
graph.set_entry_point("router")
graph.add_edge("router", "security")
graph.add_edge("router", "style")
graph.add_edge("router", "perf")
graph.add_edge("security", "report")
graph.add_edge("style", "report")
graph.add_edge("perf", "report")
graph.add_edge("report", END)

# 编译
app = graph.compile()

# 使用
async def review_code(code: str):
    result = await app.ainvoke({"code": code})
    return result["final_report"]

这个示例跑起来后，一段 100 行的代码，三个 Agent 并行执行，大概 3-5 秒出结果。如果串行执行，至少要 10 秒。

当然，这只是个基础版本。生产环境你还需要加：缓存（同样的代码不重复审查）、增量审查（只看改动的部分）、人工反馈（让用户标记误报）。但这些扩展，都是在这个架构基础上做的。

搭建多智能体协作系统

从零开始搭建一个代码审查多智能体系统

1. 1

   步骤1: 选择架构模式

   根据任务特点选择合适的架构模式
2. 2

   步骤2: 定义状态结构

   使用 TypedDict 定义多 Agent 共享状态
3. 3

   步骤3: 创建 Agent 节点

   为每个 Agent 创建独立节点函数
4. 4

   步骤4: 构建执行图

   使用 LangGraph StateGraph 构建执行流程
5. 5

   步骤5: 添加状态管理

   使用 output_key 避免竞态条件

结论

说了这么多，其实就三句话：

底层逻辑：模式选择比框架选择更重要。LangGraph、AutoGen、CrewAI 都是好工具，但如果你用 Router 模式去解决一个需要 Handoffs 的问题，再好的框架也救不了你。

中层策略：从简单开始，逐步升级。先跑通一个 Skills 或 Handoffs 的 MVP，发现瓶颈了再考虑 Subagents 或 Router。过度设计是最大的坑——我踩过，别再踩。

顶层落地：生产环境关注状态管理、性能和成本。Token 消耗、无限循环、上下文污染，这三个问题搞定，你的多 Agent 系统就能稳稳跑起来。

下一步行动：打开 LangGraph 文档，选一个模式，用 50 行代码实现一个最简单的多智能体系统。别想太多，先跑起来。

12 分钟阅读 · 发布于: 2026年3月25日 · 修改于: 2026年3月25日