持久性

LangGraph内置了一个持久化层，通过检查点器实现。当您使用检查点器编译一个图时，检查点器会在每个超级步骤保存图的 checkpoint 状态。这些检查点被保存到 thread 中，图执行后可以访问。因为 threads 允许在执行后访问图的状态，所以包括人机交互、记忆、时间旅行和容错在内的几个强大功能都是可能的。下面，我们将更详细地讨论这些概念。

LangGraph API 自动处理检查点 当使用 LangGraph API 时，您无需手动实现或配置检查点器。API 在幕后为您处理所有持久化基础设施。

线程

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

在调用带有检查点的图时，您必须在配置的 configurable 部分指定一个 thread_id。

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

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

检查点

线程在特定时间点的状态被称为检查点。检查点是每个超级步骤保存的图状态的快照，并由以下关键属性表示的 StateSnapshot 对象表示：

• config：与此检查点相关的配置。
• metadata：与此检查点相关的元数据。
• values：在此时间点状态通道的值。
• next：图下一步要执行的节点名称的元组。
• tasks：包含有关下一步要执行的任务信息的PregelTask对象的元组。如果该步骤之前已尝试执行，它将包含错误信息。如果图在节点内部被中断 动态，任务将包含与中断相关的附加数据。

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

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

import { StateGraph, START, END, MemoryServer } from "@langchain/langgraph";
import { registry } from "@langchain/langgraph/zod";
import * as z from "zod";

const State = z.object({
  foo: z.string(),
  bar: z.array(z.string()).register(registry, {
    reducer: {
      fn: (x, y) => x.concat(y),
    },
    default: () => [] as string[],
  }),
});

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: "" }, config);

import { StateGraph, START, END, MemoryServer } from "@langchain/langgraph";
import { registry } from "@langchain/langgraph/zod";
import * as z from "zod";

const State = z.object({
  foo: z.string(),
  bar: z.array(z.string()).register(registry, {
    reducer: {
      fn: (x, y) => x.concat(y),
    },
    default: () => [] as string[],
  }),
});

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: "" }, config);

在运行完图之后，我们期望看到恰好4个检查点：

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

请注意，bar 通道值包含来自两个节点的输出，因为我们有一个用于 bar 通道的reducer。

获取状态

在与保存的图状态交互时，您必须指定一个线程标识符。您可以通过调用 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: []
}

获取状态历史

您可以通过调用 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: []
      }
    ]
  }
]

回放

它还可能播放回之前的图执行。如果我们使用 invoke 一个具有 thread_id 和 checkpoint_id 的图，那么我们将 重放 在与 checkpoint_id 对应的检查点之前之前已执行的步骤，并且只执行检查点之后的步骤。

• thread_id 是线程的 ID。
• checkpoint_id 是一个标识符，它指向线程中的特定检查点。

您必须在配置的 configurable 部分调用图时传递这些参数：

const config = {
  configurable: {
    thread_id: "1",
    checkpoint_id: "0c62ca34-ac19-445d-bbb0-5b4984975b2a",
  },
};
await graph.invoke(null, config);

重要的是，LangGraph知道某个特定步骤是否之前已经执行过。如果已经执行过，LangGraph只需在图中 重放 该特定步骤，而不重新执行该步骤，但仅限于提供的 checkpoint_id 之前的步骤。所有在 checkpoint_id 之后的步骤都将执行（即，一个新的分支），即使它们之前已经执行过。有关如何进行时间旅行以了解更多关于重放的信息，请参阅这个 时间旅行指南。

更新状态

除了重新播放特定 checkpoints 中的图之外，我们还可以 编辑 图的状态。我们使用 graph.updateState() 来完成这项操作。此方法接受三个不同的参数：

config

配置应包含 thread_id 指定要更新的线程。当仅传递 thread_id 时，我们更新（或分叉）当前状态。可选地，如果我们包含 checkpoint_id 字段，那么我们将分叉所选的检查点。

values

