LangGraph Checkpointer 实战：MemorySaver、SQLite、Redis 怎么选？

本文解决的问题

• LangGraph Checkpointer 到底在状态机中保存什么核心数据？
• MemorySaver 或者说 InMemorySaver 究竟能不能在生产环境使用？
• SQLite Checkpointer 适合什么样的项目，有哪些并发与读写限制？
• Redis Checkpointer 适合什么高并发场景，需要哪些前置 Redis 模块支持？
• thread_id 和 Checkpointer 之间存在什么底层的物理寻址关系？
• 在 Human-in-the-loop 和失败恢复机制中，为什么 Checkpointer 是必不可少的？
• 个人开发者与专业团队在不同业务阶段该如何完成 Checkpointer 的架构选型？

---

适合谁读

• 已经跑通了本地 LangGraph Demo，正准备将 Agent 服务推向线上运行的后端与全栈工程师。
• 正在设计多轮对话、审批工作流、异常重试或者长周期异步执行任务的 AI 系统架构师。
• 不清楚 MemorySaver、SQLite、Redis、Postgres 各自的优缺点与适用边界，正在纠结选型的开发者。
• 需要在飞牛 NAS、轻量 VPS 或 Kubernetes 容器云上为 AI Agent 规划高可用存储架构的设计人员。

---

一、小白的硬核实战观察

在多实例或者高并发的生产环境下，Agent 的状态持久化方案是决定系统能否活下去的物理分水岭。

昨天打完一场两个小时的羽毛球，正当我大汗淋漓、衣服全湿透地在更衣室换衣服时，手机上的报警邮件突然疯狂弹窗。我部署在飞牛 NAS 容器里的一个财务账单分析 Agent 服务发生了崩溃。因为那天 NAS 上的另一个高负载任务导致系统内存耗尽，Docker 守护进程触发了 OOM 机制，直接把 Agent 容器物理杀掉了。

容器虽然通过 restart 策略自动重启了，但由于我图省事，在 compile 时传入了默认的内存 Checkpointer（也就是 Python 中的 InMemorySaver），导致当时正在挂起、等待出纳手工点击确认的 15 个长链财务审计流的状态彻底化为乌有。用户在前端看到的界面直接重置，之前跑了 20 分钟、消耗了数万 Token 换来的前置分析数据物理消失。

这笔 Token 损耗是小事，但用户对系统稳定性的信任直接降到了冰点。我只能穿着湿漉漉的衣服，坐在回程的自驾车里，反思这次惨痛的教训。在本地单机测试时，内存 Checkpointer 的开箱即用让我们产生了系统很稳的幻觉。然而，一旦涉及真实的并发环境、容器漂移、人工审核 interrupt 或者长周期运行，Checkpointer 选型和 thread_id 的设计就成了唯一的安全防线。

---

二、Checkpointer 的底层物理机制与状态解剖

Checkpointer 通过捕获有向图在每个节点执行结束时的快照，将其序列化并与唯一的 thread_id 绑定，为 Agent 提供了任务重试、时间旅行与人工审批的物理底座。

我们要明确一点：Checkpointer 解决的不是普通的缓存问题，而是 durable execution（持久化可靠执行）的问题。当我们在编译 Graph 时传入一个 checkpointer，LangGraph 会在图的每一步执行（也就是每个 node 运行完毕）之后，自动拦截并保存当前的 graph state snapshot。

这些 snapshot 是按 thread_id 组织在一起 of。一个 thread 就像是 Git 里的一个分支，记录了这个线程下所有历史运行的 commits（也就是 checkpoints）。每个 checkpoint 不仅包含了当前的 state 字典，还保存了极其丰富的物理元数据。

为了让大家有直观的感受，我们来看一下一个真实的 LangGraph checkpoint 被序列化后的简化物理数据结构：

