LangGraph中的Agent Supervisor模式

一、引言

原文地址：https://langchain-ai.github.io/langgraph/tutorials/multi_agent/agent_supervisor/

Agent Supervisor模式是LangGraph框架中的一种强大的多智能体架构设计模式，它通过建立一个中央协调机制，使多个专业化的智能体能够协同工作，共同解决复杂问题。这种模式特别适合需要不同专业领域知识协作的场景，如复杂查询处理、多领域专业知识协作等。

核心思想

专业化智能体协作：由多个专注于特定领域的智能体协同工作
中央协调控制：由主管智能体(Supervisor)负责协调所有通信和任务分配
基于上下文的决策：主管根据当前上下文和任务需求决定调用哪个智能体

主要目的

实现智能体间的专业化分工
优化任务处理流程
确保信息流的有序传递
提高复杂任务的解决效率

二、架构与组件

Agent Supervisor模式的架构包含两类主要组件：Supervisor和Agent。它们之间形成了一个星形拓扑结构，所有通信都通过中央的主管智能体进行路由。

主管智能体 (Supervisor Agent)

主管智能体是整个系统的核心，负责协调和管理所有工人智能体。

作为中心协调者

接收用户任务并进行分析

决定将任务分配给哪个工人智能体

整合工人智能体返回的结果

控制整个工作流的通信和任务调度

工人智能体 (Worker Agents)

工人智能体是专注于特定领域的执行者，每个工人智能体都配备了相应的专业工具。

每个工人智能体专注于特定领域

配备相应的专业工具

接收主管分配的任务

执行任务并将结果返回给主管

不直接与其他工人智能体通信

三、实现方法

LangGraph提供了两种实现Agent Supervisor模式的方法：使用预构建的langgraph-supervisor库和从零开始构建。

方法一：使用预构建的`langgraph-supervisor`库

这是最简单的实现方式，只需几行代码即可创建一个功能完整的Agent Supervisor系统。

from langgraph_supervisor import create_supervisor
from langchain.chat_models import init_chat_model

supervisor = create_supervisor(
    model=init_chat_model("openai:gpt-4.1"),
    agents=[research_agent, math_agent],  # 预先创建的工人智能体列表
    prompt=(
        "You are a supervisor managing two agents:\n"
        "- a research agent. Assign research-related tasks to this agent\n"
        "- a math agent. Assign math-related tasks to this agent\n"
        "Assign work to one agent at a time, do not call agents in parallel.\n"
        "Do not do any work yourself."
    ),
    add_handoff_back_messages=True,  # 在控制权交还主管时添加消息
    output_mode="full_history",  # 输出完整消息历史
).compile()

from langgraph_supervisor import create_supervisor

from langchain.chat_models import init_chat_model

