LangGraph多Agent协作模式:Supervisor与Swarm架构
引言
想象一下你正在经营一家现代化的互联网医院。患者进来时,前台分诊护士先判断"这个病人该去哪个科室"——心内科、皮肤科还是骨科;确认后再把患者引到对应的专科医生面前。这就是典型的Supervisor(主管)模式:一个总调度者负责分发任务,各个专家只专注自己的领域。
但现实里还有另一种场景:急诊科。病人推进来,A医生发现是骨折,立刻喊来骨科B医生;B医生发现病人有糖尿病史,又扭头叫来内分泌科C医生;C医生处理完,把病人交回给B医生。整个过程没有固定的"分诊台",医生之间直接对话、接力处理。这就是Swarm(蜂群)模式:Agent之间平等协作,通过"交接棒"完成复杂任务。
在构建LLM应用时,一旦任务复杂到单个Agent的上下文窗口撑不住、或者需要不同专业能力协同,我们就必须面对"多Agent如何协作"这个架构问题。LangGraph作为LangChain生态中专门做有状态图编排的框架,官方给出了两种经典范式:langgraph-supervisor 和 langgraph-swarm。很多工程师上手时只把它们当成"能跑起来的demo",但一旦进入生产就会踩坑:Supervisor陷入死循环、Swarm的上下文无限膨胀、工具调用路由错乱……
这篇文章,我会从源码级拆解这两种架构的实现差异,用生活化类比讲清楚它们的设计哲学,并给出3个可直接运行的生产级示例,最后聊聊什么时候该用哪个、以及那些官方文档不会告诉你的坑。
核心概念:Supervisor 与 Swarm 到底是什么
生活化类比
先建立一个直观模型:
| 架构 | 类比 | 调度方式 | 上下文传递 |
|---|---|---|---|
| Supervisor | 医院分诊台 | 中心化,护士决定下一个科室 | 每次只把"必要摘要"给专家 |
| Swarm | 急诊接力 | 去中心化,医生之间直接交班 | 完整对话历史一路带走 |
具体来说:
- Supervisor:有一个"主管Agent",它本身不干活,只负责看当前状态,然后决定"下一个该谁干"或"任务完成了"。所有子Agent干完活后,控制权必须交回主管,由主管再决定下一步。
- Swarm:没有中心节点。每个Agent手里有工具(包括"转交给其他Agent"的工具),干完自己的部分后,可以直接调用
transfer_to_xxx把控制权交给下一个人。谁都可以是起点,谁都可以是终点。
技术定义
在 LangGraph 的语境里,两者本质上都是由多个节点(node)和条件边(conditional edge)构成的有向图,区别在于图的拓扑结构:
Supervisor 的拓扑是星型:
主管Agent] S -->|决定路由| A1[研究员Agent] S -->|决定路由| A2[分析师Agent] S -->|决定路由| A3[写作Agent] A1 -->|返回控制权| S A2 -->|返回控制权| S A3 -->|返回控制权| S S -->|FINISH| END[结束/输出]
Swarm 的拓扑是网状:
关键差异在于:Supervisor 的每条边都要经过中心节点,Swarm 的边是Agent之间的直连。这个结构差异直接决定了它们的上下文管理策略、token消耗、以及调试难度。
源码级深度分析
要真正理解两者,必须看 langgraph-supervisor 和 langgraph-swarm 这两个包的源码实现。我以当前主流版本(0.0.x 系列)为例拆解。
1. Supervisor 的核心实现
Supervisor 的核心逻辑在 langgraph_supervisor/supervisor.py 中。它的本质是动态生成一个 Supervisor Agent,并把子Agent包装成"工具"。
简化后的核心代码如下(保留关键逻辑):
# 伪源码,展示核心机制
def create_supervisor(agents, model, prompt, output_mode="last_message"):
# 1. 把每个子Agent包装成一个工具
agent_tools = []
for agent in agents:
tool = create_handoff_tool(
agent_name=agent.name,
description=f"把任务交给 {agent.name} 处理"
)
agent_tools.append(tool)
# 2. Supervisor本身是一个ReAct风格的Agent
# 它的"工具"就是"转交给某个子Agent"
supervisor_chain = (
prompt
| model.bind_tools(agent_tools) # 关键:子Agent即工具
)
# 3. 构建图
builder = StateGraph(AgentState)
builder.add_node("supervisor", supervisor_node)
for agent in agents:
builder.add_node(agent.name, agent_node)
# 4. 星型拓扑:所有子Agent都回到supervisor
builder.add_edge(START, "supervisor")
for agent in agents:
builder.add_edge(agent.name, "supervisor") # ← 关键边
# 5. 条件路由:supervisor决定下一步
builder.add_conditional_edges(
"supervisor",
route_to_agent, # 根据supervisor的工具调用决定
{agent.name: agent.name for agent in agents} | {"FINISH": END}
)
return builder.compile()第一个关键点:子Agent即工具。 Supervisor 并不"理解"子Agent的内部逻辑,它只是通过 bind_tools 把这些Agent暴露成可调用的函数。Supervisor 的 LLM 输出一个 tool_call,LangGraph 的 ToolNode 把它翻译成"跳转到对应节点"。
第二个关键点:add_edge(agent.name, "supervisor")。 这行代码是星型拓扑的物理体现——每个子Agent执行完,无条件回到主管。这意味着:
- 子Agent无法"接力"给另一个子Agent,必须经过主管中转;
- 主管能看到所有子Agent的输出摘要,但默认看不到子Agent执行时的完整中间过程。
第三个关键点:上下文隔离。 Supervisor 传给子Agent的消息,默认是 state["messages"] 里的全部内容(可以通过 output_mode 和消息过滤定制)。但子Agent执行完返回时,通常只把最终消息写回 state,中间的工具调用细节被丢弃。这就是为什么 Supervisor 适合"分而治之"——每个子Agent在自己的沙箱里干活,脏活累活不污染主上下文。
2. Swarm 的核心实现
Swarm 的实现思路完全不同,核心在 langgraph_swarm/swarm.py。它没有中心节点,而是给每个Agent注入"转交工具",并动态修改图的边。
# 伪源码,展示核心机制
def create_swarm(agents, default_active_agent):
builder = StateGraph(AgentState)
# 1. 每个Agent都作为一个节点加入
for agent in agents:
# 2. 为每个Agent动态生成"转交给其他人"的工具
handoff_tools = [
create_handoff_tool(other.name)
for other in agents if other.name != agent.name
]
# 3. 把转交工具绑定到Agent的LLM上
agent_with_handoff = agent.bind_tools(handoff_tools)
builder.add_node(agent.name, create_agent_node(agent_with_handoff))
# 4. 动态路由:根据当前active_agent和其工具调用决定
def route(state):
# 读取state里的active_agent字段
active = state["active_agent"]
last_msg = state["messages"][-1]
# 如果Agent调用了transfer_to_xxx,就切换active_agent
if has_handoff_call(last_msg):
return get_target_from_handoff(last_msg)
return active # 否则继续当前Agent
# 5. 为每个Agent添加条件边
for agent in agents:
builder.add_conditional_edges(
agent.name,
route,
{a.name: a.name for a in agents} # 可以跳到任何Agent
)
builder.add_edge(START, default_active_agent)
return builder.compile()第一个关键点:active_agent 状态字段。 Swarm 在 state 里维护一个"当前活跃Agent"的标记。每次路由时,先看这个标记,再看最后一条消息里有没有转交工具调用。这是"接力棒"的物理实现。
第二个关键点:条件边允许任意跳转。 {a.name: a.name for a in agents} 意味着从任意Agent可以跳到任意其他Agent。这就是网状拓扑的来源。
第三个关键点:上下文全量传递。 Swarm 默认把完整消息历史传给下一个Agent。这是它和 Supervisor 最大的行为差异——接力时,下一位医生能看到前面所有医生的诊断记录、检查报告、甚至聊天记录。优点是信息不丢失,缺点是 token 消耗随对话轮次线性增长。
3. 两者的状态机对比
从状态机角度看,Supervisor 是一个两态交替机(supervisor ↔ worker),Swarm 是一个多态自由跳转机(任意 worker ↔ 任意 worker)。这个差异带来的工程后果是:
- Supervisor 的路由决策点集中在一个地方(主管的 prompt),容易调试、容易加约束(比如"最多路由5次");
- Swarm 的路由决策点分散在每个Agent身上,灵活但难以全局约束,容易出现"A交给B,B又交回A"的乒乓死循环。
实战代码
下面给出3个完整可运行示例。假设你已安装:
pip install langgraph langgraph-supervisor langgraph-swarm langchain-openai并设置 OPENAI_API_KEY 环境变量。
示例1:Supervisor 实现"研究-分析-写作"流水线
这是一个典型的内容生产场景:研究员搜集资料,分析师提炼观点,写手成文,主管协调。
"""
Supervisor 模式示例:内容生产流水线
研究员 -> 分析师 -> 写手,由主管统一调度
"""
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent
# ---------- 1. 定义各专家的工具 ----------
@tool
def search_web(query: str) -> str:
"""搜索网络资料。实际生产中替换为 Tavily/SerpAPI 等。"""
# 这里用假数据演示,真实场景应接入搜索API
return f"关于'{query}'的搜索结果:LangGraph是LangChain推出的有状态图编排框架..."
@tool
def analyze_data(text: str) -> str:
"""分析文本,提炼核心观点。"""
return f"分析结论:{text[:50]}... 的核心价值在于状态管理与多Agent协作能力。"
@tool
def write_article(points: str) -> str:
"""根据要点撰写文章。"""
return f"【成文】基于以下要点完成文章:{points}"
# ---------- 2. 创建三个专家Agent ----------
# 注意:每个Agent都绑定自己的工具,且name必须唯一
researcher = create_react_agent(
model=ChatOpenAI(model="gpt-4o-mini", temperature=0),
tools=[search_web],
name="researcher",
prompt="你是研究员,负责搜集资料。完成后把结果交给主管。",
)
analyst = create_react_agent(
model=ChatOpenAI(model="gpt-4o-mini", temperature=0),
tools=[analyze_data],
name="analyst",
prompt="你是分析师,负责提炼观点。完成后把结论交给主管。",
)
writer = create_react_agent(
model=ChatOpenAI(model="gpt-4o-mini", temperature=0),
tools=[write_article],
name="writer",
prompt="你是写手,负责成文。完成后把文章交给主管。",
)
# ---------- 3. 创建主管并编译图 ----------
# supervisor 的 prompt 决定了路由策略,这是调试的重点
workflow = create_supervisor(
agents=[researcher, analyst, writer],
model=ChatOpenAI(model="gpt-4o", temperature=0), # 主管用更强的模型
prompt=(
"你是内容生产主管。按以下顺序调度:\n"
"1. 先让 researcher 搜集资料\n"
"2. 再让 analyst 提炼观点\n"
"3. 最后让 writer 成文\n"
"每个专家完成后会向你汇报,你判断下一步。"
"如果所有步骤完成,直接回复 FINISH。"
),
)
app = workflow.compile()
# ---------- 4. 执行 ----------
if __name__ == "__main__":
result = app.invoke({
"messages": [{"role": "user", "content": "写一篇关于LangGraph多Agent协作的短文"}]
})
# 打印最终消息
for msg in result["messages"]:
print(f"[{msg.type}] {msg.content[:100]}")关键注释:主管用 gpt-4o 而专家用 gpt-4o-mini 是一种成本优化实践——路由决策需要强推理,具体执行可以降级。另外,prompt 里显式写了"完成后回复 FINISH",这是防止主管陷入无限路由的关键约束。
示例2:Swarm 实现"客服接力"场景
这是一个客服场景:售前接待→技术支持→售后处理,客户问题可能跨多个环节,Agent之间直接交接。
"""
Swarm 模式示例:客服接力
售前 -> 技术支持 -> 售后,Agent之间直接交接
"""
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph_swarm import create_swarm, create_handoff_tool
from langgraph.prebuilt import create_react_agent
# ---------- 1. 定义各环节的业务工具 ----------
@tool
def check_product_info(product: str) -> str:
"""查询产品信息。"""
return f"{product} 的规格:支持多Agent协作,版本 v0.2.x。"
@tool
def diagnose_issue(symptom: str) -> str:
"""诊断技术问题。"""
return f"针对'{symptom}'的诊断:建议检查 state 的 active_agent 字段配置。"
@tool
def process_refund(order_id: str) -> str:
"""处理退款。"""
return f"订单 {order_id} 退款已受理,3个工作日到账。"
# ---------- 2. 创建转交工具(Swarm的核心) ----------
# 每个Agent都需要知道"可以转交给谁"
to_tech = create_handoff_tool(
agent_name="tech_support",
description="当客户问题涉及技术细节时,转交给技术支持",
)
to_after_sales = create_handoff_tool(
agent_name="after_sales",
description="当客户需要退款或售后时,转交给售后",
)
to_pre_sales = create_handoff_tool(
agent_name="pre_sales",
description="当客户咨询产品信息时,转交给售前",
)
# ---------- 3. 创建三个Agent,各自绑定业务工具+转交工具 ----------
pre_sales = create_react_agent(
model=ChatOpenAI(model="gpt-4o-mini", temperature=0),
tools=[check_product_info, to_tech, to_after_sales], # 可转给技术/售后
name="pre_sales",
prompt="你是售前客服,解答产品咨询。遇到技术问题转tech_support,遇到退款转after_sales。",
)
tech_support = create_react_agent(
model=ChatOpenAI(model="gpt-4o-mini", temperature=0),
tools=[diagnose_issue, to_pre_sales, to_after_sales],
name="tech_support",
prompt="你是技术支持,诊断技术问题。解决后如需售后转after_sales。",
)
after_sales = create_react_agent(
model=ChatOpenAI(model="gpt-4o-mini", temperature=0),
tools=[process_refund, to_pre_sales, to_tech_support if False else to_tech],
name="after_sales",
prompt="你是售后客服,处理退款。处理完可转回售前做回访。",
)
# ---------- 4. 创建Swarm ----------
# default_active_agent 指定入口Agent
workflow = create_swarm(
agents=[pre_sales, tech_support, after_sales],
default_active_agent="pre_sales",
)
app = workflow.compile()
# ---------- 5. 执行 ----------
if __name__ == "__main__":
# 模拟一个跨环节的客户请求
result = app.invoke({
"messages": [{
"role": "user",
"content": "我买的产品有技术问题,而且我想退款,订单号12345"
}]
})
for msg in result["messages"]:
print(f"[{msg.type}] {msg.content[:120]}")关键注释:注意 to_tech_support if False else to_tech 这行是我故意保留的一个"坑演示"——Swarm 里转交工具的命名必须和 agent_name 严格对应,写错会导致 KeyError 或路由到不存在的节点。生产代码里应该统一用常量管理。另外,Swarm 的转交工具是双向的(售后也能转回售前),这是它灵活性的来源,也是死循环的隐患。
示例3:自定义 Supervisor,加入"最大路由次数"防死循环
官方 Supervisor 没有内置路由上限,这是生产环境最大的坑。下面展示如何通过自定义状态和条件边加一个熔断器。
"""
自定义 Supervisor:加入路由次数熔断,防止无限循环
这是生产环境的必备加固
"""
from typing import Annotated, Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import create_react_agent
from langgraph_supervisor import create_supervisor
from typing_extensions import TypedDict
MAX_ROUTES = 8 # 熔断阈值
# ---------- 1. 扩展状态,加入路由计数 ----------
class SupervisedState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
route_count: int # 记录主管路由了多少次
# ---------- 2. 创建子Agent(同示例1) ----------
@tool
def do_research(q: str) -> str:
"""研究工具"""
return f"研究结果:{q}"
researcher = create_react_agent(
model=ChatOpenAI(model="gpt-4o-mini"),
tools=[do_research],
name="researcher",
)
# ---------- 3. 手动构建带熔断的 Supervisor 图 ----------
def build_supervised_with_guard():
builder = StateGraph(SupervisedState)
# 主管节点:只做决策,不干活
supervisor_llm = ChatOpenAI(model="gpt-4o", temperature=0)
# 把子Agent包装成工具
from langgraph_supervisor.handoff import create_handoff_tool
handoff_tools = [
create_handoff_tool(agent_name="researcher", description="交给研究员")
]
supervisor_with_tools = supervisor_llm.bind_tools(handoff_tools)
def supervisor_node(state: SupervisedState):
# 熔断检查:超过上限直接结束
if state["route_count"] >= MAX_ROUTES:
return {
"messages": [AIMessage(content="[熔断] 路由次数超限,强制结束")],
"route_count": state["route_count"],
}
# 正常决策
response = supervisor_with_tools.invoke(state["messages"])
return {
"messages": [response],
"route_count": state["route_count"] + 1, # 计数+1
}
# 子Agent节点(复用官方create_react_agent)
builder.add_node("supervisor", supervisor_node)
builder.add_node("researcher", researcher)
# 星型边
builder.add_edge(START, "supervisor")
builder.add_edge("researcher", "supervisor")
# 条件路由:读工具调用 + 熔断
def route(state: SupervisedState) -> Literal["researcher", "__end__"]:
if state["route_count"] >= MAX_ROUTES:
return "__end__"
last = state["messages"][-1]
if isinstance(last, AIMessage) and last.tool_calls:
target = last.tool_calls[0]["name"]
if target == "transfer_to_researcher":
return "researcher"
return "__end__"
builder.add_conditional_edges(
"supervisor",
route,
{"researcher": "researcher", "__end__": END},
)
return builder.compile()
if __name__ == "__main__":
app = build_supervised_with_guard()
result = app.invoke({
"messages": [HumanMessage(content="帮我研究一下LangGraph")],
"route_count": 0,
})
print(f"最终路由次数:{result['route_count']}")
for m in result["messages"]:
print(f"[{m.type}] {m.content[:80]}")关键注释:这里我们没有直接调用 create_supervisor,而是手动搭图。原因是要在 state 里插入 route_count 字段并参与路由决策——官方封装没有暴露这个扩展点。这是生产环境常见的"改装"模式:先用官方 API 跑通,遇到约束需求再降到 StateGraph 手写。熔断逻辑放在 supervisor_node 和 route 两处(双保险),确保即使主管 LLM 抽风也不会无限循环。
方案对比:Supervisor vs Swarm vs 其他多Agent方案
| 维度 | Supervisor | Swarm | 单Agent+多工具 | CrewAI | AutoGen |
|---|---|---|---|---|---|
| 拓扑 | 星型(中心调度) | 网状(直连交接) | 单点 | 层级/顺序 | 对话群聊 |
| 上下文控制 | 强(专家隔离) | 弱(全量传递) | 最弱 | 中 | 中 |
| 可调试性 | 高(决策集中) | 低(决策分散) | 高 | 中 | 低 |
| 死循环风险 | 中(主管易循环) | 高(乒乓交接) | 低 | 中 | 高 |
| token消耗 | 低(摘要传递) | 高(历史累积) | 低 | 中 | 高 |
| 适用场景 | 流程明确、专家分工 | 动态协作、问题跨域 | 简单任务 | 角色扮演型任务 | 开放式讨论 |
| 上手成本 | 低 | 中 | 最低 | 中 | 高 |
核心判断法则:
- 如果任务能预先画出流程图("先A后B再C"),用 Supervisor,因为集中式路由最可控;
- 如果任务边界模糊、需要动态协作(客服、诊断、开放式研究),用 Swarm,因为强制中心化会拖慢响应;
- 如果只是一个Agent配几个工具就能搞定,别上多Agent,徒增复杂度;
- CrewAI 的"角色+任务"模型适合内容创作类,AutoGen 的"群聊"适合探索性任务,但两者的状态管理都不如 LangGraph 显式,调试成本更高。
最佳实践与避坑指南
坑1:Supervisor 无限路由
现象:主管反复把任务交给同一个专家,或者专家说"完成"主管却理解成"继续"。
根因:主管的 prompt 没有明确的终止条件,或者专家返回的消息里没有清晰的"完成信号"。
解法:
- 在主管 prompt 里写死
FINISH关键字; - 用示例3的熔断机制加
route_count; - 让每个专家在完成时返回结构化消息(如
{"status": "done", "result": ...}),主管根据status路由。
坑2:Swarm 上下文爆炸
现象:对话进行到第10轮,token 消耗突然飙升,甚至超出模型上下文窗口。
根因:Swarm 全量传递消息历史,每轮都累积。
解法:
- 在 state 里加
add_messages的 reducer,用滑动窗口只保留最近N条; - 或者用
trim_messages在转交前压缩历史; - 关键信息用
state["summary"]字段单独维护,不依赖消息历史。
from langchain_core.messages import trim_messages
# 在转交前压缩
trimmed = trim_messages(state["messages"], max_tokens=2000, strategy="last")坑3:转交工具命名不一致
现象:KeyError: 'transfer_to_tech_support' 或路由到不存在的节点。
根因:create_handoff_tool(agent_name="tech") 生成的工具名是 transfer_to_tech,但条件边里写的是 tech_support。
解法:用常量统一管理 agent 名称,并在启动时做一次断言校验。
AGENT_NAMES = {"pre_sales", "tech_support", "after_sales"}
# 启动时校验
for agent in agents:
assert agent.name in AGENT_NAMES, f"未知Agent: {agent.name}"坑4:子Agent的工具和转交工具冲突
现象:子Agent本该调业务工具,却误调了转交工具,把控制权交出去了。
根因:转交工具的描述(description)写得太宽泛,LLM 分不清什么时候该转交、什么时候该干活。
解法:转交工具的 description 要写得具体且带条件,比如"当客户明确要求退款时转交",而不是"转交给售后"。
坑5:忽略 output_mode 导致的上下文污染
现象:Supervisor 模式下,主管的上下文里塞满了子Agent的中间工具调用记录。
根因:create_supervisor 的 output_mode 默认是 last_message,但如果子Agent返回多条消息,可能全部写回。
解法:显式设置 output_mode="last_message",并在子Agent的 prompt 里要求"只返回最终结论"。
最佳实践清单
- 模型分级:主管/路由节点用强模型(gpt-4o/claude-sonnet),执行节点用弱模型(gpt-4o-mini);
- 状态显式化:把关键中间结果写进 state 的自定义字段,而不是依赖消息历史;
- 可观测性:用 LangSmith 或自建 tracing 记录每次路由决策,这是调试多Agent的唯一有效手段;
- 单测优先:每个子Agent单独测试通过后,再组装成图,否则问题定位成本指数级上升;
- 渐进式:先用单Agent+多工具,撑不住了再上 Supervisor,再撑不住了才考虑 Swarm——架构复杂度要匹配问题复杂度。
总结
回到开头的医院类比:Supervisor 是分诊台,Swarm 是急诊接力。两者没有绝对优劣,只有场景匹配。
核心要点回顾:
- Supervisor 的本质是把子Agent包装成工具,通过星型拓扑和中心路由实现"分而治之",上下文隔离好、可调试性高,但主管是单点瓶颈,容易死循环;
- Swarm 的本质是通过
active_agent状态字段和动态条件边实现Agent直连,灵活度高、适合动态协作,但上下文全量传递、决策分散、死循环风险更高; - 源码级的差异在于:Supervisor 的边是
agent -> supervisor的无条件回边,Swarm 的边是agent -> any_agent的条件边; - 生产环境必须加固:Supervisor 加路由熔断,Swarm 加历史压缩,两者都要加 Agent 名称校验和可观测性。
延伸思考:LangGraph 的这两种模式其实对应了分布式系统里的两种经典架构——中心化调度(如 Kubernetes 的 scheduler) 和 去中心化协作(如 gossip 协议)。当你理解了这层映射,就会发现多Agent系统的很多问题(单点故障、状态一致性、死循环)在分布式系统领域早有成熟解法。下一步值得研究的方向是:如何把 Supervisor 的"路由决策"做成可插拔的策略(类似 K8s 的调度器框架),以及如何用 Swarm 的"交接"机制实现 Agent 的热插拔——运行时动态增减专家,而不重启整个图。这两点,正是多Agent系统从 demo 走向生产的关键。