检查点器

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

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

使用 LangSmith 跟踪已检查点的状态，并调试您的代理如何跨会话恢复。请按照跟踪快速入门进行设置。

为什么使用检查点器

以下功能需要检查点器：

人机交互：检查点器通过允许人类检查、中断和批准图步骤来促进人机交互工作流。这些工作流需要检查点器，因为人必须能够随时查看图的状态，并且在人对状态进行任何更新后，图必须能够恢复执行。有关示例，请参阅中断。
记忆：检查点器允许交互之间存在”记忆”。在重复的人类交互（如对话）的情况下，任何后续消息都可以发送到该线程，该线程将保留对先前消息的记忆。有关如何使用检查点器添加和管理对话记忆的信息，请参阅添加记忆。
时间旅行：检查点器允许进行”时间旅行”，允许用户重放先前的图执行以审查和/或调试特定的图步骤。此外，检查点器使得可以在任意检查点分叉图状态以探索替代轨迹成为可能。
容错性：检查点提供容错和错误恢复：如果在给定的超级步骤中一个或多个节点失败，您可以从上一个成功的步骤重新启动您的图。

待处理写入：当一个图节点在给定超级步骤执行过程中失败时，LangGraph 会存储该超级步骤中成功完成的其他任何节点的待处理检查点写入。当您从该超级步骤恢复图执行时，您不需要重新运行成功的节点。

核心概念

线程

线程是分配给检查点器保存的每个检查点的唯一 ID 或线程标识符。它包含一系列运行的累积状态。当执行一次运行时，助手底层图的状态将被持久化到该线程中。使用检查点器调用图时，您必须在配置的 configurable 部分指定一个 thread_id：

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

可以检索线程的当前和历史状态。为了持久化状态，必须在执行运行之前创建线程。LangSmith API 提供了几个端点用于创建和管理线程和线程状态。有关更多详细信息，请参阅 API 参考。检查点器使用 thread_id 作为存储和检索检查点的主键。没有它，检查点器无法保存状态或在中断后恢复执行，因为检查点器使用 thread_id 来加载已保存的状态。

检查点

线程在特定时间点的状态称为检查点。检查点是在每个超级步骤保存的图状态快照，由 StateSnapshot 对象表示（有关完整字段参考，请参阅 StateSnapshot 字段）。

超级步骤

LangGraph 在每个超级步骤边界创建一个检查点。超级步骤是图的一次单独“滴答”，其中计划用于该步骤的所有节点（可能并行）执行。对于像 START -> A -> B -> END 这样的顺序图，输入、节点 A 和节点 B 分别有单独的超级步骤——并在每个之后产生一个检查点。理解超级步骤边界对于时间旅行很重要，因为您只能从检查点（即超级步骤边界）恢复执行。除了超级步骤检查点，LangGraph 还在节点（任务）级别持久化写入。当超级步骤中的每个节点完成时，其输出作为任务条目写入检查点器的 checkpoint_writes 表，并链接到进行中的检查点。这些每任务写入是待处理写入恢复的关键：如果同一超级步骤中的另一个节点失败，成功节点的写入已经持久化，在恢复时无需重新运行。然后，一旦超级步骤完成，就提交完整的状态快照。 LangGraph 还持久化超级步骤内单个节点执行的写入。这些写入作为任务存储，并用于容错：如果同一超级步骤中的另一个节点失败，当您恢复时，成功的节点写入不需要重新计算。这些任务写入不是完整的 StateSnapshot 检查点，因此时间旅行从超级步骤边界的完整检查点恢复。检查点是持久化的，可用于在稍后时间恢复线程的状态。让我们看看当调用如下简单图时会保存哪些检查点：

import { StateGraph, StateSchema, ReducedValue, START, END, MemorySaver } from "@langchain/langgraph";
import { z } from "zod/v4";

const State = new StateSchema({
  foo: z.string(),
  bar: new ReducedValue(
    z.array(z.string()).default(() => []),
    {
      inputSchema: z.array(z.string()),
      reducer: (x, y) => x.concat(y),
    }
  ),
});

const workflow = new StateGraph(State)
  .addNode("nodeA", (state) => {
    return { foo: "a", bar: ["a"] };
  })
  .addNode("nodeB", (state) => {
    return { foo: "b", bar: ["b"] };
  })
  .addEdge(START, "nodeA")
  .addEdge("nodeA", "nodeB")
  .addEdge("nodeB", END);

const checkpointer = new MemorySaver();
const graph = workflow.compile({ checkpointer });