这些值将用于更新状态。请注意，此更新被处理得与节点发出的任何更新完全相同。这意味着如果图状态中的一些通道定义了reducer函数，这些值将被传递给这些函数。这意味着 update_state 不会自动覆盖每个通道的通道值，而只会覆盖没有reducers的通道。让我们通过一个例子来解释一下。

让我们假设您已经使用以下架构定义了您的图状态（参见上面的完整示例）：

import { registry } from "@langchain/langgraph/zod";
import * as z from "zod";

const State = z.object({
  foo: z.number(),
  bar: z.array(z.string()).register(registry, {
    reducer: {
      fn: (x, y) => x.concat(y),
    },
    default: () => [] as string[],
  }),
});

让我们现在假设图当前的状

{ foo: 1, bar: ["a"] }

如果您更新状态如下：

await graph.updateState(config, { foo: 2, bar: ["b"] });

然后，图的新的状态将是：

{ foo: 2, bar: ["a", "b"] }

foo 键（通道）被完全更改（因为未指定该通道的reducer，所以 updateState 会覆盖它）。然而，已指定 bar 键的reducer，因此它将 "b" 追加到 bar 的状态中。

as_node

在调用 updateState 时，您可以可选地指定最后一个 asNode。如果您提供了它，更新将像来自节点 asNode 一样应用。如果没有提供 asNode，它将设置为最后更新状态的节点，如果不存在歧义。这之所以重要，是因为接下来要执行的步骤取决于最后给出更新的节点，因此这可以用来控制哪个节点接下来执行。有关分叉状态的更多了解，请参阅这个 时间旅行指南。

记忆存储

一个 状态模式 指定了一组在图执行过程中填充的键。如上所述，状态可以被检查点在每个图步骤写入到线程中，从而实现状态持久化。

但是，如果我们想保留一些信息 跨线程 呢？考虑一下聊天机器人的情况，我们希望在整个与该用户的 所有 聊天对话（例如，线程）中保留关于该用户的具体信息！

仅使用检查点器，我们无法在线程之间共享信息。这促使我们需要Store接口。作为一个例子，我们可以定义一个InMemoryStore来存储跨线程的用户信息。我们只需像以前一样使用检查点器编译我们的图，并使用我们新的in_memory_store变量。

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

基本用法

首先，让我们在不使用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：字符串列表，此内存类型的命名空间
• 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", "$"], // Fields to embed
  },
});

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

// Find memories about food preferences
// (This can be done after putting memories into the store)
const memories = await store.search(namespaceForMemory, {
  query: "What does the user like to eat?",
  limit: 3, // Return top 3 matches
});

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

// Store with specific fields to embed
await store.put(
  namespaceForMemory,
  uuidv4(),
  {
    food_preference: "I love Italian cuisine",
    context: "Discussing dinner plans",
  },
  { index: ["food_preference"] } // Only embed "food_preferences" field
);

// Store without embedding (still retrievable, but not searchable)
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";

// We need this because we want to enable threads (conversations)
const checkpointer = new MemorySaver();

// ... Define the graph ...

// Compile the graph with the checkpointer and store
const graph = workflow.compile({ checkpointer, store: memoryStore });

我们使用 thread_id 调用图，与之前一样，还使用 user_id，我们将使用它来命名空间我们的记忆，以便为特定用户存储，正如上面所展示的。

// Invoke the graph
const userId = "1";
const config = { configurable: { thread_id: "1", user_id: userId } };

// First let's just say hi to the AI
for await (const update of await graph.stream(
  { messages: [{ role: "user", content: "hi" }] },
  { ...config, streamMode: "updates" }
)) {
  console.log(update);
}

我们可以通过访问 config 和 store 作为节点参数，在 任何节点 中访问 memoryStore 和 user_id。以下是我们在节点中使用语义搜索查找相关记忆的方法：

import { MessagesZodMeta, Runtime } from "@langchain/langgraph";
import { BaseMessage } from "@langchain/core/messages";
import { registry } from "@langchain/langgraph/zod";
import * as z from "zod";

