跳到内容

持久性

LangGraph 内置了持久化层，通过检查点（checkpointers）实现。当您使用检查点编译图时，检查点会在每个超步保存一个图状态的checkpoint。这些检查点会保存到一个thread中，可以在图执行后访问。因为thread允许在图执行后访问图的状态，所以许多强大的功能都成为可能，包括人在回路中、内存、时间旅行和容错。有关如何在图中使用检查点的端到端示例，请参阅此操作指南。下面，我们将更详细地讨论这些概念。

Checkpoints

线程

线程是由检查点保存的每个检查点分配的唯一 ID 或 线程标识符。使用检查点调用图时，您必须将thread_id指定为 config 的configurable部分。

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

检查点

检查点是图状态在每个超步保存的快照，由具有以下关键属性的StateSnapshot对象表示

  • config: 与此检查点关联的 Config。
  • metadata: 与此检查点关联的 Metadata。
  • values: 此时状态通道的值。
  • next 图中下一步要执行的节点名称元组。
  • tasks: 包含有关要执行的下一步任务信息的PregelTask对象元组。如果该步骤先前已尝试过，它将包含错误信息。如果图从节点内部动态中断，tasks 将包含与中断相关的额外数据。

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

import { StateGraph, START, END, MemorySaver, Annotation } from "@langchain/langgraph";

const GraphAnnotation = Annotation.Root({
  foo: Annotation<string>
  bar: Annotation<string[]>({
    reducer: (a, b) => [...a, ...b],
    default: () => [],
  })
});

function nodeA(state: typeof GraphAnnotation.State) {
  return { foo: "a", bar: ["a"] };
}

function nodeB(state: typeof GraphAnnotation.State) {
  return { foo: "b", bar: ["b"] };
}

const workflow = new StateGraph(GraphAnnotation);
  .addNode("nodeA", nodeA)
  .addNode("nodeB", nodeB)
  .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。

获取状态

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

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

// Get a state snapshot for a specific checkpoint_id
const configWithCheckpoint = { configurable: { thread_id: "1", checkpoint_id: "1ef663ba-28fe-6528-8002-5a559208592c" } };
const stateWithCheckpoint = await graph.getState(configWithCheckpoint);

在我们的示例中，getState的输出将如下所示

{
  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 },
  created_at: '2024-08-29T19:19:38.821749+00:00',
  parent_config: { configurable: { thread_id: '1', checkpoint_ns: '', checkpoint_id: '1ef663ba-28f9-6ec4-8001-31981c2c39f8' } },
  tasks: []
}

获取状态历史

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

const config = { configurable: { thread_id: "1" } };
const history = await graph.getStateHistory(config);

在我们的示例中，getStateHistory的输出将如下所示

[
  {
    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 },
    created_at: '2024-08-29T19:19:38.821749+00:00',
    parent_config: { configurable: { thread_id: '1', checkpoint_ns: '', checkpoint_id: '1ef663ba-28f9-6ec4-8001-31981c2c39f8' } },
    tasks: [],
  },
  {
    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 },
    created_at: '2024-08-29T19:19:38.819946+00:00',
    parent_config: { configurable: { thread_id: '1', checkpoint_ns: '', checkpoint_id: '1ef663ba-28f4-6b4a-8000-ca575a13d36a' } },
    tasks: [{ id: '6fb7314f-f114-5413-a1f3-d37dfe98ff44', name: 'nodeB', error: null, interrupts: [] }],
  },
  // ... (other checkpoints)
]

State

重放

也可以回放先前的图执行。如果我们使用thread_id和checkpoint_idinvoking图，那么我们将从与checkpoint_id对应的检查点重放图。

  • thread_id只是线程的 ID。这总是必需的。
  • checkpoint_id 此标识符指向线程中的特定检查点。

调用图时，必须将这些作为 config 的configurable部分传入

// { configurable: { thread_id: "1" } }  // valid config
// { configurable: { thread_id: "1", checkpoint_id: "0c62ca34-ac19-445d-bbb0-5b4984975b2a" } }  // also valid config

const config = { configurable: { thread_id: "1" } };
await graph.invoke(inputs, config);

重要的是，LangGraph 知道某个检查点是否先前已执行过。如果已执行，LangGraph 会简单地重放图中的该特定步骤，而不会重新执行该步骤。请参阅这篇关于时间旅行的操作指南，了解有关重放的更多信息。

Replay

更新状态

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

config

config 应包含指定要更新哪个线程的thread_id。仅传入thread_id时，我们会更新（或分叉）当前状态。可选地，如果包含checkpoint_id字段，则我们会分叉该选定的检查点。

values

这些是将用于更新状态的值。请注意，此更新的处理方式与来自节点的任何更新完全相同。这意味着这些值将传递给reducer函数（如果为图状态中的某些通道定义了它们）。这意味着updateState不会自动覆盖每个通道的通道值，而仅覆盖没有 reducer 的通道。让我们来看一个示例。

假设您已使用以下 schema 定义了图的状态（参见上面的完整示例）

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

const GraphAnnotation = Annotation.Root({
  foo: Annotation<string>
  bar: Annotation<string[]>({
    reducer: (a, b) => [...a, ...b],
    default: () => [],
  })
});

现在假设图的当前状态是

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

如果您按如下方式更新状态

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

那么图的新状态将是

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

