跳到内容

多代理系统

代理是一个使用 LLM 来决定应用程序控制流的系统。随着你开发这些系统,它们可能会随着时间的推移变得更加复杂,从而更难管理和扩展。例如,你可能会遇到以下问题

  • 代理拥有太多工具而难以决定下一步调用哪个工具
  • 上下文对于单个代理来说变得过于复杂,难以跟踪
  • 系统中需要多个专业领域(例如,规划师、研究员、数学专家等)

为了解决这些问题,你可以考虑将应用程序分解为多个更小、独立的代理,并将它们组合成一个多代理系统。这些独立的代理可以像提示和 LLM 调用一样简单,也可以像 ReAct 代理一样复杂(以及更多!)。

使用多代理系统的主要优点是

  • 模块化:独立的代理使得开发、测试和维护代理系统更加容易。
  • 专业化:你可以创建专注于特定领域的专家代理,这有助于提高整个系统的性能。
  • 控制:你可以明确控制代理如何通信(而不是依赖于函数调用)。

多代理架构

在多代理系统中连接代理有几种方式

  • 网络:每个代理都可以与所有其他代理通信(多对多连接)。任何代理都可以决定下一步调用哪个其他代理。
  • 监督者:每个代理都与单个监督者代理通信。监督者代理决定下一步应该调用哪个代理。
  • 监督者(工具调用):这是监督者架构的一个特例。单个代理可以表示为工具。在这种情况下,监督者代理使用一个工具调用 LLM 来决定调用哪个代理工具,以及传递给这些代理的参数。
  • 分层:你可以定义一个具有监督者之上的监督者的多代理系统。这是监督者架构的推广,允许更复杂的控制流程。
  • 自定义多代理工作流程:每个代理只与代理的一个子集通信。流程的一部分是确定性的,只有一些代理可以决定下一步调用哪个其他代理。

交接

在多代理架构中,代理可以表示为图节点。每个代理节点执行其步骤,并决定是结束执行还是路由到另一个代理,包括可能路由到自身(例如,在循环中运行)。多代理交互中的常见模式是交接,其中一个代理将控制权移交给另一个代理。交接允许你指定

为了在 LangGraph 中实现交接,代理节点可以返回Command 对象,该对象允许你结合控制流和状态更新

def agent(state) -> Command[Literal["agent", "another_agent"]]:
    # the condition for routing/halting can be anything, e.g. LLM tool call / structured output, etc.
    goto = get_next_agent(...)  # 'agent' / 'another_agent'
    return Command(
        # Specify which agent to call next
        goto=goto,
        # Update the graph state
        update={"my_state_key": "my_state_value"}
    )

在更复杂的场景中,如果每个代理节点本身就是一个图(即子图),则其中一个代理子图中的节点可能希望导航到不同的代理。例如,如果你有两个代理 alicebob(父图中的子图节点),并且 alice 需要导航到 bob,你可以在 Command 对象中设置 graph=Command.PARENT