{
  "thread_id": "usr_9982:billing_audit:task_20260615_001",
  "checkpoint_id": "1ef23b8f-89a1-6a20-b001-c91823abf100",
  "current_node": "agent_review_node",
  "state": {
    "messages": [
      {
        "type": "human",
        "content": "帮我审计 5 月份的差旅发票"
      },
      {
        "type": "ai",
        "content": "",
        "tool_calls": [
          {
            "name": "fetch_invoices",
            "args": {
              "month": "2026-05"
            },
            "id": "call_tx_9982"
          }
        ]
      }
    ],
    "invoice_list": [],
    "audit_status": "pending_fetch"
  },
  "pending_action": null,
  "error_state": null,
  "metadata": {
    "source": "loop",
    "step": 2,
    "run_id": "run_0f8e91cd-89b2-44a1-b873-120aef2b001f"
  },
  "created_at": "2026-06-15T10:28:29Z"
}

从上面的结构可以看出，Checkpointer 记录了：

• 当前图所在的具体节点：告诉状态机下一次被唤醒时应该从哪个 Node 开始执行。
• 完整的 State 数据：包括对话消息历史（messages 列表）以及自定义的业务状态字段（如 invoice_list）。
• 错误状态与挂起动作：如果某个节点运行中抛出异常，或者在节点前配置了 interrupt_before 断点，这里会记录待处理的 tool call 或异常信息。
• Metadata：包含了步骤计数、单次运行的 run_id 等，供可观测性链路分析。

有了这层底座，当 Agent 因为网络波动、接口超时或容器重启而意外中断时，我们只需要传入相同的 thread_id，LangGraph 就会从底层 Checkpointer 中捞出最新的物理快照，完美恢复现场并继续未完的步骤，而不需要重新向 LLM 发起前置节点的请求。

---

三、MemorySaver / InMemorySaver：适合 Demo，不适合生产

内存型 Checkpointer 的唯一价值是零配置开箱即用，由于其不具备持久化介质，任何进程重启、多实例负载均衡都会导致状态瞬间消失。

在 Python 版本的文档中，官方示例常用的是 InMemorySaver；而在 JavaScript/TypeScript 版本中，对应的名称是 MemorySaver。它们的物理本质都是完全相同的：在进程内存中维护一个 dict 或者 Map，用以读写 checkpoint 对象。

在本地开发、跑单元测试或者快速验证一个状态机分支逻辑时，它的使用极为简便：

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph

# 初始化内存 checkpointer
memory_checkpointer = MemorySaver()

# 编译 Graph 并挂载持久化组件
graph = builder.compile(checkpointer=memory_checkpointer)

# 传入 thread_id 启动多轮交互
config = {"configurable": {"thread_id": "local_demo_thread"}}
result = graph.invoke({"input": "开始我的分析任务"}, config=config)

然而，内存 checkpointer 存在三个致命的物理缺陷，使其绝对不能走入生产部署阶段：

1. 进程内驻留限制：如果你的应用是部署在 Kubernetes 集群、PM2 进程托管或者像 Vercel 这样的 Serverless 平台，当发生容器扩缩容、节点漂移、或者进程崩溃自动拉起时，新启动的进程内部是空的。一旦用户的请求被负载均衡路由到新的实例，前置状态就会丢失，导致报错。
2. 内存容量爆满（OOM）风险：随着对话轮数和并发用户数的不断累积，保存在进程内存中的 checkpoint 列表会无限制增长。每个 checkpoint 里的 messages 历史都非常吃内存，这会直接导致应用进程因为内存泄漏或爆满被系统强制 Kill。
3. 状态无法持久审计：一旦服务发生热更新或正常维护重启，所有的历史执行链全部物理蒸发，无法用于后续的产品数据回溯与合规审计。

因此，除非是在写单机脚本或跑本地 pytest，否则在项目立项的初期，就应该把 InMemorySaver 从依赖中物理剔除。

---

四、SQLite Checkpointer：个人项目与单机部署的低成本选择

SQLite 提供了零部署成本的磁盘级持久化，支持同步与异步驱动，是单机应用与小规模工具站的理想之选，但需应对并发写入下的数据库锁屏障。

如果你的 Agent 服务目前的阶段是一个人开发的工具型站点，或者部署在一台独立的云服务器、甚至是私有的 NAS 设备上，SQLite 是一个性价比极高的选择。它既能保证进程重启后状态完好无损，又不需要你额外运维复杂的数据库集群。

