多智能体系统¶

agent 是 一个使用大语言模型（LLM）来决定应用程序控制流的系统。随着你开发这些系统，它们可能会随着时间变得越来越复杂，从而更难管理和扩展。例如，你可能会遇到以下问题：

智能体拥有太多工具，无法做出关于下一步调用哪个工具的合理决策
上下文变得过于复杂，单个智能体难以跟踪
系统中需要多个专业领域（例如计划者、研究员、数学专家等）

为了解决这些问题，你可以考虑将你的应用程序拆分成多个较小且独立的智能体，并将它们组合成一个 多智能体系统。这些独立的智能体可以简单到一个提示和一次 LLM 调用，也可以复杂到 ReAct 智能体（甚至更复杂！）。

使用多智能体系统的主要优势包括：

模块化：独立的智能体使得开发、测试和维护智能体系统更加容易。
专业化：你可以创建专注于特定领域的专家智能体，这有助于提升整个系统的性能。
控制：你可以显式地控制智能体之间的通信方式（而不是依赖函数调用）。

多智能体架构¶

在多智能体系统中，连接智能体有几种方式：

网络：每个智能体可以与其他所有智能体进行通信。任何智能体都可以决定下一步调用哪个其他智能体。
监督者：每个智能体与一个单一的监督者智能体进行通信。监督者智能体决定下一步应该调用哪个智能体。
监督者（工具调用）：这是监督者架构的一个特殊情况。单个智能体可以表示为工具。在这种情况下，监督者智能体使用一个工具调用LLM来决定应调用哪些代理工具，以及传递给这些智能体的参数。
分层：您可以定义一个具有监督者监督者的多智能体系统。这是监督者架构的一种泛化，并允许更复杂的控制流程。
自定义多智能体工作流：每个智能体仅与一部分其他智能体进行通信。流程的一部分是确定性的，只有某些智能体可以决定下一步调用哪些其他智能体。

交接¶

在多智能体架构中，智能体可以表示为图节点。每个智能体节点执行其步骤并决定是否完成执行或路由到另一个智能体，包括可能路由到自身（例如，在循环中运行）。多智能体交互中的常见模式是**交接**，其中某个智能体将控制权*交接*给另一个智能体。交接允许您指定：

目标：要导航到的目标智能体（例如，要前往的节点名称）
负载：传递给该智能体的信息（例如，状态更新）

要在LangGraph中实现交接，智能体节点可以返回 Command 对象，该对象允许您结合控制流和状态更新：

def agent(state) -> Command[Literal["agent", "another_agent"]]:
    # 路由/停止的条件可以是任何内容，例如LLM工具调用/结构化输出等。
    goto = get_next_agent(...)  # 'agent' / 'another_agent'
    return Command(
        # 指定下一步要调用的智能体
        goto=goto,
        # 更新图状态
        update={"my_state_key": "my_state_value"}
    )

在更复杂的情况下，每个智能体节点本身是一个图（即子图），其中一个智能体子图中的节点可能想要导航到不同的智能体。例如，如果您有两个智能体，alice和bob（父图中的子图节点），并且alice需要导航到bob，则可以在Command对象中设置graph=Command.PARENT：

def some_node_inside_alice(state):
    return Command(
        goto="bob",
        update={"my_state_key": "my_state_value"},
        # 指定要导航到的图（默认为当前图）
        graph=Command.PARENT,
    )

Note

如果您需要支持使用 Command(graph=Command.PARENT) 的子图通信的可视化，则需要将它们包装在一个带有 Command 注解的节点函数中，例如，而不是这样：

builder.add_node(alice)

您需要这样做：

def call_alice(state) -> Command[Literal["bob"]]:
    return alice.invoke(state)

builder.add_node("alice", call_alice)

作为工具的交接¶

最常见的智能体类型之一是工具调用智能体。对于这些类型的智能体，一种常见的模式是将交接封装在工具调用中，例如：

^{API Reference: tool}

from langchain_core.tools import tool

def transfer_to_bob():
    """转交给 bob."""
    return Command(
        # 要前往的智能体（节点）名称
        goto="bob",
        # 发送给智能体的数据
        update={"my_state_key": "my_state_value"},
        # 表示要导航到父图中的智能体节点
        graph=Command.PARENT,
    )

这是从工具更新图状态的一个特殊情况，除了状态更新外，还包含控制流。

Important

如果您想使用返回 Command 的工具，可以使用预构建的 create_react_agent / ToolNode 组件，或者实现自己的工具执行节点，收集工具返回的 Command 对象并返回它们的列表，例如：

