检查点器在每个超步骤保存图状态的快照，组织到 **线程**中。使用检查点器编译图可以启用人机交互工作流、时间旅行调试、容错执行和对话记忆。

!检查点

为什么要使用检查点器

以下功能需要检查点器：

- **Human-in-the-loop**：检查点器支持 人机交互工作流 ，允许人类检查、中断和批准图步骤。这些工作流需要检查点器，因为人们需要能够查看图在任何时间点的状态，并且图需要在人对状态进行任何更新后能够恢复执行。参见 中断 的示例。 - **记忆**：检查点器允许交互之间存在 "记忆" 。在重复的人类交互（如对话）中，任何后续消息都可以发送到该线程，它将保留对之前交互的记忆。参见 添加记忆 了解如何使用检查点器添加和管理对话记忆。 - **时间旅行**：检查点器支持 "时间旅行", allowing users to replay prior graph executions to review and / or debug specific graph steps. In addition, checkpointers make it possible to fork the graph state at arbitrary checkpoints to explore alternative trajectories. - **Fault-tolerance**：检查点提供容错和错误恢复：如果一个或多个节点在给定的超步骤失败，您可以从最后一个成功的步骤重新启动图。 <a id="pending-writes"></a> - **待处理写入**：当图节点在给定的 super-step超步骤执行中途失败时，LangGraph 会存储在该超步骤成功完成的任何其他节点的待处理检查点写入。从该超步骤恢复图执行时，您无需重新运行成功的节点。

核心概念

线程

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

使用检查点器调用图时，您 **必须** 指定 thread_id 作为 configurable 配置部分：

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

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

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

检查点

线程在特定时间点的状态称为检查点。检查点是保存在每个 super-step 处的图状态的快照，并由 StateSnapshot 对象表示（请参阅 StateSnapshot 字段 获取完整字段参考）。

Super-steps

LangGraph 在每个 **super-step** 边界创建一个检查点。超级步骤是图的一次"tick"，在该步骤中调度的所有节点都会执行（可能并行）。对于像 START -> A -> B -> END这样的顺序图，输入、节点 A 和节点 B 有单独的超级步骤——每个步骤后都会产生一个检查点。理解超级步骤边界对于 时间旅行很重要，因为只能从检查点（即超级步骤边界）恢复执行。

除了超级步骤检查点，LangGraph 还会在以下位置持久化写入： **节点（任务）级别**。当超级步骤内的每个节点完成时，其输出会被写入检查点的 checkpoint_writes 表中，作为链接到进行中检查点的任务条目。这些每个任务的写入使 待处理写入 恢复成为可能：如果同一超级步骤中的另一个节点失败，成功节点的写入已经是持久的，在恢复时不需要重新运行。完整的状态快照会在超级步骤完成时提交。

LangGraph 还会持久化超级步骤内各个节点执行的写入。这些写入存储为任务，用于容错：如果同一超级步骤中的另一个节点失败，成功节点的写入在恢复时不需要重新计算。这些任务写入不是完整的 StateSnapshot 检查点，因此时间旅行从超级步骤边界的完整检查点恢复。

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

让我们看看当按如下方式调用简单图时保存了哪些检查点：

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 channel.

检查点命名空间

每个检查点都有一个 checkpoint_ns （检查点命名空间）字段，用于标识它属于哪个图或子图：

• - **""** （空字符串）：检查点属于父（根）图。
• - **"node_name:uuid"**：检查点属于作为给定节点调用的子图。对于嵌套子图，命名空间通过 | 分隔符连接（例如， "outer_node:uuid|inner_node:uuid").

您可以通过配置从节点内部访问检查点命名空间：

from langchain_core.runnables import RunnableConfig

def my_node(state: State, config: RunnableConfig):
    checkpoint_ns = config["configurable"]["checkpoint_ns"]
    # "" for the parent graph, "node_name:uuid" for a subgraph

请参阅 子图 以获取有关使用子图状态和检查点的更多详细信息。

获取和更新状态

获取状态

与保存的图状态进行交互时，您 **必须** 指定一个 线程标识符。您可以查看 _最新_ 的图状态，方法是调用 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=()
)

StateSnapshot 字段

获取状态历史

你可以通过调用 @[graph.get_state_history(config)][获取_状态_历史]。这将返回一个 StateSnapshot objects associated with the thread ID provided in the config. Importantly, the checkpoints will be ordered chronologically with the most recent checkpoint / 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=()),),
    )
]