LangGraph 官方提供了一个独立的扩展包 langgraph-checkpoint-sqlite 用于支持 SQLite 持久化。它包含同步的 SqliteSaver 和基于 aiosqlite 实现的异步 AsyncSqliteSaver。

我们来看一个在异步应用（如 FastAPI）中标准初始化 AsyncSqliteSaver 的实战代码示例：

import asyncio
import os
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
from langgraph.graph import StateGraph

# 定义 SQLite 数据库文件路径
DB_PATH = "data/agent_checkpoints.db"
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)

async def run_agent_workflow():
    # 使用上下文管理器安全创建并管理 SQLite 连接池
    async with AsyncSqliteSaver.from_conn_string(DB_PATH) as checkpointer:
        # 执行一次性的物理建表操作（如果表不存在）
        await checkpointer.asetup()

        # 编译有向图，绑定 SQLite 异步 Checkpointer
        graph = builder.compile(checkpointer=checkpointer)

        # 定义唯一的 thread_id 进行任务追踪
        config = {"configurable": {"thread_id": "sqlite_user_task_4451"}}

        # 启动异步图执行
        async for event in graph.astream(
            {"messages": [("user", "分析这篇文档的结构")]},
            config=config,
            stream_mode="values"
        ):
            print(event)

# 启动任务
asyncio.run(run_agent_workflow())

使用 SQLite 时，你需要注意以下两个物理踩坑点：

• 数据库连接锁屏障：SQLite 本质上是一个单文件数据库，虽然它支持多进程/多线程并发读取，但是在进行高频并发写入（例如 10 个用户并发调用 Agent）时，容易触发 sqlite3.OperationalError: database is locked。为了缓解这个问题，你必须在连接字符串中加入 check_same_thread=False 参数，并在必要时开启 WAL（Write-Ahead Logging）模式，以允许读写并发。
• 数据碎片与容量膨胀：随着 Agent 运行步骤的增多，checkpoints 和 checkpoint_writes 表中的数据会快速膨胀。你需要在服务器上配置 Cron 定期执行 VACUUM 命令，否则数据库文件很容易无节制增长，吃满磁盘空间。

---

五、Redis Checkpointer：高并发与低延迟状态的利器

Redis Checkpointer 是多实例水平扩展和低延迟高并发 Agent 的首选，但因其基于内存，用作长期审计和合规追溯时需要额外挂载关系型数据库。

如果你的 Agent 服务拥有较大的日活用户，且采用多实例分布式部署在 Kubernetes 等容器集群中，用户对响应延迟（Latency）有着严苛的要求，那么基于 Redis 的持久化方案是最佳匹配。

Redis 运行在内存中，具备亚毫秒级的读写响应速度，极大地降低了状态机在每个节点转换时保存 snapshot 的开销。LangGraph 官方集成包中提供了 langgraph-checkpoint-redis，包含了 RedisSaver 和用于内存优化的 ShallowRedisSaver。

使用 Redis Saver 时存在一个强制的物理前置条件：你连接的 Redis 实例必须支持「RedisJSON」和「RediSearch」这两个官方模块。因为 LangGraph 在检索状态和回溯特定 thread 的历史快照时，需要利用 JSON 格式解析与二级索引搜索。如果你使用的是老版本的 Redis，或者没有安装这两个模块的默认 Redis 容器镜像，调用 setup 方法时会直接报错崩溃。

我们来看一下完整的 Redis Saver 实战代码：

from langgraph.checkpoint.redis import RedisSaver
from langgraph.graph import StateGraph

# Redis 连接字符串，注意在生产环境一定要开启认证和 SSL
REDIS_URI = "redis://:complex_password@192.168.1.100:6379/0"

# 初始化 RedisSaver
redis_checkpointer = RedisSaver.from_conn_string(REDIS_URI)

# 物理建索引，对于一个全新的 Redis 实例，这一步必须执行且只需执行一次
redis_checkpointer.setup()

# 编译 Graph 绑定 Redis 持久化层
graph = builder.compile(checkpointer=redis_checkpointer)

