层级智能体

Supervisor 模式适合 3-5 个 worker 的小团队。当 worker 数量到十几个、任务跨多个领域时，单层 Supervisor 就力不从心了——光"派活"这一步 LLM 都会选错。这时要用 层级智能体（Hierarchical Teams）：把 worker 分组，每组有自己的 Supervisor，上面再有一个总 Supervisor。

一、什么是层级智能体

类比公司组织架构：

顶层 Supervisor（=CEO）：只决定"这事归哪个部门"。
部门 Supervisor（=部门经理）：决定"这事组内谁干"。
Worker（=员工）：实际执行。

mermaid

flowchart TB
    TOP[顶层 Supervisor<br/>工程总监]
    TOP --> FE_TEAM[前端组 Supervisor]
    TOP --> BE_TEAM[后端组 Supervisor]
    FE_TEAM --> FE1[React 工程师]
    FE_TEAM --> FE2[Vue 工程师]
    BE_TEAM --> BE1[Python 工程师]
    BE_TEAM --> BE2[Java 工程师]

何时用层级，何时用扁平

场景	推荐
worker ≤ 5 个，任务单一	扁平 Supervisor
worker 6-15 个，跨 2-3 个领域	两层层级
worker > 15 个，多领域	三层及以上
想快速验证原型	扁平，别一上来就层级

新手原则：从扁平开始，等 supervisor 路由准确率下降再分层。层级越多越难调试，别过度设计。

二、用子图实现每个团队

LangGraph 的层级实现思路：每个"团队"是一个编译好的子图，作为外层图的一个节点。子图内部自己有 supervisor + workers，对外只暴露输入输出。

2.1 状态设计

层级的状态有个关键点：每层有自己的状态 schema，但消息要互通。最简单做法是各层都用 messages 字段：

python

from typing import TypedDict, Annotated, Literal

class TeamState(TypedDict):
    messages: Annotated[list, "add"]
    next: str   # 下一步路由，宽松点用 str

2.2 子团队：前端组

前端组内部是个迷你 Supervisor 系统，有 React 和 Vue 两个 worker：

python

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import create_react_agent

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

# 前端 supervisor 的路由器
from pydantic import BaseModel
class FeRoute(BaseModel):
    next: Literal["react", "vue", "FINISH"]

fe_router = llm.with_structured_output(FeRoute)
FE_PROMPT = """你是前端组长。react 负责 React 问题，vue 负责 Vue 问题。
答完了返回 FINISH。"""

def fe_supervisor(state: TeamState):
    msgs = [SystemMessage(content=FE_PROMPT)] + state["messages"]
    return {"next": fe_router.invoke(msgs).next}

react_agent = create_react_agent(llm, [], prompt="你是 React 专家。")
vue_agent = create_react_agent(llm, [], prompt="你是 Vue 专家。")

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

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

# 组装前端子图
fe_builder = StateGraph(TeamState)
fe_builder.add_node("supervisor", fe_supervisor)
fe_builder.add_node("react", react_node)
fe_builder.add_node("vue", vue_node)
fe_builder.add_edge(START, "supervisor")
fe_builder.add_conditional_edges("supervisor", lambda s: s["next"],
    {"react": "react", "vue": "vue", "FINISH": END})
fe_builder.add_edge("react", "supervisor")
fe_builder.add_edge("vue", "supervisor")
fe_team = fe_builder.compile()

后端组结构完全一样，把 react/vue 换成 python/java，prompt 改一下即可（代码略，参考配套示例）。

2.3 顶层 Supervisor

顶层只关心"这事归前端还是后端"，不关心具体谁干：

python

class TopRoute(BaseModel):
    next: Literal["fe_team", "be_team", "FINISH"]

top_router = llm.with_structured_output(TopRoute)
TOP_PROMPT = """你是工程总监。
- 前端问题（React/Vue/页面/CSS）交给 fe_team。
- 后端问题（Python/Java/API/数据库）交给 be_team。
- 完成返回 FINISH。"""

