LangGraph 内置了一个通过检查点器实现的持久化层。当你使用检查点器编译图时，检查点器会在每个超级步骤保存图状态的检查点。这些检查点被保存到一个线程中，可以在图执行后访问。因为线程允许在执行后访问图的状态，所以包括人在回路、记忆、时间旅行和容错在内的几个强大功能都成为可能。下面，我们将更详细地讨论这些概念。

Agent Server 自动处理检查点 当使用 Agent Server 时，你不需要手动实现或配置检查点器。服务器会在幕后为你处理所有的持久化基础设施。

线程

线程是检查点器为每个保存的检查点分配的唯一 ID 或线程标识符。它包含一系列运行的累积状态。当运行被执行时，助手底层图的状态将被持久化到线程中。

当使用检查点器调用图时，你必须在配置的可配置部分指定一个thread_id：

{"configurable": {"thread_id": "1"}}

可以检索线程的当前和历史状态。为了持久化状态，必须在执行运行之前创建线程。LangSmith API 提供了几个用于创建和管理线程及线程状态的端点。更多详情请参阅 API 参考。

检查点器使用thread_id作为存储和检索检查点的主键。没有它，检查点器无法保存状态或在中断后恢复执行，因为检查点器使用thread_id来加载保存的状态。

检查点

线程在特定时间点的状态称为检查点。检查点是在每个超级步骤保存的图状态快照，由StateSnapshot对象表示，具有以下关键属性：

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

检查点被持久化，并可用于在以后恢复线程的状态。

让我们看看当调用一个简单图时保存了哪些检查点：

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.runnables import RunnableConfig
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: RunnableConfig = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": "", "bar":[]}, config)

运行图后，我们预计会看到恰好 4 个检查点：

• 下一个要执行的节点为 START 的空检查点
• 包含用户输入 {'foo': '', 'bar': []} 且下一个要执行的节点为 node_a 的检查点
• 包含 node_a 的输出 {'foo': 'a', 'bar': ['a']} 且下一个要执行的节点为 node_b 的检查点
• 包含 node_b 的输出 {'foo': 'b', 'bar': ['a', 'b']} 且没有下一个要执行节点的检查点

请注意，bar 通道的值包含来自两个节点的输出，因为我们为 bar 通道设置了归约器。

获取状态

与保存的图状态交互时，你必须指定一个线程标识符。你可以通过调用 graph.get_state(config) 来查看图的_最新_状态。这将返回一个 StateSnapshot 对象，该对象对应于配置中提供的线程 ID 关联的最新检查点，或者如果提供了，则对应于该线程的检查点 ID 关联的检查点。

# get the latest state snapshot
config = {"configurable": {"thread_id": "1"}}
graph.get_state(config)

# get a state snapshot for a specific checkpoint_id
config = {"configurable": {"thread_id": "1", "checkpoint_id": "1ef663ba-28fe-6528-8002-5a559208592c"}}
graph.get_state(config)

在我们的示例中，get_state 的输出将如下所示：

StateSnapshot(
    values={'foo': 'b', 'bar': ['a', 'b']},
    next=(),
    config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28fe-6528-8002-5a559208592c'}},
    metadata={'source': 'loop', 'writes': {'node_b': {'foo': 'b', 'bar': ['b']}}, 'step': 2},
    created_at='2024-08-29T19:19:38.821749+00:00',
    parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f9-6ec4-8001-31981c2c39f8'}}, tasks=()
)

获取状态历史

你可以通过调用 graph.get_state_history(config) 来获取给定线程的图执行完整历史。这将返回与配置中提供的线程 ID 关联的 StateSnapshot 对象列表。重要的是，检查点将按时间顺序排列，最新的检查点 / StateSnapshot 位于列表首位。

config = {"configurable": {"thread_id": "1"}}
list(graph.get_state_history(config))

在我们的示例中，get_state_history 的输出将如下所示：

