Langgraph的持久化管理-记忆模块

持久化就是记忆，就是能让AI记住用户的定制化的核心。

LangGraph 有一个内置的持久层，通过校验指针实现。使用校验指针编译图形时，校验指针会在每个超级步骤superstep中保存图形状态的校验点checkpoint。这些检查点被保存到一个线程中，在图形执行后可以访问该线程。由于线程允许在图形执行后访问图形的状态，因此包括人在回路、内存、时间旅行和容错在内的多种强大功能都成为可能。有关如何在图形中添加和使用定点器的端到端示例，请参阅本指南。下面，我们将详细讨论这些概念。

这个checkpoint的概念非常重要，需要详细理解。

checkpoint

检查点是在每个超级步骤中保存的图形状态快照，由具有以下关键属性的 StateSnapshot 对象表示：
config: Config associated with this checkpoint.
metadata: 与该检查点相关的元数据。
values: 此时状态通道的值。
next 图中下一个要执行的节点名称元组。
tasks:PregelTask 对象的元组，包含下一个待执行任务的信息。如果之前尝试过该步骤，则会包含错误信息。如果图形在节点内被动态中断，任务将包含与中断相关的附加数据。

示例

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from typing import Annotated
from typing_extensions import TypedDict
from operator import add

class State(TypedDict):
    foo: str
    bar: Annotated[list[str], add]

def node_a(state: State):
    return {"foo": "a", "bar": ["a"]}

def node_b(state: State):
    return {"foo": "b", "bar": ["b"]}


workflow = StateGraph(State)
workflow.add_node(node_a)
workflow.add_node(node_b)
workflow.add_edge(START, "node_a")
workflow.add_edge("node_a", "node_b")
workflow.add_edge("node_b", END)

checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": ""}, config)

这里有四个检查点
1、以 START 为下一个执行节点的空检查点
2、用户输入 {‘foo’: ‘’, ‘bar’： []}，并将 node_a 作为下一个要执行的节点
3、node_a的输出进行检查点 {‘foo’: ‘a’, ‘bar’： [‘a’]}，并将 node_b 作为下一个要执行的节点
4、node_b {‘foo’: ‘b’, ‘bar’： [‘a’、’b’]}的输出作为检查点，并且没有下一个节点要执行

好了以上就是你需要理解的概念，我们可以回到主要问题：
如何为graph添加线程级持久性了。

如何为您的图添加线程级持久性

许多AI应用程序需要内存以在多个交互之间共享上下文。在LangGraph中，这种内存可以通过在任何 StateGraph 中添加 线程级持久性 来实现。

在创建任何LangGraph图时，您可以通过在编译图时添加 检查点保存器 来设置其状态保持：

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph.compile(checkpointer=checkpointer)

这里不涉及多个对话或用户之间共享的内存（跨线程持久性）。如果想，看这里https://langchain-ai.github.io/langgraph/how-tos/persistence/#add-long-term-memory

下面我们搞个例子，方便理解。
基础步骤我就跳过，具体查看https://www.aidoczh.com/langgraph/how-tos/persistence/
1、安装所需的包
2、为 OpenAI（我们将使用的 LLM）和 Tavily（我们将使用的搜索工具）设置 API 密钥。
3、定义图，可以定义我们的 StateGraph 并添加我们的模型调用节点
4、添加持久性

from langgraph.checkpoint.memory import MemorySaver

# 创建一个内存保存器实例，用于存储图的状态和执行历史
# MemorySaver是一个内存中的检查点保存器，它将图的状态保存在内存中，适用于单个会话或测试环境
memory = MemorySaver()

# 使用memory作为checkpointer编译图
# 这样图的每个执行状态都会被保存到内存中，使得我们可以在多次交互之间保持状态
# 这实现了线程级的持久性，即同一个对话线程中的状态可以被保留
graph = builder.compile(checkpointer=memory)

# 调用原理：
# 1. 当图被调用时，checkpointer会根据thread_id查找是否存在之前的状态
# 2. 如果存在，会从上次的检查点恢复状态，继续执行
# 3. 在图执行过程中，每个节点执行后都会创建检查点，保存当前状态和下一个要执行的节点
# 4. 检查点包含：当前状态数据、执行历史和下一个要执行的节点信息
# 5. 通过config参数中的thread_id来区分不同的对话线程
# 6. 这使得同一个图实例可以同时处理多个独立的对话，每个对话都有自己的状态

# If you're using LangGraph Cloud or LangGraph Studio, you don't need to pass the checkpointer when compiling the graph, since it's done automatically.

定义具有持久性的图表

要为具有子图的图添加持久性，您只需在编译父图时传递一个检查点保存器。LangGraph将自动将检查点保存器传播到子子图。

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