!状态

查找特定检查点

你可以筛选状态历史以查找符合特定条件的检查点：

history = list(graph.get_state_history(config))

# Find the checkpoint before a specific node executed
before_node_b = next(s for s in history if s.next == ("node_b",))

# Find a checkpoint by step number
step_2 = next(s for s in history if s.metadata["step"] == 2)

# Find checkpoints created by update_state
forks = [s for s in history if s.metadata["source"] == "update"]

# Find the checkpoint where an interrupt occurred
interrupted = next(
    s for s in history
    if s.tasks and any(t.interrupts for t in s.tasks)
)

重放

重放会重新执行先前检查点的步骤。使用先前的 checkpoint_id 重新运行该检查点后的节点。检查点之前的节点会被跳过（其结果已保存）。检查点之后的节点会重新执行，包括任何 LLM 调用、API 请求或 中断 ——这些在重放期间总是会被重新触发。

参见 时间旅行 关于重放过去执行的完整详细信息和代码示例。

!重放

更新状态

您可以使用 @[ 来编辑图状态update_state]。这会创建一个包含更新值的新检查点——它不会修改原始检查点。更新被视为与节点更新相同：值通过 reducer 函数（当已定义时），因此带有 reducer 的通道 _累积_ 值而不是覆盖它们。

您可以选择指定 as_node 来控制更新被视为来自哪个节点，这会影响下一个执行的节点。请参阅 时间旅行： as_node 了解更多详情。

!更新

持久性模式

LangGraph 支持三种持久性模式，让您在性能和数据一致性之间取得平衡。您可以在调用任何图执行方法时指定持久性模式：

graph.stream(
    {"input": "test"},
    durability="sync"
)

持久性模式从最不持久到最持久依次如下：

• * "exit"：LangGraph 仅在图执行退出时（无论成功完成、出现错误还是由于人工介入中断）保存更改。这为长时间运行的图提供了最佳性能，但意味着不会保存中间状态，因此您无法从系统故障（如进程崩溃）中恢复执行。
• * "async"：LangGraph 在下一步执行时异步保存更改。这提供了良好的性能和持久性，但如果进程在执行过程中崩溃，LangGraph 可能无法写入检查点。
• * "sync"：LangGraph 在下一步开始前同步保存更改。这确保 LangGraph 在继续执行前写入每个检查点，以牺牲部分性能为代价提供高持久性。

优化检查点存储

默认情况下，LangGraph 检查点在每个超步骤写入每个状态通道的完整值。对于带有大量累积的长运行线程（如多轮对话），这会随着时间推移产生显著的存储增长。

DeltaChannel 仅存储增量差值而不是完整的累积值，从而显著减少追加密集型通道的检查点大小。请参阅 DeltaChannel 了解使用方法和存储与延迟的权衡。

Checkpointer 库

在底层，检查点由符合 @[ 接口的 checkpointer 对象提供支持BaseCheckpointSaver] 接口的 checkpointer 对象提供支持。LangGraph 提供了多个 checkpointer 实现，全部通过独立的可安装库实现。

• * langgraph-checkpoint：检查点保存器的基础接口（BaseCheckpointSaver) and serialization/deserialization interface (SerializerProtocol）。包含内存中检查点实现（InMemorySaver）用于实验。LangGraph 自带 langgraph-checkpoint included.
• * langgraph-checkpoint-sqlite：使用 SQLite 数据库的 LangGraph 检查点实现（SqliteSaver / AsyncSqliteSaver）。适合实验和本地工作流。需要单独安装。
• * langgraph-checkpoint-postgres：使用 Postgres 数据库的高级检查点（PostgresSaver / AsyncPostgresSaver），用于 LangSmith。适合生产环境使用。需要单独安装。
• * langchain-azure-cosmosdb：使用 Azure Cosmos DB for NoSQL 的 LangGraph 检查点实现（CosmosDBSaverSync / CosmosDBSaver）。适合在 Azure 上生产使用。支持同步和异步操作，以及 Microsoft Entra ID 身份验证。需要单独安装。

检查点接口

每个检查点都遵循 BaseCheckpointSaver 接口并实现以下方法：

• * .put - 存储检查点及其配置和元数据。
• * .put_writes - 存储与检查点关联的中间写入（即 待处理写入).
• * .get_tuple - 使用给定配置获取检查点元组（thread_id 和 checkpoint_id）。用于填充 StateSnapshot in graph.get_state().
• * .list - 列出匹配给定配置和过滤条件的检查点。用于填充状态历史 graph.get_state_history()