def call_tools(state):
    ...
    commands = [tools_by_name[tool_call["name"]].invoke(tool_call) for tool_call in tool_calls]
    return commands

现在我们详细看看不同的多智能体架构。

网络¶

在此架构中，智能体被定义为图节点。每个智能体可以与其他所有智能体通信（多对多连接），并可以决定下一步调用哪个智能体。这种架构适用于没有明确的智能体层次结构或没有特定顺序要求的问题。

^{API Reference: ChatOpenAI | Command | StateGraph | START | END}

from typing import Literal
from langchain_openai import ChatOpenAI
from langgraph.types import Command
from langgraph.graph import StateGraph, MessagesState, START, END

model = ChatOpenAI()

def agent_1(state: MessagesState) -> Command[Literal["agent_2", "agent_3", END]]:
    # 您可以将状态的相关部分传递给 LLM（例如，state["messages"]）
    # 以确定下一步调用哪个智能体。一种常见模式是调用模型
    # 并使用结构化输出（例如，强制它返回一个包含 "next_agent" 字段的输出）
    response = model.invoke(...)
    # 根据 LLM 的决策路由到其中一个智能体或退出
    # 如果 LLM 返回 "__end__"，图将完成执行
    return Command(
        goto=response["next_agent"],
        update={"messages": [response["content"]]},
    )

def agent_2(state: MessagesState) -> Command[Literal["agent_1", "agent_3", END]]:
    response = model.invoke(...)
    return Command(
        goto=response["next_agent"],
        update={"messages": [response["content"]]},
    )

def agent_3(state: MessagesState) -> Command[Literal["agent_1", "agent_2", END]]:
    ...
    return Command(
        goto=response["next_agent"],
        update={"messages": [response["content"]]},
    )

builder = StateGraph(MessagesState)
builder.add_node(agent_1)
builder.add_node(agent_2)
builder.add_node(agent_3)

builder.add_edge(START, "agent_1")
network = builder.compile()

监督者¶

在此架构中，我们将智能体定义为节点，并添加一个监督者节点（LLM），它决定下一步应调用哪些智能体节点。我们使用 Command 来根据监督者的决策将执行路由到适当的智能体节点。此架构也适合并行运行多个智能体或使用 map-reduce 模式。

^{API Reference: ChatOpenAI | Command | StateGraph | START | END}

from typing import Literal
from langchain_openai import ChatOpenAI
from langgraph.types import Command
from langgraph.graph import StateGraph, MessagesState, START, END

model = ChatOpenAI()

def supervisor(state: MessagesState) -> Command[Literal["agent_1", "agent_2", END]]:
    # 您可以将状态的相关部分传递给 LLM（例如，state["messages"]）
    # 以确定下一步调用哪个智能体。一种常见模式是调用模型
    # 并使用结构化输出（例如，强制它返回一个包含 "next_agent" 字段的输出）
    response = model.invoke(...)
    # 根据监督者的决策路由到其中一个智能体或退出
    # 如果监督者返回 "__end__"，图将完成执行
    return Command(goto=response["next_agent"])

def agent_1(state: MessagesState) -> Command[Literal["supervisor"]]:
    # 您可以将状态的相关部分传递给 LLM（例如，state["messages"]）
    # 并添加任何额外逻辑（不同模型、自定义提示、结构化输出等）
    response = model.invoke(...)
    return Command(
        goto="supervisor",
        update={"messages": [response]},
    )

def agent_2(state: MessagesState) -> Command[Literal["supervisor"]]:
    response = model.invoke(...)
    return Command(
        goto="supervisor",
        update={"messages": [response]},
    )

builder = StateGraph(MessagesState)
builder.add_node(supervisor)
builder.add_node(agent_1)
builder.add_node(agent_2)

builder.add_edge(START, "supervisor")

supervisor = builder.compile()

查看这个教程以了解监督者多智能体架构的示例。

监督者（工具调用）¶

在这个监督者架构的变种中，我们定义了一个负责调用子智能体的监督者智能体。子智能体以工具的形式暴露给监督者，而监督者智能体决定下一步调用哪个工具。监督者智能体遵循一个标准实现，作为一个在 while 循环中运行的 LLM，直到它决定停止调用工具。

^{API Reference: ChatOpenAI | InjectedState | create_react_agent}

from typing import Annotated
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import InjectedState, create_react_agent

model = ChatOpenAI()