def top_supervisor(state: TeamState):
    msgs = [SystemMessage(content=TOP_PROMPT)] + state["messages"]
    return {"next": top_router.invoke(msgs).next}

2.4 把子图作为节点

子图编译后是个 CompiledStateGraph，可以直接当节点加进外层图：

python

top_builder = StateGraph(TeamState)
top_builder.add_node("supervisor", top_supervisor)
top_builder.add_node("fe_team", fe_team)   # 整个子图当一个节点
top_builder.add_node("be_team", be_team)   # 假设已编译

top_builder.add_edge(START, "supervisor")
top_builder.add_conditional_edges("supervisor", lambda s: s["next"],
    {"fe_team": "fe_team", "be_team": "be_team", "FINISH": END})
top_builder.add_edge("fe_team", "supervisor")
top_builder.add_edge("be_team", "supervisor")

org = top_builder.compile()

三、完整可运行示例

为节省篇幅，下面只画前端组 + 后端组各一个 worker，但结构完整：

python

from typing import TypedDict, Annotated, Literal
from pydantic import BaseModel
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

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
class TeamState(TypedDict):
    messages: Annotated[list, "add"]
    next: str

# ----- 前端组（仅 React worker）-----
class FeRoute(BaseModel):
    next: Literal["react", "FINISH"]
fe_router = llm.with_structured_output(FeRoute)
def fe_supervisor(state):
    msgs = [SystemMessage(content="你是前端组长，react 干活，完成返回 FINISH。")] + state["messages"]
    return {"next": fe_router.invoke(msgs).next}
react_agent = create_react_agent(llm, [], prompt="你是 React 专家。")
def react_node(state):
    r = react_agent.invoke({"messages": state["messages"]})
    return {"messages": [r["messages"][-1]]}
fe_builder = StateGraph(TeamState)
fe_builder.add_node("supervisor", fe_supervisor)
fe_builder.add_node("react", react_node)
fe_builder.add_edge(START, "supervisor")
fe_builder.add_conditional_edges("supervisor", lambda s: s["next"],
    {"react": "react", "FINISH": END})
fe_builder.add_edge("react", "supervisor")
fe_team = fe_builder.compile()

# ----- 后端组（仅 Python worker）-----
class BeRoute(BaseModel):
    next: Literal["python", "FINISH"]
be_router = llm.with_structured_output(BeRoute)
def be_supervisor(state):
    msgs = [SystemMessage(content="你是后端组长，python 干活，完成返回 FINISH。")] + state["messages"]
    return {"next": be_router.invoke(msgs).next}
py_agent = create_react_agent(llm, [], prompt="你是 Python 专家。")
def python_node(state):
    r = py_agent.invoke({"messages": state["messages"]})
    return {"messages": [r["messages"][-1]]}
be_builder = StateGraph(TeamState)
be_builder.add_node("supervisor", be_supervisor)
be_builder.add_node("python", python_node)
be_builder.add_edge(START, "supervisor")
be_builder.add_conditional_edges("supervisor", lambda s: s["next"],
    {"python": "python", "FINISH": END})
be_builder.add_edge("python", "supervisor")
be_team = be_builder.compile()

# ----- 顶层 -----
class TopRoute(BaseModel):
    next: Literal["fe_team", "be_team", "FINISH"]
top_router = llm.with_structured_output(TopRoute)
def top_supervisor(state):
    msgs = [SystemMessage(content="前端问题交 fe_team，后端问题交 be_team，完成 FINISH。")] + state["messages"]
    return {"next": top_router.invoke(msgs).next}

top_builder = StateGraph(TeamState)
top_builder.add_node("supervisor", top_supervisor)
top_builder.add_node("fe_team", fe_team)
top_builder.add_node("be_team", be_team)
top_builder.add_edge(START, "supervisor")
top_builder.add_conditional_edges("supervisor", lambda s: s["next"],
    {"fe_team": "fe_team", "be_team": "be_team", "FINISH": END})