如果检查点与异步图执行一起使用（即通过 .ainvoke, .astream, .abatch执行图），则将使用上述方法的异步版本（.aput, .aput_writes, .aget_tuple, .alist).

序列化器

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

langgraph_checkpoint 定义 protocol 用于实现序列化器，提供默认实现（JsonPlusSerializer），可处理多种类型，包括 LangChain 和 LangGraph 基元、datetime、枚举等。

使用序列化 pickle

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

如果要为 msgpack 编码器当前不支持的对象（如 Pandas 数据帧）回退到 pickle， 可以使用 pickle_fallback JsonPlusSerializer:

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 的实例传递给 serde 任意 BaseCheckpointSaver 实现中的参数。最简单的创建加密序列化器的方法是通过 @[from_pycryptodome_aes，它从 LANGGRAPH_AES_KEY 环境变量读取 AES 密钥（或接受 key 参数）：

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.

构建自定义检查点

本节介绍实现 BaseCheckpointSaver 从头开始为自定义存储后端。如果您已有可工作的检查点且仅需要添加增量通道支持，请跳至 增量通道支持.

概述

LangGraph 的持久层构建在两个存储抽象之上：

• 检查点表 — 每个超步一行；存储序列化的图状态（channel_values, channel_versions, versions_seen）并链接到其父检查点。
• 写入表 — 每个超步内的节点输出一行；存储 (task_id, channel, value) 链接到检查点的元组。

您的检查点管理这两个表。 put 写入检查点行； put_writes 写入节点输出行； get_tuple 将两者读回 CheckpointTuple.

基础契约

子类化 BaseCheckpointSaver 并实现这五个方法。所有方法都是必需的——缺少基本方法会在运行时引发 NotImplementedError 错误。

from collections.abc import AsyncIterator, Iterator, Sequence
from typing import Any
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import (
    BaseCheckpointSaver,
    ChannelVersions,
    Checkpoint,
    CheckpointMetadata,
    CheckpointTuple,
)

class MyCheckpointer(BaseCheckpointSaver):
    async def aput(
        self,
        config: RunnableConfig,
        checkpoint: Checkpoint,
        metadata: CheckpointMetadata,
        new_versions: ChannelVersions,
    ) -> RunnableConfig:
        ...

    async def aput_writes(
        self,
        config: RunnableConfig,
        writes: Sequence[tuple[str, Any]],
        task_id: str,
        task_path: str = "",
    ) -> None:
        ...

    async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
        ...

    async def alist(
        self,
        config: RunnableConfig | None,
        *,
        filter: dict[str, Any] | None = None,
        before: RunnableConfig | None = None,
        limit: int | None = None,
    ) -> AsyncIterator[CheckpointTuple]:
        ...
        yield  # make this an async generator

    async def adelete_thread(self, thread_id: str) -> None:
        ...

put / aput

存储一个检查点行。返回包含已存储内容的更新配置 checkpoint_id.

关键要求：

• - 使用以下方式序列化检查点 self.serde.dumps_typed(checkpoint) — 这会处理所有 LangGraph 原生类型，包括 _DeltaSnapshot 增量通道使用的 blob。
• - 存储 metadata 完整存储 — 不要剥离未知键。LangGraph 在次要版本中添加新的元数据字段（例如 counters_since_delta_snapshot 用于增量通道）；在次要版本中静默丢弃这些字段会破坏功能。
• - 存储 config["configurable"].get("checkpoint_id") 作为父检查点 ID，以便 get_tuple 可以填充 parent_config.

async def aput(self, config, checkpoint, metadata, new_versions):
    thread_id = config["configurable"]["thread_id"]
    checkpoint_ns = config["configurable"]["checkpoint_ns"]
    checkpoint_id = checkpoint["id"]
    parent_id = config["configurable"].get("checkpoint_id")

    type_, blob = self.serde.dumps_typed(checkpoint)
    serialized_metadata = self.serde.dumps_typed(metadata)

    await self.db.execute(
        "INSERT INTO checkpoints (...) VALUES (...)",
        thread_id, checkpoint_ns, checkpoint_id, parent_id,
        type_, blob, *serialized_metadata,
    )
    return {
        "configurable": {
            "thread_id": thread_id,
            "checkpoint_ns": checkpoint_ns,
            "checkpoint_id": checkpoint_id,
        }
    }

put_writes / aput_写入

为当前超步内的单个任务存储节点输出行。这些行通过以下方式链接到检查点 (thread_id, checkpoint_ns, checkpoint_id).

async def aput_writes(self, config, writes, task_id, task_path=""):
    thread_id = config["configurable"]["thread_id"]
    checkpoint_ns = config["configurable"]["checkpoint_ns"]
    checkpoint_id = config["configurable"]["checkpoint_id"]

    rows = []
    for idx, (channel, value) in enumerate(writes):
        type_, blob = self.serde.dumps_typed(value)
        final_idx = WRITES_IDX_MAP.get(channel, idx)
        rows.append((thread_id, checkpoint_ns, checkpoint_id,
                      task_id, task_path, final_idx, channel, type_, blob))

    await self.db.executemany("INSERT INTO writes (...) VALUES (...)", rows)

导入 WRITES_IDX_MAP 从 langgraph.checkpoint.base。它将特殊通道（__error__, __interrupt__等）映射到保留的负索引，以免与常规写入索引冲突。

get_tuple / aget_元组

检索检查点。配置可能包含：

• - **No checkpoint_id** — 返回该线程 + 命名空间的最新检查点。
• - **一个特定的 checkpoint_id** — 返回该确切检查点。

两条路径都必须正常工作。 specific-id 路径用于时间旅行，并且——关键的是——用于在每次图调用时重建增量通道状态（见 增量通道支持）。损坏的 specific-id 查询会静默破坏增量通道状态。

async def aget_tuple(self, config):
    thread_id = config["configurable"]["thread_id"]
    checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
    checkpoint_id = config["configurable"].get("checkpoint_id")

    if checkpoint_id:
        row = await self.db.fetchone(
            "SELECT * FROM checkpoints "
            "WHERE thread_id=? AND checkpoint_ns=? AND checkpoint_id=?",
            thread_id, checkpoint_ns, checkpoint_id,
        )
    else:
        row = await self.db.fetchone(
            "SELECT * FROM checkpoints "
            "WHERE thread_id=? AND checkpoint_ns=? "
            "ORDER BY checkpoint_id DESC LIMIT 1",
            thread_id, checkpoint_ns,
        )

    if row is None:
        return None

    writes = await self.db.fetchall(
        "SELECT task_id, channel, type, value FROM writes "
        "WHERE thread_id=? AND checkpoint_ns=? AND checkpoint_id=? "
        "ORDER BY task_id, idx",
        thread_id, checkpoint_ns, row["checkpoint_id"],
    )
    pending_writes = [
        (w["task_id"], w["channel"], self.serde.loads_typed((w["type"], w["value"])))
        for w in writes
    ]

    checkpoint = self.serde.loads_typed((row["type"], row["blob"]))
    metadata = self.serde.loads_typed((row["metadata_type"], row["metadata"]))

    parent_config = None
    if row["parent_checkpoint_id"]:
        parent_config = {
            "configurable": {
                "thread_id": thread_id,
                "checkpoint_ns": checkpoint_ns,
                "checkpoint_id": row["parent_checkpoint_id"],
            }
        }

    return CheckpointTuple(
        config={
            "configurable": {
                "thread_id": thread_id,
                "checkpoint_ns": checkpoint_ns,
                "checkpoint_id": row["checkpoint_id"],
            }
        },
        checkpoint=checkpoint,
        metadata=metadata,
        parent_config=parent_config,
        pending_writes=pending_writes,
    )

list / alist

返回线程的检查点，最新的在前。遵循 before （仅返回比该配置的 checkpoint_id更旧的检查点）以及 limit.

删除_thread / adelete_线程

删除线程的所有检查点和写入。检查点行和写入行都必须删除。

Row key / index design

存储和索引检查点的方式会直接影响正确性和性能。

推荐模式（SQL）：

CREATE TABLE checkpoints (
    thread_id          TEXT NOT NULL,
    checkpoint_ns      TEXT NOT NULL DEFAULT '',
    checkpoint_id      TEXT NOT NULL,   -- ULID, lexicographically sortable newest-last
    parent_checkpoint_id TEXT,
    type               TEXT,
    checkpoint         BYTEA,
    metadata           JSONB,
    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
);

CREATE TABLE writes (
    thread_id     TEXT NOT NULL,
    checkpoint_ns TEXT NOT NULL DEFAULT '',
    checkpoint_id TEXT NOT NULL,
    task_id       TEXT NOT NULL,
    task_path     TEXT NOT NULL DEFAULT '',
    idx           INTEGER NOT NULL,
    channel       TEXT NOT NULL,
    type          TEXT,
    value         BYTEA,
    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx)
);

