AI Study Online
AI教程

掌握多智能体协作：5种核心模式与手把手代码实战

5 min read

分类： AI教程 · 难度： 中级

为什么多智能体协作很重要

在AI开发领域，单个Agent在处理复杂动态任务时往往会遇到瓶颈。这就是多智能体协作的用武之地。与僵化的基于函数的工作流不同，多智能体系统赋予每个Agent独立的LLM推理上下文和决策自主权，使它们能够动态协调。本文将深入探讨五种核心协作模式——Supervisor、Hierarchical、Swarm、Sequential Chain和Network——并附上使用LangGraph和AutoGen的实用代码示例。我们还将探讨通信协议、安全措施以及何时选择多Agent vs. 单Agent架构。

在深入探讨模式之前，让我们先弄清楚为什么多智能体系统值得投入。与传统的函数链式调用（中央控制器僵化地按A到B到C的顺序调用函数）不同，多Agent设置提供：

  • 独立推理：每个Agent自主决策，无需中央控制器预定义所有分支。
  • 动态协作：Agent根据实时上下文调整交互，而非固定工作流。
  • 聚焦上下文：每个Agent的提示保持简洁，通过避免无关噪声提高推理质量。

多Agent适合以下场景：

  • 任务需要跨学科角色（例如，一个包含产品经理、开发者和测试人员的软件团队）。
  • 工作流不可预测（没有预先定义的逐步流程）。
  • 并行处理或上下文窗口限制（单个Agent无法容纳所有必要信息）需要扩展。

五种核心多智能体协作模式

1. Supervisor模式：集中式任务编排

可以将它想象成一个"项目经理"Agent，将任务委派给专门的子Agent并汇总结果。

用例

构建一个包含研究人员（查找数据）、分析师（解读数据）和作者（创建内容）的内容生产流水线。

代码实现 (LangGraph)

from langgraph.supervisor import create_react_agent, create_supervisor

# 第一步：创建专门Agent
researcher = create_react_agent({
    "model": "gpt-4o",
    "name": "researcher",
    "prompt": "你的角色是研究和收集事实信息。"
})

analyst = create_react_agent({
    "model": "gpt-4o",
    "name": "analyst",
    "prompt": "你的角色是分析数据并识别洞见。"
})

writer = create_react_agent({
    "model": "gpt-4o",
    "name": "writer",
    "prompt": "你的角色是将洞见转化为引人入胜的内容。"
})

# 第二步：组装Supervisor
supervisor = create_supervisor({
    "model": "gpt-4o",
    "agents": [researcher, analyst, writer],
    "prompt": "你是一个项目经理。委派任务给最合适的Agent并汇总结果。"
})

# 第三步：运行工作流
result = supervisor.invoke("创建一篇关于2026年AI Agent协作趋势的博客文章")
print(result)

2. Hierarchical模式：嵌套多层编排

这是Supervisor的扩展版——想象"CEO到部门主管到团队成员"——适合大型复杂系统。

用例

一个包含CEO主管、技术/设计部门负责人和个人贡献者（前端、后端、测试人员）的软件开发团队。

代码实现 (LangGraph)

from langgraph.supervisor import create_supervisor

# 第一步：创建个人贡献者
frontend_dev = create_react_agent({
    "model": "gpt-4o",
    "name": "frontend_dev",
    "prompt": "专注于React/TypeScript开发。"
})

backend_dev = create_react_agent({
    "model": "gpt-4o",
    "name": "backend_dev",
    "prompt": "专注于Node.js/Express后端开发。"
})

# 第二步：创建部门主管
tech_supervisor = create_supervisor({
    "model": "gpt-4o",
    "agents": [frontend_dev, backend_dev],
    "prompt": "管理技术团队：委派编码任务并审查工作。"
})

design_supervisor = create_react_agent({
    "model": "gpt-4o",
    "name": "design_supervisor",
    "prompt": "监督UI/UX设计，确保与产品目标一致。"
})

# 第三步：创建CEO级主管
ceo_supervisor = create_supervisor({
    "model": "gpt-4o",
    "agents": [tech_supervisor, design_supervisor],
    "prompt": "监督整个项目：协调技术和设计团队。"
})