# 设置 thread_id 并唤醒执行
config = {"configurable": {"thread_id": "redis_thread_9872"}}
graph.invoke({"messages": [("user", "执行系统健康检查")]}, config=config)

在 Redis 架构选型中，我们要面临以下几个不可忽视的设计考量：

• 内存容量保护（ShallowRedisSaver）：默认的 RedisSaver 会完整保存整个 thread 的所有 checkpoint 历史。如果一个 Agent 和用户聊了几百轮，每次状态都会被原封不动保存。这会导致 Redis 内存迅速被吃满。为此，可以使用 ShallowRedisSaver，它采用覆盖机制，每个 thread_id 在 Redis 中只保留最新的那一个 checkpoint 节点。这样虽然无法进行“时间旅行（time travel）”去加载历史版本，但内存占用是恒定的。
• 内存淘汰与持久化策略：生产环境的 Redis 实例一定要关闭 allkeys-lru 或 volatile-lru 内存淘汰机制。如果因为内存不足，Redis 自动把某些活跃 thread 的最新 checkpoint 物理剔除了，那么用户的 Agent 流程会直接在下一轮对话中崩溃报错。此外，必须开启 AOF（Append Only File）持久化，并设置每秒同步（everysec），以防 Redis 服务进程崩溃导致物理断电时丢失未落盘的状态。

---

六、Postgres：正式生产的终极物理归宿

PostgresCheckpointer 通过强大的关系型数据库事务、JSONB 高效检索与高可用容灾能力，为大型分布式 Agent 系统提供了最安全可靠的持久化保障。

当你需要开发一个面向企业级多租户、长任务生命周期、且需要面临监管合规、数据审计以及历史调用分析的 Agent 平台时，Postgres 是唯一的物理归宿。

实际上，LangChain 官方的高级托管服务 LangSmith Agent Server 默认就是使用 Postgres 作为底层 checkpoint data 的存储介质。通过 langgraph-checkpoint-postgres 包，我们可以非常方便地构建出高可用的企业级 Agent。

在这里，我必须分享一个我踩了整整一天的极其隐秘的物理坑：如果你不想使用默认的连接字符串工厂，而是希望手动管理连接池（例如在 FastAPI 中复用已有的数据库连接池），你「必须」在初始化连接时开启以下两项配置：

1. autocommit=True：因为 LangGraph 的建表语句和状态更新依赖非阻塞事务提交，如果不开启，.setup() 方法会在创建 checkpoints 表时陷入永久锁死或者直接抛出事务异常。
2. row_factory=dict_row：LangGraph 内部在读取数据库行时，全部是基于列名键值对进行物理提取的（例如 row["checkpoint_id"]）。如果你用的是默认的 tuple 行工厂，会导致解析数据时抛出 KeyError 异常而崩溃。

下面是规范的手动管理连接并初始化 PostgresSaver 的生产级代码：

import psycopg
from psycopg.rows import dict_row
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://agent_admin:secure_db_pass@192.168.1.150:5432/agent_store"

# 建立数据库连接，必须配置 autocommit 和 dict_row
conn = psycopg.connect(DB_URI, autocommit=True, row_factory=dict_row)

# 将连接对象传给 PostgresSaver
postgres_checkpointer = PostgresSaver(conn)

# 创建 checkpoints 相关的物理表结构
postgres_checkpointer.setup()

# 编译 Graph，开启生产级安全策略
graph = builder.compile(checkpointer=postgres_checkpointer)

# 传入符合企业级规范的 thread_id
config = {
    "configurable": {
        "thread_id": "org_100:user_200:task_abc",
        "strict_msgpack": True # 开启此选项，可物理限制反序列化安全类型
    }
}

为了防范由于第三方库反序列化漏洞造成的远程代码执行（RCE）物理风险，在生产环境中，一定要设置环境变量 LANGGRAPH_STRICT_MSGPACK=true。这样可以确保在从 Postgres 中读取并还原 State 时，只对可信的内置数据类型进行还原。

---

七、深度对比：四大 Checkpointer 后端对标

在架构选型时，我们需要通过对比各个方案的延迟、并发性、容灾能力与运维复杂度，来决定最契合当前业务阶段的持久化后端。