const MessagesZodState = z.object({
  messages: z
    .array(z.custom<BaseMessage>())
    .register(registry, MessagesZodMeta),
});

const updateMemory = async (
  state: z.infer<typeof MessagesZodState>,
  runtime: Runtime<{ user_id: string }>,
) => {
  // Get the user id from the config
  const userId = runtime.context?.user_id;
  if (!userId) throw new Error("User ID is required");

  // Namespace the memory
  const namespace = [userId, "memories"];

  // ... Analyze conversation and create a new memory

  // Create a new memory ID
  const memoryId = uuidv4();

  // We create a new memory
  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 = async (
  state: z.infer<typeof MessagesZodState>,
  config: LangGraphRunnableConfig,
  store: BaseStore
) => {
  // Get the user id from the config
  const userId = config.configurable?.user_id;

  // Namespace the memory
  const namespace = [userId, "memories"];

  // Search based on the most recent message
  const memories = await store.search(namespace, {
    query: state.messages[state.messages.length - 1].content,
    limit: 3,
  });
  const info = memories.map((d) => d.value.memory).join("\n");

  // ... Use memories in the model call
};

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

// Invoke the graph
const config = { configurable: { thread_id: "2", user_id: "1" } };

// Let's say hi again
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：LangGraph检查点的实现，使用SQLite数据库(SqliteSaver)。适用于实验和本地工作流程。需要单独安装。
• @langchain/langgraph-checkpoint-postgres：使用Postgres数据库(PostgresSaver)的高级检查点，用于LangSmith。适用于生产环境。需要单独安装。

检查点接口

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

• .put - 存储带有其配置和元数据的检查点。
• .putWrites - 存储与检查点相关联的中间写入操作（即待写入操作）。
• .getTuple - 使用给定的配置（thread_id 和 checkpoint_id）获取检查点元组。这用于在 graph.getState() 中填充 StateSnapshot。
• .list - 列出与给定配置和筛选标准匹配的检查点。这用于在 graph.getStateHistory() 中填充状态历史记录。

序列化器

当检查点保存图状态时，它们需要将状态中的通道值进行序列化。这是通过使用序列化对象来完成的。

@langchain/langgraph-checkpoint 定义了实现序列化器的协议，并提供了一个默认实现，该实现可以处理各种类型，包括 LangChain 和 LangGraph 原语、日期时间、枚举等。

功能

人工增强循环

首先，检查点器通过允许人类检查、中断和批准图步骤，促进了人机交互工作流程。这些工作流程需要检查点器，因为人类必须能够在任何时候查看图的状态，并且图必须在人类对状态进行任何更新后能够恢复执行。请参阅如何指南以获取示例。

记忆

其次，检查点器允许在交互之间进行”记忆”。在重复的人机交互（如对话）的情况下，任何后续消息都可以发送到该线程，该线程将保留对之前消息的记忆。有关如何使用检查点器添加和管理对话记忆的信息，请参阅添加记忆。

时间旅行

第三，检查点允许进行 “时间旅行”，使用户能够回放之前的图执行以审查和/或调试特定的图步骤。此外，检查点使得在任意检查点分叉图状态成为可能，以探索替代轨迹。

容错性

最后，检查点机制还提供了容错和错误恢复功能：如果在某个超级步骤中一个或多个节点失败，您可以从最后一个成功的步骤重新启动您的图。此外，当图节点在给定超级步骤的执行过程中失败时，LangGraph会存储来自在该超级步骤中成功完成的任何其他节点的挂起检查点写入，这样，无论何时我们从该超级步骤恢复图执行，我们都不会重新运行成功的节点。

待写入

此外，当图节点在给定的超级步骤中执行过程中失败时，LangGraph会存储在该超级步骤中成功完成的任何其他节点的挂起检查点写入，这样，无论何时我们从该超级步骤恢复图执行，我们都不会重新运行成功的节点。

---

在GitHub上编辑此页面的源代码。

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