# 第四步：执行
result = ceo_supervisor.invoke("用React和Node.js构建一个任务管理应用")
print(result)

3. Swarm模式：对等自主交接

Agent作为平等参与者，通过handoff机制动态传递控制权——就像接力赛一样，每个Agent决定下一个接棒者是谁。

用例

研究人员、评论者和作者根据需要在彼此间传递工作的内容创建流程。

代码实现 (LangGraph)

from langgraph.swarm import create_handoff, create_swarm

# 第一步：创建交接工具
handoff_tool = create_handoff()

# 第二步：创建具有交接能力的Agent
researcher = create_react_agent({
    "model": "gpt-4o",
    "name": "researcher",
    "tools": [handoff_tool],
    "prompt": "研究主题，完成后交接给评论者。"
})

critic = create_react_agent({
    "model": "gpt-4o",
    "name": "critic",
    "tools": [handoff_tool],
    "prompt": "评论研究成果，完成后交接给作者（或返回给研究人员如果不完整）。"
})

writer = create_react_agent({
    "model": "gpt-4o",
    "name": "writer",
    "tools": [handoff_tool],
    "prompt": "将有评论的研究成果转化为博客文章。"
})

# 第三步：组装Swarm
swarm = create_swarm({
    "agents": [researcher, critic, writer],
    "default_active_agent": "researcher"
})

# 第四步：运行
result = swarm.invoke("写一篇关于多Agent安全最佳实践的文章")
print(result)

4. Sequential Chain模式：固定流水线工作流

一个严格的线性流水线——非常适合有严格定义步骤的任务（例如，研究、大纲、草稿、编辑）。

用例

具有严格顺序步骤的内容创建。

代码实现 (LangGraph)

from langgraph.graph import StateGraph, START, END

# 定义状态模式（步骤间传递的数据）
class ContentState:
    research: str
    outline: str
    draft: str
    edited: str
    final_content: str

# 构建图
graph = StateGraph(ContentState)

# 将每个步骤定义为节点
def research(state):
    state.research = "AI Agent协作趋势包括多协议支持和安全进步。"
    return state

def outline(state):
    state.outline = "1. 引言
2. 关键模式
3. 协议
4. 安全
5. 结论"
    return state

def draft(state):
    state.draft = f"基于研究：{state.research}
大纲：{state.outline}
[草稿内容]"
    return state

def edit(state):
    state.edited = state.draft.replace("[草稿内容]", "润色后的内容...")
    return state

def finalize(state):
    state.final_content = state.edited + "

---
发布于2026-06-10"
    return state

# 添加节点并定义固定序列
graph.add_node("research", research)
graph.add_node("outline", outline)
graph.add_node("draft", draft)
graph.add_node("edit", edit)
graph.add_node("finalize", finalize)

graph.add_edge(START, "research")
graph.add_edge("research", "outline")
graph.add_edge("outline", "draft")
graph.add_edge("draft", "edit")
graph.add_edge("edit", "finalize")
graph.add_edge("finalize", END)

# 编译并运行
app = graph.compile()
result = app.invoke({})
print(result.final_content)

5. Network模式：自由形式路由

Agent自由连接，通过一个"路由器"Agent使用LLM决定下一步——非常适合探索性、非结构化的任务。

用例

一个规划者、执行者和审查者灵活协作的动态问题解决工作流。

代码实现 (LangGraph)

from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI

