多智能体系统

一个代理是“使用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)

交接作为工具

最常见的代理类型之一是ReAct风格的工具调用代理。对于这种类型的代理，常见的模式是将交接封装在一个工具调用中，例如：

def transfer_to_bob(state):
    """转移至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: 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模式。

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_ Yöntem 1)
builder.add_node(agent_2)

builder.add_edge(START, "supervisor")

supervisor = builder.compile()

查看此tutorial以了解监督者多代理架构示例。

监督者（工具调用）

在此版本的监督者架构中，我们将单个代理定义为**工具**，并在监督者节点中使用工具调用的LLM。这可以通过一个ReAct-风格的代理实现，该代理由两个节点组成——一个LLM节点（监督者）和一个执行工具的节点（在这种情况下为代理）。

API Reference: 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: 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: 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")

代理之间的通信

构建多代理系统时最重要的事情之一是确定代理如何进行通信。这有几个不同的考虑因素：

• 代理是通过图状态还是工具调用进行通信？
• 如果两个代理具有不同的状态模式，怎么办？
• 如何在共享消息列表上进行通信？

图状态与工具调用

代理之间传递的“负载”是什么？在上述大多数架构中，代理通过图状态进行通信。而在带有工具调用的监督者的情况下，负载则是工具调用参数。

图状态

要通过图状态进行通信，必须将各个代理定义为图节点。这些可以作为函数添加，也可以作为整个子图添加。在图执行的每一步中，代理节点都会接收当前的图状态，执行代理代码，然后将更新后的状态传递给下一个节点。

通常情况下，代理节点共享一个单一的状态模式。然而，你可能希望设计具有不同状态模式的代理节点。

不同状态模式

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

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

共享消息列表

代理之间最常用的通信方式是通过共享状态通道，通常是消息列表。这假设至少有一个共享的通道（键）存在于状态中。当通过共享消息列表进行通信时，还需要考虑以下问题：代理是否应该分享完整的思考过程历史还是只分享最终结果？

分享完整的历史

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

分享最终结果

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

对于被调用为工具的代理，监督者根据工具模式确定输入。此外，LangGraph允许传递状态以在运行时传递给单个工具，因此下属代理可以在需要时访问父图状态。

Comments