持久化

LangGraph 具有内置的持久化层，可将图状态保存为检查点。当您使用检查点保存器编译图时，图状态的快照会在执行的每一步保存，并组织成线程。这支持人机回环工作流、对话记忆、时间旅行调试和容错执行。

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

为何使用持久化

持久化是以下功能所必需的：

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

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

核心概念

线程

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

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

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

检查点

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

超级步骤

LangGraph 在每个超级步骤边界创建检查点。超级步骤是图的单个“滴答”，其中为该步骤计划的所有节点都会执行（可能并行）。对于像START -> A -> B -> END这样的顺序图，输入、节点 A 和节点 B 分别有单独的超级步骤——每个步骤后都会生成一个检查点。理解超级步骤边界对于时间旅行很重要，因为您只能从检查点（即超级步骤边界）恢复执行。检查点会被持久化，并可用于在以后恢复线程的状态。让我们看看在调用以下简单图时保存了哪些检查点：

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

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

获取状态历史

您可以通过调用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。

内存存储

状态模式指定了一组在图执行时填充的键。如上所述，状态可以由检查点保存器在每个图步骤写入线程，从而实现状态持久化。如果我们想跨线程保留某些信息怎么办？考虑一个聊天机器人的情况，我们希望在与该用户的所有聊天对话（即线程）中保留有关用户的特定信息！仅使用检查点保存器，我们无法跨线程共享信息。这就需要Store接口。作为说明，我们可以定义一个InMemoryStore来跨线程存储有关用户的信息。我们只需像以前一样使用检查点保存器编译图，并传递存储。

LangGraph API 自动处理存储 使用 LangGraph API 时，您无需手动实现或配置存储。API 会在后台为您处理所有存储基础设施。

InMemoryStore 适用于开发和测试。对于生产环境，请使用持久存储，如PostgresStore或RedisStore。所有实现都扩展了BaseStore，这是在节点函数签名中使用的类型注解。

基本用法

首先，让我们在不使用 LangGraph 的情况下单独展示这一点。

import { MemoryStore } from "@langchain/langgraph";

const memoryStore = new MemoryStore();

记忆按tuple命名空间划分，在此特定示例中将是(<user_id>, "memories")。命名空间可以是任意长度并表示任何内容，不一定特定于用户。

const userId = "1";
const namespaceForMemory = [userId, "memories"];

我们使用store.put方法将记忆保存到存储中的命名空间。执行此操作时，我们指定命名空间（如上定义）和记忆的键值对：键只是记忆的唯一标识符（memory_id），值（字典）是记忆本身。

import { v4 as uuidv4 } from "uuid";

const memoryId = uuidv4();
const memory = { food_preference: "I like pizza" };
await memoryStore.put(namespaceForMemory, memoryId, memory);

我们可以使用store.search方法读取命名空间中的记忆，这将返回给定用户的所有记忆作为列表。最近的记忆位于列表的末尾。

const memories = await memoryStore.search(namespaceForMemory);
memories[memories.length - 1];

// {
//   value: { food_preference: 'I like pizza' },
//   key: '07e0caf4-1631-47b7-b15f-65515d4c1843',
//   namespace: ['1', 'memories'],
//   createdAt: '2024-10-02T17:22:31.590602+00:00',
//   updatedAt: '2024-10-02T17:22:31.590605+00:00'
// }

它具有的属性：

value：此记忆的值
key：此命名空间中此记忆的唯一键
namespace：字符串元组，此记忆类型的命名空间
虽然类型是tuple，但在转换为 JSON 时可能序列化为列表（例如，['1', 'memories']）。
createdAt：此记忆创建时的时间戳
updatedAt：此记忆更新时的时间戳

语义搜索

除了简单检索外，存储还支持语义搜索，允许您根据含义而不是精确匹配查找记忆。为此，请使用嵌入模型配置存储：

import { OpenAIEmbeddings } from "@langchain/openai";

const store = new InMemoryStore({
  index: {
    embeddings: new OpenAIEmbeddings({ model: "text-embedding-3-small" }),
    dims: 1536,
    fields: ["food_preference", "$"], // 要嵌入的字段
  },
});

现在搜索时，您可以使用自然语言查询来查找相关记忆：

// 查找有关食物偏好的记忆
// （这可以在将记忆放入存储后完成）
const memories = await store.search(namespaceForMemory, {
  query: "What does the user like to eat?",
  limit: 3, // 返回前 3 个匹配项
});

您可以通过配置fields参数或在存储记忆时指定index参数来控制记忆的哪些部分被嵌入：

// 存储时指定要嵌入的特定字段
await store.put(
  namespaceForMemory,
  uuidv4(),
  {
    food_preference: "I love Italian cuisine",
    context: "Discussing dinner plans",
  },
  { index: ["food_preference"] } // 仅嵌入 "food_preferences" 字段
);

// 存储时不嵌入（仍可检索，但不可搜索）
await store.put(
  namespaceForMemory,
  uuidv4(),
  { system_info: "Last updated: 2024-01-01" },
  { index: false }
);

在 LangGraph 中使用

准备好这一切后，我们在 LangGraph 中使用memoryStore。memoryStore与检查点保存器协同工作：检查点保存器将状态保存到线程（如上所述），而memoryStore允许我们存储任意信息以供跨线程访问。我们如下编译图，同时使用检查点保存器和memoryStore。