因为 checkpoint_id 是一个 ULID，它按字典序排序——值越大越新。"获取最新"是 ORDER BY checkpoint_id DESC LIMIT 1；"按 id 获取"是对主键的等值查询。

对于非 SQL 存储： 同样的原则也适用。无论使用什么键方案，按 (thread_id, checkpoint_ns, checkpoint_id) 的直接查找必须是 O(1) 或接近 O(1)。避免使用那种只能通过扫描线程的所有行来按 id 查找检查点的设计。

序列化

始终使用 self.serde （继承自 BaseCheckpointSaver，默认为 JsonPlusSerializer)，用于检查点、写入和元数据。请勿使用 pickle 直接用于元数据——它可以工作，但 JsonPlusSerializer 生成人类可读的输出，并更好地处理版本控制。

JsonPlusSerializer 自动处理所有 LangGraph 原生类型： - _DeltaSnapshot — 增量通道使用的哨兵 blob（msgpack 扩展代码 7） - Pydantic v2 模型、数据类、numpy 数组、日期时间、枚举等

如果您编写自定义序列化器，请确保它可以往返转换 _DeltaSnapshot 从 langgraph.checkpoint.serde.types.

扩展功能

这些方法是可选的，但可以解锁额外的 Agent Server 功能。如果您的存储后端能够高效支持，则实现它们。