def some_node_inside_alice(state):
    return Command(
        goto="bob",
        update={"my_state_key": "my_state_value"},
        # specify which graph to navigate to (defaults to the current graph)
        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):
    """Transfer to 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[tool_call["name"]].invoke(tool_call) for tool_call in tool_calls]
    return commands

现在让我们更仔细地看看不同的多代理架构。

网络

在这种架构中,代理被定义为图节点。每个代理可以与所有其他代理通信(多对多连接),并且可以决定下一步调用哪个代理。这种架构适用于没有明确的代理层级或特定调用顺序的问题。

API 参考: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]]:
    # you can pass relevant parts of the state to the LLM (e.g., state["messages"])
    # to determine which agent to call next. a common pattern is to call the model
    # with a structured output (e.g. force it to return an output with a "next_agent" field)
    response = model.invoke(...)
    # route to one of the agents or exit based on the LLM's decision
    # if the LLM returns "__end__", the graph will finish execution
    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 参考: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]]:
    # you can pass relevant parts of the state to the LLM (e.g., state["messages"])
    # to determine which agent to call next. a common pattern is to call the model
    # with a structured output (e.g. force it to return an output with a "next_agent" field)
    response = model.invoke(...)
    # route to one of the agents or exit based on the supervisor's decision
    # if the supervisor returns "__end__", the graph will finish execution
    return Command(goto=response["next_agent"])

def agent_1(state: MessagesState) -> Command[Literal["supervisor"]]:
    # you can pass relevant parts of the state to the LLM (e.g., state["messages"])
    # and add any additional logic (different models, custom prompts, structured output, etc.)
    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 节点(监督者)和一个执行工具(在此例中为代理)的工具调用节点。

API 参考:ChatOpenAI | InjectedState | create_react_agent

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

model = ChatOpenAI()

# this is the agent function that will be called as tool
# notice that you can pass the state to the tool via InjectedState annotation
def agent_1(state: Annotated[dict, InjectedState]):
    # you can pass relevant parts of the state to the LLM (e.g., state["messages"])
    # and add any additional logic (different models, custom prompts, structured output, etc.)
    response = model.invoke(...)
    # return the LLM response as a string (expected tool response format)
    # this will be automatically turned to ToolMessage
    # by the prebuilt create_react_agent (supervisor)
    return response.content

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

tools = [agent_1, agent_2]
# the simplest way to build a supervisor w/ tool-calling is to use prebuilt ReAct agent graph
# that consists of a tool-calling LLM node (i.e. supervisor) and a tool-executing node
supervisor = create_react_agent(model, tools)

分层

随着你向系统中添加更多代理,监督者可能很难管理所有代理。监督者可能会开始对下一步调用哪个代理做出糟糕的决策,或者上下文对于单个监督者来说变得过于复杂,难以跟踪。换句话说,你最终会遇到最初促使采用多代理架构的相同问题。

为了解决这个问题,你可以分层设计你的系统。例如,你可以创建由单个监督者管理的独立的专业代理团队,以及一个顶层监督者来管理这些团队。

API 参考: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()

# define team 1 (same as the single supervisor example above)

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

# define team 2 (same as the single supervisor example above)
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()


# define top-level supervisor

builder = StateGraph(MessagesState)
def top_level_supervisor(state: MessagesState) -> Command[Literal["team_1_graph", "team_2_graph", END]]:
    # you can pass relevant parts of the state to the LLM (e.g., state["messages"])
    # to determine which team to call next. a common pattern is to call the model
    # with a structured output (e.g. force it to return an output with a "next_team" field)
    response = model.invoke(...)
    # route to one of the teams or exit based on the supervisor's decision
    # if the supervisor returns "__end__", the graph will finish execution
    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 参考: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)
# define the flow explicitly
builder.add_edge(START, "agent_1")
builder.add_edge("agent_1", "agent_2")

代理之间的通信

构建多代理系统时最重要的事情是弄清楚代理如何通信。有一些不同的考虑因素

图状态 vs 工具调用

在代理之间传递的“负载”是什么?在上面讨论的大多数架构中,代理通过图状态进行通信。在使用工具调用的监督者的情况下,负载是工具调用参数。

图状态

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

通常代理节点共享一个单一的状态模式。但是,你可能希望设计具有不同状态模式的代理节点。

不同的状态模式

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

  • 定义具有单独状态模式的子图代理。如果子图和父图之间没有共享状态键(通道),则添加输入/输出转换非常重要,以便父图知道如何与子图通信。
  • 定义代理节点函数,其私有输入状态模式与整个图状态模式不同。这允许传递执行该特定代理所需的信息。

共享消息列表

代理之间最常见的通信方式是通过共享状态通道,通常是消息列表。这假设状态中总是至少有一个通道(键)由代理共享。通过共享消息列表进行通信时,还有一个额外的考虑因素:代理应该共享其思维过程的完整历史记录还是只共享最终结果

共享完整历史记录

代理可以与其他所有代理共享其思维过程的完整历史记录(即“草稿板”)。这个“草稿板”通常看起来像一个消息列表。共享完整思维过程的好处是它可以帮助其他代理做出更好的决策,并提高整个系统的推理能力。缺点是随着代理数量及其复杂性的增加,“草稿板”会迅速增长,可能需要额外的内存管理策略。

共享最终结果

代理可以拥有自己的私有“草稿板”,并只与其余代理共享最终结果。这种方法可能更适用于代理数量众多或代理更复杂的系统。在这种情况下,你需要定义具有不同状态模式的代理

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

评论