LangGraph 教程：初学者综合指南

使用大型语言模型 (LLM) 构建应用程序，为创建复杂的交互式系统提供了激动人心的机会。但随着这些应用程序的复杂性不断增长，尤其是涉及多个 LLM 协同工作的应用程序，新的挑战也随之而来。我们如何管理这些代理之间的信息流？如何确保它们无缝交互并保持一致地理解当前任务？这正是LangGraph 的用武之地。

理解 LangGraph

LangGraph是 LangChain 生态系统中一个强大的库，它为构建和管理多代理 LLM 应用程序提供了优雅的解决方案。通过将工作流表示为循环图，LangGraph 允许开发人员协调多个 LLM 代理之间的交互，确保顺畅的通信和复杂任务的高效执行。

LangChain 擅长创建线性计算任务链（即有向无环图，简称 DAG），而 LangGraph 则引入了将循环纳入这些工作流的能力。这项创新使开发者能够构建更复杂、适应性更强的系统，并模拟智能代理的动态特性，使其能够重新审视任务并根据不断变化的信息做出决策。

关键概念

图结构

LangGraph 设计的核心是基于图形的方式呈现应用程序的工作流程。该图包含两个主要元素：

节点——工作的基石：LangGraph 中的每个节点代表应用程序中一个特定的工作单元或操作。这些节点本质上是封装特定任务的 Python 函数。该任务可能涉及多种操作，例如：

与 LLM 直接沟通以进行文本生成、摘要或其他基于语言的任务。
与外部工具和 API 交互以获取数据或在现实世界中执行操作。
通过格式化、过滤或转换等过程来处理数据。
与用户互动以收集输入或显示信息。
边缘——引导信息和控制的流动：边是 LangGraph 中的连接组织，用于建立信息流的路径并指示操作顺序。LangGraph 支持多种边类型：

简单边：表示从一个节点到另一个节点的直接且无条件的流动。第一个节点的输出作为后续节点的输入，从而形成线性级数。
条件边：条件边引入了一层动态性，使工作流能够根据特定节点操作的结果进行分支。例如，根据用户的响应，图可能会决定终止交互或继续调用工具。这种决策能力对于创建能够适应不同情况的应用程序至关重要。我们将在本文的后面部分看到一个这样的示例。

状态管理

管理多智能体系统的一个关键方面是确保所有智能体在对任务当前状态达成共识的情况下运行。LangGraph 通过自动状态管理解决了这个问题。这意味着该库会在智能体执行任务时自动处理中央状态对象的跟踪和更新。

此状态对象充当关键信息的存储库，这些信息需要在工作流的不同点之间访问。这些信息可能包括：

对话历史：在聊天机器人应用程序中，状态可以存储用户和机器人之间正在进行的对话，从而实现上下文感知的响应。
上下文数据：与当前任务相关的信息，如用户偏好、过去的行为或相关的外部数据，可以存储在状态中，以供代理用于决策。
内部变量：代理可能使用状态来跟踪指导其行为和决策的内部标志、计数器或其他变量。

LangGraph 入门

安装

要开始使用 LangGraph，您需要安装它。

要安装 LangGraph，请打开终端或命令提示符并运行以下命令：

pip install -U langgraph

此命令将下载并安装最新版本的 LangGraph。此-U标志可确保您获取最新版本。

在 LangGraph 中创建基本聊天机器人

这个例子是理解 LangGraph 基本概念的一个很好的起点。

导入必要的库：首先从 LangGraph 和其他相关库导入所需的类和模块。

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' will store the chatbot conversation history.

     # The 'add_messages' function ensures new messages are appended to the list.

     messages: Annotated[list, add_messages]



 # Create an instance of the StateGraph, passing in the State class

 graph_builder = StateGraph(State)

初始化 LLM：实例化您选择的 LLM 模型，并提供所有必要的 API 密钥或配置参数。此 LLM 将用于支持聊天机器人的响应。

#pip install -U langchain_anthropic

 from langchain_anthropic import ChatAnthropic



 llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")

创建聊天机器人节点：定义一个 Python 函数，封装聊天机器人节点的逻辑。该函数将当前状态作为输入，并根据 LLM 的输出生成响应。

def chatbot(state: State):

     # Use the LLM to generate a response based on the current conversation history.

     response = llm.invoke(state["messages"])



     # Return the updated state with the new message appended

     return {"messages": [response]}



 # Add the 'chatbot' node to the graph,

 graph_builder.add_node("chatbot", chatbot)

定义入口点和终点：指定图表内工作流的起点和终点。