为了方便大家一目了然地进行决策，我将这四种典型的 Checkpointer 整理成对标表格：

物理维度	MemorySaver	SQLite	Redis	Postgres
适合场景	教程 Demo、本地单元测试	个人工具站、单机部署、轻量后台	高并发短会话、低延迟分布式服务	分布式多用户、企业级长任务审计
运维成本	零成本，无外部依赖	极低，单文件管理	中等，需额外配置内存模块	高，需要高可用关系型数据库
并发能力	仅限单进程内并发	较弱，多线程写入易锁库	极强，高并发吞吐	强，支持连接池与高并发事务
容灾持久化	重启即物理蒸发	磁盘级持久化，需做好备份	内存为主，依赖 AOF 磁盘同步	物理落盘，成熟的备份与恢复机制
时间旅行支持	支持完整历史回溯	支持完整历史回溯	可配置完整或 Shallow 覆盖模式	支持完整历史，便于 JSONB 查询分析
物理限制	受限于进程分配的 RAM 大小	写入锁限制，不适合高并发	必须是 Redis Stack 8.0 以上	必须配置 autocommit 与 dict_row

---

八、thread_id 与状态隔离的防串线设计

thread_id 是 Checkpointer 读取状态的物理主键，如果不做严格的租户隔离和属主强校验，系统将不可避免地面临越权访问与上下文串线灾难。

在前几天的开发总结中，我也深刻探讨过这个问题。如果所有的用户请求无脑共用一个 thread_id，或者只用 user_id 作为 thread_id，那么在高并发场景下，A 用户的操作快照会在 Checkpointer 物理表里直接把 B 用户的前置状态覆盖掉，导致状态机的上下文瞬间串线。

我们在设计生产接口时，必须将 thread_id 规范化为复合键。例如，我们规定： thread_id = user_id + business_type + task_id

同时，API 网关层不能直接把客户端传来的 thread_id 扔给有向图编译层，必须构建一层基于属主校验的防越权物理屏障。以下是我在网关拦截层设计的强校验伪代码：

# API 属主强校验层
def process_incoming_request(user_context: dict, client_payload: dict):
    current_user_id = user_context["user_id"]
    requested_thread_id = client_payload["thread_id"]

    # 物理截取校验：thread_id 的第一段必须是当前已登录的用户 ID
    thread_parts = requested_thread_id.split(":")
    if len(thread_parts) < 3 or thread_parts[0] != current_user_id:
        raise PermissionError("物理安全阻断：试图读取不属于当前登录用户的 Checkpoint 状态")

    config = {
        "configurable": {
            "thread_id": requested_thread_id,
            "user_id": current_user_id
        }
    }

    # 从数据库中拉取并恢复
    return graph.invoke({"input": client_payload["prompt"]}, config=config)

---

九、常见坑与常见报错 (Error Logs)

规避常见编译与运行报错，是保证 Checkpointer 稳定发挥状态恢复作用的必要前提。

在我的实际填坑经历中，以下这几个报错信息是出镜率最高的。如果你的控制台弹出了类似的 Log，可以直接对照我的物理排查方案解决：

1. SQLite 锁库报错

sqlite3.OperationalError: database is locked

• 发生原因：在高并发请求下，多个线程试图同时写入 SQLite 的 checkpoint_writes 表。
• 解决方案：在配置 AsyncSqliteSaver.from_conn_string 时，确保设置连接超时时间（timeout=30），并在 SQLite 数据库层面配置 WAL 模式。

2. Postgres 连接挂起与配置报错

psycopg.errors.ActiveSqlTransaction: CREATE TABLE cannot run inside a transaction block

或者在读取数据时抛出：

KeyError: 'checkpoint_id'

• 发生原因：没有显式配置 autocommit=True 导致 DDL 语句锁死，或者没有配置 row_factory=dict_row 导致 LangGraph 无法以字典形式物理提取数据行。
• 解决方案：在建立 psycopg 连接对象时，强制指定 autocommit=True, row_factory=dict_row。

3. Redis 索引缺失报错

redis.exceptions.ResponseError: Cannot create index: ft.create require RediSearch module v2.0+

