Supervisor 模式

多智能体系统里提到四种协作模式，其中 Supervisor 主管模式 是最常用、最稳定的。本章详解它的原理与实现。

一、什么是 Supervisor 模式

Supervisor 模式像一个团队里的项目经理：

主管（Supervisor）自己不干活，只负责"分工"——看完任务后决定交给哪个 worker。
Worker 各自是独立的 agent（可以是 ReAct、子图、甚至另一个 Supervisor）。
Worker 干完活把结果汇报给主管，主管再决定"够不够，要不要再派活"。

mermaid

flowchart TB
    U([用户]) --> S[Supervisor 主管<br/>决定下一步交给谁]
    S -->|路由| W1[Worker1 搜索]
    S -->|路由| W2[Worker2 计算]
    S -->|路由| W3[Worker3 写作]
    W1 --> S
    W2 --> S
    W3 --> S
    S -->|任务完成| E([END])

为什么用它

可控：所有决策集中在一个 LLM 调用里，好调试。
可扩展：加 worker 只需加节点 + 改路由表，不动其他 worker。
职责清晰：worker 只管自己的事，不用知道全局。

二、状态设计

Supervisor 要做路由决策，状态里需要一个 next 字段记录"下一步去哪"：

python

from typing import TypedDict, Annotated, Literal
from langgraph.graph import MessagesState

class TeamState(TypedDict):
    messages: Annotated[list, "add"]
    # next 必须是这几个值之一，Literal 约束防止笔误
    next: Literal["searcher", "calculator", "FINISH"]

三、Supervisor 节点：用 LLM 做路由

Supervisor 节点的本质是：把当前消息喂给 LLM，让 LLM 输出"下一步该找谁"。最稳的做法是让 LLM 用结构化输出返回 next 字段。

python

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 用 with_structured_output 强制 LLM 返回结构化结果
from pydantic import BaseModel

class Route(BaseModel):
    next: Literal["searcher", "calculator", "FINISH"]

router_llm = llm.with_structured_output(Route)

SUPERVISOR_PROMPT = """你是团队主管，负责把任务分给合适的成员：
- searcher：负责查资料、搜索信息。
- calculator：负责数学计算。
- 任务已完成、不需要再分配时，返回 FINISH。

根据当前对话判断下一步交给谁。如果已经有最终答案，返回 FINISH。"""

def supervisor_node(state: TeamState):
    msgs = [SystemMessage(content=SUPERVISOR_PROMPT)] + state["messages"]
    decision = router_llm.invoke(msgs)
    return {"next": decision.next}

with_structured_output 是 LangChain 的利器，让 LLM 输出符合 Pydantic schema 的对象，避免自己解析字符串。

四、Worker 节点

每个 worker 就是一个独立的 ReAct agent。我们做两个：搜索员和计算员。

python

from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def search(query: str) -> str:
    """搜索资料。返回一段文字。"""
    return f"【{query}】的搜索结果：这是一种2024年流行的AI框架。"

@tool
def calc(expr: str) -> str:
    """计算数学表达式。"""
    if not set(expr) <= set("0123456789+-*/.() "):
        return "非法字符"
    try:
        return str(eval(expr))
    except Exception as e:
        return str(e)

searcher = create_react_agent(
    llm, tools=[search],
    prompt="你是搜索员，用 search 工具回答问题。",
)
calculator = create_react_agent(
    llm, tools=[calc],
    prompt="你是计算员，用 calc 工具做数学运算。",
)

def searcher_node(state: TeamState):
    result = searcher.invoke({"messages": state["messages"]})
    return {"messages": [result["messages"][-1]]}

def calculator_node(state: TeamState):
    result = calculator.invoke({"messages": state["messages"]})
    return {"messages": [result["messages"][-1]]}

五、拼装 Supervisor 图

关键在于条件边：根据 next 字段路由到对应 worker，或结束。

python

from langgraph.graph import StateGraph, START, END