# 这是作为工具调用的智能体函数
# 注意，您可以通过 InjectedState 注解将状态传递给工具
def agent_1(state: Annotated[dict, InjectedState]):
    # 您可以将状态的相关部分传递给 LLM（例如，state["messages"]）
    # 并添加任何额外逻辑（不同模型、自定义提示、结构化输出等）
    response = model.invoke(...)
    # 将 LLM 响应作为字符串返回（预期的工具响应格式）
    # 这将自动转换为 ToolMessage
    # 通过预构建的 create_react_agent（监督者）
    return response.content

def agent_2(state: Annotated[dict, InjectedState]):
    response = model.invoke(...)
    return response.content

tools = [agent_1, agent_2]
# 构建一个带有工具调用的监督者的最简单方法是使用预构建的 ReAct 智能体图
# 它由一个工具调用 LLM 节点（即监督者）和一个工具执行节点组成
supervisor = create_react_agent(model, tools)

分层¶

随着您向系统中添加更多的智能体，监督者可能会难以管理所有的智能体。监督者可能会开始做出关于下一个要调用的智能体的糟糕决策，或者上下文可能会变得过于复杂，以至于一个监督者无法跟踪。换句话说，您最终会遇到最初促使多智能体架构的问题。

为了解决这个问题，您可以 分层地 设计您的系统。例如，您可以创建由个别监督者管理的独立、专门化的智能体团队，并且有一个顶层的监督者来管理这些团队。

^{API Reference: ChatOpenAI | StateGraph | START | END | Command}

from typing import Literal
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.types import Command
model = ChatOpenAI()

# 定义团队 1（与上面的单个监督者示例相同）

def team_1_supervisor(state: MessagesState) -> Command[Literal["team_1_agent_1", "team_1_agent_2", END]]:
    response = model.invoke(...)
    return Command(goto=response["next_agent"])

def team_1_agent_1(state: MessagesState) -> Command[Literal["team_1_supervisor"]]:
    response = model.invoke(...)
    return Command(goto="team_1_supervisor", update={"messages": [response]})

def team_1_agent_2(state: MessagesState) -> Command[Literal["team_1_supervisor"]]:
    response = model.invoke(...)
    return Command(goto="team_1_supervisor", update={"messages": [response]})

team_1_builder = StateGraph(Team1State)
team_1_builder.add_node(team_1_supervisor)
team_1_builder.add_node(team_1_agent_1)
team_1_builder.add_node(team_1_agent_2)
team_1_builder.add_edge(START, "team_1_supervisor")
team_1_graph = team_1_builder.compile()

# 定义团队 2（与上面的单个监督者示例相同）
class Team2State(MessagesState):
    next: Literal["team_2_agent_1", "team_2_agent_2", "__end__"]

def team_2_supervisor(state: Team2State):
    ...

def team_2_agent_1(state: Team2State):
    ...

def team_2_agent_2(state: Team2State):
    ...

team_2_builder = StateGraph(Team2State)
...
team_2_graph = team_2_builder.compile()


# 定义顶层监督者

builder = StateGraph(MessagesState)
def top_level_supervisor(state: MessagesState) -> Command[Literal["team_1_graph", "team_2_graph", END]]:
    # 您可以将状态的相关部分传递给 LLM（例如，state["messages"]）
    # 以确定下一步调用哪个团队。一种常见模式是调用模型
    # 并使用结构化输出（例如，强制它返回一个包含 "next_team" 字段的输出）
    response = model.invoke(...)
    # 根据监督者的决策路由到其中一个团队或退出
    # 如果监督者返回 "__end__"，图将完成执行
    return Command(goto=response["next_team"])

builder = StateGraph(MessagesState)
builder.add_node(top_level_supervisor)
builder.add_node("team_1_graph", team_1_graph)
builder.add_node("team_2_graph", team_2_graph)
builder.add_edge(START, "top_level_supervisor")
builder.add_edge("team_1_graph", "top_level_supervisor")
builder.add_edge("team_2_graph", "top_level_supervisor")
graph = builder.compile()

自定义多智能体工作流¶

在此架构中，我们将单独的智能体添加为图节点，并提前在自定义工作流中定义调用智能体的顺序。在 LangGraph 中，工作流可以通过两种方式定义：

显式控制流（正常边）：LangGraph 允许您通过正常图边显式定义应用程序的控制流（即，智能体之间通信的顺序）。这是上述架构中最确定性的变体——我们总是提前知道下一步将调用哪个智能体。
动态控制流（Command）：在 LangGraph 中，您可以允许 LLM 决定应用程序控制流的部分。这可以通过使用 Command 来实现。这种情况的一个特例是监督者工具调用架构。在这种情况下，驱动监督者智能体的工具调用 LLM 将决定工具（智能体）被调用的顺序。

^{API Reference: ChatOpenAI | StateGraph | START}

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START

model = ChatOpenAI()

