持久化

状态丢了

你有没有遇到过这样的情况？

开始之前

核心概念

理解持久化之前，先搞清三个关键词

步骤与快照

对于 START → A → B → END 的简单图，执行一次会产生 4 个检查点

// 定义带 reducer 的状态
import { StateSchema, MemorySaver, StateGraph, START, END } from "@langchain/langgraph";
import { z } from "zod";

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

// 用 checkpointer 编译——这一行就启用了持久化
const checkpointer = new MemorySaver();
const graph = workflow.compile({ checkpointer });

📄 完整代码：05_memory_saver.ts

使用方式

从创建到运行——MemorySaver 的完整用法

// 完整示例：一个带持久化的简单图
import { StateSchema, StateGraph, MemorySaver, START, END } from "@langchain/langgraph";
import { z } from "zod";

const State = new StateSchema({
  messages: z.array(z.string()).default(() => []),
});

const nodeA = (state) => {
  console.log("Node A executing, messages:", state.messages);
  return { messages: ["from A"] };  // reducer 自动 concat
};

const nodeB = (state) => {
  console.log("Node B executing, messages:", state.messages);
  return { messages: ["from B"] };
};

// 关键一步：编译时传入 checkpointer
const graph = new StateGraph(State)
  .addNode("a", nodeA)
  .addNode("b", nodeB)
  .addEdge(START, "a")
  .addEdge("a", "b")
  .addEdge("b", END)
  .compile({ checkpointer: new MemorySaver() });

// 运行时必须指定 thread_id
const config = { configurable: { thread_id: "thread-1" } };

const result1 = await graph.invoke({ messages: ["initial"] }, config);
console.log(result1.messages);  // ["initial", "from A", "from B"]

// 第二次调用——状态会延续！
const result2 = await graph.invoke({ messages: ["second call"] }, config);
console.log(result2.messages);  // ["initial", "from A", "from B", "second call", "from A", "from B"]

📄 完整代码：05_memory_saver.ts

获取状态

用 getState() 查看当前线程的完整状态快照

// 获取当前线程的最新状态
const config = { configurable: { thread_id: "thread-1" } };
const snapshot = await graph.getState(config);

console.log(snapshot.values);     // 当前状态值
console.log(snapshot.next);       // 下一个要执行的节点列表
console.log(snapshot.config);     // 配置信息（含 thread_id）
console.log(snapshot.metadata);   // 元数据（步骤号、来源等）
console.log(snapshot.createdAt);  // 创建时间
console.log(snapshot.parentConfig); // 父检查点的 config（用于回溯）
console.log(snapshot.tasks);      // 当前待执行的任务列表

指定 checkpoint_id 获取历史检查点

// 获取特定检查点的状态（而非最新）
const historyConfig = {
  configurable: {
    thread_id: "thread-1",
    checkpoint_id: "具体的检查点 ID",  // 从 metadata 中获取
  },
};
const oldSnapshot = await graph.getState(historyConfig);
console.log("历史状态:", oldSnapshot.values);

从历史中精准定位某个检查点

// 收集全部历史
const history = [];
for await (const state of graph.getStateHistory(config)) {
  history.push(state);
}

// 按待执行节点查找——"找到 nodeB 即将执行前的那个检查点"
const beforeNodeB = history.find((s) => s.next.includes("nodeB"));

// 按步骤号查找——"找到第 2 步的状态"
const step2 = history.find((s) => s.metadata.step === 2);

// 按来源查找——"找到所有由 updateState 创建的分支"
const forks = history.filter((s) => s.metadata.source === "update");

📄 完整代码：05_state_snapshot.ts

快照字段

StateSnapshot 的每个字段代表什么？

更新状态

用 updateState() 直接修改检查点中的状态值

// 直接修改当前线程的状态
const config = { configurable: { thread_id: "thread-1" } };

await graph.updateState(config, {
  foo: "updated value",
});

// 验证修改
const snapshot = await graph.getState(config);
console.log(snapshot.values.foo);  // "updated value"

指定 as_node 参数——让更新看起来像某个节点产生的

// 模拟 nodeA 的输出
await graph.updateState(config, {
  messages: ["injected by human"],
}, "nodeA");  // as_node: "nodeA"

// 现在状态就像 nodeA 返回了 "injected by human" 一样
// 下一步 next 会变成 nodeA 之后的节点

