LangGraph: un marco para la construcción y ejecución de agentes de IA basado en la topología de grafos acíclicos dirigidos

Base de conocimientos de IAPublicado hace 6 meses Círculo de intercambio de inteligencia artificial

2.5K 00

人工智能（AI）是一个迅速发展的领域。语言模型已经发展到能够让AI Agent执行复杂任务和做出复杂决策。然而，随着这些Agent的技能不断提升，支持它们的基础设施难以跟上。

LangGraph，这是一个革命性的库，旨在革新AI Agent的 构建和运行时 Implantación.

LangGraph 提出了一个新的Agent构建和运行时构造范式，它将Agent工作流视为基于循环图拓扑的结构。

LangGraph 是什么

LangChain 的一个重要价值主张是能够轻松创建自定义链。然而，LangChain 缺乏一种简单的方法来在这些链中引入循环。实际上，这些链是有向无环图（DAG）—— 就像大多数数据编排框架一样。

人们构建复杂的LLM应用时常见的一个模式是：在运行时中引入循环。这些循环常常借助LLM来推理循环中的下一步行动。这本质上相当于在for循环中运行LLM。这类系统通常被称为Agent。

最简单的形式是一个有两个步骤的循环：

调用LLM确定采取哪些行动，或给用户什么回应
执行给定的操作，然后返回到第 1 步

这些步骤会重复，直到生成最终响应。这就是驱动 LangChain核心AgentExecutor 的循环，这是一个非常简单的循环,也最关键，是因为通过它将几乎所有的决策和推理能力交给了LLM。

需要注意的是，这个过程通常需要更多的控制。比如强制Agent首先调用特定工具、对工具的调用方式做更多控制、根据Agent所处的状态做不同的提示等

这些受控的流程，LangChain称之为“状态机”

这些状态机具有循环的能力,然而，在该循环的构造方面，需要一定的人工干预。

LangGraph 是通过将这些状态机指定为图的方式来创建它们。

LangGraph 建立在 LangChain 基础上，并与 LangChain 生态系统完全互操作。它主要通过引入一种简单的方法来创建循环图而增加了新价值。这在创建Agent runtimes是非常有用的。

这种方法使Agent能够展现出比其前辈的线性执行模型更多的可变且细微的行为。利用图论，LangGraph为开发复杂的网络化Agent系统提供了新的途径。

LangGraph：基于有向无环图拓扑的AI Agent构建与执行框架图：一个用于生成代码的LangGraph

一个例子：动态问答系统

toma ：用户可能是一步一步的提供给模型信息，而不是一次性发送所有的信息。

prescripción ：使用循环图：

让 LLM 生成初步回答。
如果答案不完整或需要更多信息，生成新问题并继续询问用户。
直到用户确认问题已解决。

循环图的结构

节点（Nodes）：每个节点代表一个具体的任务或操作，比如文本生成、问题回答、数据处理等。
边（Edges）：节点之间的连接表示任务执行的先后顺序，可能带有条件判断。
循环路径（Looping Path）：当某个节点的输出需要反馈回之前的节点，或需要反复执行时，就会形成循环。
终止条件（Stopping Condition）：循环图会设定某种终止条件，例如结果的质量满足要求、达到设定的次数限制，或者外部触发的停止信号。

为什么使用 LangGraph ？

destreza ：随着AI Agent的发展，开发者需要对Agent运行时进行更多控制，以实现个性化的行动计划和决策程序。
AI推理的循环性质 ：许多复杂的LLM应用在使用链式推理等策略时依赖于循环执行。LangGraph 提供了一个自然的框架，用于建模这些循环过程。
sistema de inteligencia múltiple ：随着多智能体工作流变得越来越普遍，对能够高效管理和协调多个自主智能体的系统的需求也在增加。

使用代码来看 LangGraph 如何工作

LangGraph 的功能基于几个基本元素：