# For this basic chatbot, the 'chatbot' node is both the entry and finish point

 graph_builder.add_edge(START, "chatbot")

 graph_builder.add_edge("chatbot", END)

编译图表：通过编译图表创建可运行的实例。

graph = graph_builder.compile()

可视化图形：通过运行简单的 Python 代码，您可以可视化带有节点和边的图形。

from IPython.display import Image, display



 try:

     display(Image(graph.get_graph().draw_mermaid_png()))

 except Exception:

     # This requires some extra dependencies and is optional

     pass

运行聊天机器人：实现一个循环来与用户交互，将他们的输入提供给图表并显示聊天机器人的响应。

while True:

     user_input = input("User: ")

     if user_input.lower() in ["quit", "exit", "q"]:

         print("Goodbye!")

         break



     # Process user input through the LangGraph

     for event in graph.stream({"messages": [("user", user_input)]}):

         for value in event.values():

             print("Assistant:", value["messages"][-1].content)

此代码片段提供了 LangGraph 聊天机器人的基本结构。您可以通过添加更复杂的状态管理和不同的 LLM 模型，或连接到外部工具和 API 来扩展此结构。关键在于为不同的任务定义清晰的节点，并使用边来建立聊天机器人内部所需的信息流和控制。

高级 LangGraph 技术

工具集成

将工具集成到您的 LangGraph 聊天机器人中可以显著增强其功能，使其能够按照您喜欢的方式访问和处理信息。

通过工具集成增强我们的基本聊天机器人

让我们修改上一节中创建的基本聊天机器人，使其包含一个可以在网络上搜索信息的工具。我们将使用TavilySearchResults中的工具。本示例需要Tavily API 密钥。langchain_community.tools.tavily_search

#pip install -U tavily-python langchain_community

from typing import Annotated

from langchain_anthropic import ChatAnthropic

from langchain_community.tools.tavily_search import TavilySearchResults

from langchain_core.messages import BaseMessage

from typing_extensions import TypedDict



from langgraph.graph import StateGraph

from langgraph.graph.message import add_messages

from langgraph.prebuilt import ToolNode, tools_condition



class State(TypedDict):

    messages: Annotated[list, add_messages]



graph_builder = StateGraph(State)



tool = TavilySearchResults(max_results=2)

tools = [tool]

llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")

llm_with_tools = llm.bind_tools(tools)



def chatbot(state: State):

    return {"messages": [llm_with_tools.invoke(state["messages"])]}



graph_builder.add_node("chatbot", chatbot)



tool_node = ToolNode(tools=[tool])

graph_builder.add_node("tools", tool_node)



graph_builder.add_conditional_edges(

    "chatbot",

    tools_condition,

)

# Any time a tool is called, we return to the chatbot to decide the next step

graph_builder.add_edge("tools", "chatbot")

graph_builder.set_entry_point("chatbot")

graph = graph_builder.compile()

解释：

导入工具：导入必要的工具类，在本例中为TavilySearchResults。
定义并绑定工具：创建工具实例并使用将其绑定到 LLM llm.bind_tools()。这将告知 LLM 可用的工具及其用法。
创建 ToolNode：实例化一个ToolNode，传入可用工具列表。
将 ToolNode 添加到图表中：使用将其合并ToolNode到 LangGraph 中graph_builder.add_node()。
条件路由：根据graph_builder.add_conditional_edges()LLM 是否决定调用工具来设置路由逻辑。该tools_condition函数检查 LLM 的响应是否包含工具调用指令。
环回：执行该工具后，用于graph_builder.add_edge()将流引导回节点chatbot，以允许对话继续。

现在，当您运行聊天机器人并提出需要外部信息的问题时，LLM 可以选择调用网络搜索工具，检索相关数据并将其合并到其响应中。

为聊天机器人添加记忆

记忆对于创建可以通过记住过去的互动进行有意义的对话的聊天机器人至关重要。

LangGraph 的检查点系统

Checkpointer：当你编译你的 LangGraph 时，你可以提供一个checkpointer对象。该对象负责保存图在不同时间点的状态。
线程 ID：每次调用图表时，您都需要提供一个thread_id。此 ID 用于checkpointer跟踪不同的对话线程。
自动保存和加载： LangGraph 会在给定的每个图执行步骤后自动保存状态thread_id。当您使用相同的再次调用该图时thread_id，它会自动加载已保存的状态，使聊天机器人能够从中断的地方继续对话。

使用检查点实现内存

基于前面的代码，下面介绍如何使用 LangGraph 的检查点添加内存：

# ... (Previous code to define State, graph_builder, nodes, and edges)



from langgraph.checkpoint.memory import MemorySaver



