LangGraph 状态持久化：PyMySQLSaver 实战指南

做 LangGraph 应用，状态管理这件事真的别等到线上出事才回头补。我之前把一个客服工单机器人丢到测试环境跑，一开始图省事，直接用内存态MemorySaver，Demo 阶段一点问题没有。结果服务一重启，用户刚填了一半的工单信息全丢了，对方还以为程序出了什么大 Bug，气得直接在群里截图。后来我把PyMySQLSaver接进去，用 MySQL 做 LangGraph 的持久化 checkpointer，才算真正把状态管理的可靠性补上了。这篇就把我这次实战的整个思考、接线过程和踩坑记录写出来，给正在琢磨 LangGraph 状态管理方案的同学一个参考。

1. 为什么必须把状态从内存里搬出来

1.1 内存态的三个痛点：重启、并发、审计

很多人跑 LangGraph 的第一个 Hello World，用的都是MemorySaver。它的实现很简单，所有 checkpoint 都存在进程内存里，读取速度极快，代码也省事。但一旦进入稍微正式一点的环境，它的短板会暴露得特别明显。

首先就是重启丢状态。任何一次发布、回滚、OOM 重启，进程内存里的检查点全部清空。对单轮问答影响不大，但对多轮对话、工单流程这种需要跨请求保持上下文的应用来说，等同于用户每次都要从头开始。其次，内存态无法跨实例共享。只要负载均衡后面挂了两个副本，同一个用户请求被路由到不同实例，状态就对不上，表现就是对话上下文错乱。最后一个是审计问题。内存里没有持久化记录，出了问题你想回放某次执行的完整轨迹，完全没有依据。

所以“把状态持久化”不是锦上添花，而是做真实业务的基本前提。LangGraph 本身也提供了 checkpointer 抽象，内置了 SQLite、Postgres 等实现，核心思路一致：在图执行的每个关键节点之间把状态存下来，让图可以暂停、恢复、重试。

1.2 为什么我选了 MySQL 而不是文件型方案

选型的时候其实犹豫过。SQLite 方案部署最简单，一个文件搞定，但并发写入能力太弱。我们这边客服机器人并发场景虽然不算极端，但多个 worker 同时写一个 SQLite 文件，锁竞争就很明显，高并发下容易出现database is locked。Postgres 方案很成熟，生态也好，但如果团队已经有 MySQL 基础设施，为 LangGraph 单独引一套 Postgres，运维成本不划算。

PyMySQLSaver 的价值在于，它把 LangGraph 的 checkpoint 逻辑无缝对接到 MySQL 上。MySQL 的 InnoDB 提供了事务、行级锁、主从复制这些能力，既能保证写入的原子性，又能通过主从架构做高可用。对绝大多数已经有 MySQL 业务库的团队来说，这就是“零新依赖”的最优解。

2. Checkpoint 在 LangGraph 里到底是怎么存进去的

2.1 一个 checkpoint 的生命周期

要理解 PyMySQLSaver 做了什么，得先明白 LangGraph 的 checkpoint 机制。你可以把 LangGraph 的一整次图执行想象成一条流水线，每个节点是一个工位，节点与节点之间的产物就是状态。checkpoint就是某个工位完成后的“快照”，记录了当前所有状态值、执行到哪一步、父级 checkpoint 是谁。

默认配置下，LangGraph 在每次节点执行结束后都会生成一个新的 checkpoint，存起来后继续执行下一个节点。如果中途进程崩溃，重启后只需要拿着同一个thread_id去找最近的 checkpoint，就能从断点继续，而不是从头开始。PyMySQLSaver 就是这套机制的 MySQL 落地版本：它接收 LangGraph 引擎传过来的 checkpoint 数据，序列化后写进 MySQL 表里；需要恢复时，再按thread_id和checkpoint_id查出来反序列化，还原成完整状态。

这个设计里最关键的一点是thread_id。它类似于业务里的会话 ID，多个请求之间靠它来识别“这是同一段对话”。没有这个 ID，图就不知道自己该恢复哪一段执行过程。

2.2 PyMySQLSaver 的存储结构与序列化约定

PyMySQLSaver 落地到 MySQL 后，核心会用到一张checkpoints表，关键字段大致如下：