foo键（通道）完全改变了（因为没有为该通道指定 reducer，所以updateState会覆盖它）。但是，为bar键指定了一个 reducer，因此它会将"b"追加到bar的状态中。

作为节点

调用updateState时可以可选地指定的最后一个参数是第三个位置参数asNode。如果提供了此参数，更新将如同来自节点asNode一样应用。如果未提供asNode，则如果不是模棱两可，它将被设置为上次更新状态的节点。这之所以重要，是因为下一步要执行的步骤取决于最后一个给出更新的节点，因此这可用于控制下一个执行的节点。请参阅这篇关于时间旅行的操作指南，了解有关分叉状态的更多信息。

Update

内存存储

Update

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

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

仅靠检查点，我们无法在跨线程共享信息。这促使了对Store接口的需求。作为说明，我们可以定义一个InMemoryStore来存储跨线程的用户信息。首先，让我们在不使用 LangGraph 的情况下单独展示这一点。

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

const inMemoryStore = new InMemoryStore();

内存由一个tuple命名空间化，在此特定示例中，它将是[<user_id>, "memories"]。命名空间的长度可以是任意的，可以表示任何内容，不一定特定于用户。

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

我们使用store.put方法将内存保存到存储中的命名空间。执行此操作时，我们会指定命名空间（如上定义）以及内存的键值对：键只是内存的唯一标识符（memoryId），值（一个对象）是内存本身。

import { v4 as uuid4 } from 'uuid';

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

我们可以使用store.search读取我们命名空间中的内存，这将返回给定用户的所有内存列表。最新的内存在列表的末尾。

const memories = await inMemoryStore.search(namespaceForMemory);
console.log(memories.at(-1));

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

检索到的内存具有以下属性

  • value: 此内存的值（本身是一个字典）
  • key: 此内存在此命名空间中的 UUID
  • namespace: 字符串列表，此内存类型的命名空间
  • created_at: 此内存创建时的时间戳
  • updated_at: 此内存更新时的时间戳

有了这些，我们就可以在 LangGraph 中使用inMemoryStore。inMemoryStore与检查点协同工作：检查点将状态保存到线程中（如上所述），而inMemoryStore允许我们存储任意信息以供在跨线程访问。我们按如下方式编译包含检查点和inMemoryStore的图。

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 = builder.compile({
  checkpointer,
  store: inMemoryStore
});

我们像之前一样使用thread_id调用图，同时使用user_id，这将用于将我们的内存命名空间化到此特定用户，如上所示。

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

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

for await (const update of stream) {
  console.log(update);
}

我们可以通过将config: LangGraphRunnableConfig作为节点参数传入，在任何节点中访问inMemoryStore和user_id。然后，就像我们上面看到的那样，只需使用put方法将内存保存到存储中即可。

import {
  type LangGraphRunnableConfig,
  MessagesAnnotation,
} from "@langchain/langgraph";

const updateMemory = async (
  state: typeof MessagesAnnotation.State,
  config: LangGraphRunnableConfig
) => {
  // Get the store instance from the config
  const store = config.store;

  // Get the user id from the config
  const userId = config.configurable.user_id;

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

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

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

  // We create a new memory
  await store.put(namespace, memoryId, { memory });
};

如上所示，我们还可以在任何节点中访问存储并使用search获取内存。请记住，内存作为对象列表返回，可以转换为字典。

const memories = inMemoryStore.search(namespaceForMemory);
console.log(memories.at(-1));

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

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

const callModel = async (
  state: typeof StateAnnotation.State,
  config: LangGraphRunnableConfig
) => {
  const store = config.store;

  // Get the user id from the config
  const userId = config.configurable.user_id;

  // Get the memories for the user from the store
  const memories = await store.search([userId, "memories"]);
  const info = memories.map((memory) => {
    return JSON.stringify(memory.value);
  }).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
const stream = await graph.stream(
  { messages: [{ role: "user", content: "hi, tell me about my memories" }] },
  { ...config, streamMode: "updates" },
);

for await (const update of stream) {
  console.log(update);
}

当我们使用 LangGraph API 时，无论是在本地（例如，在 LangGraph Studio 中）还是使用 LangGraph Cloud，内存存储默认可用，并且在图编译期间无需指定。

检查点库

在底层，检查点功能由符合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)，在 LangGraph Cloud 中使用。非常适合在生产环境中使用。需要单独安装。

检查点接口

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

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

序列化器

当检查点保存图状态时，需要序列化状态中的通道值。这是通过序列化器对象完成的。@langchain/langgraph-checkpoint定义了一个实现序列化器的协议，并提供了一个默认实现，该实现可以处理各种类型，包括 LangChain 和 LangGraph 基本类型、日期时间、枚举等等。

功能特性

人在回路中

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

内存

其次，检查点允许在交互之间具有“记忆”。在重复的人类交互（如对话）的情况下，任何后续消息都可以发送到该线程，该线程将保留其对先前消息的记忆。有关如何使用检查点添加和管理对话记忆的端到端示例，请参阅此操作指南。

时间旅行

第三，检查点允许“时间旅行”，允许用户重放先前的图执行，以查看和/或调试特定的图步骤。此外，检查点还可以分叉任意检查点处的图状态，以探索替代路径。

容错

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

待定写入

此外，当图节点在给定超步执行过程中失败时，LangGraph 会存储在该超步中成功完成的任何其他节点的待定检查点写入，以便当我们在该超步恢复图执行时，不再重新运行成功节点。