# Create a MemorySaver object to act as the checkpointer

memory = MemorySaver()



# Compile the graph, passing in the 'memory' object as the checkpointer

graph = graph_builder.compile(checkpointer=memory)



# ... (Rest of the code to run the chatbot)

解释：

导入 MemorySaver：MemorySaver从中导入类langgraph.checkpoint.memory。
创建 MemorySaver对象：实例化一个MemorySaver对象，它将处理保存和加载图形的状态。
传递给 compile()：编译图形时，将memory对象作为checkpointer参数传递。

现在，当您运行聊天机器人时，首先，使用thread_id作为此对话的密钥：

config = {"configurable": {"thread_id": "1"}}

每个唯一的thread_id对话都会有其历史存储。

现在开始对话：

while True:

    user_input = input("User: ")

    if user_input.lower() in ["quit", "exit", "q"]:

        print("Goodbye!")

        break



    # Process user input through the LangGraph

    for event in graph.stream({"messages": [("user", user_input)]}, config):

        for value in event.values():

            print("Assistant:", value["messages"][-1].content)

注意：调用我们的图表时，配置作为第二个位置参数提供。

人在循环中

当您想要在 AI 应用程序中加入人工监督、验证或决策时，人机交互工作流程至关重要。

利用中断实现人机交互

interrupt_before以下代码演示了如何使用 LangGraph 的or功能实现人机交互interrupt_after。具体细节如下：

from typing import Annotated



from langchain_anthropic import ChatAnthropic

from langchain_community.tools.tavily_search import TavilySearchResults

from langchain_core.messages import BaseMessage

from typing_extensions import TypedDict



from langgraph.checkpoint.memory import MemorySaver

from langgraph.graph import StateGraph

from langgraph.graph.message import add_messages

from langgraph.prebuilt import ToolNode, tools_condition



class State(TypedDict):

    messages: Annotated[list, add_messages]



graph_builder = StateGraph(State)



tool = TavilySearchResults(max_results=2)

tools = [tool]

llm = ChatAnthropic(model="claude-3-5-sonnet-20240620")

llm_with_tools = llm.bind_tools(tools)



def chatbot(state: State):

    return {"messages": [llm_with_tools.invoke(state["messages"])]}



graph_builder.add_node("chatbot", chatbot)



tool_node = ToolNode(tools=[tool])

graph_builder.add_node("tools", tool_node)



graph_builder.add_conditional_edges(

    "chatbot",

    tools_condition,

)

graph_builder.add_edge("tools", "chatbot")

graph_builder.set_entry_point("chatbot")



memory = MemorySaver()

graph = graph_builder.compile(

    checkpointer=memory,

    # This is new!

    interrupt_before=["tools"],

    # Note: can also interrupt __after__ actions, if desired.

    # interrupt_after=["tools"]

)

解释：

在这个特定的例子中，计算图将在执行tools节点之前暂停。该tools节点负责运行LLM在其轮次中可能请求的任何工具。通过在这个节点中断，你实际上可以允许人类执行以下任一操作：

批准工具调用：人工可以审核 LLM 想要进行的工具调用及其输入。如果人工认为合适，只需允许图表继续运行，工具就会被执行。
修改工具调用：如果人类认为需要调整 LLM 的工具调用（例如，优化搜索查询），他们可以修改图的状态，然后恢复执行。
绕过工具调用：人类可能会认为该工具没有必要。也许他们已经找到了 LLM 想要查找的答案。在这种情况下，他们可以用适当的信息更新图状态，LLM 会像工具返回该信息一样接收这些信息。

LangGraph 的实际用途

LangGraph 能够管理状态、协调多个代理并允许人工反馈，让您构建比简单的问答机器人更复杂、互动性更强的 AI 系统。以下是 LangGraph 的一些用途：

更智能的客户服务：想象一下，一个用于在线购物的聊天机器人，能够记住你过去的订单和偏好。它可以回答关于产品的问题，追踪你的发货情况，甚至在需要时帮你联系人工客服。
AI 研究助理：需要研究项目方面的帮助吗？基于 LangGraph 的助理可以搜索大量学术论文和文章，总结主要发现，甚至帮你整理笔记。
个性化学习： LangGraph 可以为下一代教育平台赋能。想象一下，一个系统能够适应你的学习风格，识别你需要额外帮助的领域，并推荐个性化资源。
精简业务：许多业务流程涉及多个步骤和人员。LangGraph 可以自动化部分工作流程，例如安排审批文件、分析数据或管理项目。

这些例子突出了LangGraph 如何帮助弥合人工智能能力与现实世界复杂性之间的差距。