def agent_1(state: MessagesState):
    response = model.invoke(...)
    return {"messages": [response]}

def agent_2(state: MessagesState):
    response = model.invoke(...)
    return {"messages": [response]}

builder = StateGraph(MessagesState)
builder.add_node(agent_1)
builder.add_node(agent_2)
# 显式定义流程
builder.add_edge(START, "agent_1")
builder.add_edge("agent_1", "agent_2")

通信与状态管理¶

在构建多智能体系统时，最重要的是弄清楚智能体之间如何进行通信。

智能体之间通信的一种常见且通用的方式是通过消息列表。这引发了一些问题：

智能体是通过交接（handoffs）还是工具调用（tool calls）进行通信？
从一个智能体传递到另一个智能体的消息是什么？
交接是如何在消息列表中表示的？
如何为子智能体管理状态？

此外，如果你正在处理更复杂的智能体，或者希望将单个智能体的状态与多智能体系统状态分开，则可能需要使用不同的状态模式。

交接 vs 工具调用¶

在智能体之间传递的“有效载荷”是什么？在上述大多数架构中，智能体通过交接进行通信，并将图状态作为交接有效载荷的一部分。具体来说，智能体通过图状态传递消息列表。在具有工具调用的监督器的情况下，有效载荷是工具调用参数。

智能体之间的消息传递¶

智能体之间最常见的通信方式是通过共享状态通道，通常是一个消息列表。这假设至少有一个共享的通道（键）存在于状态中（例如 messages）。当通过共享的消息列表进行通信时，还有一个额外的考虑：智能体应该分享其思考过程的完整历史，还是只分享最终结果？

分享完整的思考过程¶

智能体可以**分享其思考过程的完整历史**（即“草稿本”）给所有其他智能体。这个“草稿本”通常看起来像一个消息列表。分享完整思考过程的好处是，它可能帮助其他智能体做出更好的决策，并提高整个系统的推理能力。缺点是，随着智能体数量和复杂性的增加，“草稿本”会迅速增长，可能需要额外的策略来进行内存管理。

只分享最终结果¶

智能体可以拥有自己的私有“草稿本”，并只**分享最终结果**给其他智能体。这种方法可能更适合具有许多智能体或更复杂智能体的系统。在这种情况下，你需要定义具有不同状态模式的智能体。

对于作为工具调用的智能体，监督器根据工具模式确定输入。此外，LangGraph允许在运行时向单个工具传递状态，因此如果需要，下属智能体可以访问父级状态。

在消息中标识智能体名称¶

在较长的消息历史记录中，标识某条AI消息来自哪个智能体可能是有帮助的。一些LLM提供者（如OpenAI）支持在消息中添加一个name参数——你可以使用它来将智能体名称附加到消息上。如果不支持该功能，可以考虑手动将智能体名称注入消息内容中，例如 <agent>alice</agent><message>来自alice的消息</message>。

在消息历史中表示交接¶

交接通常通过LLM调用专门的交接工具完成。这被表示为一个AI消息，其中包含工具调用，并传递给下一个智能体（LLM）。大多数LLM提供者不支持接收带有工具调用的AI消息，**除非**有对应的工具消息。

因此你有两个选项：

在消息列表中添加一个额外的工具消息，例如：“已成功转交给智能体X”
移除包含工具调用的AI消息

在实践中，我们看到大多数开发者选择第一种方法。

子智能体的状态管理¶

一种常见的做法是让多个智能体在一个共享的消息列表上进行通信，但只将它们的最终消息添加到列表中。这意味着任何中间消息（例如工具调用）都不会保存在这个列表中。

如果你确实想保存这些消息，以便将来如果调用这个特定的子智能体，可以将它们传回，那么有两种高层次的方法来实现这一点：

将这些消息存储在共享的消息列表中，但在将其传递给子智能体的LLM之前过滤该列表。例如，你可以选择过滤掉来自**其他**智能体的所有工具调用。
在子智能体的图状态中为每个智能体（例如 alice_messages）存储一个单独的消息列表。这将是他们对消息历史的“视图”。

使用不同的状态模式¶

一个智能体可能需要与其他智能体使用不同的状态模式。例如，搜索智能体可能只需要跟踪查询和检索到的文档。在LangGraph中有两种方法可以实现这一点：

定义子图智能体，使用独立的状态模式。如果子图与父图之间没有共享的状态键（通道），则需要添加输入/输出转换，以使父图知道如何与子图通信。
定义具有私有输入状态模式的智能体节点函数，该模式与整体图状态模式不同。这使得可以传递仅用于执行该特定智能体所需的信息。