Agent Server 在启动时自动检测您的检查点实现支持哪些功能，并激活相应的特性。

增量通道支持

DeltaChannel 是一个归约器通道，它仅在检查点 blob 中存储一个哨兵 (MISSING)，而不是完整的通道值。状态通过重放祖先写入来通过归约器重建。这使得检查点 blob 每个步骤为 O(1)，而不是像 messages 那样随时间累积的 O(N)。

运行时需求

当加载一个其增量通道在 channel_values中缺失的检查点时，LangGraph 调用 saver.get_delta_channel_history(config=config, channels=[...])。这返回每个通道的：

• - **writes** — 祖先链中对该通道的所有写入，按最旧优先排列，直到最近的快照。
• - **seed** （可选）— 最近祖先处存储的 _DeltaSnapshot blob；如果遍历到根仍未找到快照则不存在。

然后运行时调用 channel.from_checkpoint(seed) 和 channel.replay_writes(writes) 来重建实时值。

默认实现

BaseCheckpointSaver 提供了一个默认的 get_delta_channel_history ，适用于任何正确的 get_tuple implementation:

# Simplified from BaseCheckpointSaver
def get_delta_channel_history(self, *, config, channels):
    target = self.get_tuple(config)          # load the head checkpoint
    cursor = target.parent_config            # walk from its parent
    collected = {ch: [] for ch in channels}
    seed = {}
    remaining = set(channels)

    while cursor and remaining:
        tup = self.get_tuple(cursor)         # ← requires correct by-id lookup
        if tup is None:
            break
        for write in reversed(tup.pending_writes or []):
            if write[1] in remaining:
                collected[write[1]].append(write)
        for ch in list(remaining):
            if ch in tup.checkpoint["channel_values"]:
                seed[ch] = tup.checkpoint["channel_values"][ch]
                remaining.discard(ch)
        cursor = tup.parent_config

    return {
        ch: {"writes": list(reversed(collected[ch])), **({"seed": seed[ch]} if ch in seed else {})}
        for ch in channels
    }

关键依赖： get_tuple(cursor) 始终使用特定的 checkpoint_id （父级的 id）调用。如果该查找返回 None，则遍历立即停止，每个增量通道重建为空——静默地，没有错误。这就是为什么 get_tuple 中的特定 id 路径必须正确。

性能覆盖

默认遍历为每个祖先检查点发出一 get_tuple 次调用。对于具有良好查询支持的后端，覆盖 get_delta_channel_history （及其异步版本）以两次查询检索祖先链和写入：