top_builder.add_edge("fe_team", "supervisor")
top_builder.add_edge("be_team", "supervisor")
org = top_builder.compile()

# 测试
result = org.invoke(
    {"messages": [HumanMessage(content="用 Python 写一个 hello world")], "next": "be_team"},
    config={"recursion_limit": 50},
)
print(result["messages"][-1].content)

流程：顶层 supervisor 判断是后端问题→交给后端组→后端组长派 python worker→worker 回答→回到顶层→FINISH。

四、层级结构图

mermaid

flowchart TB
    S([START]) --> TOP[顶层 Supervisor]
    TOP -->|前端| FE[前端组子图]
    TOP -->|后端| BE[后端组子图]
    TOP -->|完成| E([END])
    FE --> TOP
    BE --> TOP

    subgraph 前端组内部
        FES[组长] --> RE[react]
        FES --> VU[vue]
        RE --> FES
        VU --> FES
    end
    subgraph 后端组内部
        BES[组长] --> PY[python]
        BES --> JA[java]
        PY --> BES
        JA --> BES
    end

五、状态在层级间流转

关键技巧：

子图作为节点时，外层会把同名字段传进去。所以内外都用 messages 字段就能自动传递。
子图的最终输出（最后一条消息）会作为该节点的输出回到外层。
如果想让子图只返回"摘要"而非全部内部对话，在子图入口节点里先压缩 messages。

六、常见踩坑

1. 层级过深难调试

三层以上时，一个错误可能来自任何一层，定位极痛苦。建议：

每层都加 checkpointer，用 LangGraph Studio 或打断点逐层看状态。
每层 next 字段加日志。
别为了"看起来高级"加层。

2. 状态传递丢失

子图用的状态 schema 和外层不一致时，字段会"消失"。比如外层有 user_id，子图 schema 没这个字段，就传不进去。解决：子图 schema 显式包含需要的字段，或在节点函数里手动取值。

3. 递归上限爆掉

每层循环都算步数，三层嵌套很容易 50 步打不住。复杂系统设 recursion_limit=100 起步。

4. 顶层 supervisor 不知道子团队能力

如果顶层 prompt 没说清楚"前端组会什么、后端组会什么"，它会瞎派。把每个子团队的能力清单写进顶层 prompt。

七、小结

层级 = 多级 Supervisor 嵌套，适合大型多领域任务。
实现核心：每个子团队是一个编译好的子图，作为外层节点。
状态靠同名字段（如 messages）在层级间自动流转。
别滥用：扁平能搞定就别分层，调试成本很高。

下一章我们换话题，讲智能体的记忆机制——怎么让 agent 记住用户的长期偏好。

层级智能体 ​

一、什么是层级智能体 ​

何时用层级，何时用扁平 ​

二、用子图实现每个团队 ​

2.1 状态设计 ​

2.2 子团队：前端组 ​

2.3 顶层 Supervisor ​

2.4 把子图作为节点 ​

三、完整可运行示例 ​

四、层级结构图 ​

五、状态在层级间流转 ​

六、常见踩坑 ​

1. 层级过深难调试 ​

2. 状态传递丢失 ​

3. 递归上限爆掉 ​

4. 顶层 supervisor 不知道子团队能力 ​

七、小结 ​

层级智能体

一、什么是层级智能体

何时用层级，何时用扁平

二、用子图实现每个团队

2.1 状态设计

2.2 子团队：前端组

2.3 顶层 Supervisor

2.4 把子图作为节点

三、完整可运行示例

四、层级结构图

五、状态在层级间流转

六、常见踩坑

1. 层级过深难调试

2. 状态传递丢失

3. 递归上限爆掉

4. 顶层 supervisor 不知道子团队能力

七、小结