[
    StateSnapshot(
        values={'foo': 'b', 'bar': ['a', 'b']},
        next=(),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28fe-6528-8002-5a559208592c'}},
        metadata={'source': 'loop', 'writes': {'node_b': {'foo': 'b', 'bar': ['b']}}, 'step': 2},
        created_at='2024-08-29T19:19:38.821749+00:00',
        parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f9-6ec4-8001-31981c2c39f8'}},
        tasks=(),
    ),
    StateSnapshot(
        values={'foo': 'a', 'bar': ['a']},
        next=('node_b',),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f9-6ec4-8001-31981c2c39f8'}},
        metadata={'source': 'loop', 'writes': {'node_a': {'foo': 'a', 'bar': ['a']}}, 'step': 1},
        created_at='2024-08-29T19:19:38.819946+00:00',
        parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f4-6b4a-8000-ca575a13d36a'}},
        tasks=(PregelTask(id='6fb7314f-f114-5413-a1f3-d37dfe98ff44', name='node_b', error=None, interrupts=()),),
    ),
    StateSnapshot(
        values={'foo': '', 'bar': []},
        next=('node_a',),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f4-6b4a-8000-ca575a13d36a'}},
        metadata={'source': 'loop', 'writes': None, 'step': 0},
        created_at='2024-08-29T19:19:38.817813+00:00',
        parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f0-6c66-bfff-6723431e8481'}},
        tasks=(PregelTask(id='f1b14528-5ee5-579c-949b-23ef9bfbed58', name='node_a', error=None, interrupts=()),),
    ),
    StateSnapshot(
        values={'bar': []},
        next=('__start__',),
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f0-6c66-bfff-6723431e8481'}},
        metadata={'source': 'input', 'writes': {'foo': ''}, 'step': -1},
        created_at='2024-08-29T19:19:38.816205+00:00',
        parent_config=None,
        tasks=(PregelTask(id='6d27aa2e-d72b-5504-a36f-8620e54a76dd', name='__start__', error=None, interrupts=()),),
    )
]

重放

也可以回放先前的图执行。如果我们使用 thread_id 和 checkpoint_id 来 invoke 一个图，那么我们将_重放_对应于 checkpoint_id 的检查点_之前_已执行的步骤，并且只执行检查点_之后_的步骤。

• thread_id 是线程的 ID。
• checkpoint_id 是指向线程内特定检查点的标识符。

你必须在调用图时将这些作为配置的可配置部分传递：

config = {"configurable": {"thread_id": "1", "checkpoint_id": "0c62ca34-ac19-445d-bbb0-5b4984975b2a"}}
graph.invoke(None, config=config)

重要的是，LangGraph 知道某个特定步骤是否之前已执行过。如果是，LangGraph 只是简单地_重放_图中的该特定步骤，并且不重新执行该步骤，但仅针对提供的 checkpoint_id _之前_的步骤。checkpoint_id _之后_的所有步骤都将被执行（即一个新的分支），即使它们之前已经执行过。请参阅此关于时间旅行的操作指南以了解更多关于重放的信息。

更新状态

除了从特定的检查点重放图之外，我们还可以_编辑_图状态。我们使用 update_state 来实现这一点。该方法接受三个不同的参数：

config

配置应包含指定要更新哪个线程的 thread_id。当只传递 thread_id 时，我们更新（或分支）当前状态。或者，如果我们包含 checkpoint_id 字段，那么我们将分支该选定的检查点。

values

这些是用于更新状态的值。请注意，此更新与来自节点的任何更新完全相同。这意味着这些值将被传递给归约器函数，如果它们为图状态中的某些通道定义了。这意味着 update_state 不会自动覆盖每个通道的通道值，而只会覆盖没有归约器的通道。让我们看一个例子。

假设你已使用以下模式定义了图的状态（参见上面的完整示例）：

from typing import Annotated
from typing_extensions import TypedDict
from operator import add

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

现在假设图的当前状态是

{"foo": 1, "bar": ["a"]}

如果你按如下方式更新状态：

graph.update_state(config, {"foo": 2, "bar": ["b"]})

那么图的新状态将是：

{"foo": 2, "bar": ["a", "b"]}

foo 键（通道）被完全更改（因为该通道没有指定归约器，所以 update_state 会覆盖它）。但是，为 bar 键指定了归约器，因此它将 "b" 附加到 bar 的状态。

as_node

调用 update_state 时可以可选指定的最后一件事是 as_node。如果你提供了它，更新将应用为好像来自节点 as_node。如果未提供 as_node，它将设置为最后更新状态的节点（如果不模糊）。这很重要的原因是，接下来要执行的步骤取决于最后给出更新的节点，因此这可以用来控制哪个节点接下来执行。请参阅此关于时间旅行的操作指南以了解更多关于分支状态的信息。