表结构本身不复杂，但序列化方式很容易被忽略。LangGraph 默认的序列化器是JsonPlusSerializer，它能在标准 JSON 基础上处理datetime、Decimal等常见类型。如果你的状态里塞了自定义类对象、bytes之类的数据，默认序列化器可能直接抛异常，或者存进去再读出来变成奇怪的东西。我的建议是：放进 state 的数据，尽量保持 JSON 友好，复杂业务对象只存 ID，具体数据靠 ID 到业务库去查。

3. 接入 PyMySQLSaver 的完整过程

3.1 安装、建库、初始化

PyMySQLSaver 属于 LangGraph 官方维护的扩展包，安装命令很简单：

pip install "langgraph-checkpoint-mysql"

如果你已经装过langgraph，这个包会自动把核心依赖带齐。安装完成后，先在 MySQL 里建一个专用库，不建议直接跟业务表混在同一个库里，方便后面做备份和权限隔离。

CREATE DATABASE langgraph_state DEFAULT CHARACTER SET utf8mb4;

接下来是初始化 saver。PyMySQLSaver 提供了从连接字符串直接初始化的方式，它会自动帮你建表，这一步非常省事。

from langgraph.checkpoint.mysql import PyMySQLSaver saver = PyMySQLSaver.from_conn_string( "mysql://user:password@127.0.0.1:3306/langgraph_state?charset=utf8mb4" ) # 自动创建 checkpoints 表及配套表 saver.setup()

注意连接串里的charset=utf8mb4一定不能省。LangGraph 状态里一旦有中文，没有这个参数就会出现乱码，而且是那种存进去正常、读出来全是问号的诡异问题。

3.2 连接池参数怎么给才不踩坑

直接用from_conn_string很简单，但它内部创建的连接可能没有针对高并发做优化。我在实际项目里更推荐用 SQLAlchemy 引擎显式传入，这样能精细控制连接池行为。