builder = StateGraph(TeamState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("searcher", searcher_node)
builder.add_node("calculator", calculator_node)

builder.add_edge(START, "supervisor")

# supervisor 根据 next 路由
builder.add_conditional_edges(
    "supervisor",
    lambda state: state["next"],
    {
        "searcher": "searcher",
        "calculator": "calculator",
        "FINISH": END,
    },
)

# worker 干完回到 supervisor
builder.add_edge("searcher", "supervisor")
builder.add_edge("calculator", "supervisor")

team = builder.compile()

六、完整可运行示例

python

from typing import TypedDict, Annotated, Literal
from pydantic import BaseModel
from langchain_core.tools import tool
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import create_react_agent

class TeamState(TypedDict):
    messages: Annotated[list, "add"]
    next: Literal["searcher", "calculator", "FINISH"]

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class Route(BaseModel):
    next: Literal["searcher", "calculator", "FINISH"]

router_llm = llm.with_structured_output(Route)
SUPERVISOR_PROMPT = """你是团队主管。
- searcher 查资料；calculator 做计算。
- 有最终答案就返回 FINISH。"""

def supervisor_node(state: TeamState):
    msgs = [SystemMessage(content=SUPERVISOR_PROMPT)] + state["messages"]
    return {"next": router_llm.invoke(msgs).next}

@tool
def search(query: str) -> str:
    """搜索资料。"""
    return f"【{query}】：2024年流行的AI框架。"

@tool
def calc(expr: str) -> str:
    """计算数学表达式。"""
    try:
        return str(eval(expr))
    except Exception as e:
        return str(e)

searcher = create_react_agent(llm, [search], prompt="你是搜索员。")
calculator = create_react_agent(llm, [calc], prompt="你是计算员。")

def searcher_node(state: TeamState):
    r = searcher.invoke({"messages": state["messages"]})
    return {"messages": [r["messages"][-1]]}

def calculator_node(state: TeamState):
    r = calculator.invoke({"messages": state["messages"]})
    return {"messages": [r["messages"][-1]]}

builder = StateGraph(TeamState)
builder.add_node("supervisor", supervisor_node)
builder.add_node("searcher", searcher_node)
builder.add_node("calculator", calculator_node)
builder.add_edge(START, "supervisor")
builder.add_conditional_edges("supervisor", lambda s: s["next"],
    {"searcher": "searcher", "calculator": "calculator", "FINISH": END})
builder.add_edge("searcher", "supervisor")
builder.add_edge("calculator", "supervisor")
team = builder.compile()

result = team.invoke({
    "messages": [HumanMessage(content="查一下 LangGraph 是什么，然后算 25*4+10")],
    "next": "searcher",
}, config={"recursion_limit": 30})

print(result["messages"][-1].content)

预期流程：supervisor 先派 searcher→搜索员返回资料→supervisor 看还差计算→派 calculator→计算员返回结果→supervisor 判断完成→END。

七、langgraph-supervisor 库（可选）

社区有现成的 langgraph-supervisor 包，封装了上面的模式：

bash

pip install langgraph-supervisor -i https://pypi.tuna.tsinghua.edu.cn/simple

python

from langgraph_supervisor import create_supervisor

team = create_supervisor(
    model=llm,
    agents=[searcher, calculator],
    prompt="你是主管，按需分配任务。",
).compile()

少写点样板代码，但底层逻辑和手搓一样。建议先手搓理解，再用库提效。

八、与 LangGraph4j 对比

LangGraph4j 同样支持 Supervisor 模式，对应关系：

概念	LangGraph (Python)	LangGraph4j (Java)
状态	`TypedDict`	`record`/`Map`
结构化输出	`with_structured_output(Pydantic)`	`withStructuredOutput(Class)`
条件边	`add_conditional_edges`	`addConditionalEdges`
子 agent	`create_react_agent`	`ReactAgent`

逻辑一致，差异主要在类型声明语法。

九、常见踩坑

1. Supervisor 决策不稳定

LLM 路由有时会"抽风"，同样问题两次选不同的 worker。对策：

temperature=0。
prompt 里加示例（few-shot），明确"什么问题选谁"。
用 with_structured_output 强约束，避免解析失败。

2. Worker 结果没写回共享状态

worker 节点返回的 messages 必须放进外层共享状态，否则 supervisor 看不到。新手常犯的错是 worker 内部状态改了但没 return，或 return 了错误的字段名。

3. 永远不返回 FINISH

supervisor 一直派活，死循环。务必在 prompt 里强调"已有答案就 FINISH"，并设 recursion_limit。

4. Worker 之间不能直接通信

标准 Supervisor 模式下，worker 只能通过 supervisor 中转。如果两个 worker 需要频繁交互，考虑把它们合并成一个，或改用层级模式。

十、小结

Supervisor = 主管 LLM 做路由 + worker 各司其职 + 共享状态串联。
关键技术：with_structured_output 做路由决策、next 字段 + 条件边做调度。
完整示例：搜索员 + 计算员，由 supervisor 调度。
踩坑：决策不稳、结果丢失、不结束、worker 间通信受限。

任务再大、worker 再多时，就要升级到层级智能体了。

Supervisor 模式 ​

一、什么是 Supervisor 模式 ​

为什么用它 ​

二、状态设计 ​

三、Supervisor 节点：用 LLM 做路由 ​

四、Worker 节点 ​

五、拼装 Supervisor 图 ​

六、完整可运行示例 ​

七、langgraph-supervisor 库（可选） ​

八、与 LangGraph4j 对比 ​

九、常见踩坑 ​

1. Supervisor 决策不稳定 ​

2. Worker 结果没写回共享状态 ​

3. 永远不返回 FINISH ​

4. Worker 之间不能直接通信 ​

十、小结 ​

Supervisor 模式

一、什么是 Supervisor 模式

为什么用它

二、状态设计

三、Supervisor 节点：用 LLM 做路由

四、Worker 节点

五、拼装 Supervisor 图

六、完整可运行示例

七、langgraph-supervisor 库（可选）

八、与 LangGraph4j 对比

九、常见踩坑

1. Supervisor 决策不稳定

2. Worker 结果没写回共享状态

3. 永远不返回 FINISH

4. Worker 之间不能直接通信

十、小结