📄 完整代码：05_state_snapshot.ts

跨线程存储

检查点是线程内的，InMemoryStore 用于跨线程共享数据

检查点绑定在单个线程上，不同线程之间无法共享状态。如果你需要跨线程的记忆——比如用户偏好、长期知识库——就需要 InMemoryStore。

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

// 创建跨线程存储
const store = new InMemoryStore();

// 编译时传入 store
const graph = workflow.compile({
  checkpointer: new MemorySaver(),
  store,  // 同时使用 checkpointer 和 store
});

// 在图外部操作存储——命名空间：[userId, "memories"]
const userId = "user-123";
const namespace = [userId, "memories"];

// 写入记忆
await store.put(namespace, "memory-1", {
  content: "用户偏好深色主题",
  type: "preference",
});

// 搜索记忆
const results = await store.search(namespace, {
  query: "用户喜欢什么",
});
console.log(results);  // [{ key: "memory-1", value: { content: "..." }, ... }]

📄 完整代码：05_cross_thread_store.ts

语义搜索

给 InMemoryStore 加上向量嵌入，实现语义级别的记忆检索

默认的 store.search() 是关键词匹配。如果你需要"意思相近"的搜索——比如搜"用户喜欢什么"能匹配到"偏好深色主题"——就需要接入嵌入模型。

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

// 创建带嵌入的存储
const embeddings = new OpenAIEmbeddings({
  model: "text-embedding-3-small",
});

const store = new InMemoryStore({
  index: {
    dims: 1536,  // 嵌入维度，取决于模型
    embeddings: embeddings,  // 嵌入模型实例（字段名是复数 embeddings）
  },
});

// 现在 search 会做语义匹配
const results = await store.search(["user-123", "memories"], {
  query: "用户的视觉偏好",
});
// 能匹配到 "用户偏好深色主题"，即使没有关键词重叠

📄 完整代码：05_semantic_search.ts

节点内使用

在节点函数中通过 runtime 参数访问跨线程存储

编译时传入了 store，节点内部就能通过第二个参数的 runtime 字段访问它。不需要闭包或全局变量。

// 节点签名：async (state, runtime) => { ... }
const recallMemories = async (state, runtime) => {
  const userId = state.userId;

  // 通过 runtime.store 访问跨线程存储
  const memories = await runtime.store?.search(
    [userId, "memories"],
    { query: state.userInput }
  );

  const memoryText = memories
    ?.map((m) => m.value.content)
    .join("\n") || "无历史记忆";

  return { context: memoryText };
};

// 保存新记忆的节点
const saveMemory = async (state, runtime) => {
  const userId = state.userId;

  await runtime.store?.put(
    [userId, "memories"],
    `memory-${Date.now()}`,
    { content: state.newFact, type: "fact" }
  );

  return { newFact: "" };  // 清空，避免重复保存
};

存储后端

根据场景选择合适的 Checkpointer 实现

// 开发
import { MemorySaver } from "@langchain/langgraph";
.compile({ checkpointer: new MemorySaver() })

// 生产
import { PostgresSaver } from "...";
.compile({ checkpointer: new PostgresSaver(pool) })

排错指南

持久化使用中的常见问题与修复方案

动手练习

构建一个带持久化和跨线程记忆的对话图

// 参考答案：练习 1 & 2
import { StateSchema, StateGraph, MemorySaver, START, END } from "@langchain/langgraph";
import { z } from "zod";

const State = new StateSchema({
  messages: z.array(z.string()).default(() => []),
});

const graph = new StateGraph(State)
  .addNode("greet", (s) => ({ messages: [`Hello, ${s.name}!`] }))
  .addNode("farewell", (s) => ({ messages: ["Goodbye!"] }))
  .addEdge(START, "greet")
  .addEdge("greet", "farewell")
  .addEdge("farewell", END)
  .compile({ checkpointer: new MemorySaver() });

const config = { configurable: { thread_id: "demo" } };

// 第一次调用
await graph.invoke({ messages: [], name: "Alice" }, config);
const snap = await graph.getState(config);
console.log(snap.values.messages);  // ["Hello, Alice!", "Goodbye!"]

// 遍历历史
for await (const state of graph.getStateHistory(config)) {
  console.log(`Step ${state.metadata?.step}:`, state.next, state.values);
}

本章回顾

持久化是 LangGraph 高级功能的基石