记忆存储

状态模式指定了一组在图执行时填充的键。如上所述，状态可以由检查点器在每个图步骤写入线程，从而实现状态持久化。

但是，如果我们想_跨线程_保留一些信息呢？考虑聊天机器人的情况，我们希望在该用户的所有聊天对话（例如，线程）中保留关于该用户的特定信息！

仅使用检查点器，我们无法跨线程共享信息。这引出了对 Store 接口的需求。作为说明，我们可以定义一个 InMemoryStore 来跨线程存储关于用户的信息。我们只需像以前一样使用检查点器编译我们的图，并加上我们的新 in_memory_store 变量。

LangGraph API 自动处理存储 当使用 LangGraph API 时，你不需要手动实现或配置存储。API 会在幕后为你处理所有的存储基础设施。

基本用法

首先，让我们在不使用 LangGraph 的情况下单独展示这一点。

from langgraph.store.memory import InMemoryStore
in_memory_store = InMemoryStore()

记忆通过一个元组进行命名空间划分，在这个特定示例中将是 (<user_id>, "memories")。命名空间可以是任意长度并代表任何内容，不一定必须是用户特定的。

user_id = "1"
namespace_for_memory = (user_id, "memories")

我们使用 store.put 方法将记忆保存到我们命名空间中的存储。当我们这样做时，我们指定命名空间（如上定义）以及记忆的键值对：键只是记忆的唯一标识符（memory_id），值（一个字典）是记忆本身。

memory_id = str(uuid.uuid4())
memory = {"food_preference" : "I like pizza"}
in_memory_store.put(namespace_for_memory, memory_id, memory)

我们可以使用 store.search 方法读取我们命名空间中的记忆，该方法将返回给定用户的所有记忆作为列表。最新的记忆是列表中的最后一个。

memories = in_memory_store.search(namespace_for_memory)
memories[-1].dict()
{'value': {'food_preference': 'I like pizza'},
 'key': '07e0caf4-1631-47b7-b15f-65515d4c1843',
 'namespace': ['1', 'memories'],
 'created_at': '2024-10-02T17:22:31.590602+00:00',
 'updated_at': '2024-10-02T17:22:31.590605+00:00'}

每种记忆类型都是一个具有某些属性的 Python 类（Item）。我们可以通过如上所示的 .dict 转换将其作为字典访问。

它具有的属性是：

• value：此记忆的值（本身是一个字典）
• key：此记忆在此命名空间中的唯一键
• namespace：字符串列表，此记忆类型的命名空间
• created_at：此记忆创建时的时间戳
• updated_at：此记忆更新时的时间戳

语义搜索

除了简单的检索，存储还支持语义搜索，允许你根据含义而不是精确匹配来查找记忆。要启用此功能，请使用嵌入模型配置存储：

from langchain.embeddings import init_embeddings

store = InMemoryStore(
    index={
        "embed": init_embeddings("openai:text-embedding-3-small"),  # Embedding provider
        "dims": 1536,                              # Embedding dimensions
        "fields": ["food_preference", "$"]              # Fields to embed
    }
)

现在搜索时，你可以使用自然语言查询来查找相关记忆：

# Find memories about food preferences
# (This can be done after putting memories into the store)
memories = store.search(
    namespace_for_memory,
    query="What does the user like to eat?",
    limit=3  # Return top 3 matches
)

你可以通过配置 fields 参数或在存储记忆时指定 index 参数来控制记忆的哪些部分被嵌入：

# Store with specific fields to embed
store.put(
    namespace_for_memory,
    str(uuid.uuid4()),
    {
        "food_preference": "I love Italian cuisine",
        "context": "Discussing dinner plans"
    },
    index=["food_preference"]  # Only embed "food_preferences" field
)

# Store without embedding (still retrievable, but not searchable)
store.put(
    namespace_for_memory,
    str(uuid.uuid4()),
    {"system_info": "Last updated: 2024-01-01"},
    index=False
)

在 LangGraph 中使用

有了这一切，我们在 LangGraph 中使用 in_memory_store。in_memory_store 与检查点器协同工作：检查点器将状态保存到线程，如上所述，而 in_memory_store 允许我们存储任意信息以供_跨_线程访问。我们使用检查点器和 in_memory_store 编译图，如下所示。

