PythonAI算法
基于 LangGraph 构建多代理应用程序详解
本文介绍 LangGraph 框架及其在多代理应用中的应用。通过定义节点（代理）和边（逻辑路由），实现心理学家、社会学家和经济学家三个角色的协作流程。利用状态图和条件边管理循环执行，结合 Tavily 搜索工具增强能力。展示了如何通过 LangGraph 创建动态工作流，支持复杂推理任务。重点讲解了节点初始化、路由器逻辑、状态图编译及运行调试过程，并提供了最佳实践建议，帮助开发者构建高效的多智能体系统。
本文介绍 LangGraph 框架及其在多代理应用中的应用。通过定义节点（代理）和边（逻辑路由），实现心理学家、社会学家和经济学家三个角色的协作流程。利用状态图和条件边管理循环执行，结合 Tavily 搜索工具增强能力。展示了如何通过 LangGraph 创建动态工作流，支持复杂推理任务。重点讲解了节点初始化、路由器逻辑、状态图编译及运行调试过程，并提供了最佳实践建议，帮助开发者构建高效的多智能体系统。
LangChain 推出了 LangGraph，这是一个新的库，基于图形框架和运行时循环的概念，用于管理多代理应用程序。它允许开发者创建具有循环的链，解决了 LangChain 之前缺乏的在运行时进行循环推理的能力。
LangChain 是当前主流的人工智能框架之一。LangGraph 作为其新推出的库，利用图框架和运行时循环的概念，允许创建具有循环的链。该库使用图形这一数学对象来表示构建由大型语言模型（LLM）驱动的应用程序所需的典型组件，其中节点代表代理，边代表连接和关系。本文将介绍如何使用 LangGraph 设计多代理图的示例，并演示其在多代理应用程序中创建动态工作流的潜力。
近年来，LangChain 已成为市场上最受欢迎的人工智能框架之一。这个轻量级的框架帮助开发人员使用向量数据库、内存、提示、工具和代理等所有相关组件，构建由大语言模型（LLM）支持的应用程序。LangChain 的主要特点是能够轻松创建所谓的'链'。链是一系列组件，使用 LLM 处理用户的输入和输出。一条链可以由不同类型的组件组成，例如提示、检索器、处理器和工具。一条链也可以嵌套在另一个链中以创建更复杂的应用程序。
然而，这些传统的链缺乏将循环引入其运行时的能力。这意味着没有现成的框架使 LLM 能够在类似 for 循环的场景中推理出下一个最佳操作。随着多智能体应用程序概念的兴起——展示不同智能体的应用程序，每个智能体都有特定的个性和访问工具——变得越来越现实和主流（参见 AutoGen 等库项目的兴起），LangChain 的开发人员引入了一个新的库，使管理这类代理应用程序变得更加容易。这个新库于 2024 年 1 月推出，名为 LangGraph，它基于图形这一数学对象，作为 LLM 驱动应用程序的代表性框架。
在动手实践之前，首先介绍一下图的数学概念以及为什么它是一个非常适合循环链的框架。
每当需要模拟和分析复杂系统时，使用能够捕获代理之间关系的模型就至关重要。这些关系——以及它们的相对权重和分布——可以用图表的数学表示来建模。
图是一组对象的表示，其中一些对象对通过链接连接。这些链接通常称为边或弧，它们连接的对象称为顶点或节点。图形用于对对象之间的关系进行建模，例如社交网络、道路网络或化学中分子之间的连接。
一般来说，只要一个系统可以用图来建模，就可以说这个系统是一个网络。网络科学是一门学科，其目标是理解其底层结构是图的现象。也就是说，在流行病学领域，网络被用来模拟疾病的传播，其中节点是个体，链接代表个体之间的联系。
在 LangGraph 中，图形框架用于表示构建 LLM 支持的应用程序所需的典型组件：
在这一部分，我们将构建一个涉及 3 个角色的多代理应用程序，以及他们的个性（使用 LLMs 的术语表、系统消息）：
接下来将探究社交媒体与心理健康之间的相关性。为了构建该图，使用了 LangGraph 存储库中的示例代码，并对其进行调整以设计上述三个代理的个性。在本文中，将介绍修改后的脚本。
如上一节中提到的，为了构建图，需要两个主要成分：节点和边。接下来介绍背后的逻辑。
节点可以是具有特定任务或个性的代理，也可以是应用程序可以使用给定工具执行的动作。在本文的示例中，已经设置了 3 个代理——心理学家、社会学家和经济学家——以及一个工具——搜索引擎 Tavily。
为了初始化图表，首先设置模型的基本结构（这里利用 Azure OpenAI GPT-4）和工具，代码如下：
# 导入 langchain.schema 模块中的 HumanMessage 和 AIMessage 类
from langchain.schema import HumanMessage, AIMessage
# 导入 langchain_openai 模块中的 AzureChatOpenAI 类
from langchain_openai import AzureChatOpenAI
# 导入 dotenv 模块，用于加载环境变量
from dotenv import load_dotenv
import os
# 加载环境变量文件，通常是 .env 文件，其中包含了敏感信息如 API 密钥等
load_dotenv()
# 设置环境变量，这些变量通常在 .env 文件中定义
os.environ["AZURE_OPENAI_API_KEY"] = "your_api_key_here"
os.environ["AZURE_OPENAI_ENDPOINT"] = "your_endpoint_here"
os.environ["TAVILY_API_KEY"] = "your_tavily_key_here"
# 创建一个 AzureChatOpenAI 实例，用于与 Azure Chat OpenAI 服务交互
llm = AzureChatOpenAI(
openai_api_version="2023-07-01-preview",
azure_deployment="gpt-4",
)
# 导入 langchain_community.tools.tavily_search 模块中的 TavilySearchResults 类
from langchain_community.tools.tavily_search import TavilySearchResults
# 创建一个 TavilySearchResults 实例，用于执行 Tavily 搜索操作
tavily_tool = TavilySearchResults(max_results=5) # 最多返回 5 个搜索结果
# 创建一个工具列表，将 Tavily 搜索工具添加到列表中
tools = [tavily_tool]
# 创建一个 ToolExecutor 实例，用于执行工具列表中的操作
from langchain.agents import ToolExecutor
tool_executor = ToolExecutor(tools)
现在使用代理的个性来初始化代理（以及将它们分配给每个节点的函数）。注意：create_agent 和 AgentState 需要根据具体环境定义，此处为示意。
import functools
# 定义一个辅助函数，用于为给定的代理创建节点
def agent_node(state, agent, name):
# 调用代理的 invoke 方法，传入状态 state
result = agent.invoke(state)
# 将代理的输出转换为合适的格式，以便追加到全局状态中
if isinstance(result, FunctionMessage): # 如果结果是一个 FunctionMessage 类型
pass # 这里暂时没有操作，可能是后续会有特定的处理
else:
# 如果不是 FunctionMessage 类型，将其转换为 HumanMessage 类型
result = HumanMessage(
**result.dict(exclude={"type", "name"}), # 将 result 对象转换为字典，排除 "type" 和 "name" 键
name=name # 设置消息的发送者名称
)
return {
"messages": [result], # 将转换后的结果作为消息列表的一部分返回
# 因为我们有一个严格的工作流程，我们可以
# 跟踪发送者，这样我们就知道接下来要传递给谁。
"sender": name, # 返回发送者名称
}
# 创建心理学家代理和节点
psychologist_agent = create_agent(
llm,
[tavily_tool],
system_message="一个将神经科学理论无缝融入对话的专家心理学家，解开人类行为和情感的复杂性。"
)
psychologist_node = functools.partial(agent_node, agent=psychologist_agent, name="Psychologist")
# 以类似的方式创建社会学家代理和节点
sociologist_agent = create_agent(
llm,
[tavily_tool],
system_message="一个擅长剖析社会模式，调查集体心理的敏锐社会学家。关注群体效应而非个体效应。"
)
sociologist_node = functools.partial(agent_node, agent=sociologist_agent, name="Sociologist")
# 以类似的方式创建经济学家代理和节点
economist_agent = create_agent(
llm,
[tavily_tool],
system_message="一个务实的经济学家，量化无形资产，将趋势与经济影响精确地联系起来。"
)
economist_node = functools.partial(agent_node, agent=economist_agent, name="Economist")
现在需要一个逻辑来应用于边，以确定循环的执行模式。为此，将创建一个具有以下结构的路由器：
# 定义一个名为 router 的函数，用于决定对话流程的下一步
def router(state):
# 这个函数是流程的路由器
messages = state["messages"] # 从状态 state 中获取消息列表
last_message = messages[-1] # 获取列表中的最后一个消息
# 检查最后一个消息是否包含 "function_call" 关键字
if "function_call" in last_message.additional_kwargs:
# 如果上一个代理正在调用一个工具
return "call_tool" # 返回 "call_tool"，表示下一步是调用工具
# 检查最后一个消息的内容是否包含 "FINAL ANSWER" 文本
if "FINAL ANSWER" in last_message.content:
# 如果任何一个代理决定工作已经完成
return "end" # 返回 "end"，表示结束对话流程
# 如果以上条件都不满足
return "continue" # 返回 "continue"，表示继续对话流程
给定循环中的状态，下一步可能采取三种行动：
现在已经具备了建立图形框架的所有要素。可以用路由器初始化节点（3 个代理 +1 个工具）和边，设置一个循环如下：心理学家→社会学家→经济学家。代码如下：
# 创建一个名为 workflow 的 StateGraph 实例，用于定义工作流程
workflow = StateGraph(AgentState)
# 向工作流程中添加节点
workflow.add_node("Psychologist", psychologist_node) # 添加心理学家节点
workflow.add_node("Sociologist", sociologist_node) # 添加社会学家节点
workflow.add_node("Economist", economist_node) # 添加经济学家节点
workflow.add_node("call_tool", tool_node) # 添加调用工具的节点
# 为心理学家节点添加条件边
# 根据 router 函数的返回值，决定下一步的动作
workflow.add_conditional_edges(
"Psychologist", # 从心理学家节点开始
router, # 使用 router 函数作为条件判断
{
"continue": "Sociologist", # 如果返回 "continue"，则跳转到社会学家节点
"call_tool": "call_tool", # 如果返回 "call_tool"，则跳转到调用工具的节点
"end": END # 如果返回 "end"，则结束流程
}
)
# 为社会学家节点添加条件边
workflow.add_conditional_edges(
"Sociologist", # 从社会学家节点开始
router, # 使用 router 函数作为条件判断
{
"continue": "Economist", # 如果返回 "continue"，则跳转到经济学家节点
"call_tool": "call_tool", # 如果返回 "call_tool"，则跳转到调用工具的节点
"end": END # 如果返回 "end"，则结束流程
}
)
# 为经济学家节点添加条件边
workflow.add_conditional_edges(
"Economist", # 从经济学家节点开始
router, # 使用 router 函数作为条件判断
{
"continue": "Psychologist", # 如果返回 "continue"，则跳转到心理学家节点
"call_tool": ,
: END
}
)
workflow.add_conditional_edges(
,
x: x[],
{
: ,
: ,
:
}
)
workflow.set_entry_point()
graph = workflow.()
接下来初始化图表。将整个对话存储到一个称为输出的列表中，然后打印每个代理的推理，代码如下所示：
# 初始化一个空列表，用于存储输出结果
outputs = []
# 使用 graph.stream 方法处理初始状态，并生成一个流
for s in graph.stream(
{
"messages": [
HumanMessage( # 创建一个 HumanMessage 对象
content="Provide me with meaningful insights about the effects of social media on mental health, covering individual effects, collective effects and macroeconomics effects. Once you provided me with these three perspectives, finish."
)
],
},
# 配置选项，设置在图中执行的最大步数为 150
{"recursion_limit": 150},
):
# 打印当前步骤的状态
print(s)
# 将当前步骤的状态添加到输出结果列表中
outputs.append(s)
# 打印分隔线，用于在输出中区分不同的步骤
print("----")
输出内容会显示每一步的状态流转。也可以仅打印最终答案，如下所示：
print(outputs[-1]["__end__"]["messages"][-1].content)
在实际开发中，使用 LangGraph 构建多代理应用时需注意以下几点：
AgentState 定义清晰，避免状态冲突。状态应包含必要的上下文信息，如消息历史、中间结果等。recursion_limit 防止无限循环。在多代理协作中，若代理间互相依赖过强，可能导致死锁。router 中正确处理工具调用的响应。LangGraph 已经成为开发多代理应用程序的强大工具，其基于图形的方法具有明显的优势，可以创建多个独立代理无缝协作的动态工作流程。这些由 LLM 提供支持的代理可以共享状态或独立工作，并将最终响应传递给彼此。
此外，通过对工具和职责进行分组，LangGraph 还提高了效率。每个代理都专注于特定的任务，从而获得更好的结果。单独的提示允许定制，代理甚至可以由单独的微调语言模型提供支持。
即使 LangGraph 还处于早期开发阶段，它在快速发展的多代理应用程序领域是一个有极大发展前景的框架。通过合理设计节点和边，开发者可以构建出灵活、可扩展的智能体系统，适应各种复杂的业务场景。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online