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 的拓扑是星型:

graph TD U[用户输入] --> S[Supervisor
主管Agent] S -->|决定路由| A1[研究员Agent] S -->|决定路由| A2[分析师Agent] S -->|决定路由| A3[写作Agent] A1 -->|返回控制权| S A2 -->|返回控制权| S A3 -->|返回控制权| S S -->|FINISH| END[结束/输出]

Swarm 的拓扑是网状:

graph TD U[用户输入] --> A1[研究员Agent] A1 -->|transfer_to_analyst| A2[分析师Agent] A2 -->|transfer_to_writer| A3[写作Agent] A3 -->|transfer_to_researcher| A1 A1 -->|任务完成| END[结束/输出] A2 -->|任务完成| END A3 -->|任务完成| END

关键差异在于: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 没有明确的终止条件,或者专家返回的消息里没有清晰的"完成信号"。

解法:

  1. 在主管 prompt 里写死 FINISH 关键字;
  2. 用示例3的熔断机制加 route_count;
  3. 让每个专家在完成时返回结构化消息(如 {"status": "done", "result": ...}),主管根据 status 路由。

坑2:Swarm 上下文爆炸

现象:对话进行到第10轮,token 消耗突然飙升,甚至超出模型上下文窗口。

根因:Swarm 全量传递消息历史,每轮都累积。

解法:

  1. 在 state 里加 add_messages 的 reducer,用滑动窗口只保留最近N条;
  2. 或者用 trim_messages 在转交前压缩历史;
  3. 关键信息用 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 里要求"只返回最终结论"。

最佳实践清单

  1. 模型分级:主管/路由节点用强模型(gpt-4o/claude-sonnet),执行节点用弱模型(gpt-4o-mini);
  2. 状态显式化:把关键中间结果写进 state 的自定义字段,而不是依赖消息历史;
  3. 可观测性:用 LangSmith 或自建 tracing 记录每次路由决策,这是调试多Agent的唯一有效手段;
  4. 单测优先:每个子Agent单独测试通过后,再组装成图,否则问题定位成本指数级上升;
  5. 渐进式:先用单Agent+多工具,撑不住了再上 Supervisor,再撑不住了才考虑 Swarm——架构复杂度要匹配问题复杂度。

总结

回到开头的医院类比:Supervisor 是分诊台,Swarm 是急诊接力。两者没有绝对优劣,只有场景匹配。

核心要点回顾:

  1. Supervisor 的本质是把子Agent包装成工具,通过星型拓扑和中心路由实现"分而治之",上下文隔离好、可调试性高,但主管是单点瓶颈,容易死循环;
  2. Swarm 的本质是通过 active_agent 状态字段和动态条件边实现Agent直连,灵活度高、适合动态协作,但上下文全量传递、决策分散、死循环风险更高;
  3. 源码级的差异在于:Supervisor 的边是 agent -> supervisor 的无条件回边,Swarm 的边是 agent -> any_agent 的条件边;
  4. 生产环境必须加固:Supervisor 加路由熔断,Swarm 加历史压缩,两者都要加 Agent 名称校验和可观测性。

延伸思考:LangGraph 的这两种模式其实对应了分布式系统里的两种经典架构——中心化调度(如 Kubernetes 的 scheduler) 和 去中心化协作(如 gossip 协议)。当你理解了这层映射,就会发现多Agent系统的很多问题(单点故障、状态一致性、死循环)在分布式系统领域早有成熟解法。下一步值得研究的方向是:如何把 Supervisor 的"路由决策"做成可插拔的策略(类似 K8s 的调度器框架),以及如何用 Swarm 的"交接"机制实现 Agent 的热插拔——运行时动态增减专家,而不重启整个图。这两点,正是多Agent系统从 demo 走向生产的关键。