from langgraph.checkpoint.memory import InMemorySaver

# We need this because we want to enable threads (conversations)
checkpointer = InMemorySaver()

# ... Define the graph ...

# Compile the graph with the checkpointer and store
graph = graph.compile(checkpointer=checkpointer, store=in_memory_store)

我们像之前一样，使用 thread_id 来调用图，同时还会使用 user_id，正如我们上面展示的，我们将用这个 user_id 来为这个特定用户的记忆进行命名空间划分。

# Invoke the graph
user_id = "1"
config = {"configurable": {"thread_id": "1", "user_id": user_id}}

# First let's just say hi to the AI
for update in graph.stream(
    {"messages": [{"role": "user", "content": "hi"}]}, config, stream_mode="updates"
):
    print(update)

我们可以在 任何节点 中通过传递 store: BaseStore 和 config: RunnableConfig 作为节点参数来访问 in_memory_store 和 user_id。以下是如何在节点中使用语义搜索来查找相关记忆的示例：

def update_memory(state: MessagesState, config: RunnableConfig, *, store: BaseStore):

    # Get the user id from the config
    user_id = config["configurable"]["user_id"]

    # Namespace the memory
    namespace = (user_id, "memories")

    # ... Analyze conversation and create a new memory

    # Create a new memory ID
    memory_id = str(uuid.uuid4())

    # We create a new memory
    store.put(namespace, memory_id, {"memory": memory})

正如我们上面展示的，我们也可以在任何节点中访问存储，并使用 store.search 方法来获取记忆。请注意，记忆会以可转换为字典的对象列表形式返回。

memories[-1].dict()
{'value': {'food_preference': 'I like pizza'},
 'key': '07e0caf4-1631-47b7-b15f-65515d4c1843',
 'namespace': ['1', 'memories'],
 'created_at': '2024-10-02T17:22:31.590602+00:00',
 'updated_at': '2024-10-02T17:22:31.590605+00:00'}

我们可以访问这些记忆并在模型调用中使用它们。

