LangGraph检查点和Send机制

示例代码

注：将ppt后缀名改为zip后解压

python版本：3.12.4

检查点

检查点是每个超级步骤保存的图状态的快照，并由StateSnapshot对象表示，具有以下关键属性：

• config：与此检查点相关的配置。
• metadata：与此检查点相关的元数据。
• values：此时状态通道的值。
• next：将要在图中执行的下一个节点名称的元组。
• tasks：包含有关要执行的下一个任务信息的PregelTask对象的元组。如果该步骤之前尝试过，它将包含错误信息。如果图在节点内部被动态中断，任务将包含与中断相关的其他数据。

官方文档地址：https://langchain-ai.github.io/langgraph/concepts/persistence/#checkpoints

中文文档地址：https://www.aidoczh.com/langgraph/concepts/persistence/#_3

LangGraph 具有一个内置的持久化层，通过检查点实现。当您将检查点与图形一起使用时，您可以与该图形的状态进行交互。当您将检查点与图形一起使用时，您可以与图形的状态进行交互并管理它。检查点在每个超级步骤中保存图形状态的检查点，从而实现一些强大的功能

首先，检查点通过允许人类检查、中断和批准步骤来促进人机交互工作流工作流。检查点对于这些工作流是必需的，因为人类必须能够在任何时候查看图形的状态，并且图形必须能够在人类对状态进行任何更新后恢复执行。

其次，它允许在交互之间进行“记忆”。您可以使用检查点创建线程并在图形执行后保存线程的状态。在重复的人类交互（例如对话）的情况下，任何后续消息都可以发送到该检查点，该检查点将保留对其以前消息的记忆。

许多 AI 应用程序需要内存来跨多个交互共享上下文。在 LangGraph 中，通过 检查点 为任何 StateGraph 提供内存。

在创建任何 LangGraph 工作流时，您可以通过以下方式设置它们以持久保存其状态

1. 一个 检查点，例如 AsyncSqliteSaver
2. 在编译图时调用 compile(checkpointer=my_checkpointer)。

示例

# 导入 asyncio 模块，用于处理异步编程
import asyncio

# 从 langgraph.checkpoint.sqlite.aio 模块中导入 AsyncSqliteSaver 类，它用于异步保存检查点到 SQLite 数据库
# pip install langgraph.checkpoint.sqlite
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver

# 从 langgraph.graph 模块中导入 StateGraph 类，它用于构建状态图
from langgraph.graph import StateGraph

async def main():
    # 创建一个 StateGraph 对象，节点的值类型为 int
    builder = StateGraph(int)

    # 添加一个名为 "add_one" 的节点，该节点的功能是将输入值加 1
    builder.add_node("add_one", lambda x: x + 1)

    # 设置 "add_one" 节点为状态图的入口点
    builder.set_entry_point("add_one")

    # 设置 "add_one" 节点为状态图的结束点
    builder.set_finish_point("add_one")

    # 使用异步上下文管理器创建一个 AsyncSqliteSaver 对象，并连接到名为 "checkpoints.db" 的 SQLite 数据库
    async with AsyncSqliteSaver.from_conn_string("checkpoints.db") as memory:
        # 编译状态图，并使用 memory 作为检查点保存器
        graph = builder.compile(checkpointer=memory)