BaseCheckpointSaver（base/__init__.py:176）定义了一组方法，其中四个是核心、后端必须实现（基类里全是 raise NotImplementedError）：

另有 get（base:227）是 get_tuple 的便捷包装、get_next_version（base:692）生成版本号，以及每个方法的 a 前缀异步版（aget_tuple/aput/...）。

基类里它只是一句"你得自己实现"，但签名和文档就是契约 base/__init__.py:239：

# base/__init__.py:239
def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
    """Fetch a checkpoint tuple using the given configuration.
    Returns: The requested checkpoint tuple, or `None` if not found.
    Raises: NotImplementedError: Implement this method in your custom checkpoint saver.
    """
    raise NotImplementedError

# base/__init__.py:227 —— get 是它的便捷包装
def get(self, config: RunnableConfig) -> Checkpoint | None:
    if value := self.get_tuple(config):
        return value.checkpoint

base/__init__.py:277：

# base/__init__.py:277
def put(
    self,
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
    new_versions: ChannelVersions,     # ← 本次写入相比上一档，哪些通道涨了版本
) -> RunnableConfig:
    """Store a checkpoint with its configuration and metadata.
    Returns: RunnableConfig: Updated configuration after storing the checkpoint.
    """
    raise NotImplementedError

base/__init__.py:300：

# base/__init__.py:300
def put_writes(
    self,
    config: RunnableConfig,
    writes: Sequence[tuple[str, Any]],   # [(channel, value), ...] 这个任务产生的写
    task_id: str,                        # 哪个任务写的
    task_path: str = "",                 # 任务在图中的路径（子图/Send 时区分）
) -> None:
    """Store intermediate writes linked to a checkpoint."""
    raise NotImplementedError

相关的还有个常量 base/__init__.py:795，它定义了几种"特殊写"的负数索引：

# base/__init__.py:795
WRITES_IDX_MAP = {ERROR: -1, SCHEDULED: -2, INTERRUPT: -3, RESUME: -4}
# 普通写按它在 writes 列表里的下标（0,1,2...）；
# 特殊写（错误、已调度、中断、恢复）映射到负数下标，避免和普通写撞键。

base/__init__.py:253：

# base/__init__.py:253
def list(
    self,
    config: RunnableConfig | None,
    *,
    filter: dict[str, Any] | None = None,   # 按 metadata 过滤（如 {"source":"loop"}）
    before: RunnableConfig | None = None,    # 只列早于某档的（分页/翻历史）
    limit: int | None = None,                # 最多返回几份
) -> Iterator[CheckpointTuple]:
    """List checkpoints that match the given criteria."""
    raise NotImplementedError

这是基类里唯一有默认实现的核心方法 base/__init__.py:692——因为版本递增有个通用逻辑：

# base/__init__.py:692
def get_next_version(self, current: V | None, channel: None) -> V:
    """Generate the next version ID for a channel.
    Default is to use integer versions, incrementing by `1`."""
    if isinstance(current, str):
        raise NotImplementedError          # 字符串版本得子类自己实现
    elif current is None:
        return 1                           # 第一次：从 1 开始
    else:
        return current + 1                 # 之后：+1

你会发现每个方法都有个 a 开头的孪生兄弟：aget_tuple（base:429）、aput（base:468）、aput_writes（base:491）、alist（base:443）。基类文档里专门提醒：

# base/__init__.py:204 (类文档节选)
# Note: When creating a custom checkpoint saver, consider implementing async
#       versions to avoid blocking the main thread.

# base/__init__.py:443 —— 注意 alist 这个空 yield 小技巧
async def alist(self, config, *, filter=None, before=None, limit=None):
    raise NotImplementedError
    yield                    # ← 让函数在语法上成为 async generator

✋ 10 分钟动手

# 1. 通读四个核心方法的签名与文档
sed -n '227,318p' libs/checkpoint/langgraph/checkpoint/base/__init__.py

# 2. 看 get_next_version 默认实现 + WRITES_IDX_MAP
sed -n '692,712p' libs/checkpoint/langgraph/checkpoint/base/__init__.py
sed -n '788,808p' libs/checkpoint/langgraph/checkpoint/base/__init__.py

# 3. 观察 list/before 分页效果
python - <<'PY'
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
g=StateGraph(dict); g.add_node("a",lambda s:{"n":s.get("n",0)+1})
g.add_edge(START,"a"); g.add_edge("a",END)
app=g.compile(checkpointer=InMemorySaver()); cfg={"configurable":{"thread_id":"t"}}
for _ in range(3): app.invoke({"n":0}, cfg)
for t in app.checkpointer.list(cfg): print(t.metadata.get("step"), t.config["configurable"]["checkpoint_id"][:8])
PY