const config = { configurable: { thread_id: "1" } };
await graph.invoke({ foo: "", bar: [] }, config);

运行图后，将恰好有 4 个检查点：

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

请注意，bar 通道的值包含来自两个节点的输出，因为此示例中 bar 通道有一个归约器。

检查点命名空间

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

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

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

import { RunnableConfig } from "@langchain/core/runnables";

function myNode(state: typeof State.Type, config: RunnableConfig) {
  const checkpointNs = config.configurable?.checkpoint_ns;
  // 父图返回 ""，子图返回 "node_name:uuid"
}

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

获取和更新状态

获取状态

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

// 获取最新的状态快照
const config = { configurable: { thread_id: "1" } };
await graph.getState(config);

// 获取特定检查点 ID 的状态快照
const config = {
  configurable: {
    thread_id: "1",
    checkpoint_id: "1ef663ba-28fe-6528-8002-5a559208592c",
  },
};
await graph.getState(config);

在此示例中，getState 的输出将如下所示：

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: { nodeB: { foo: 'b', bar: ['b'] } },
    step: 2
  },
  createdAt: '2024-08-29T19:19:38.821749+00:00',
  parentConfig: {
    configurable: {
      thread_id: '1',
      checkpoint_ns: '',
      checkpoint_id: '1ef663ba-28f9-6ec4-8001-31981c2c39f8'
    }
  },
  tasks: []
}

StateSnapshot 字段

字段	类型	描述
`values`	`object`	此检查点的状态通道值。
`next`	`string[]`	下一个要执行的节点名称。空的 `[]` 表示图已完成。
`config`	`object`	包含 `thread_id`、`checkpoint_ns` 和 `checkpoint_id`。
`metadata`	`object`	执行元数据。包含 `source`（`"input"`、`"loop"` 或 `"update"`）、`writes`（节点输出）和 `step`（超级步骤计数器）。
`createdAt`	`string`	此检查点创建时的 ISO 8601 时间戳。
`parentConfig`	`object \| null`	前一个检查点的配置。第一个检查点为 `null`。
`tasks`	`PregelTask[]`	此步骤要执行的任务。每个任务都有 `id`、`name`、`error`、`interrupts`，以及可选的 `state`（子图快照，当使用 `subgraphs: true` 时）。

获取状态历史

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

const config = { configurable: { thread_id: "1" } };
for await (const state of graph.getStateHistory(config)) {
  console.log(state);
}

在此示例中，getStateHistory 的输出将如下所示：

[
  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: { nodeB: { foo: 'b', bar: ['b'] } },
      step: 2
    },
    createdAt: '2024-08-29T19:19:38.821749+00:00',
    parentConfig: {
      configurable: {
        thread_id: '1',
        checkpoint_ns: '',
        checkpoint_id: '1ef663ba-28f9-6ec4-8001-31981c2c39f8'
      }
    },
    tasks: []
  },
  StateSnapshot {
    values: { foo: 'a', bar: ['a'] },
    next: ['nodeB'],
    config: {
      configurable: {
        thread_id: '1',
        checkpoint_ns: '',
        checkpoint_id: '1ef663ba-28f9-6ec4-8001-31981c2c39f8'
      }
    },
    metadata: {
      source: 'loop',
      writes: { nodeA: { foo: 'a', bar: ['a'] } },
      step: 1
    },
    createdAt: '2024-08-29T19:19:38.819946+00:00',
    parentConfig: {
      configurable: {
        thread_id: '1',
        checkpoint_ns: '',
        checkpoint_id: '1ef663ba-28f4-6b4a-8000-ca575a13d36a'
      }
    },
    tasks: [
      PregelTask {
        id: '6fb7314f-f114-5413-a1f3-d37dfe98ff44',
        name: 'nodeB',
        error: null,
        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: null,
      step: 0
    },
    createdAt: '2024-08-29T19:19:38.817813+00:00',
    parentConfig: {
      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: null,
        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
    },
    createdAt: '2024-08-29T19:19:38.816205+00:00',
    parentConfig: null,
    tasks: [
      PregelTask {
        id: '6d27aa2e-d72b-5504-a36f-8620e54a76dd',
        name: '__start__',
        error: null,
        interrupts: []
      }
    ]
  }
]

查找特定检查点

您可以过滤状态历史记录以查找匹配特定条件的检查点：

const history: StateSnapshot[] = [];
for await (const state of graph.getStateHistory(config)) {
  history.push(state);
}

// 查找特定节点执行前的检查点
const beforeNodeB = history.find((s) => s.next.includes("nodeB"));

// 按步骤号查找检查点
const step2 = history.find((s) => s.metadata.step === 2);