• 发生原因：使用的 Redis 镜像为普通的 Redis Server，缺少 RedisJSON 或 RediSearch 核心模块。
• 解决方案：将 Redis 镜像替换为 redis/redis-stack-server:latest，或者向 Redis 云托管服务商确认已开启这两个模块插件。

4. 序列化失败报错

TypeError: Object of type CustomToolResult is not JSON serializable

• 发生原因：你的 Tool 返回的对象是一个自定义的 Python 类实例（比如直接返回了某个数据库 ORM 对象），而没有将其转换为基础的 dict 或 JSON 字符串。Checkpointer 在序列化图状态保存到磁盘时，无法将该复杂对象转化为字节流。
• 解决方案：在 Tool 函数中，确保所有返回给 Node 的数据都是标准的基本类型（str、int、dict、list）。

---

十、上线检查清单与 FAQ

上线检查清单

• 你是否已经将 MemorySaver / InMemorySaver 彻底从生产依赖中剔除？
• 如果选用 SQLite，是否已经开启了 WAL 模式，并配置了数据自动 VACUUM 脚本？
• 如果选用 Redis，是否选用了 redis-stack 镜像，并关闭了自动内存淘汰淘汰机制以防丢失活跃状态？
• 如果选用 Postgres，连接初始化是否强制开启了 autocommit=True 和 row_factory=dict_row？
• 是否在环境变量里配置了 LANGGRAPH_STRICT_MSGPACK=true 保证反序列化安全？
• 你的 API 接口层是否对客户端传入的 thread_id 进行了强校验，防止多租户状态越权串线？
• 当 Tool 执行抛出异常时，Checkpointer 是否能正确保存 error_state 并挂起，以支持后续重试？

---

FAQ

MemorySaver 既然不适合生产，那它为什么还是官方示例里的常客？

因为它的唯一目的是降低教学门槛。官方为了让初学者不需要安装配置 SQLite、Redis 或 Postgres 就能快速跑通 Agent demo，所以默认使用内存 Saver。但在真实的商业级应用中，进程内存是不安全的临时介质。

如果我的 SQLite 文件损坏了，Agent 还能恢复吗？

如果数据库文件损坏且没有备份，该文件里的所有 thread 状态就彻底丢失了，Agent 只能选择重新开辟新的 thread 从头运行。对于使用 SQLite 的项目，建议每天深夜定时将 .db 文件备份归档到物理冷存储中（比如飞牛 NAS 的独立存储盘）。

Redis 的 ShallowRedisSaver 和标准 RedisSaver 到底怎么选？

如果你的业务场景不需要用户能“点击历史消息卡片恢复到特定历史版本”（即不需要 Time Travel），只要求 Agent 能记住当前最近一轮的会话状态，推荐使用 ShallowRedisSaver，它能帮助你节省 80% 以上的 Redis 内存消耗。

Postgres 的 Checkpointer 读写频繁是否会拖慢 Agent 的响应？

在高并发写入时，Postgres 的 I/O 开销确实比 Redis 的纯内存写入要大。如果性能成为瓶颈，可以在应用架构中采用读写分离，或者将短期对话状态用 Redis checkpointer 承载，只在流程彻底归档、或者触发 interrupt 等关键节点时，异步同步到 Postgres 做长期持久化审计。

---

系列导航

LangGraph 生产级 Agent 编排实战系列：

• 第 1 篇：Supervisor / Worker
• 第 2 篇：状态隔离
• 第 3 篇：Human-in-the-loop
• 第 4 篇：失败恢复
• 第 5 篇：Observability
• 第 6 篇：Checkpointer
• 第 7 篇：Subgraph

继续阅读

• LangGraph 状态隔离实战：thread_id、session_id、user_id 怎么设计？
• LangGraph Human-in-the-loop 实战：多智能体审批流怎么做？
• LangGraph 多智能体失败恢复：Tool Error、Timeout 与重试策略
• LangGraph Observability 实战：如何追踪每个 Agent 的决策路径？
• LangGraph 多智能体协作实战：Supervisor、Worker 与状态交接怎么设计？
• LangGraph Memory and Checkpointing for Production AI Agents