使用 LangGraph 和监督智能体构建多智能体路由器

将一个智能体置于真实产品面前，它很快就会承担过多任务。让同一个智能体负责客户支持、订单追踪和退款处理，结果它在三方面都会表现平平。其系统提示词会变成长达2000字的冗长条款，内容诸如“如果用户询问X，则执行Y，但如果同时涉及Z，则……”。解决之道并非打造更聪明的智能体，而是使用更多规模较小的智能体，并加上一个负责在它们之间分配任务的监督者。这可以说是目前智能体AI中最广泛采用的生产模式，而且一个可运行的版本仅需一段简短的LangGraph代码即可实现。

单一巨型智能体的问题

假设你在开发一个能够回答一般问题、进行算术运算以及搜索网络获取最新信息的助手。一种诱人的设计方式是使用一个智能体、一个包含所有指令的庞大提示词，再将所有工具都附加在其中。但随着工具和指令的不断增加，模型会变得难以判断该使用哪种工具，其推理能力也会逐渐减弱。调试工作则会变成一场噩梦，因为当所有内容都集中在同一个提示词中时，你无法找出它为何选择了错误的工具。

解决方法是划分职责：一个智能体负责算术运算，另一个负责信息检索，还有一个监督者——实际上就是一个路由器——负责决定由哪位专家处理请求的各个部分。

监督者模式

想象一下客服中心。你打电话提出问题，却无法直接接通专家。首先会有接待员接听，判断你需要的是账单处理、技术支持还是销售服务，然后再为你转接。接待员相当于主管，而那三个团队则是相应的专业客服人员。主管从不亲自回答问题，其唯一职责就是挑选合适的人来处理。

从视觉上看是这样的：

                    ┌─────────────┐
                    │    User     │
                    │   Request   │
                    └──────┬──────┘
                           │
                           ▼
                   ┌───────────────┐
                   │   Supervisor  │
                   │  (routes task)│
                   └───────┬───────┘
                    ┌──────┴──────┐
                    ▼             ▼
             ┌────────────┐ ┌────────────┐
             │  Research  │ │    Math    │
             │   Agent    │ │   Agent    │
             └──────┬─────┘ └──────┬─────┘
                    │              │
                    └──────┬───────┘
                           ▼
                   ┌───────────────┐
                   │   Supervisor  │
                   │ (final answer)│
                   └───────────────┘

监督器负责接收任务，将其转交给合适的专家，获取处理结果后，再决定是进一步传递还是直接将最终答案提供给用户。

推荐的构建步骤

在编写代码之前需要了解的一点是：LangChain团队现在建议直接从工具调用层面构建监督器模式，而无需仅依赖旧版的langgraph-supervisor辅助库，因为工具调用能更好地掌控上下文构建过程。下面首先介绍预构建的快速版本，随后是手动实现方式，以便清晰了解其内部运作机制。

步骤1：安装相关包

pip install langgraph langchain-openai langgraph-supervisor

第2步：培养专家

从引进人才开始：

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

第一个专家负责处理数学问题，其工具为两个简单的函数。首先是加法：

# Specialist 1: handles math
def add(a: float, b: float) -> float:
    """Add two numbers together."""
    return a + b

接着是乘法：

def multiply(a: float, b: float) -> float:
    """Multiply two numbers together."""
    return a * b

将它们封装在具有名称的 ReAct 智能体中，并通过提示词限定其功能角色：

math_agent = create_react_agent(
    model="openai:gpt-4o",
    tools=[add, multiply],
    name="math_expert",
    prompt="You are a math specialist. Solve numeric problems precisely."
)

第二位专家负责研究工作。他的工具之一就是搜索功能，此处仅展示其框架：

# Specialist 2: handles research
def web_search(query: str) -> str:
    """Search the web for current information."""
    # In production, replace this with a real search API call
    return f"Search results for '{query}': [your real search integration here]"

也要将其用代理封装起来：

research_agent = create_react_agent(
    model="openai:gpt-4o",
    tools=[web_search],
    name="research_expert",
    prompt="You are a research specialist. Find accurate, current information."
)

每个智能体仅负责一项任务，并且只使用该任务所需的工具，因此不会产生混淆，也不会出现提示过载的情况。示例中使用了OpenAI模型，不过像Groq或OpenRouter这样的免费替代方案也同样适用。

步骤3：构建监督器

导入辅助工具：

from langgraph_supervisor import create_supervisor

然后为其提供两位专家，即一个模型和一个提示词，告知它在何时使用每个专家：

workflow = create_supervisor(
    [research_agent, math_agent],
    model=ChatOpenAI(model="gpt-4o"),
    prompt=(
        "You are a team supervisor managing a research expert and a math expert. "
        "For questions requiring current information or web lookups, use research_expert. "
        "For anything involving calculation, use math_expert. "
        "Route each request to exactly the right specialist."
    )
)

将工作流编译为可运行的应用程序：

app = workflow.compile()

第4步：运行它

发送一道数学题：

result = app.invoke({
    "messages": [
        {"role": "user", "content": "What's 1,847 multiplied by 293?"}
    ]
})

并打印最终消息：

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

以下是其内部的工作流程：

