Skip to content

多智能体系统

一个智能体是_一个使用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,
    )

注意

如果你需要支持使用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,
    )

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

重要

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

def call_tools(state):
    ...
    commands = [tools_by_name[call["name"].invoke(call, config={"coerce_tool_content": False}) for tool_call in tool_calls]
    return commands

现在，让我们更仔细地看看不同的多智能体架构。

网络

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

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

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.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()

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

监督者（工具调用）

在这种监督者架构的变体中，我们将个别代理定义为**工具**，并在监督节点中使用调用工具的LLM。这可以实现为一种ReAct风格的代理，包含两个节点——一个LLM节点（监督者）和一个执行工具的节点（在此情况下为代理）。

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)

层次化

随着你向系统中添加更多代理，监督者可能会难以管理所有代理。监督者可能会开始做出错误的决策，以至于不知应该调用哪个代理，且上下文可能变得过于复杂，以至于单个监督者无法跟踪。换句话说，你会遇到最初促使多代理架构出现的相同问题。

为了解决这个问题，你可以将系统设计为_层次化_。例如，你可以创建由个别监督者管理的独立、专门的代理团队，并有一个顶级监督者来管理这些团队。

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

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):
    # 你可以将状态的相关部分传递给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)
builder.add_node(team_2_graph)

builder.add_edge(START, "top_level_supervisor")
graph = builder.compile()

自定义多代理工作流

在这种架构中，我们将个别代理添加为图节点，并预先定义代理调用的顺序，形成自定义工作流。在LangGraph中，工作流可以通过两种方式定义：

  • 显式控制流（普通边缘）：LangGraph允许你显式定义应用程序的控制流（即代理之间通信的顺序）通过普通图边缘。这是上述架构中最确定性的变体——我们总是提前知道下一个将被调用的代理。

  • 动态控制流（命令）：在LangGraph中，你可以允许LLM决定应用程序的一部分控制流。这可以通过使用Command实现。其特殊情况是监督工具调用架构。在该情况下，为监督代理提供动力的调用工具的LLM将决定工具（代理）被调用的顺序。

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 允许在运行时传递状态给每个工具，因此下级代理可以在需要时访问父状态。

优云智算