async def aget_delta_channel_history(self, *, config, channels):
    if not channels:
        return {}

    thread_id = config["configurable"]["thread_id"]
    checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
    checkpoint_id = config["configurable"]["checkpoint_id"]

    # Stage 1: stream ancestors newest-first until every channel has a seed
    ancestors = await self.db.fetchall(
        "SELECT checkpoint_id, parent_checkpoint_id, type, checkpoint "
        "FROM checkpoints "
        "WHERE thread_id=? AND checkpoint_ns=? AND checkpoint_id < ? "
        "ORDER BY checkpoint_id DESC",
        thread_id, checkpoint_ns, checkpoint_id,
    )

    chain_by_ch: dict[str, list[str]] = {ch: [] for ch in channels}
    seed_by_ch: dict[str, Any] = {}
    remaining = set(channels)
    cur_id = config["configurable"]["checkpoint_id"]

    for row in ancestors:
        if not remaining:
            break
        parent_id = row["parent_checkpoint_id"]
        ckpt = self.serde.loads_typed((row["type"], row["checkpoint"]))
        cv = ckpt.get("channel_values") or {}
        for ch in list(remaining):
            chain_by_ch[ch].append(row["checkpoint_id"])
            if ch in cv:
                seed_by_ch[ch] = cv[ch]
                remaining.discard(ch)
        cur_id = parent_id

    # Stage 2: fetch writes for each channel's ancestor chain in one query
    result: dict[str, DeltaChannelHistory] = {}
    for ch in channels:
        chain = chain_by_ch[ch]
        if not chain:
            entry: DeltaChannelHistory = {"writes": []}
            if ch in seed_by_ch:
                entry["seed"] = seed_by_ch[ch]
            result[ch] = entry
            continue

        write_rows = await self.db.fetchall(
            f"SELECT checkpoint_id, task_id, idx, type, value FROM writes "
            f"WHERE thread_id=? AND checkpoint_ns=? AND channel=? "
            f"AND checkpoint_id IN ({','.join('?' * len(chain))})"
            f"ORDER BY checkpoint_id, task_id, idx",
            thread_id, checkpoint_ns, ch, *chain,
        )
        writes_by_cid: dict[str, list[PendingWrite]] = {}
        for row in write_rows:
            cid = row["checkpoint_id"]
            value = self.serde.loads_typed((row["type"], row["value"]))
            writes_by_cid.setdefault(cid, []).append((row["task_id"], ch, value))

        # chain is newest-first; iterate oldest-first to get correct replay order
        collected: list[PendingWrite] = []
        for cid in reversed(chain):
            collected.extend(writes_by_cid.get(cid, []))

        entry = {"writes": collected}
        if ch in seed_by_ch:
            entry["seed"] = seed_by_ch[ch]
        result[ch] = entry

    return result

使用增量通道进行修剪

DeltaChannel 状态不在单个检查点中是自包含的——它依赖于到最近的 _DeltaSnapshot的祖先写入链。如果您实现 prune or delete_for_runs，则不得删除存活检查点的增量通道所依赖的写入行。

安全选项：

1. **修剪前遍历** — 对于每个要保留的检查点，遍历其祖先链并标记到最近的 _DeltaSnapshot 为止的所有写入行为不可删除。
2. **修剪前强制生成快照** — 在您保留的检查点上重写 channel_values[ch] = _DeltaSnapshot(reconstructed_value) ，然后自由删除祖先。
3. **跳过增量通道线程的修剪** — 如果您尚不需要修剪，这是最安全的短期选项。

复制带有增量通道的线程

在实现 copy_thread时，复制完整的祖先链——而不仅仅是头检查点。目标线程必须具有至少回溯到一个 _DeltaSnapshot 的写入行用于每个增量通道，否则这些通道在复制后将重建为空。

使用一致性套件进行测试

langgraph-checkpoint-conformance 根据完整合约验证您的实现，包括增量通道历史记录：

pip install langgraph-checkpoint-conformance

from langgraph.checkpoint.conformance import checkpointer_test, validate

@checkpointer_test(name="MyCheckpointer")
async def my_checkpointer():
    async with MyCheckpointer.create() as saver:
        yield saver

async def main():
    report = await validate(my_checkpointer)
    report.print_report()
    # Fails the process if any base capability is missing or broken
    if not report.passed_all_base():
        raise RuntimeError("Checkpointer failed conformance suite")

asyncio.run(main())

该套件自动检测您的检查点实现实现了哪些扩展功能（包括 aget_delta_channel_history）并为每个功能运行相关测试。在发布前将其作为 CI 的一部分运行。