1. 监管器读取请求内容。
2. 它发现这是一个计算任务而非研究问题。
3. 它将任务交给 math_expert 处理。
4. math_expert 使用其 multiply 工具计算出结果。
5. 监管器重新接管控制权并返回结果。

对于研究类问题，比如大型科技公司的当前员工人数，同一监管器会将其转交给 research_expert。代码完全相同，只有任务分配方式不同，而这一决策由大语言模型自行判断。

这些任务转交工具的真正作用

这部分内容消除了那种“魔法感”。监督器并非通过某种隐藏的LangGraph机制来处理任务，而是像普通工具调用代理一样，通过工具调用来完成工作。在幕后，create_supervisor会为每位专家创建一个对应的传递工具，其名称诸如transfer_to_research_expert和transfer_to_math_expert。监督器模型将这些视为可调用的工具，与查询天气信息并无不同；因此当它判断需要开展研究时，只需调用相应的传递函数即可。这一设计令人安心：监督器其实并非什么新概念，它不过是常规的代理循环而已，只不过某些工具是由其他代理而非API调用来实现的。

手动构建它

如果您希望对路由进行完全控制——在即将投入生产时这是个明智之举——那么这里有一个基于原始 StateGraph 的手动实现版本。首先从导入部分开始：

from typing import Annotated, Sequence, TypedDict
import operator
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph, END

定义共享状态，其中包含消息列表以及下一个节点的名称：

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    next: str

监控节点查看最新的消息：

def supervisor_node(state: AgentState):
    """Decide which specialist should act next."""
    last_message = state["messages"][-1].content

在实际应用中，该决策会通过调用大型语言模型来作出。而在这里则简化为简单的关键词检查，这样路由逻辑就依然清晰可见：

# In production, this decision comes from an LLM call —
    # simplified here to show the routing logic clearly
    if any(word in last_message.lower() for word in ["calculate", "multiply", "add", "sum"]):
        return {"next": "math_agent"}
    else:
        return {"next": "research_agent"}

每位专家都会获得一个节点，该节点会调用其对应的代理，然后将控制权交还给监督器。先来算数学问题：

def math_node(state: AgentState):
    result = math_agent.invoke(state)
    return {"messages": [result], "next": "supervisor"}

然后进行研究：

def research_node(state: AgentState):
    result = research_agent.invoke(state)
    return {"messages": [result], "next": "supervisor"}

接下来，通过注册节点将图表连接起来：

# Wire the graph together
workflow = StateGraph(AgentState)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("math_agent", math_node)
workflow.add_node("research_agent", research_node)

设置入口点：

workflow.set_entry_point("supervisor")

添加根据主管决策生成的条件边：

workflow.add_conditional_edges(
    "supervisor",
    lambda state: state["next"],
    {"math_agent": "math_agent", "research_agent": "research_agent", "end": END}
)

之后将每位专家送回给主管：

workflow.add_edge("math_agent", "supervisor")
workflow.add_edge("research_agent", "supervisor")

然后编译：

app = workflow.compile()

虽然文字更多，但没有任何内容被隐藏。你可以清楚地看到状态是如何流转的、每个路由决策是如何做出的，以及应在何处添加自己的逻辑，比如日志记录、约束条件或自定义终止条件。

最容易过时的状态设计选择

根据实际生产经验给出的警告：状态架构是日后最难修改的决策。人们很容易一开始就试图把所有可能用到的字段都放入一个扁平的字典中，涵盖消息、用户上下文以及中间结果。对于两个智能体来说这倒没问题，但当有八到十个节点都在读写同一个共享状态时，问题就变得十分棘手，因为LangGraph的类型检查非常严格，要整理臃肿的状态就必须修改每个节点。即便只是两个智能体的系统，也应从第一天起就精心设计架构，明确每个智能体真正需要读写的内容，将其他所有内容都排除在共享状态之外。

你将遇到的失败

投入实际使用的监督系统往往会出现这种模式。想象一下监督系统分配任务给专家处理，专家完成之后，监督系统却认为结果不达标而将其退回。专家再次修改，但新版本反而比最初更差。监督系统依然不满意，再次要求修改，除非采取相应措施，否则这种情况可能会无限循环。

解决办法是对监督系统的来回迭代设置上限，可以在状态中使用简单的计数器，并设定明确的、可确定的停止条件。仅依靠大语言模型对是否合格的判断作为唯一的停止标准并不可靠。

何时不应使用它

要诚实地评估适用性。如果任务仅仅是先检索信息再生成结果，比如普通的RAG式问答系统，那么使用监督节点就显得多余了。在实际上并不存在需要路由决策的情况下，额外的状态管理和路由开销毫无意义。只有当真正存在不同的专家角色以及真正的路由选择时，才考虑使用监督节点模式，切勿默认采用它。

总结

• 随着复杂度增加，单一的过载代理会无法正常工作。
• 监督节点是一种负责将任务分配给各专家角色的路由代理。
• create_supervisor是预先构建好的快速路由方式。
• 在底层，任务传递实际上只是普通的工具调用而已。
• 手动创建的StateGraph可以提供完全的路由控制能力。
• 两个实际应用中的经验教训：尽早设计状态结构，并限制修改循环的次数。