// 查找由 updateState 创建的检查点
const forks = history.filter((s) => s.metadata.source === "update");

// 查找发生中断的检查点
const interrupted = history.find(
  (s) => s.tasks.length > 0 && s.tasks.some((t) => t.interrupts.length > 0)
);

重放

重放会重新执行来自先前检查点的步骤。使用先前的 checkpoint_id 调用图，以重新运行该检查点之后的节点。检查点之前的节点将被跳过（其结果已保存）。检查点之后的节点将重新执行，包括任何 LLM 调用、API 请求或中断——在重放期间这些中断总是会重新触发。有关重放过去执行的完整详细信息和代码示例，请参阅时间旅行。

更新状态

您可以使用 graph.updateState() 编辑图状态。这会创建一个带有更新值的新检查点——它不会修改原始检查点。该更新被视为与节点更新相同：如果定义了归约器函数，值会通过归约器函数传递，因此带有归约器的通道会_累积_值而不是覆盖它们。您可以选择指定 asNode 来控制该更新被视为来自哪个节点，这会影响接下来哪个节点执行。有关详细信息，请参阅时间旅行：asNode。

持久性模式

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

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

持久性模式，从最不持久到最持久，如下所示：

"exit"：LangGraph 仅在图执行退出时（成功、出错或由于人机交互中断）持久化更改。这为长时间运行的图提供了最佳性能，但意味着中间状态不会被保存，因此您无法从系统故障（如进程崩溃）中恢复执行中的状态。
"async"：LangGraph 在下一步执行时异步持久化更改。这提供了良好的性能和持久性，但如果进程在执行期间崩溃，存在 LangGraph 不写入检查点的微小风险。
"sync"：LangGraph 在下一步开始之前同步持久化更改。这确保了 LangGraph 在继续执行之前写入每个检查点，以一些性能开销为代价提供了高持久性。

优化检查点存储

检查点器库

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

@langchain/langgraph-checkpoint：检查点器保存器的基础接口 (BaseCheckpointSaver) 以及序列化/反序列化接口 (SerializerProtocol)。包含用于实验的内存检查点器实现 (MemorySaver)。LangGraph 自带 @langchain/langgraph-checkpoint。
@langchain/langgraph-checkpoint-sqlite：使用 SQLite 数据库 (SqliteSaver) 的 LangGraph 检查点器实现。适用于实验和本地工作流。需要单独安装。
@langchain/langgraph-checkpoint-postgres：使用 Postgres 数据库 (PostgresSaver) 的高级检查点器，在 LangSmith 中使用。适用于在生产环境中使用。需要单独安装。
@langchain/langgraph-checkpoint-mongodb：由 MongoDB 支持的高级检查点器 (MongoDBSaver) 和长期记忆存储 (MongoDBStore)。该存储支持跨线程持久化，并具有可选的集成向量搜索。适用于生产环境。需要单独安装。
@langchain/langgraph-checkpoint-redis：使用 Redis 数据库 (RedisSaver) 的高级检查点器。适用于在生产环境中使用。需要单独安装。

检查点器接口

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

.put - 使用其配置和元数据存储一个检查点。
.putWrites - 存储链接到检查点的中间写入（即待处理写入）。
.getTuple - 使用给定配置（thread_id 和 checkpoint_id）获取检查点元组。用于在 graph.getState() 中填充 StateSnapshot。
.list - 列出匹配给定配置和过滤条件的检查点。用于在 graph.getStateHistory() 中填充状态历史。

构建自定义检查点器

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

Get started

Capabilities

Production

Frontend

LangGraph APIs

为什么使用检查点器

核心概念

线程

检查点

超级步骤

检查点命名空间

获取和更新状态

获取状态

StateSnapshot 字段

获取状态历史

查找特定检查点

重放

更新状态

持久性模式

优化检查点存储

检查点器库

检查点器接口

构建自定义检查点器

​为什么使用检查点器

​核心概念

​线程

​检查点

​超级步骤

​检查点命名空间

​获取和更新状态

​获取状态

​StateSnapshot 字段

​获取状态历史

​查找特定检查点

​重放

​更新状态

​持久性模式

​优化检查点存储

​检查点器库

​检查点器接口

​构建自定义检查点器

为什么使用检查点器

核心概念

线程

检查点

超级步骤

检查点命名空间

获取和更新状态

获取状态

StateSnapshot 字段

获取状态历史

查找特定检查点

重放

更新状态

持久性模式

优化检查点存储

检查点器库

检查点器接口

构建自定义检查点器