子图的定义：https://langchain-ai.github.io/langgraph/concepts/low_level/#subgraphs

我觉得还是很好理解的，一般子图会应用在多Agent的情况，每个Agent可以看作一个子图，然后由主图控制协调多个Agent的执行。当前我们用的不多，但是是很关键的定义。

使用内存检查点保存器

内存检查点保存器是最简单的检查点保存器，它将图的状态保存在内存中。

如何为您的图添加跨线程持久性

这个概念对我们很重要，所有的AI应用都需要跨线程持久性，否则无法实现多轮对话。Chatgpt就是用类似这种方法进行多线程管理。
LangGraph 还允许您在多个线程之间持久化数据。例如，您可以将用户（他们的姓名或偏好）信息存储在共享内存中，并在新的对话线程中重复使用这些信息。

在本指南中，我们将展示如何构建和使用一个实现了使用Store接口的共享内存的图。

https://www.aidoczh.com/langgraph/how-tos/cross-thread-persistence/#_2

前面的步骤还是一样：
1、装安装包、设置API
2、定义存储

定义存储

在这个例子中，我们将创建一个图形，可以检索用户的偏好信息。我们将通过定义一个 InMemoryStore 来实现

• 这是一个可以在内存中存储数据并查询数据的对象。然后，我们将在编译图形时传递存储对象。这允许图形中的每个节点访问存储：当你定义节点函数时，可以定义 store 关键字参数，LangGraph 将自动传递你编译图形时使用的存储对象。

在使用 Store 接口存储对象时，你需要定义两个东西：

• 对象的命名空间， 一个元组（类似于目录）
• 对象键（类似于文件名）

在我们的例子中，我们将使用 (“memories”, ) 作为命名空间，并为每个新记忆使用随机 UUID 作为键。

重要的是，为了确定用户，我们将通过节点函数的 config 关键字参数传递 user_id。
让我们首先定义一个已经填充了一些关于用户记忆的 InMemoryStore。

from langgraph.store.memory import InMemoryStore
from langchain_openai import OpenAIEmbeddings

in_memory_store = InMemoryStore(
    index={
        "embed": OpenAIEmbeddings(model="text-embedding-3-small"),
        "dims": 1536,
    }
)

创建图形

import uuid
from typing import Annotated
from typing_extensions import TypedDict

from langchain_anthropic import ChatAnthropic
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.base import BaseStore


model = ChatAnthropic(model="claude-3-5-sonnet-20240620")  # 初始化Claude模型


# 注意：我们将Store参数传递给节点函数 -- 这是我们编译图时使用的Store
def call_model(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
    user_id = config["configurable"]["user_id"]  # 从配置中获取用户ID
    namespace = ("memories", user_id)  # 创建命名空间，用于存储特定用户的记忆
    memories = store.search(namespace, query=str(state["messages"][-1].content))  # 根据用户最新消息搜索相关记忆
    info = "\n".join([d.value["data"] for d in memories])  # 将找到的记忆合并为文本
    system_msg = f"You are a helpful assistant talking to the user. User info: {info}"  # 创建系统消息，包含用户信息

    # 如果用户要求记住信息，则存储新的记忆
    last_message = state["messages"][-1]
    if "remember" in last_message.content.lower():
        memory = "User name is Bob"  # 示例记忆内容
        store.put(namespace, str(uuid.uuid4()), {"data": memory})  # 使用UUID作为键存储记忆

    response = model.invoke(
        [{"type": "system", "content": system_msg}] + state["messages"]  # 调用模型，传入系统消息和对话历史
    )
    return {"messages": response}  # 返回模型响应


builder = StateGraph(MessagesState)  # 创建状态图
builder.add_node("call_model", call_model)  # 添加模型调用节点
builder.add_edge(START, "call_model")  # 设置图的起始边

# 注意：在编译图时传递store对象，实现跨线程持久性
graph = builder.compile(checkpointer=MemorySaver(), store=in_memory_store)
# 如果使用LangGraph Cloud或LangGraph Studio，不需要手动传递store或checkpointer，因为它会自动完成。

# 这段代码展示了如何实现跨线程持久性：
# 1. 创建一个共享的InMemoryStore存储对象
# 2. 在节点函数中通过store参数访问存储
# 3. 使用(namespace, key)的方式组织数据，其中namespace包含用户ID
# 4. 在编译图时传入store对象，使其在所有会话中共享
# 5. 通过这种方式，即使在不同的对话线程中，也能访问同一用户的历史记忆

# 调用图

后面验证步骤我就不写了

后面还有如何使用用Postgres检查点进行持久性，如何使用 MongoDB 检查点进行持久性如何使用 Redis 创建自定义检查点
我就等之后用到再更新吧