一文读懂如何使用图框架构建多代理应用程序

点击下方卡片，关注“AI前沿速递”公众号

各种重磅干货，第一时间送达

简述

LangChain 推出了 LangGraph，这是一个新的库，基于图形框架和运行时循环的概念，用于管理多代理应用程序。

摘要

LangChain 是当前主流的一个人工智能框架，它引入了一个名为 LangGraph 的新库，用于管理多代理应用程序。该库基于图框架和运行时循环的概念，允许创建具有循环的链，这是LangChain之前所缺乏的。使用图形这一数学对象来表示构建由 LLMs（大型语言模型）驱动的应用程序所需的典型组件，其中节点代表代理，边代表连接和关系。本文介绍了如何使用 LangGraph 设计多代理图的示例，并演示了其在多代理应用程序中创建动态工作流的潜力。

LangGraph 入门

去年，LangChain 已成为市场上最受欢迎的人工智能框架之一。这个轻量级的框架可帮助开发人员使用所有相关组件（向量数据库、内存、提示、工具和代理）构建由大语言模型（Large Language Model，LLM） 支持的应用程序。LangChain 的主要特点是能够轻松创建所谓的链。链是一系列组件，使用 LLM 处理用户的输入和输出。一条链可以由不同类型的组件组成，例如提示、检索器、处理器和工具。一条链也可以嵌套在另一个链中以创建更复杂的应用程序。

然而，这些链缺乏将循环引入其运行时的能力，这意味着没有现成的框架使 LLM 能够在类似 for 循环的场景中推理出下一个最佳操作。由于多智能体应用程序的概念——展示不同智能体的应用程序，每个智能体都有特定的个性和访问工具——变得越来越现实和主流（参见 AutoGen 等库项目的兴起），LangChain 的开发人员引入了一个新的库，使管理这类代理应用程序变得更加容易。这个新库于 2024 年 1 月推出，名为 LangGraph，它基于图形这一数学对象，作为 LLM 驱动应用程序的代表性框架。

在动手实践之前，首先介绍一下图的数学概念以及为什么它是一个非常适合循环链的框架。

图表简介

每当需要模拟和分析复杂系统时，使用能够捕获代理之间关系的模型就至关重要。这些关系——以及它们的相对权重和分布——可以用图表的数学表示来建模。

图是一组对象的表示，其中一些对象对通过链接连接。这些链接通常称为边或弧，它们连接的对象称为顶点或节点。图形用于对对象之间的关系进行建模，例如社交网络、道路网络或化学中分子之间的连接。

一般来说，只要一个系统可以用图来建模，就可以说这个系统是一个网络。网络科学是一门学科，其目标是理解其底层结构是图的现象。也就是说，在流行病学领域，网络被用来模拟疾病的传播，其中节点是个体，链接代表个体之间的联系。

在 LangGraph 中，图形框架用于表示构建 LLM 支持的应用程序所需的典型组件。进一步来说：

1. 节点→类似于游戏中的“演员”或代理。每个人都可以从事不同的活动。也就是说，可以有一个包含 3 个节点的图表，代表心理学家、社会学家和经济学家正在进行一项关于社交媒体对心理健康影响的大规模调查（还可以添加第四个节点，代表这三个专家的监督者）。
2. 边→代表节点之间的连接和关系。它们可以是在节点之间承载信息的连接，也可以是确定激活节点的条件及其执行顺序的条件边。

构建多智能体图

在这一部分，我们将构建一个涉及 3 个角色的多代理应用程序，以及他们的个性（使用 LLMs 的术语表、系统消息）：

1. 心理学家：“一位心理学家专家，将神经科学理论无缝地融入对话中，揭示了人类行为和情感的复杂性。”
2. 社会学家：“一位目光敏锐的社会学家，擅长剖析社会模式，研究社交媒体与集体心理之间的相互作用。”
3. 经济学家：“一位务实的经济学家，量化无形资产，精确地将社交媒体趋势与经济影响联系起来。”

接下来将探究社交媒体与心理健康之间的相关性。为了构建该图，使用了 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

#加载环境变量文件，通常是 .env 文件，其中包含了敏感信息如API密钥等
load_dotenv()

#设置环境变量，这些变量通常在 .env 文件中定义
os.environ["AZURE_OPENAI_API_KEY"]  # Azure OpenAI API 的密钥
os.environ["AZURE_OPENAI_ENDPOINT"]  # Azure OpenAI 的服务端点
os.environ["TAVILY_API_KEY"]  # Tavily API 的密钥

#创建一个 AzureChatOpenAI 实例，用于与 Azure Chat OpenAI 服务交互
llm = AzureChatOpenAI(
    openai_api_version="2023-07-01-preview",  # 使用的 OpenAI API 版本
    azure_deployment="gpt-4",  # 指定使用的 Azure 部署，这里假设是 "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 实例，用于执行工具列表中的操作
tool_executor = ToolExecutor(tools)

现在使用代理的个性来初始化代理（以及将它们分配给每个节点的函数），代码如下：
import functools  # 导入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,  # 返回发送者名称
    }

#创建心理学家代理和节点
#使用create_agent函数创建一个代理，传入llm和工具列表[tavily_tool]
#system_message参数定义了心理学家的角色描述
psychologist_agent = create_agent(
    llm,
    [tavily_tool],
    system_message="一个将神经科学理论无缝融入对话的专家心理学家，解开人类行为和情感的复杂性。"
)
#使用functools.partial创建一个偏函数，将心理学家代理和名称"Psychologist"固定下来
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"，表示继续对话流程

给定循环中的状态，下一步可能采取三种行动：

1. 如果前一个操作 (-1) 需要调用工具（给出“函数调用”输出），则会调用其中一个可用的工具；
2. 如果前一个动作产生最终输出（“FINAN ANSWER”），则循环结束；
3. 否则，循环将按照在初始化图表时定义的顺序继续。

现在已经具备了建立图形框架的所有要素。可以用路由器初始化节点（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": "call_tool",  # 如果返回"call_tool"，则跳转到调用工具的节点
        "end": END  # 如果返回"end"，则结束流程
    }
)

#为调用工具的节点添加条件边
#根据发送者名称决定下一步的动作
workflow.add_conditional_edges(
    "call_tool",  # 从调用工具的节点开始
    lambda x: x["sender"],  # 使用lambda函数作为条件判断，获取发送者名称
    {
        "Psychologist": "Psychologist",  # 如果发送者是心理学家，则跳转到心理学家节点
        "Sociologist": "Sociologist",  # 如果发送者是社会学家，则跳转到社会学家节点
        "Economist": "Economist"  # 如果发送者是经济学家，则跳转到经济学家节点
    }
)

#设置工作流程的入口点为心理学家节点
workflow.set_entry_point("Psychologist")

#编译工作流程图
graph = workflow.compile()

接下来初始化图表。将整个对话存储到一个称为输出的列表中，然后打印每个代理的推理，代码如下所示：
#初始化一个空列表，用于存储输出结果
outputs = []

#使用graph.stream方法处理初始状态，并生成一个流
#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 已经成为开发多代理应用程序的强大工具，其基于图形的方法具有明显的优势，可以创建多个独立代理无缝协作的动态工作流程。这些由 LLM 提供支持的代理可以共享状态或独立工作，并将最终响应传递给彼此。

此外，通过对工具和职责进行分组，LangGraph 还提高了效率。每个代理都专注于特定的任务，从而获得更好的结果。单独的提示允许定制，代理甚至可以由单独的微调语言模型提供支持。

即使 LangGraph 还处于早期开发阶段，它在快速发展的多代理应用程序领域是一个有极大发展前景的框架。