使用LangGraph构建多Agent系统架构！|agent|spiderlinebreakbuilder|大模型|工作流|应用程序|系统架构|调用

点击下方“JavaEdge”，选择“设为星标”

第一时间关注技术干货！

免责声明~ 任何文章不要过度深思！万事万物都经不起审视，因为世上没有同样的成长环境，也没有同样的认知水平，更「没有适用于所有人的解决方案」；别急着评判文章列出的观点，只需代入其中，适度审视一番自己即可，能「跳脱出来从外人的角度看看现在的自己处在什么样的阶段」才不为俗人。怎么想、怎么做，全在乎自己「不断实践中寻找适合自己的大道」

0 前言

Agent是一个使用大语言模型决定应用程序控制流的系统。随着这些系统的开发，它们随时间推移变得复杂，使管理和扩展更困难。如你可能会遇到：

Agent拥有太多的工具可供使用，对接下来应该调用哪个工具做出糟糕决策
上下文过于复杂，以至于单个Agent无法跟踪
系统中需要多个专业领域（例如规划者、研究员、数学专家等）。

为解决这些问题，你可能考虑将应用程序拆分成多个更小、独立的代理，并将它们组合成一个多Agent系统。这些独立的Agent可以简单到一个提示和一个LLM调用，或者复杂到像一个ReActAgent（甚至更多！）。

1 多Agent系统的好处

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

2 多Agent架构

多Agent系统中有几种方式连接Agent：

网络：每个Agent都可与其他Agent通信。任何Agent都可以决定接下来调用哪个其他Agent
监督者：每个Agent与一个监督者Agent通信。监督者Agent决定接下来应该调用哪个Agent。
监督者（工具调用）：这是监督者架构的一个特殊情况。个别Agent可以被表示为工具。在这种情况下，监督者Agent使用一个工具调用LLM来决定调用哪个Agent工具，以及传递哪些参数给这些Agent。
层次结构：你可以定义一个有监督者的多Agent系统。这是监督者架构的概括，并允许更复杂的控制流。
自定义多Agent工作流：每个Agent只与Agent子集中的其他Agent通信。流程的部分是确定性的，只有一些Agent可以决定接下来调用哪个其他Agent。

网络

这种架构中，Agent被定义为图节点。每个Agent都可以与每个其他Agent通信（多对多连接），并且可以决定接下来调用哪个Agent。虽然非常灵活，但随着Agent数量的增加，这种架构扩展性并不好：

很难强制执行接下来应该调用哪个Agent
很难确定应该在Agent之间传递多少信息

建议生产避免使用这架构，而是使用以下架构之一。

监督者

这种架构中，定义Agent为节点，并添加一个监督者节点（LLM），它决定接下来应该调用哪个Agent节点。使用条件边根据监督者的决策将执行路由到适当的Agent节点。这种架构也适用于并行运行多个Agent或使用map-reduce模式。

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

model = ChatOpenAI()

class AgentState(MessagesState):
next: Literal["agent_1", "agent_2"]

def supervisor(state: AgentState):
response = model.invoke(...)
return {"next": response["next_agent"]}

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

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

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

builder.add_edge(START, "supervisor")
# 根据监督者的决策路由到Agent之一或退出
builder.add_conditional_edges("supervisor", lambda state: state["next"])
builder.add_edge("agent_1", "supervisor")
builder.add_edge("agent_2", "supervisor")

supervisor = builder.compile()

教程以获取有关监督者多Agent架构的示例。

监督者（工具调用）

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

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

model = ChatOpenAI()

def agent_1(state: Annotated[dict, InjectedState]):
tool_message = ...
return {"messages": [tool_message]}

def agent_2(state: Annotated[dict, InjectedState]):
tool_message = ...
return {"messages": [tool_message]}

tools = [agent_1, agent_2]
supervisor = create_react_agent(model, tools)

自定义多Agent工作流

在这种架构中，我们添加个别Agent作为图节点，并提前定义Agent被调用的顺序，以自定义工作流。在LangGraph中，工作流可以以两种方式定义：

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

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

3 Agent之间通信

构建多Agent系统时最重要的事情是弄清楚Agent如何通信。有几个不同的考虑因素：

Agent是通过图状态或工具调用进行通信的吗？
如果两个Agent有不同的状态模式怎么办？
如何通过共享消息列表进行通信？

3.1 图状态与工具调用

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

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

通常，Agent节点共享一个单一的状态模式。然而，你可能想要设计具有不同状态模式的Agent节点。

3.2 不同的状态模式

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

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

3.3 共享消息列表

Agent之间通信的最常见方式是通过共享状态通道，通常是消息列表。这假设状态中至少有一个通道（键）由Agent共享。当通过共享消息列表通信时，还有一个额外的考虑因素：Agent是共享完整的历史记录还是仅共享最终结果？

Agent可以共享他们的思维过程的完整历史记录（即“草稿垫”）与其他所有Agent。这种“草稿垫”通常看起来像一个消息列表。共享完整思维过程的好处是，它可能有助于其他Agent做出更好的决策，提高整个系统的整体推理能力。缺点是，随着Agent数量和复杂性的增长，“草稿垫”将迅速增长，可能需要额外的策略进行内存管理。

共享最终结果

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

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

关注我，紧跟本系列专栏文章，咱们下篇再续！

★ 作者简介：魔都架构师，多家大厂后端一线研发经验，在分布式系统设计、数据平台架构和AI应用开发等领域都有丰富实践经验。各大技术社区头部专家博主。具有丰富的引领团队经验，深厚业务架构和解决方案的积累。负责：中央/分销预订系统性能优化活动&券等营销中台建设交易平台及数据中台等架构和开发设计车联网核心平台-物联网连接平台、大数据平台架构设计及优化 LLM Agent应用开发区块链应用开发大数据开发挖掘经验推荐系统项目目前主攻市级软件项目设计、构建服务全社会的应用系统。 ”

参考：