def call_model(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
    # Get the user id from the config
    user_id = config["configurable"]["user_id"]

    # Namespace the memory
    namespace = (user_id, "memories")

    # Search based on the most recent message
    memories = store.search(
        namespace,
        query=state["messages"][-1].content,
        limit=3
    )
    info = "\n".join([d.value["memory"] for d in memories])

    # ... Use memories in the model call

如果我们创建一个新线程，只要 user_id 相同，我们仍然可以访问相同的记忆。

# Invoke the graph
config = {"configurable": {"thread_id": "2", "user_id": "1"}}

# Let's say hi again
for update in graph.stream(
    {"messages": [{"role": "user", "content": "hi, tell me about my memories"}]}, config, stream_mode="updates"
):
    print(update)

当我们使用 LangSmith 时，无论是在本地（例如在 Studio 中）还是 使用 LangSmith 托管，基础存储默认都是可用的，无需在图编译时指定。但是，要启用语义搜索，您 确实 需要在 langgraph.json 文件中配置索引设置。例如：

{
    ...
    "store": {
        "index": {
            "embed": "openai:text-embeddings-3-small",
            "dims": 1536,
            "fields": ["$"]
        }
    }
}

有关更多详细信息和配置选项，请参阅 部署指南。

检查点器库

在底层，检查点功能由符合 BaseCheckpointSaver 接口的检查点器对象提供支持。LangGraph 提供了几种检查点器实现，都是通过独立的、可安装的库实现的：

• langgraph-checkpoint：检查点保存器的基础接口（BaseCheckpointSaver）和序列化/反序列化接口（SerializerProtocol）。包含用于实验的内存检查点器实现（InMemorySaver）。LangGraph 内置了 langgraph-checkpoint。
• langgraph-checkpoint-sqlite：使用 SQLite 数据库的 LangGraph 检查点器实现（SqliteSaver / AsyncSqliteSaver）。适用于实验和本地工作流。需要单独安装。
• langgraph-checkpoint-postgres：使用 Postgres 数据库的高级检查点器（PostgresSaver / AsyncPostgresSaver），在 LangSmith 中使用。适用于生产环境。需要单独安装。

检查点器接口

每个检查点器都符合 BaseCheckpointSaver 接口，并实现以下方法：

• .put - 存储一个检查点及其配置和元数据。
• .put_writes - 存储与检查点关联的中间写入（即 待定写入）。
• .get_tuple - 根据给定配置（thread_id 和 checkpoint_id）获取检查点元组。这用于在 graph.get_state() 中填充 StateSnapshot。
• .list - 列出符合给定配置和筛选条件的检查点。这用于在 graph.get_state_history() 中填充状态历史记录。

如果检查点器用于异步图执行（即通过 .ainvoke、.astream、.abatch 执行图），则将使用上述方法的异步版本（.aput、.aput_writes、.aget_tuple、.alist）。

要异步运行图，您可以使用 InMemorySaver，或者 Sqlite/Postgres 检查点器的异步版本——AsyncSqliteSaver / AsyncPostgresSaver 检查点器。

序列化器

当检查点器保存图状态时，它们需要序列化状态中的通道值。这是通过序列化器对象完成的。

langgraph_checkpoint 定义了用于实现序列化器的 协议，并提供了一个默认实现（JsonPlusSerializer），该实现处理多种类型，包括 LangChain 和 LangGraph 原语、日期时间、枚举等。

使用 pickle 进行序列化

默认的序列化器 JsonPlusSerializer 在底层使用 ormsgpack 和 JSON，这不适用于所有类型的对象。

如果您希望对我们当前的 msgpack 编码器不支持的对象（例如 Pandas 数据框）回退到 pickle，可以使用 JsonPlusSerializer 的 pickle_fallback 参数：

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer

# ... Define the graph ...
graph.compile(
    checkpointer=InMemorySaver(serde=JsonPlusSerializer(pickle_fallback=True))
)

加密

检查点器可以选择性地加密所有持久化的状态。要启用此功能，请将 EncryptedSerializer 的实例传递给任何 BaseCheckpointSaver 实现的 serde 参数。创建加密序列化器最简单的方法是通过 from_pycryptodome_aes，它从 LANGGRAPH_AES_KEY 环境变量中读取 AES 密钥（或接受 key 参数）：

import sqlite3

from langgraph.checkpoint.serde.encrypted import EncryptedSerializer
from langgraph.checkpoint.sqlite import SqliteSaver

serde = EncryptedSerializer.from_pycryptodome_aes()  # reads LANGGRAPH_AES_KEY
checkpointer = SqliteSaver(sqlite3.connect("checkpoint.db"), serde=serde)

from langgraph.checkpoint.serde.encrypted import EncryptedSerializer
from langgraph.checkpoint.postgres import PostgresSaver

serde = EncryptedSerializer.from_pycryptodome_aes()
checkpointer = PostgresSaver.from_conn_string("postgresql://...", serde=serde)
checkpointer.setup()

在 LangSmith 上运行时，只要存在 LANGGRAPH_AES_KEY，加密就会自动启用，因此您只需要提供环境变量。其他加密方案可以通过实现 CipherProtocol 并将其提供给 EncryptedSerializer 来使用。

功能

人在回路

首先，检查点器通过允许人类检查、中断和批准图步骤，促进了 人在回路工作流。这些工作流需要检查点器，因为人类必须能够随时查看图的状态，并且图必须在人类对状态进行任何更新后能够恢复执行。有关示例，请参阅 操作指南。

记忆

其次，检查点器允许在交互之间实现 "记忆"。在重复的人类交互（如对话）中，任何后续消息都可以发送到该线程，该线程将保留对先前消息的记忆。有关如何使用检查点器添加和管理对话记忆的信息，请参阅 添加记忆。

时间旅行

第三，检查点器允许 "时间旅行"，使用户能够重放先前的图执行，以审查和/或调试特定的图步骤。此外，检查点器使得可以在任意检查点处分叉图状态，以探索不同的轨迹。

容错性

最后，检查点还提供了容错性和错误恢复能力：如果一个或多个节点在某个超步中失败，您可以从最后一个成功的步骤重新启动图。此外，当图节点在某个超步中执行失败时，LangGraph 会存储该超步中任何其他成功完成的节点的待定检查点写入，这样无论何时我们从该超步恢复图执行，都不会重新运行成功的节点。

待定写入

此外，当图节点在某个超步中执行失败时，LangGraph 会存储该超步中任何其他成功完成的节点的待定检查点写入，这样无论何时我们从该超步恢复图执行，都不会重新运行成功的节点。