在入门和演示阶段，我们使用的是：

new MemorySaver()

它足够简单，也非常适合理解机制。但一旦进入真实工程环境，就会立刻遇到一个问题：内存是靠不住的。

• 服务重启，短期记忆就没了
• 多实例部署，内存无法共享
• 无法支持稳定的时间旅行和恢复

这也是为什么官方在生产环境示例中，直接给出了 PostgresSaver。

那么问题来了：如果我们的技术栈不是 PostgreSQL，而是 MongoDB，该怎么办？

答案：自定义saver

自定义Checkpointer分析

Checkpointer 在 LangGraph 中负责什么？

它只关心一件事：Graph 在某一个 thread、某一个时刻，State 是什么样子。

从职责上看，Checkpointer 主要负责：

1. 保存 checkpoint（状态快照）
2. 恢复最新 checkpoint（继续执行）
3. 在需要时支持回溯（时间旅行）

也就是说：

• 它不关心你存的是聊天、订单还是流程状态
• 它只关心：我能不能把这个状态完整地存下来，并且准确地再拿出来

在 LangGraph 中，自定义 Checkpointer 的方式是：继承 BaseCheckpointSaver，并实现必要的方法。

MongoDB 是否适合做 Checkpointer？

Checkpoint 数据本身有几个明显特征：

• 以 thread_id 为主键维度
• append-only（不断追加）
• 强调时间顺序
• 单条数据体积可能不小（完整 State）

从数据模型角度看：

• MongoDB 非常适合这种日志或快照型数据，因为 Mongo 天然擅长存“结构可能变化的快照文档”。
• 特别是在已有 Mongo 技术栈的团队中，引入成本极低

‼️需要你会mongodb

我们为 MongoDB 中的 checkpoint 设计了这样一种文档结构：

interface MongoCheckpointDoc {
  thread_id: string;
  checkpoint_ns: string;
  checkpoint_id: string;
  parent_checkpoint_id?: string;
  type: "checkpoint";
  checkpoint: Buffer;
  checkpoint_type: string;
  metadata: Buffer;
  metadata_type: string;
  created_at: Date;
}

这里有几个非常关键的字段：

• thread_id：短期记忆的边界，同一个 thread 的 checkpoint 构成一条时间线
• checkpoint_id / parent_checkpoint_id：用于构建 checkpoint 链，是时间旅行的基础
• checkpoint / metadata：实际的 State 快照与元信息，以二进制形式存储，避免结构耦合

MongoSaver的核心实现讲解

1. 继承 BaseCheckpointSaver

export class MongoSaver extends BaseCheckpointSaver

这一点非常重要：

• LangGraph 并不直接依赖你的数据库
• 它只依赖 BaseCheckpointSaver 定义的行为

2. 保存 checkpoint：put

async put(config, checkpoint, metadata)

这个方法的职责只有一句话：把当前 Graph 执行完成后的状态，完整地存下来。

在实现中，需要做几件事：

• 从 config 中读取 thread_id
• 使用 LangGraph 内置的 serde 进行序列化
• 将结果作为一条新文档插入 MongoDB

复用 LangGraph 提供的 serde 来做序列化

这样可以保证：

• State 结构变化时不会炸
• 与官方 Saver 行为保持一致

3. 恢复 checkpoint：getTuple

async getTuple(config)

这个方法负责：在 Graph 执行前，恢复某一个 thread 的状态。

具体步骤：

• 根据 thread_id 和 checkpoint_ns 查询
• 如果指定了 checkpoint_id，则精确恢复
• 否则恢复最新的一条

最终返回的是一个 CheckpointTuple，这是 LangGraph 内部统一使用的结构。

4. 其它方法为什么可以先简化

对以下方法做了简化处理：

• putWrites
• list

原因并不是它们不重要，而是它们主要用于：

• interrupt
• human-in-the-loop
• 并行执行

暂时用不到

5. 如何在 Graph 中使用 MongoSaver

一旦 MongoSaver 实现完成，接入方式非常简单：

const checkpointer = new MongoSaver(collection);

const graph = builder.compile({
  checkpointer,
});

注意这一点：Graph 本身的逻辑，一行都不需要改。这正是 LangGraph 设计得非常优秀的地方。

// MongoSaver.ts

import { BaseCheckpointSaver } from "@langchain/langgraph";
import { Collection, type Document } from "mongodb";
import type { RunnableConfig } from "@langchain/core/runnables";
import type {
  Checkpoint,
  CheckpointMetadata,
  CheckpointTuple,
  PendingWrite,
} from "@langchain/langgraph-checkpoint";

