检查点（Checkpointers）

LangGraph 检查点保存器在每个步骤将图状态保存为检查点，从而实现持久化、人机协同和容错执行。

检查点保存器（checkpointer）在每个超步（super-step）保存图状态的快照，并组织成线程（threads）。使用检查点保存器编译图，可启用人机协同工作流、时间旅行调试、容错执行以及对话记忆。

Checkpoints

INFO

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

TIP

使用 LangSmith 跟踪检查点状态并调试智能体如何跨会话恢复。按照跟踪快速入门进行设置。

为什么使用检查点保存器

以下特性需要使用检查点保存器：

人机协同（Human-in-the-loop）：检查点保存器通过允许人类检查、中断和批准图步骤来促进人机协同工作流。这些工作流需要检查点保存器，因为人员必须能够在任何时间点查看图的状态，并且图必须能够在人员对状态进行任何更新后恢复执行。有关示例，请参阅中断。
记忆（Memory）：检查点保存器支持交互之间的“记忆”。在重复的人类交互（如对话）情况下，任何后续消息都可以发送到该线程，该线程将保留对之前消息的记忆。有关如何使用检查点保存器添加和管理对话记忆的信息，请参阅添加记忆。
时间旅行（Time travel）：检查点保存器支持“时间旅行”，允许用户重放先前的图执行，以审查和/或调试特定的图步骤。此外，检查点保存器使得可以在任意检查点分支派生（fork）图状态，以探索替代轨迹。
容错（Fault-tolerance）：检查点提供容错和错误恢复：如果一个或多个节点在给定的超步失败，你可以从上一个成功的步骤重新启动你的图。
未决写入（Pending writes）：当图节点在给定的超步执行中途失败时，LangGraph 会存储在该超步成功完成的任何其他节点的未决检查点写入。当你从该超步恢复图执行时，无需重新运行成功的节点。

核心概念

线程（Threads）

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

当使用检查点保存器调用图时，你必须指定 thread_id 作为配置中 configurable 部分的一部分：

typescript

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

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

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

检查点（Checkpoints）

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

超步（Super-steps）

LangGraph 在每个超步边界创建检查点。超步是图的单个“时钟刻度（tick）”，其中为该步骤调度的所有节点同时执行（可能是并行的）。对于像 START -> A -> B -> END 这样的顺序图，输入、节点 A 和节点 B 分别有单独的超步——在每个超步之后都会生成一个检查点。理解超步边界对于时间旅行非常重要，因为你只能从检查点（即超步边界）恢复执行。

除了超步检查点之外，LangGraph 还在节点（任务）级别持久化写入。当超步内的每个节点完成时，其输出将作为链接到正在进行的检查点的任务条目写入检查点保存器的 checkpoint_writes 表。正是这些逐任务写入实现了未决写入恢复：如果同一超步中的另一个节点失败，成功节点的写入已经是持久的，并且在恢复时无需重新运行。一旦超步完成，完整的状态快照就会被提交。

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

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

让我们看看如下调用简单图时会保存哪些检查点：

typescript

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 通道设置了 reducer。

检查点命名空间（Checkpoint namespace）

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

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

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

typescript

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

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

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

获取与更新状态

获取状态（Get state）

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

typescript

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

// 获取特定 checkpoint_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` 时的子图快照）。

获取状态历史（Get state history）

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

typescript

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: []
      }
    ]
  }
]

State

查找特定检查点

你可以过滤状态历史记录以查找符合特定条件的检查点：

typescript

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)
);

重放（Replay）

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

有关重放先前执行的完整详细信息和代码示例，请参阅时间旅行。

Replay

更新状态（Update state）

你可以使用 graph.updateState() 编辑图状态。这会创建一个具有更新值的新检查点——它不会修改原始检查点。该更新与节点更新的处理方式相同：定义时值会通过 reducer 函数传递，因此带有 reducer 的通道会累加值而不是覆盖它们。

你可以选择指定 asNode 来控制更新被视为来自哪个节点，这会影响接下来执行的节点。有关详细信息，请参阅时间旅行：asNode。

Update

持久化级别模式（Durability modes）

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

typescript

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

从持久化程度最低到最高，模式如下：

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

优化检查点存储

检查点库生态（Checkpointer libraries）

在底层，检查点由符合 BaseCheckpointSaver 接口的检查点保存器对象驱动。LangGraph 提供了几种检查点保存器实现，全部通过独立的、可安装的库实现。

@langchain/langgraph-checkpoint: 检查点保存器的基本接口（BaseCheckpointSaver）和序列化/反序列化接口（SerializerProtocol）。包含用于实验的内存中检查点保存器实现（MemorySaver）。LangGraph 随附 @langchain/langgraph-checkpoint。
@langchain/langgraph-checkpoint-sqlite: 使用 SQLite 数据库的 LangGraph 检查点保存器实现（SqliteSaver）。非常适合实验和本地工作流。需要单独安装。
@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() 中的状态历史记录。

检查点（Checkpointers） ​

为什么使用检查点保存器 ​

核心概念 ​

线程（Threads） ​

检查点（Checkpoints） ​

超步（Super-steps） ​

检查点命名空间（Checkpoint namespace） ​

获取与更新状态 ​

获取状态（Get state） ​

StateSnapshot 字段 ​

获取状态历史（Get state history） ​

查找特定检查点 ​

重放（Replay） ​

更新状态（Update state） ​

持久化级别模式（Durability modes） ​

优化检查点存储 ​

检查点库生态（Checkpointer libraries） ​

检查点接口 ​

构建自定义检查点保存器 ​

检查点（Checkpointers）

为什么使用检查点保存器

核心概念

线程（Threads）

检查点（Checkpoints）

超步（Super-steps）

检查点命名空间（Checkpoint namespace）

获取与更新状态

获取状态（Get state）

StateSnapshot 字段

获取状态历史（Get state history）

查找特定检查点

重放（Replay）

更新状态（Update state）

持久化级别模式（Durability modes）

优化检查点存储

检查点库生态（Checkpointer libraries）

检查点接口

构建自定义检查点保存器