Multi-Agent系统正在从"能协作"向"会协作"演进,而决定协作质量的核心变量不是模型能力,而是通信拓扑——Agent之间如何传递信息、谁先说谁后说、消息是点对点还是广播。2025年EMNLP Industry Track发表的AMAS论文证实:同一组Agent,仅切换通信拓扑,在Crossword任务上准确率可波动3-4个百分点。这意味着拓扑选择已成为多智能体系统的"第二维调参"。本文系统拆解Star、Mesh、Ring、Pipeline、Hierarchical、Broadcast、Swarm七大通信拓扑范式,结合LangGraph、CrewAI、AutoGen/AG2、OpenAI Agents SDK、ESAA、AMAS、Claude Code Swarm七大引擎的实现差异,给出生产级选型建议。
| 拓扑范式 | 信息流向 | 中心节点 | 容错性 | 延迟特征 | 典型框架 |
|---|---|---|---|---|---|
| Star(星形) | Hub ↔ Spoke | 有(Orchestrator) | Hub单点故障 | O(1) 路由 | OpenAI Agents SDK、AG2 GroupChat |
| Mesh(网状) | 任意 ↔ 任意 | 无 | 高(多路径冗余) | O(n²) 消息膨胀 | AutoGen自由对话、AG2 |
| Ring(环形) | 单向循环 | 无 | 断链即瘫痪 | O(n) 固定延迟 | AutoGen RoundRobin |
| Pipeline(流水线) | A→B→C→... | 无 | 上游故障级联 | O(n) 串行累加 | CrewAI Sequential |
| Hierarchical(层级) | 树形递归 | 有(根节点) | 分支可隔离 | O(log n) 路由 | CrewAI Hierarchical、LangGraph子图 |
| Broadcast(广播) | 一对多/多对多 | 有(Event Bus) | 高(解耦) | O(1) 事件投递 | ESAA Event Sourcing、Zylos EDA |
| Swarm(群智) | 动态自适应 | 无/弱 | 自愈型 | 非确定性 | AMAS、Claude Code Swarm、OpenAI Swarm |
Star拓扑是最直觉也最常见的设计——一个中心Orchestrator负责路由所有消息,周围Agent只与中心通信,彼此不直接对话。
OpenAI Agents SDK的Handoff机制是Star拓扑的典型实现。每个Agent注册时声明自己的能力和handoff列表,Orchestrator根据用户意图和Agent返回的transfer_to_agent指令进行路由。Microsoft AG2的GroupChat同样采用Star结构:所有Agent的输出先发给GroupChatManager,由Manager决定下一个发言者。
# OpenAI Agents SDK — Star拓扑Handoff示例
from agents import Agent, Runner
triage_agent = Agent(
name="Triage",
instructions="分析用户问题,决定转交给哪个专家",
handoffs=["billing_agent", "tech_agent", "sales_agent"]
)
billing_agent = Agent(
name="Billing",
instructions="处理账单相关问题",
handoffs=["triage_agent"] # 只能回到中心
)
result = await Runner.run(triage_agent, messages=[...])
关键特征:Agent之间绝对不直接通信。Billing Agent需要Tech Agent的数据时,必须先handoff回Triage,由Triage再路由到Tech。这种"一切经手中心"的设计牺牲了效率,但换来了全局可见性和审计追踪。
Star拓扑的最大风险是Hub成为认知瓶颈。当Agent数量超过5-6个,Orchestrator的context window会被历史消息填满,路由决策质量急剧下降。AG2的解法是让GroupChatManager使用独立的轻量模型做路由,不与工作Agent共享context。OpenAI Agents SDK则通过context_variables机制控制信息粒度,避免全量消息广播。
Mesh拓扑中每个Agent可以直接与任意其他Agent通信,没有中心路由节点。这最接近人类团队的自由讨论模式。
AutoGen的ConversableAgent支持Agent间直接send和receive消息。在Mesh模式下,Agent A可以直接调用Agent B的generate_reply,无需经过第三方。AG2的NestedChats进一步支持Agent在对话中嵌套调用其他Agent,形成多层Mesh。
# AutoGen — Mesh拓扑自由对话
from autogen import ConversableAgent
researcher = ConversableAgent("researcher", ...)
coder = ConversableAgent("coder", ...)
reviewer = ConversableAgent("reviewer", ...)
# 任意Agent间直接对话,无中心路由
researcher.initiate_chat(coder, message="这是调研结果,请实现")
coder.initiate_chat(reviewer, message="代码已完成,请审查")
reviewer.initiate_chat(researcher, message="发现设计缺陷,需要重新调研")
Mesh的优势是信息传递路径最短——Agent可以即时获取所需信息,无需中间人转述。但代价是消息爆炸:n个Agent的完全Mesh会产生O(n²)条潜在通信路径。EMNLP 2025发表的研究表明,在8个Agent的Mesh网络中,无效信息传播导致的Token浪费可达总消耗的40%以上。
Mesh适合短周期、高创造性的任务,如头脑风暴、方案辩论、创意发散。不适合结构化流程任务——Mesh无法保证执行顺序,容易陷入"讨论循环"(Agent A让B做,B又让A做)。CrewAI刻意不支持Mesh,其创始人指出"无序讨论是Agent系统最常见的失败模式"。
Ring拓扑让Agent按固定顺序依次发言,最后一位发言者的输出作为第一位的新输入,形成闭环。
AutoGen for .NET和AG2都提供了RoundRobin GroupChat实现。核心是一个next_speaker_index计数器,每轮递增,循环选择下一个发言Agent。Microsoft Semantic Kernel的AgentGroupChat使用SelectionStrategy控制发言顺序,其默认实现就是Ring模式。
# AG2 — Ring拓扑RoundRobin
from ag2 import AssistantAgent, GroupChat, GroupChatManager
agent_a = AssistantAgent("A", llm_config=...)
agent_b = AssistantAgent("B", llm_config=...)
agent_c = AssistantAgent("C", llm_config=...)
groupchat = GroupChat(
agents=[agent_a, agent_b, agent_c],
messages=[],
max_round=9, # 每个Agent发言3轮
speaker_selection_method="round_robin" # Ring模式
)
manager = GroupChatManager(groupchat=groupchat, llm_config=...)
manager.initiate_chat(agent_a, message="开始讨论")
Ring的哲学是绝对公平——每个Agent获得相等的发言机会,不存在"声音大的Agent主导对话"。但僵化性也是显而易见的:如果轮到Agent B时它没有新信息需要贡献,它被迫输出一条"我同意A的观点"这种无效消息,浪费Token且增加噪声。
AG2允许设置max_consecutive_auto_reply来限制Agent连续无效发言。更高级的做法是让每个Agent在发言前先做"有无新信息"的判断,无新信息则skip,但这需要额外的LLM调用,成本未必更低。
Pipeline拓扑将Agent串联成线性链条,每个Agent的输出严格作为下一个Agent的输入。这是最易理解也最脆弱的拓扑。
CrewAI的Process.SEQUENTIAL是Pipeline的标准实现。任务按定义顺序依次分配给Agent,前一个Task的output自动作为后一个Task的context。LangGraph的线性StateGraph(无分支条件)同样实现Pipeline语义。
# CrewAI — Pipeline拓扑
from crewai import Agent, Task, Crew, Process
researcher = Agent(role="研究员", goal="收集信息")
writer = Agent(role="作者", goal="撰写文章")
editor = Agent(role="编辑", goal="润色审核")
task1 = Task(description="调研主题", agent=researcher)
task2 = Task(description="基于调研写初稿", agent=writer, context=[task1])
task3 = Task(description="编辑润色", agent=editor, context=[task2])
crew = Crew(
agents=[researcher, writer, editor],
tasks=[task1, task2, task3],
process=Process.SEQUENTIAL # Pipeline模式
)
Pipeline的核心优势是确定性——给定相同输入,输出路径完全可预测。这对合规审计、金融报告、医疗诊断等场景至关重要。但脆弱性在于级联失败:Pipeline中任何一个Agent的输出质量下降,下游所有Agent都会被污染。研究数据显示,4级Pipeline的末端Agent接收到的有效信息仅为原始信息的60-70%,存在严重的信息衰减问题。
LangGraph通过State对象传递完整状态(而非仅前一步输出)来缓解衰减。CrewAI的memory=True参数让后续Agent可以回溯所有前序输出。但根本解法是将Pipeline长度控制在3-4级以内,超过5级的Pipeline建议拆分为并行分支后合并。
Hierarchical拓扑引入管理层级——顶层Manager负责规划和分派,中层Agent管理子任务,底层Worker执行具体操作。
CrewAI的Process.HIERARCHICAL自动创建一个Manager Agent,由它分析任务、拆分子任务、分配给Worker。LangGraph通过Subgraph节点实现层级——父图中的一个节点本身是一个完整的子图,拥有独立的State空间。
# LangGraph — Hierarchical子图
from langgraph.graph import StateGraph, START, END
# 子图:代码生成Pipeline
sub_graph = StateGraph(CodeState)
sub_graph.add_node("design", design_agent)
sub_graph.add_node("implement", code_agent)
sub_graph.add_node("test", test_agent)
sub_graph.add_edge("design", "implement")
sub_graph.add_edge("implement", "test")
code_subgraph = sub_graph.compile()
# 父图:Manager路由
parent = StateGraph(MainState)
parent.add_node("planner", planner_agent)
parent.add_node("code_team", code_subgraph) # 嵌套子图
parent.add_node("doc_team", doc_subgraph) # 另一个子图
parent.add_conditional_edges("planner", route_to_team)
Hierarchical的核心价值是关注点分离——每个层级只处理自己关心的抽象级别。Manager不需要知道代码怎么写,只需要知道"这个需求应该给代码团队"。这使得系统可以扩展到10+Agent而不失控。
实测表明,Hierarchical拓扑的Manager Agent通常消耗总Token量的15-25%。一个3层、8个Worker的层级系统,Manager和Sub-Manager的管理开销约占总成本30%。但换来的是:Worker的context window压力显著降低(每个Worker只看到分配给自己的子任务),总有效Token利用率反而提升。
Broadcast拓扑是最"不像Agent系统"的架构——Agent之间不直接对话,而是通过Event Bus发布和订阅事件。这借鉴了微服务架构中的Event Sourcing和CQRS模式。
ESAA(Event Sourcing for Autonomous Agents)架构是这一范式的代表。其核心原则是:Agent只发出Intention(意图),不直接修改状态。所有意图被写入append-only的Event Store,由Orchestrator验证后才产生实际效果。Read Model(如roadmap.json)是Event Store的确定性投影,可以被任何Agent订阅。
ESAA事件流:
Agent A发出Intention → Event Store追加事件 → Orchestrator验证
→ 投影更新Read Model → Agent B感知到状态变更 → B发出新Intention
# ESAA核心流程(简化)
class Orchestrator:
def admit(self, agent_result):
# 1. 验证Agent输出是否符合合约
if not self.contract.validate(agent_result):
return {"status": "rejected", "reason": "合约违规"}
# 2. 追加事件到Event Store
event = self.event_store.append(
event_type=agent_result.intention,
payload=agent_result.data,
agent=agent_result.agent_id
)
# 3. 投影到Read Model
self.projector.apply(event)
# 4. 广播给订阅者
self.event_bus.publish(event)
Zylos AI在2026年3月发布的Event-Driven Architecture for AI Agent Systems报告中指出:Broadcast拓扑将Agent之间的耦合度降低了80%,但引入了最终一致性问题——Agent B读到的状态可能不是最新的。对于需要强一致性的场景(如金融交易),需要在Event Bus之上叠加Saga补偿模式。
Broadcast最适合长周期、多参与者的工作流:软件工程项目(多Agent协作开发)、内容生产流水线、持续集成/部署。ESAA论文显示,在软件工程Agent场景中,Event Sourcing将Agent间的冲突率从传统架构的12%降至1.3%。
Swarm拓扑是最前沿也最不确定的范式——通信拓扑不是固定的,而是根据输入动态生成或切换。每个任务可能使用不同的Agent交互图谱。
AMAS(Adaptively Determining Communication Topology)框架是这一方向的学术标杆,发表于EMNLP 2025。其核心创新是两层拓扑学习:
AMAS动态拓扑选择流程:
输入任务 → 图选择器LLM → 候选拓扑评分 → 选出最优拓扑
→ 按该拓扑组织Agent交互 → 执行任务 → 输出结果
实测数据:在LLaMA-3 8B上,AMAS在Crossword任务达到48.3%(vs GPTSwarm的44.7%),Game-of-24达到37.4%(vs 34.3%),MMLU、BBH、HumanEval均有2-5个百分点的提升。
Claude Code在2026年1月通过1096个commit引入了Swarm模式。与AMAS的学术路径不同,Claude Code Swarm采用任务驱动拓扑:Orchestrator分析任务后动态创建Worker Agent拓扑,任务完成后拓扑销毁。Worker之间可以通过共享文件系统和内存进行有限通信,但主要信息流仍经过Orchestrator。
# Claude Code Swarm模式(概念示意)
swarm_config = {
"topology": "dynamic", # 动态拓扑
"max_workers": 8,
"spawn_rule": "task_complexity > threshold",
"communication": "shared_memory + orchestrator_routing",
"lifecycle": "task_scoped" # 任务结束即销毁
}
Swarm拓扑的根本挑战是可解释性。当系统自动选择了一个"Mesh + Pipeline混合"拓扑来处理某个任务,开发者很难理解为什么这个拓扑是最优的。AMAS的图选择器虽然是LLM驱动的,但选择逻辑本身是一个黑盒。在生产环境中,无法解释的拓扑决策可能导致难以复现的错误。
flowchart TD
A[Multi-Agent任务] --> B{Agent数量}
B -->|≤3| C{需要严格顺序?}
B -->|4-8| D{需要中心管控?}
B -->|>8| E{有管理层级?}
C -->|是| F[Pipeline拓扑]
C -->|否| G{需要自由讨论?}
G -->|是| H[Mesh拓扑]
G -->|否| I[Ring拓扑]
D -->|是| J{信息流复杂度?}
D -->|否| K[Mesh拓扑]
J -->|高| L[Star拓扑]
J -->|中| M[Hierarchical拓扑]
E -->|是| N[Hierarchical拓扑]
E -->|否| O{需要异步解耦?}
O -->|是| P[Broadcast拓扑]
O -->|否| Q{任务类型多变?}
Q -->|是| R[Swarm动态拓扑]
Q -->|否| S[Hierarchical拓扑]
style F fill:#4CAF50
style H fill:#2196F3
style I fill:#FF9800
style L fill:#9C27B0
style M fill:#00BCD4
style N fill:#00BCD4
style P fill:#E91E63
style R fill:#FF5722
| 评估维度 | Star | Mesh | Ring | Pipeline | Hierarchical | Broadcast | Swarm |
|---|---|---|---|---|---|---|---|
| 可解释性 | ★★★★★ | ★★ | ★★★★ | ★★★★★ | ★★★★ | ★★★ | ★★ |
| 可扩展性 | ★★ | ★ | ★★★ | ★★★ | ★★★★★ | ★★★★★ | ★★★★ |
| 容错性 | ★★ | ★★★★★ | ★ | ★★ | ★★★★ | ★★★★★ | ★★★★ |
| 延迟确定性 | ★★★★ | ★★ | ★★★★★ | ★★★★★ | ★★★ | ★★ | ★ |
| Token效率 | ★★★★ | ★ | ★★★ | ★★★★ | ★★★★ | ★★★★ | ★★★ |
| 上手难度 | 极低 | 低 | 极低 | 极低 | 中 | 高 | 极高 |
| 生产成熟度 | ★★★★★ | ★★★ | ★★★★ | ★★★★★ | ★★★★ | ★★★ | ★★ |
| 框架 | 原生支持拓扑 | 可扩展拓扑 | 切换成本 |
|---|---|---|---|
| LangGraph | Pipeline、Hierarchical | Star、Mesh、Ring | 低(重写图定义) |
| CrewAI | Pipeline、Hierarchical | Star(通过Manager) | 中(需改Process类型) |
| AG2/AutoGen | Ring、Mesh、Star | Hierarchical | 中(需自定义Manager) |
| OpenAI Agents SDK | Star(Handoff)、Pipeline | Hierarchical | 低(嵌套Handoff) |
| ESAA | Broadcast | 全部(Event驱动) | 高(需实现Event映射) |
| AMAS | Swarm(动态) | 学习任意拓扑 | 极高(需训练图选择器) |
| Claude Code Swarm | Swarm、Star | Hierarchical | 低(配置式切换) |
推荐Star拓扑 + OpenAI Agents SDK。客服场景需要明确的审计追踪(谁回答了什么),Handoff机制天然提供全链路日志。Triage Agent作为Hub,专业Agent作为Spoke。避免使用Mesh——客服场景不允许Agent间自由讨论导致给用户发送矛盾信息。
推荐Pipeline拓扑 + CrewAI。调研→撰写→编辑→SEO优化,步骤固定、依赖明确。CrewAI的Sequential Process直接映射。如果需要并行分支(如同时生成图文),升级为Hierarchical拓扑,用Manager协调两个并行Pipeline。
推荐Hierarchical + Broadcast混合。用Hierarchical拆分模块(前端/后端/测试/QA),模块内用Broadcast(ESAA模式)管理Agent间的代码变更。ESAA的Event Sourcing确保每次代码修改都有不可篡改的审计记录,Orchestrator合约防止Agent越权修改其他模块的代码。
推荐Swarm拓扑 + AMAS思路。如果Agent数量和任务模式变化很大,固定拓扑必然出现局部最优。AMAS的per-input图选择策略可以让系统自适应。但生产部署前建议先用离线模式收集一批任务的候选拓扑,人工验证后再上线。
推荐Mesh拓扑 + AG2。创意场景需要低延迟、高频率的Agent间通信,Mesh的信息传递路径最短。设置max_round防止无限循环,用silent=True参数抑制无效Agent的输出。建议搭配一个"观察者Agent"监控讨论质量,在对话陷入循环时主动介入。
| 拓扑方向 | 训练营定位 | 目标学员 | 定价区间 |
|---|---|---|---|
| Star + OpenAI SDK | "从零构建企业级Agent客服" | 初中级开发者 | ¥999-1999 |
| Pipeline + CrewAI | "AI内容工厂:自动化生产流水线" | 运营/内容从业者 | ¥699-1299 |
| Hierarchical + LangGraph | "大规模Agent系统架构设计" | 高级工程师/架构师 | ¥2999-4999 |
| Broadcast + ESAA | "Agent事件驱动架构实战" | 后端工程师 | ¥1999-2999 |
| Swarm + AMAS | "自适应群智系统前沿实战" | 研究员/AI工程师 | ¥3999-5999 |
[!tip] 51学通信知识星球 星主:爱卫生(微信:gprshome201101)——公众号「51学通信」创办者,19 年通信行业运维与讲师经验,著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书(各大电商平台有售)。 星球定位:深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域,主打运维岗位与信令分析。 当前规模:5200 篇主题 / 990 个回答,198 元/年。 加入福利:加入星球后请加爱卫生微信,拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题,另有群内分享。 AI 加持:51学通信知识星球已作为 skill 接入 WorkBuddy,使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例,做总结、找文档和 PPT 等场景全覆盖。
