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

!检查点

为什么要使用检查点器

以下功能需要检查点器：

- **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 检查点，因此时间旅行从超级步骤边界的完整检查点恢复。

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

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

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

检查点命名空间

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

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

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

function myNode(state: typeof State.Type, config: RunnableConfig) {
  const checkpointNs = config.configurable?.checkpoint_ns;
  // "" for the parent graph, "node_name:uuid" for a subgraph
}

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

获取和更新状态

获取状态

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

// get the latest state snapshot
const config = { configurable: { thread_id: "1" } };
await graph.getState(config);

// get a state snapshot for a specific 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 字段

获取状态历史

你可以通过调用 graph.getStateHistory(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 列表，其中

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

// Find the checkpoint before a specific node executed
const beforeNodeB = history.find((s) => s.next.includes("nodeB"));

// Find a checkpoint by step number
const step2 = history.find((s) => s.metadata.step === 2);

// Find checkpoints created by updateState
const forks = history.filter((s) => s.metadata.source === "update");

// Find the checkpoint where an interrupt occurred
const interrupted = history.find(
  (s) => s.tasks.length > 0 && s.tasks.some((t) => t.interrupts.length > 0)
);

重放

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

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

!重放

更新状态

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

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

!更新

持久性模式

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

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

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

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

优化检查点存储

Checkpointer 库

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

• * @langchain/langgraph-checkpoint：检查点保存器的基础接口（BaseCheckpointSaver) and serialization/deserialization interface (SerializerProtocol）。包含内存中检查点实现（MemorySaver）用于实验。LangGraph 自带 @langchain/langgraph-checkpoint included.
• * @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）。这用于填充 StateSnapshot in graph.getState().
• * .list - 列出与给定配置和过滤条件匹配的检查点。这用于填充状态历史记录 graph.getStateHistory()

构建自定义检查点