supervisor = create_supervisor(

model=init_chat_model("openai:gpt-4.1"),

agents=[research_agent, math_agent], # 预先创建的工人智能体列表

prompt=(

"You are a supervisor managing two agents:\n"

"- a research agent. Assign research-related tasks to this agent\n"

"- a math agent. Assign math-related tasks to this agent\n"

"Assign work to one agent at a time, do not call agents in parallel.\n"

"Do not do any work yourself."

add_handoff_back_messages=True, # 在控制权交还主管时添加消息

output_mode="full_history", # 输出完整消息历史

).compile()

方法二：从零开始构建

从零开始构建Agent Supervisor系统需要更多的代码，但提供了更高的灵活性和控制力。主要步骤包括：

创建工人智能体
设置智能体通信（交接工具）
创建主管智能体
创建多智能体图

四、代码示例与实现细节

1. 创建工人智能体

以下是两个专业化工人智能体的示例：研究智能体和数学智能体。

研究智能体示例：

from langchain_tavily import TavilySearch
from langgraph.prebuilt import create_react_agent

web_search = TavilySearch(max_results=3)
research_agent = create_react_agent(
    model="openai:gpt-4.1",
    tools=[web_search],
    prompt=(
        "You are a research agent.\n\n"
        "INSTRUCTIONS:\n"
        "- Assist ONLY with research-related tasks, DO NOT do any math\n"
        "- After you're done with your tasks, respond to the supervisor directly\n"
        "- Respond ONLY with the results of your work, do NOT include ANY other text."
    ),
    name="research_agent",
)

from langchain_tavily import TavilySearch

from langgraph.prebuilt import create_react_agent

web_search = TavilySearch(max_results=3)

research_agent = create_react_agent(

model="openai:gpt-4.1",

tools=[web_search],

prompt=(

"You are a research agent.\n\n"

"INSTRUCTIONS:\n"

"- Assist ONLY with research-related tasks, DO NOT do any math\n"

"- After you're done with your tasks, respond to the supervisor directly\n"

"- Respond ONLY with the results of your work, do NOT include ANY other text."

name="research_agent",

)

数学智能体示例：

def add(a: float, b: float):
    """Add two numbers."""
    return a + b

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

def divide(a: float, b: float):
    """Divide two numbers."""
    return a / b

math_agent = create_react_agent(
    model="openai:gpt-4.1",
    tools=[add, multiply, divide],
    prompt=(
        "You are a math agent.\n\n"
        "INSTRUCTIONS:\n"
        "- Assist ONLY with math-related tasks\n"
        "- After you're done with your tasks, respond to the supervisor directly\n"
        "- Respond ONLY with the results of your work, do NOT include ANY other text."
    ),
    name="math_agent",
)

def add(a: float, b: float):

"""Add two numbers."""

return a + b

def multiply(a: float, b: float):

"""Multiply two numbers."""

return a * b

def divide(a: float, b: float):

"""Divide two numbers."""

return a / b

math_agent = create_react_agent(

model="openai:gpt-4.1",

tools=[add, multiply, divide],

prompt=(

"You are a math agent.\n\n"

"INSTRUCTIONS:\n"

"- Assist ONLY with math-related tasks\n"

"- After you're done with your tasks, respond to the supervisor directly\n"

"- Respond ONLY with the results of your work, do NOT include ANY other text."

name="math_agent",

)

2. 设置智能体通信（交接工具）

交接工具是实现智能体之间通信的关键机制，它允许主管智能体将控制权转移给工人智能体。

from typing import Annotated
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import InjectedState
from langgraph.graph import StateGraph, START, MessagesState
from langgraph.types import Command

def create_handoff_tool(*, agent_name: str, description: str | None = None):
    name = f"transfer_to_{agent_name}"
    description = description or f"Ask {agent_name} for help."

    @tool(name, description=description)
    def handoff_tool(
        state: Annotated[MessagesState, InjectedState],
        tool_call_id: Annotated[str, InjectedToolCallId],
    ) -> Command:
        tool_message = {
            "role": "tool",
            "content": f"Successfully transferred to {agent_name}",
            "name": name,
            "tool_call_id": tool_call_id,
        }
        return Command(
            goto=agent_name,
            update={**state, "messages": state["messages"] + [tool_message]},
            graph=Command.PARENT,
        )
    return handoff_tool

assign_to_research_agent = create_handoff_tool(
    agent_name="research_agent", 
    description="Assign task to a researcher agent."
)
assign_to_math_agent = create_handoff_tool(
    agent_name="math_agent", 
    description="Assign task to a math agent."
)

from typing import Annotated

from langchain_core.tools import tool, InjectedToolCallId

from langgraph.prebuilt import InjectedState

from langgraph.graph import StateGraph, START, MessagesState

from langgraph.types import Command

def create_handoff_tool(*, agent_name: str, description: str | None = None):

name = f"transfer_to_{agent_name}"

description = description or f"Ask {agent_name} for help."

@tool(name, description=description)

def handoff_tool(

state: Annotated[MessagesState, InjectedState],

tool_call_id: Annotated[str, InjectedToolCallId],

) -> Command:

tool_message = {

"role": "tool",

"content": f"Successfully transferred to {agent_name}",

"name": name,

"tool_call_id": tool_call_id,

}

return Command(

goto=agent_name,

update={**state, "messages": state["messages"] + [tool_message]},

graph=Command.PARENT,

)

return handoff_tool

assign_to_research_agent = create_handoff_tool(

agent_name="research_agent",

description="Assign task to a researcher agent."

)

assign_to_math_agent = create_handoff_tool(

agent_name="math_agent",

description="Assign task to a math agent."

)

3. 创建主管智能体

主管智能体配备了交接工具，使其能够将任务分配给不同的工人智能体。

supervisor_agent = create_react_agent(
    model="openai:gpt-4.1",
    tools=[assign_to_research_agent, assign_to_math_agent],
    prompt=(
        "You are a supervisor managing two agents:\n"
        "- a research agent. Assign research-related tasks to this agent\n"
        "- a math agent. Assign math-related tasks to this agent\n"
        "Assign work to one agent at a time, do not call agents in parallel.\n"
        "Do not do any work yourself."
    ),
    name="supervisor",
)

supervisor_agent = create_react_agent(

model="openai:gpt-4.1",

tools=[assign_to_research_agent, assign_to_math_agent],

prompt=(

"You are a supervisor managing two agents:\n"

"- a research agent. Assign research-related tasks to this agent\n"

"- a math agent. Assign math-related tasks to this agent\n"

"Assign work to one agent at a time, do not call agents in parallel.\n"

"Do not do any work yourself."

name="supervisor",

)

4. 创建多智能体图

多智能体图定义了智能体之间的通信流程和控制流转移。

from langgraph.graph import END

supervisor = (
    StateGraph(MessagesState)
    .add_node(supervisor_agent, destinations=("research_agent", "math_agent", END))
    .add_node(research_agent)
    .add_node(math_agent)
    .add_edge(START, "supervisor")
    .add_edge("research_agent", "supervisor")
    .add_edge("math_agent", "supervisor")
    .compile()
)

from langgraph.graph import END

supervisor = (

StateGraph(MessagesState)

.add_node(supervisor_agent, destinations=("research_agent", "math_agent", END))

.add_node(research_agent)

.add_node(math_agent)

.add_edge(START, "supervisor")

.add_edge("research_agent", "supervisor")

.add_edge("math_agent", "supervisor")

.compile()

)

五、高级功能：精细任务委托

为了更精细地控制任务委托，可以使用Send()原语，它允许主管智能体在交接过程中直接向工人智能体发送特定数据。

from langgraph.types import Send

def create_task_description_handoff_tool(
    *, agent_name: str, description: str | None = None
):
    name = f"transfer_to_{agent_name}"
    description = description or f"Ask {agent_name} for help."

    @tool(name, description=description)
    def handoff_tool(
        task_description: Annotated[
            str,
            "Description of what the next agent should do, including all of the relevant context.",
        ],
        state: Annotated[MessagesState, InjectedState],
    ) -> Command:
        task_description_message = {"role": "user", "content": task_description}
        agent_input = {**state, "messages": [task_description_message]}
        return Command(
            goto=[Send(agent_name, agent_input)],
            graph=Command.PARENT,
        )
    return handoff_tool

from langgraph.types import Send

def create_task_description_handoff_tool(

*, agent_name: str, description: str | None = None

name = f"transfer_to_{agent_name}"

description = description or f"Ask {agent_name} for help."

@tool(name, description=description)

def handoff_tool(

task_description: Annotated[

str,

"Description of what the next agent should do, including all of the relevant context.",

state: Annotated[MessagesState, InjectedState],

) -> Command:

task_description_message = {"role": "user", "content": task_description}

agent_input = {**state, "messages": [task_description_message]}

return Command(

goto=[Send(agent_name, agent_input)],

graph=Command.PARENT,

)

return handoff_tool

六、关键要点

完整消息历史

默认情况下，主管智能体将所有工人智能体消息（包括其内部工具调用循环）附加到完整消息历史中，使主管能够了解全局情况。

输入控制

可以使用Send()原语在交接过程中直接向工人智能体发送特定数据，而不是发送完整的图状态，实现更精确的输入控制。

输出控制

可以通过将智能体包装在单独的节点函数中来控制有多少智能体的内部消息历史被添加到整体主管消息历史中。

交接机制

使用Command原语实现智能体之间的控制权转移，包括goto（指定要跳转到的智能体节点）、update（更新父图状态）和graph=Command.PARENT（指示LangGraph导航到父多智能体图中的智能体节点）。

七、使用场景

Agent Supervisor模式适用于以下场景：

复杂查询处理：如”查找美国和纽约州2024年GDP，计算纽约州GDP占美国GDP的百分比”，需要先研究数据再进行数学计算。

多领域专业知识协作：需要不同领域专家（如研究、数学、编程、写作等）协作完成的任务。

任务分解与协调：将复杂任务分解为子任务，并由专门的智能体处理各个子任务。

信息流控制：需要严格控制信息流向和任务分配的场景，确保每个智能体只接收到与其相关的信息。

专业化工具使用：不同智能体配备不同工具集，通过主管智能体的协调实现最优工具选择。

八、总结

Agent Supervisor模式是构建复杂多智能体系统的强大架构，通过中央协调和专业化分工，能够有效处理需要多种能力协作的复杂任务。它提供了灵活的实现选项，从简单的预构建库到完全自定义的实现，可以根据具体需求进行选择。通过合理设计主管智能体和工人智能体，以及它们之间的通信机制，可以构建出高效、可靠的多智能体系统。

Heart.Think.Do