// 存储到mongodb里面的文档的类型
interface MongoCheckpointDoc {
  thread_id: string; // 线程 ID，用于标识一次对话
  checkpoint_ns: string; // 检查点命名空间，通常为空或用于区分不同类型的检查点
  checkpoint_id: string; // 检查点 ID，通常是 UUID
  parent_checkpoint_id?: string; // 父检查点 ID，用于构建检查点链
  type: "checkpoint"; // 文档类型，固定为 "checkpoint"
  checkpoint: Buffer; // 序列化后的检查点数据，存储为二进制 Buffer
  checkpoint_type: string; // 检查点数据的序列化类型（如 json）
  metadata: Buffer; // 序列化后的元数据，存储为二进制 Buffer
  metadata_type: string; // 元数据的序列化类型
  created_at: Date; // 创建时间
}

export class MongoSaver extends BaseCheckpointSaver {
  // 定义了一个私有属性
  private collection: Collection<Document>;

  constructor(collection: Collection<Document>, serde?: any) {
    super(serde);
    this.collection = collection;
  }

  // 将检查点写入到mongodb里面
  // 核心思路：从检查点里面提取信息，组装成一个doc对象，调用mongodb相应方法进行存储
  async put(
    config: RunnableConfig,
    checkpoint: Checkpoint,
    metadata: CheckpointMetadata,
  ): Promise<RunnableConfig> {
    // 提取thread_id、checkpoint_ns 等信息
    const thread_id = config.configurable?.thread_id;
    const checkpoint_ns = config.configurable?.checkpoint_ns ?? "";

    if (!thread_id) throw new Error("存储checkpoint时需要提供thread_id");

    // 整个checkpoint，有一些可以直接存，有一些数据需要转为二进制
    // 返回值一个是类型，一个转换后的数据
    const [cType, cData] = await this.serde.dumpsTyped(checkpoint);
    const [mType, mData] = await this.serde.dumpsTyped(metadata);

    // 目前，要存储到mongodb里面的各个数据就准备好了
    // 下一步需要做一个数据的组装，组装成一个文档对象
    const doc: MongoCheckpointDoc = {
      thread_id,
      checkpoint_ns,
      checkpoint_id: checkpoint.id,
      parent_checkpoint_id: config.configurable?.checkpoint_id,
      type: "checkpoint",
      checkpoint: Buffer.from(cData),
      checkpoint_type: cType,
      metadata: Buffer.from(mData),
      metadata_type: mType,
      created_at: new Date(),
    };

    // 回头就需要将上面的 doc 对象添加到 mongodb 里面对应的 collection（表） 中
    await this.collection.insertOne(doc);

    // 返回新的运行配置
    return {
      configurable: {
        thread_id,
        checkpoint_ns,
        checkpoint_id: checkpoint.id,
      },
    };
  }

  // 从mongodb中获取指定的检查点. 恢复某一个 thread 的状态
  async getTuple(config: RunnableConfig): Promise<CheckpointTuple | undefined> {
    const thread_id = config.configurable?.thread_id;
    const checkpoint_id = config.configurable?.checkpoint_id;
    const checkpoint_ns = config.configurable?.checkpoint_ns ?? "";

    if (!thread_id) throw new Error("没有提供thread_id，无法恢复检查点");

    // 构建 mongodb 的 query 对象（负责查询的）
    // 根据 thread_id 和 checkpoint_ns 查询
    // default: 恢复最新的一条
    const query: any = {
      thread_id,
      checkpoint_ns,
      type: "checkpoint",
    };

    // 如果指定了 checkpoint_id，则精确恢复
    if (checkpoint_id) query.checkpoint_id = checkpoint_id;

    // 执行查询
    const doc = (await this.collection.findOne(query, {
      sort: {
        _id: -1, // 按照 _id 进行一个倒序排列，因为默认要获取最新的
      },
    })) as unknown as MongoCheckpointDoc;

    // 目前 doc 拿到的是一个文档对象
    // 需要将 doc 这个文档对象还原为 checkpoint
    if (!doc) return undefined;

    // 将 checkpoint 以及 metadata 这两个二进制数据重新转换回来
    const checkpoint = await this.serde.loadsTyped(
      doc.checkpoint_type,
      doc.checkpoint.toString(),
    );

    const metadata = await this.serde.loadsTyped(
      doc.metadata_type,
      doc.metadata.toString(),
    );

    // 接下来就可以返回符合 langgraph 要求的 CheckpointTuple 对象
    return {
      config: {
        configurable: {
          thread_id,
          checkpoint_ns,
          checkpoint_id: doc.checkpoint_id,
        },
      },
      checkpoint,
      metadata,
      parentConfig: doc.parent_checkpoint_id
        ? {
            configurable: {
              thread_id,
              checkpoint_ns,
              checkpoint_id: doc.parent_checkpoint_id,
            },
          }
        : undefined,
      pendingWrites: [],
    };
  }

  // 保存中间写入操作（pending writes）
  async putWrites(
    _config: RunnableConfig,
    _writes: PendingWrite[],
    _taskId: string,
  ): Promise<void> {}

  // 提供历史列表
  async *list() {}

  // 删除对应thread_id的所有检查点
  async deleteThread(threadId: string): Promise<void> {
    await this.collection.deleteMany({
      thread_id: threadId,
    });
  }
}

-EOF-