Nodes: 这些是函数或Agent的工具。
Edges: 定义Agent系统中执行和数据流向的路径，连接节点。
StateGraph : LangGraph 通过管理和更新状态对象来允许在执行周期之间保持持久数据，数据在节点之间流动。

图：LangGraph的基本工作流程

StateGraph 状态图

StateGraph 是一个表示图的类。您通过传入 state 定义来初始化此类。该状态定义表示一个中心状态对象，随着时间的推移不断更新。此状态由图中的节点更新，节点以键值存储的形式返回对该状态属性的操作。

import os  # 导入操作系统模块
os.environ['OpenAI_API_KEY'] = 'hk-iwtbi1e427'
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
class State(TypedDict):
    # messages 的类型为“list”。Annotated中的“add_messages” 定义了如何更新messages
    # 在本例中，messages是追加到list中的，而不是覆盖
    messages: Annotated[list, add_messages]
# 创建StateGraph的实例，传入State类
graph_builder = StateGraph(State)

Nodes 节点

Crear un StateGraph 后，可以使用 graph.add_node(name, value) 语法添加节点。 name 参数是一个字符串，是节点name。 value 参数是一个函数或可运行的 LCEL，将被调用。

这个函数/LCEL 接受一个与 State 对象具有相同形式的字典作为输入，并输出一个使用 State 对象的键来更新的字典。

from langchain_openai import ChatOpenAI  # 从langchain_openai模块导入ChatOpenAI
llm = ChatOpenAI(  # 创建ChatOpenAI的实例
    model="gpt-4o-mini",  # 指定模型
    temperature=0,  # 设置温度
    base_url="https://api.openai-hk.com/v1"# 设置基础URL
)
def chatbot(state: State):# 定义chatbot函数，接受State类型的参数
    """
    处理聊天机器人请求并返回消息。
    参数:
    state (State): 包含消息的状态。
    返回:
    dict: 机器人生成的消息的字典。
    """
    return {"messages": [llm.invoke(state["messages"])]}  # 返回包含消息的字典
# 第一个参数是节点name
# 第二个参数是被调用的函数或对象
graph_builder.add_node("chatbot", chatbot)  # 向图中添加节点，节点名称为"chatbot"

Edges 边

表示其中一个节点始终在另一个节点之后被调用

# 添加从START到"chatbot"的边
graph_builder.add_edge(START, "chatbot")  
# 添加从"chatbot"到END的边
graph_builder.add_edge("chatbot", END)

Compile 编译

定义我们的图形后，我们可以将其编译为可运行的！暴露与 LangChain 相同的方法（ .invoke (matemáticas) género .stream (matemáticas) género .astream_log 等），允许以与链相同的方式调用它。

graph = graph_builder.compile()  # 编译图

estar en movimiento

def stream_graph_updates(user_input: str):  # 定义stream_graph_updates函数，接受用户输入
    """
    使用user_input流式更新图的消息。
    参数:
    user_input (str): 用户输入的消息。
    返回:
    None
    """
    # 遍历图的流事件，传入用户消息以获取模型的响应
    for event in graph.stream({"messages": [("user", user_input)]}):  # 遍历图的流事件
        for value in event.values():  # 遍历事件的值
            print("Assistant:", value["messages"][-1].content)  # 打印助手的最后一条消息
whileTrue:
    try:
        user_input = input("User: ")  # 获取用户输入
        if user_input.lower() in ["quit", "exit", "q"]:  # 检查用户是否想退出
            print("Goodbye!")
            break# 退出循环
        # 调用stream_graph_updates函数处理用户输入
        stream_graph_updates(user_input)
    except:  # 捕获异常
        # fallback if input() is not available
        user_input = "What do you know about LangGraph?"# 设置默认用户输入
        print("User: " + user_input)  # 打印用户输入
        stream_graph_updates(user_input)  # 调用stream_graph_updates函数处理默认输入
        break# 退出循环