def router(state):
    llm = ChatOpenAI(model="gpt-4o")
    response = llm.invoke(f"当前状态：{state}
选择下一个Agent：planner、executor、reviewer还是end？")
    return {"next_agent": response.content.strip().lower()}

def planner(state):
    state.plan = "开发一个具有三层结构的多Agent安全框架。"
    return state

def executor(state):
    state.code = "def secure_multi_agent(): ..."
    return state

def reviewer(state):
    state.feedback = "代码看起来不错，但需要添加输入验证。"
    return state

# 构建图
graph = StateGraph(dict)
graph.add_node("router", router)
graph.add_node("planner", planner)
graph.add_node("executor", executor)
graph.add_node("reviewer", reviewer)

graph.add_edge(START, "router")
graph.add_conditional_edges(
    "router",
    lambda s: s["next_agent"],
    {
        "planner": "planner",
        "executor": "executor",
        "reviewer": "reviewer",
        "end": END
    }
)
# 每个Agent执行完后返回路由器
graph.add_edge("planner", "router")
graph.add_edge("executor", "router")
graph.add_edge("reviewer", "router")

app = graph.compile()
result = app.invoke({})
print(result)

关键通信协议

多智能体系统依赖协议来确保互操作性：

1. Agent-to-Agent (A2A) 协议

  • 目的：使来自不同框架/厂商的Agent能够相互通信。
  • 核心概念：AgentCard — 一个描述Agent能力、端点和认证方法的JSON对象。

示例结构

{
  "name": "researcher-agent",
  "capabilities": ["web-search", "data-synthesis"],
  "endpoint": "https://agent-service.com/researcher",
  "auth": {
    "type": "api-key",
    "key": "secure_123"
  }
}

2. 模型上下文协议 (MCP)

  • 目的：管理Agent如何与外部工具和数据源交互。
  • 能力：
    • Tools：访问API（例如天气、数据库）。
    • Resources：访问文档/文件。
    • Prompts：预定义的提示模板。
  • 架构：LLM应用通过MCP Client连接到暴露这些资源的MCP Server。

多智能体系统安全

多Agent设置引入了独特的安全风险——以下是如何缓解它们：

  1. 结构化消息：在Agent之间强制使用JSON或结构化数据（不传原始文本）以防止提示注入。
def validate_message(message):
    if not isinstance(message, dict):
        raise SecurityError("只允许结构化消息")
    return message
  1. 工具白名单：将Agent限制在预先批准的工具范围内。
ALLOWED_TOOLS = ["web-search", "calculator"]
def check_tool(tool_name):
    if tool_name not in ALLOWED_TOOLS:
        raise SecurityError(f"工具 {tool_name} 未授权")
    return tool_name
  1. Token预算：限制每个Agent的Token使用量，防止无限循环。
def check_token_usage(usage):
    if usage > 10000:
        raise ResourceError("Token超过限制")
    return usage

何时使用多Agent vs. 单Agent

任务特征单Agent多Agent
步骤固定不变推荐过度设计
需要跨学科角色上下文混乱推荐
工作流不可预测无法处理分支推荐
单次交互足够推荐额外开销
成本优先Token使用量低成本高2-5倍

常见问题

问：Supervisor模式和Hierarchical模式有什么区别？

Supervisor模式使用单一管理者监督多个工作Agent。Hierarchical模式在此基础上扩展为多层管理——想象CEO、部门负责人、再到个人贡献者。当需要扩展到5-7个以上Agent时使用Hierarchical模式。

问：可以同时使用多个模式吗？

是的，实际系统通常组合使用多种模式。例如，在顶层使用Supervisor模式，在某个子组内使用Swarm模式。LangGraph支持嵌套图来实现这一点。

问：对于初学者来说，LangGraph和AutoGen哪个框架更好？

LangGraph基于图的API和清晰的文档降低了学习门槛。AutoGen提供了更多灵活性，但需要对Agent通信有更深入的理解。建议先用LangGraph进行原型开发，在生产环境中切换到AutoGen。

下一篇： 多智能体协作架构：终极版问答工具 →

分享这篇文章

相关文章

AI教程入门

ChatGPT基础：界面、设置和你的前10个提示词

ChatGPT新手？这里有完整的初学者指南——账户设置、界面导览、需要配置的设置，以及10个入门提示词让你从第一天就获得真正价值。

5分钟阅读
ChatGPT入门基础
AI教程入门

如何写出真正有效的提示词：5点框架

模糊的提示词只能得到平庸的回答。掌握5点提示框架——角色、上下文、任务、格式、约束——从任何AI工具中获得显著更好的结果。

5分钟阅读
提示词提示工程框架