import { MemorySaver } from "@langchain/langgraph";

// 我们需要这个，因为我们想要启用线程（对话）
const checkpointer = new MemorySaver();

// ... 定义图 ...

// 使用检查点保存器和存储编译图
const graph = workflow.compile({ checkpointer, store: memoryStore });

我们像以前一样使用thread_id调用图，并使用user_id，我们将使用它来为该特定用户命名空间化我们的记忆，如上所示。

// 调用图
const userId = "1";
const config = { configurable: { thread_id: "1" }, context: { userId } };

// 首先，我们只是向 AI 问好
for await (const update of await graph.stream(
  { messages: [{ role: "user", content: "hi" }] },
  { ...config, streamMode: "updates" }
)) {
  console.log(update);
}

您可以使用runtime参数在_任何节点_中访问存储和userId。以下是如何使用它来保存记忆：

import { StateSchema, MessagesValue, Runtime } from "@langchain/langgraph";
import { v4 as uuidv4 } from "uuid";

const MessagesState = new StateSchema({
  messages: MessagesValue,
});

const updateMemory: GraphNode<typeof MessagesState> = async (state, runtime) => {
  // 从配置中获取用户 ID
  const userId = runtime.context?.user_id;
  if (!userId) throw new Error("User ID is required");

  // 命名空间化记忆
  const namespace = [userId, "memories"];

  // ... 分析对话并创建新记忆
  const memory = "Some memory content";

  // 创建新的记忆 ID
  const memoryId = uuidv4();

  // 我们创建一个新记忆
  await runtime.store?.put(namespace, memoryId, { memory });
};

如上所示，我们也可以在任何节点中访问存储，并使用store.search方法获取记忆。回想一下，记忆作为对象列表返回，可以转换为字典。

memories[memories.length - 1];
// {
//   value: { food_preference: 'I like pizza' },
//   key: '07e0caf4-1631-47b7-b15f-65515d4c1843',
//   namespace: ['1', 'memories'],
//   createdAt: '2024-10-02T17:22:31.590602+00:00',
//   updatedAt: '2024-10-02T17:22:31.590605+00:00'
// }

我们可以访问记忆并在模型调用中使用它们。

const callModel: GraphNode<typeof MessagesState> = async (state, runtime) => {
  // 从配置中获取用户 ID
  const userId = runtime.context?.user_id;

  // 命名空间化记忆
  const namespace = [userId, "memories"];

  // 基于最新消息搜索
  const memories = await runtime.store?.search(namespace, {
    query: state.messages[state.messages.length - 1].content,
    limit: 3,
  });
  const info = memories.map((d) => d.value.memory).join("\n");

  // ... 在模型调用中使用记忆
};

如果我们创建一个新线程，只要user_id相同，我们仍然可以访问相同的记忆。

// 调用图
const config = { configurable: { thread_id: "2" }, context: { userId: "1" } };

// 再次问好
for await (const update of await graph.stream(
  { messages: [{ role: "user", content: "hi, tell me about my memories" }] },
  { ...config, streamMode: "updates" }
)) {
  console.log(update);
}

当我们使用 LangSmith 时，无论是本地（例如，在Studio中）还是由 LangSmith 托管，基础存储默认可用，无需在图编译期间指定。但是，要启用语义搜索，您确实需要在langgraph.json文件中配置索引设置。例如：

{
    ...
    "store": {
        "index": {
            "embed": "openai:text-embeddings-3-small",
            "dims": 1536,
            "fields": ["$"]
        }
    }
}

有关更多详细信息和配置选项，请参见部署指南。

检查点保存器库

在底层，检查点保存由符合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）。适用于生产环境。需要单独安装。
@langchain/langgraph-checkpoint-redis：使用 Redis 数据库的高级检查点保存器（RedisSaver）。适用于生产环境。需要单独安装。

检查点保存器接口

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

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

在 GitHub 上编辑此页面或提交问题。

通过 MCP 将这些文档连接到 Claude、VSCode 等以获取实时答案。

Get started

Capabilities

Production

Frontend

LangGraph APIs

为何使用持久化

核心概念

线程

检查点

超级步骤

检查点命名空间

获取和更新状态

获取状态

StateSnapshot 字段

获取状态历史

查找特定检查点

重放

更新状态

内存存储

基本用法

语义搜索

在 LangGraph 中使用

检查点保存器库

检查点保存器接口

Get started

Capabilities

Production

Frontend

LangGraph APIs

​为何使用持久化

​核心概念

​线程

​检查点

​超级步骤

​检查点命名空间

​获取和更新状态

​获取状态

​StateSnapshot 字段

​获取状态历史

​查找特定检查点

​重放

​更新状态

​内存存储

​基本用法

​语义搜索

​在 LangGraph 中使用

​检查点保存器库

​检查点保存器接口

为何使用持久化

核心概念

线程

检查点

超级步骤

检查点命名空间

获取和更新状态

获取状态

StateSnapshot 字段

获取状态历史

查找特定检查点

重放

更新状态

内存存储

基本用法

语义搜索

在 LangGraph 中使用

检查点保存器库

检查点保存器接口