from sqlalchemy import create_engine from langgraph.checkpoint.mysql import PyMySQLSaver engine = create_engine( "mysql+pymysql://user:password@127.0.0.1:3306/langgraph_state?charset=utf8mb4", pool_size=5, # 连接池保持的最小连接数 max_overflow=10, # 峰值时可额外创建的连接数 pool_pre_ping=True, # 每次取连接前探活，避免用到失效连接 pool_recycle=3600, # 连接超过1小时强制回收重建 echo=False, ) saver = PyMySQLSaver(engine) saver.setup()

这里我想重点说下pool_pre_ping和pool_recycle。MySQL 默认的wait_timeout通常是 8 小时，但中间只要发生一次网络抖动、MySQL 重启，连接就可能已经失效，SQLAlchemy 连接池却不知道。没有pool_pre_ping的话，你会在运行到某个节点时突然报Lost connection to MySQL server during query，并且还不是必现，排查起来特别崩溃。pool_recycle则是主动让长连接定期重建，减少被服务端掐断的概率。

3.3 把 checkpointer 挂进图的代码示例

初始化完成后，接入图本身就三行代码的事。

from langgraph.graph import StateGraph builder = StateGraph(ConversationState) builder.add_node("collect_info", collect_info) builder.add_node("confirm_order", confirm_order) builder.add_edge("collect_info", "confirm_order") # 关键一步：编译时传入 checkpointer app = builder.compile(checkpointer=saver) config = {"configurable": {"thread_id": "order-10086"}} result = app.invoke({"input": "我想订一杯美式"}, config)

之后每次调用带上同一个thread_id，LangGraph 就会自动从 MySQL 里读取该会话的历史状态，继续往后执行。这里有一个初学者经常混淆的点：thread_id不是图节点里的参数，它是传给config的。图内部如果需要读取当前会话 ID，可以从config["configurable"]["thread_id"]里取，而不是在 state 里自己维护一个字段。

如果需要在服务重启后恢复某个会话，代码更简单，甚至不需要重新跑到上次断点，直接查询状态即可：

state_snapshot = app.get_state(config) print(state_snapshot.values)

这个接口在做“会话找回”“人工查看当前流程走到哪一步”这类功能时非常有用。

4. 高可靠性背后的几个关键细节

4.1 事务保障：要么写完整，要么不写

“高可靠性”这个词很容易变成口号，落到技术层面，首先要靠数据库事务保证 checkpoint 写入的原子性。LangGraph 引擎在调用 saver 写入 checkpoint 时，会把当前执行步的检查点连同元数据一起，在一个事务里写入。InnoDB 的原子性保证了要么全部落盘，要么全部不落盘，不会出现 checkpoint 写了一半、元数据却丢了这种情况。

这在实际运行中非常重要。比如图执行到第三个节点写 checkpoint 的瞬间，数据库连接断了。如果没有事务，可能出现checkpoints表里有一条残缺记录，下次恢复时读出来状态不完整，下游节点拿到的 state 缺字段，引发更难排查的运行时报错。有了事务，这个中途失败的操作会被整体回滚，重启后系统会自动回退到上一个完整的 checkpoint，用户最多重试一次，不会出现脏数据。

4.2 进程重启与多副本恢复

进程重启的恢复流程，我实际走了一遍之后才真正理解 checkpointer 的设计意图。某次版本发布，新代码有 bug，多个节点执行到一半就崩了。修复后服务重新拉起，用户并没有反馈说“我得从头再来”，因为他们下一次请求还是带着原来的thread_id，LangGraph 启动后自动去 MySQL 查最近的 checkpoint，很自然地从上次完成的节点继续往下走。

多副本场景更是 MySQL 方案的主场。客户端请求经过负载均衡，同一用户的不同请求可能打到不同实例上，但只要所有实例连的是同一个 MySQL，状态就是共享的，不存在“A 实例不知道 B 实例干了什么”的问题。部署结构上，每个 Python 进程仍持有自己的连接池，互相独立，数据库层做统一状态收敛。

不过要注意，同一个thread_id的并发写入尽量要避免。如果两个不同节点同时触发同一个会话的 checkpoint 写入，可能出现后写覆盖前写的情况。LangGraph 本身的执行模型默认是单线程推进一个图实例，所以常规用法不会触发这个问题，但你如果自己写了并发的任务分发逻辑，就要保证同一个thread_id不会被两个执行上下文同时操作。

4.3 中断恢复与人工介入状态

LangGraph 的持久化 checkpoint 还有一个隐藏价值，就是支持真正意义上的人工介入。我们客服机器人在确认订单之前，需要人工审核一下金额，这个场景可以在节点里用interrupt中断执行。

from langgraph.types import interrupt def confirm_order(state): # 中断图执行，等待外部输入 decision = interrupt({"order": state["order_preview"]}) return {"confirmed": decision["approved"]}

图执行到这里会暂停，生成的 checkpoint 已经落库。等人工在后台点击“通过”或“驳回”后，再带上同一个thread_id继续invoke，图会从刚才中断的地方恢复执行，而不是重头再来。我把这个功能叫做“人机协同的断点续传”，在审批流、复杂工单、风险控制这类场景里特别有用。

这个能力的前提就是 checkpointer 必须持久化。如果还用内存态，中断后进程一重启，恢复就无从谈起。所以 PyMySQLSaver 不只是解决“防丢失”，它直接解锁了一批业务形态。

5. 线上最容易翻车的四个问题及排查

5.1 MySQL server has gone away

这是个高频报错，而且出现时机很随机。我先说结论：大概率是连接池里的连接被 MySQL 服务端断开了，但客户端不知道，拿着死连接去执行查询。

我的排查链路分三步走。第一步，确认是不是wait_timeout太短导致空闲连接被回收，执行SHOW VARIABLES LIKE 'wait_timeout'查看。第二步，检查 SQLAlchemy 引擎有没有配pool_pre_ping=True，没有就先加上。第三步，把pool_recycle调到小于 MySQL 的wait_timeout，比如数据库超时时间是 8 小时，连接池回收就设 1 小时，确保连接一定在服务端掐断前被重做。

这一步操作完之后，MySQL server has gone away基本绝迹。如果还出现，就检查网络层，看是否有防火墙、负载均衡设备掐了空闲连接。

5.2 中文乱码和序列化异常

乱码问题基本都出在连接串没指定charset=utf8mb4。MySQL 默认字符集如果还是老旧的latin1，中文状态存进去再读出来就会变成一串问号。

序列化异常则是另一类。LangGraph 的默认序列化器能处理 JSON 通用的数据类型，以及datetime、UUID这类常见扩展，但如果你在 state 里放了自定义对象，比如某个机器学习模型实例，写入 checkpoint 时会直接抛异常。我的处理原则是：

• state 里只放纯数据，不放对象实例；
• 复杂对象先用序列化工具转成 dict 再放进去；
• 读取时按需在节点内部重建对象。

这样既保证了 checkpoint 可持久化，也让每个节点更容易调试。

5.3 checkpoint 表无限膨胀

checkpoints表的写入频率比我想象的高。默认情况下，每个节点执行完都会写一条，意味着一个 10 节点的图跑完一轮，表里就多 10 条记录。如果业务流程长、调用量大，表膨胀速度非常快，不仅占空间，还会拖慢按thread_id查最新 checkpoint 的速度。

我的做法是写一个定时清理任务，按业务场景保留策略删除过期检查点。比如客服会话只保留最后 7 天，工单流程结束后保留最新 3 个断点用于回放，其余全部清理。

-- 清理示例：删除指定时间之前的所有 checkpoints DELETE FROM checkpoints WHERE checkpoint_id < DATE_SUB(NOW(), INTERVAL 7 DAY);

清理任务低频跑就行，比如每天凌晨一次。注意别和大业务高峰重叠，避免锁竞争。

5.4 同一 thread 并发写冲突

前面提过同一个thread_id不应该被并发执行，但多副本架构里偶尔还是会因为代码 bug 出现这种情况。表现是，A 实例刚写入 checkpoint，B 实例紧接着把同一个会话的 checkpoint 覆盖了，下游节点读到的状态是“混合”的，逻辑上完全错乱。

最稳妥的防御是在业务入口做会话锁。同一个thread_id的请求强制路由到同一个实例，或者用 Redis 分布式锁保证同一时刻只有一个 worker 在推进这个会话。虽然 PyMySQLSaver 本身有事务机制，但它是保证单次写入的原子性，不是保证业务层面的执行互斥，这两者不能混为一谈。

6. 状态管理方案选型：PyMySQLSaver 不是唯一答案

6.1 常见 Saver 横向对比

如果你的应用还处于原型阶段，MemorySaver完全够用，别一上来就搞数据库，反而影响迭代速度。当你开始考虑多副本部署，或者需要支持跨重启恢复，就应该切换成 MySQL 或 Postgres 方案。

6.2 我选型时的判断标准

我实际选型主要看三个问题：团队有没有现成的数据库基础设施？业务对状态一致性的要求有多高？运维是否愿意为状态存储额外引入新组件？

对大多数已经有 MySQL 的团队，PyMySQLSaver 是性价比最高的选择。不需要额外维护一套新数据库，MySQL 的主从备份、监控告警体系都能直接复用。如果团队本身已经在用 Postgres，那直接用 PostgresSaver 也很合理，核心思路完全一样，只是存储层不同。

6.3 什么时候需要 SQLite

单机部署的小工具、本地桌面应用、或者是离线脚本，用SqliteSaver就足够。它不需要单独起数据库服务，一个文件搞定状态持久化，配合 Python 的sqlite3标准库，零运维成本。

但要注意，SQLite 的并发写性能和 MySQL 不是一个量级。如果同一个 LangGraph 应用会被多个进程同时使用，并且状态写入频繁，SQLite 文件锁会很快成为瓶颈。这种情况下直接上 MySQL，别在 SQLite 上做性能优化，性价比太低。

我自己现在这个项目的状态管理，触达用户的全链路走 MySQL 主库，备份库用来做只读分析，基本满足需求。等并发量再涨一个量级，大概率会引入读写分离，或者把状态表按thread_id做分片。这种演进路径是 MySQL 方案天然支持的，也是我当初选它的一个重要原因。

回到开头那个失忆的客服机器人，接入 PyMySQLSaver 之后再没出现过“用户眼巴巴看着进度条回到零点”的情况。如果你也在为 LangGraph 应用的状态持久化烦恼，先从最小可用的 MySQL checkpoint 接起来，再逐步处理连接池、清理策略这些细节，会比一开始就追求完美方案稳得多。