LangGraph状态持久化实战：用PyMySQLSaver解决进程重启丢失上下文

讲个真实的坑。我之前用 LangGraph 搭本地 Agent 服务调试得好好的一重启进程对话上下文全没了。排查半天发现是checkpointer配成了默认的内存实现进程一死状态就跟着蒸发。后来换成了 PyMySQLSaver把检查点落到 MySQL问题才彻底解决。这篇博文就把整个过程拆开讲为什么要做状态持久化、PyMySQLSaver 的核心机制、怎么接、以及我实际踩过的坑。如果你正在用 LangGraph 做多轮对话、长流程 Agent或者想把状态从进程内存里“搬”到数据库里这篇内容可以直接照着改。1. 为什么我会盯上 PyMySQLSaver状态丢了才懂持久化的价值1.1 没有 checkpointer 的 LangGraph 有多脆弱先说个最基础的问题LangGraph 本身是一个图编排框架StateGraph 里的每个节点都会更新状态但这些状态默认只存在于内存里。只要进程活着一切看起来都正常进程一重启内存释放所有对话记录、中间变量、已执行步骤的信息全部归零。我最初踩这个坑是在做一个客服工单处理的 Agent。流程里有个节点叫“收集用户意向”用户说完问题Agent 需要记住用户已经填过哪些字段然后在后续节点里继续追问。单次请求内没问题但一旦用户隔几分钟再发一条消息进程已经重启过前面收集的信息全部丢失Agent 又开始从头问一遍。这个问题的根源不是代码逻辑而是 LangGraph 的检查点机制没有接上持久化后端。LangGraph 允许你在编译图的时候传入checkpointer它会在每次节点执行前、执行后自动保存一份快照。如果没有配置就等于没存档玩游戏随时可能重来。1.2 状态管理方案选型对比LangGraph 生态里常用的 checkpointer 大概有这几种方案存储位置持久化适用场景明显短板MemorySaver进程内存否本地调试、单元测试进程重启即丢SqliteSaver本地文件是单机演示、小流量原型并发能力弱不适合多实例PostgresSaverPostgreSQL是生产级适合已有 PG 团队需要引入 PG 运维PyMySQLSaverMySQL是生产级适合已有 MySQL 团队需要设计表结构依赖连接池稳定我最终选 PyMySQLSaver不是因为 Postgres 不好而是我们团队本来就有一套成熟的 MySQL 主从环境监控、备份、权限管理都是现成的。为多一个功能再引入一套 PostgreSQL运维成本是实打实的增长。MySQL 的 BLOB/JSON 字段也足够存放 LangGraph 的检查点数据性能在中小流量下完全够用。另外提一句SqliteSaver 在单机小项目里确实方便但 LangGraph 的检查点写入机制需要支持并发连接SQLite 在并发写场景下会出现database is locked的问题。我之前用 SQLite 做联调时一旦多个请求同时触发 checkpoint 写入报错概率很高。所以只要你有跑服务的打算尽量直接上 MySQL。1.3 MySQL 凭什么能扛住检查点写入LangGraph 的检查点写入本质上是一种“追加式”操作每执行一次节点就往表里插一条新的 checkpoint 记录。这个负载模式对 MySQL 非常友好InnoDB 的主键顺序插入性能很高基本不会出现随机写导致的页分裂问题。另外MySQL 支持事务LangGraph 的检查点写入会包裹在事务里。这样可以保证图执行的原子性如果某个节点执行到一半挂了数据库里要么有完整的检查点要么没有不会出现“写了一半下次恢复时数据对不上”的脏状态。你可能会问检查点写入频繁会不会拖慢图执行实测下来单个 checkpoint 的大小通常在几 KB 到几十 KB看状态里塞了什么东西。对于普通 Agent 场景每轮对话写入一次检查点的延迟在毫秒级可以接受。如果状态里放了超大的 embedding 数组或者长文本那就要考虑精简状态、只保存必要字段否则数据库写入的负担会明显上升。2. PyMySQLSaver 的运行机制与核心概念2.1 checkpointer 到底是什么我们可以把 LangGraph 的图理解成一条流水线节点是流水线上的工位状态是工位之间传递的物料。默认情况下物料只存在内存里流水线一断电物料全没了。checkpointer 就是给每个工位装了一台“自动拍照机”每完成一个操作就拍一张快照快照存档到数据库。LangGraph 不是只存最后一版状态而是存一个版本的链条。每次节点执行完都会生成一个新的checkpoint_id并且记录它对应的parent_checkpoint_id。这样以后不仅能恢复最新状态还能回溯到任意一个历史步骤方便排查“上下文是从哪一步开始错的”。这个机制在调试里特别有用。举个例子Agent 在第三步推理出错如果你没有历史检查点只能靠日志猜。但有了检查点链条你可以直接把状态恢复到第三步执行前把当时的输入原样放回去重新跑一遍复现问题就变得非常容易。2.2 thread_id 与会话恢复thread_id是 PyMySQLSaver 恢复状态的关键钥匙。LangGraph 每次调用graph.invoke()或graph.stream()时都会从一个config里读取thread_id然后拿着这个 ID 去数据库里找最近的 checkpoint。简单理解thread_id就是会话 ID。同一个thread_id的多次调用共享状态不同thread_id之间完全隔离。如果你的业务是一个用户一个会话那就用用户 ID 加会话 ID 拼成一个唯一串当thread_id如果是工单流程可以用工单号如果是任务流水线可以用任务 ID。实际操作中很容易犯一个错每次请求都生成新的thread_id导致 LangGraph 永远找不到上一轮的状态。这不是 PyMySQLSaver 的问题而是会话标识设计的问题。线程 ID 必须是业务上稳定的、可复现的不能是随机数。2.3 表结构与序列化方式PyMySQLSaver 内部会维护一张检查点表我实际用到的表结构大致是这样的CREATE TABLE checkpoints ( thread_id varchar(255) NOT NULL, checkpoint_ns varchar(255) NOT NULL DEFAULT , checkpoint_id varchar(255) NOT NULL, parent_checkpoint_id varchar(255) DEFAULT NULL, type varchar(255) DEFAULT NULL, checkpoint blob NOT NULL, metadata blob NOT NULL, PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;不同版本的表结构可能有出入核心思想是一致的以thread_id区分会话以checkpoint_id区分版本checkpoint字段保存 JSON 序列化后的完整状态metadata保存执行步骤信息、来源节点、写入时间等附加说明。这里值得注意一点checkpoint字段保存的内容不是简单的“最终结果”而是 LangGraph 内部所有 channel 的值。如果你在 State 里定义了三个字段那么每个字段都能在 checkpoint 里找到对应值。这也就是为什么 LangGraph 能精确恢复“执行到哪一步、当时所有变量是什么”。序列化格式以 JSON 为主所以 State 里的数据必须能 JSON 序列化。如果 State 里有自定义对象、函数、文件句柄之类的东西写入数据库时会直接报错。我在第四节会专门讲这个问题。3. 实操落地用 PyMySQLSaver 实现跨进程状态恢复3.1 环境准备与安装先装依赖。我这里用的是 PyMySQL 驱动配合 SQLAlchemy 连接池所以需要安装这几个包pip install langgraph langgraph-checkpoint-mysql pymysql SQLAlchemy如果你的 LangGraph 版本比较新包名和 API 可能有微调建议安装后先跑一下python -c from langgraph.checkpoint.mysql import PyMySQLSaver验证导入是否正常。API 以你安装版本的官方文档为准我这个是实战经验不是版本说明书。接下来在 MySQL 里建一个库并给应用分配专属账号CREATE DATABASE langgraph_state DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; CREATE USER langgraph% IDENTIFIED BY your_password; GRANT ALL PRIVILEGES ON langgraph_state.* TO langgraph%; FLUSH PRIVILEGES;表不用手动建PyMySQLSaver 首次初始化时会尝试自动创建。但如果你的数据库账号没有 DDL 权限就需要手动执行类似 2.3 节里的建表语句。我建议生产环境还是手动建表把表结构纳入版本管理避免运行时自动 DDL 引发权限问题。3.2 编写第一个带持久化的 StateGraph下面这个例子很简单但能完整演示持久化的效果。我定义了一个计数器状态每调用一次节点计数器加一并记录当前是哪一次推理。from typing import TypedDict from langgraph.checkpoint.mysql import PyMySQLSaver from langgraph.graph import StateGraph, START, END class State(TypedDict): counter: int last_message: str def increment(state: State) - dict: n state.get(counter, 0) 1 return {counter: n, last_message: f这是第 {n} 次推理} builder StateGraph(State) builder.add_node(increment, increment) builder.add_edge(START, increment) builder.add_edge(increment, END) # 关键通过 PyMySQLSaver 连接 MySQL saver PyMySQLSaver.from_conn_string( mysqlpymysql://langgraph:your_password127.0.0.1:3306/langgraph_state ) graph builder.compile(checkpointersaver) config {configurable: {thread_id: demo-session-001}} # 第一次调用 result graph.invoke({}, config) print(result[counter]) # 输出 1 print(result[last_message]) # 输出 这是第 1 次推理 # 第二次调用同一个 thread_id result graph.invoke({}, config) print(result[counter]) # 输出 2注意graph.invoke()第二参必须传config而且config里必须有configurable.thread_id。否则 checkpointer 不知道该去数据库里找哪个会话的状态结果就会退化成普通内存调用。from_conn_string是 PyMySQLSaver 提供的快捷方法本质上它会内部创建一个 SQLAlchemy engine。你也可以显式创建 engine 再传给PyMySQLSaverfrom sqlalchemy import create_engine from langgraph.checkpoint.mysql import PyMySQLSaver engine create_engine( mysqlpymysql://langgraph:your_password127.0.0.1:3306/langgraph_state, pool_pre_pingTrue, pool_recycle3600, ) saver PyMySQLSaver(engine)显式创建 engine 的好处是连接池参数完全由你控制这在多实例部署时非常重要。后面排查章节我会展开讲连接池问题。3.3 验证状态恢复同进程与跨进程刚才的例子只验证了同进程内的状态恢复看起来似乎和内存模式没区别。真正的差异在“跨进程”场景。你可以把下面这段代码单独存成一个recover.py随便新开一个 Python 进程运行from langgraph.checkpoint.mysql import PyMySQLSaver from langgraph.graph import StateGraph, START, END class State(TypedDict): counter: int last_message: str def increment(state: State) - dict: n state.get(counter, 0) 1 return {counter: n, last_message: f这是第 {n} 次推理} builder StateGraph(State) builder.add_node(increment, increment) builder.add_edge(START, increment) builder.add_edge(increment, END) saver PyMySQLSaver.from_conn_string( mysqlpymysql://langgraph:your_password127.0.0.1:3306/langgraph_state ) graph builder.compile(checkpointersaver) # 同一个 thread_id但这里是全新进程 config {configurable: {thread_id: demo-session-001}} result graph.invoke({}, config) print(result[counter]) # 输出 3说明前两次的状态被恢复了如果上面的代码输出3说明 LangGraph 成功从 MySQL 里读到了前面两轮的状态然后在最新状态基础上继续加一。这个能力对生产环境的意义是决定性的你的服务可以随意滚动发布、扩缩容用户会话不会因为后端进程重启而中断。如果你想去数据库里亲眼看一下可以执行SELECT thread_id, checkpoint_id, parent_checkpoint_id, LENGTH(checkpoint) AS size FROM langgraph_state.checkpoints WHERE thread_id demo-session-001 ORDER BY checkpoint_id DESC LIMIT 5;你会看到一条条 checkpoint 记录最新的那一条里面counter字段的值就是3。3.4 接入 LLM 对话场景的改造计数器例子毕竟太简单真实业务里更常见的是多轮对话。LangGraph 官方推荐在聊天场景里使用MessagesState它会自动维护一个messages列表每次新消息追加进去。结合 PyMySQLSaver整个对话历史就会被持久化到 MySQL。from typing import TypedDict from langchain_core.messages import AIMessage, HumanMessage, SystemMessage from langgraph.checkpoint.mysql import PyMySQLSaver from langgraph.graph import StateGraph, START, END from langgraph.graph.message import MessagesState class ChatState(MessagesState): # 可以额外扩展业务字段比如用户意图、当前步骤 user_intent: str def chat_node(state: ChatState) - dict: # 这里替换成你的模型调用逻辑 messages state[messages] last_user_message messages[-1].content return { messages: [AIMessage(contentf我收到你的消息了{last_user_message})], user_intent: handle_query, } builder StateGraph(ChatState) builder.add_node(chat, chat_node) builder.add_edge(START, chat) builder.add_edge(chat, END) saver PyMySQLSaver.from_conn_string( mysqlpymysql://langgraph:your_password127.0.0.1:3306/langgraph_state ) graph builder.compile(checkpointersaver) config {configurable: {thread_id: customer-9527}} messages [ SystemMessage(content你是一个客服助手), HumanMessage(content我想查一下订单状态), ] response graph.invoke({messages: messages}, config) print(response[messages][-1].content)第二次再来一条消息时只要thread_id还是customer-9527state[messages]会自动带上之前的历史消息。你可以把上一轮的结果原样打印出来看看LangGraph 会自动恢复。这里有一个实践建议thread_id不要直接用裸的用户 ID最好拼上会话编号。比如f{user_id}:{session_id}。因为一个用户可能同时开多个会话如果用裸用户 ID两个会话会互相串状态。我见过不止一次这种线上事故都是因为线程 ID 设计得太随意。4. 实战中的问题排查与维护心得4.1 高频报错与解决办法速查表下面是这段时间我实际遇到过的几类典型问题整理成了一张速查表报错现象常见原因解决办法Table langgraph_state.checkpoints doesnt exist数据库账号没有自动建表权限手动执行建表语句或给账号授予 DDL 权限NonJSONSerializableReturnErrorState 里塞了无法 JSON 序列化的对象只保存可序列化数据自定义对象转成 dict 后存入Lost connection to MySQL server during query连接空闲超时连接池里的连接失效设置pool_pre_pingTruepool_recycle小于 MySQL wait_timeoutDeadlock found when trying to get lock多个实例同时写同一个 thread_id用唯一thread_id约束业务并发必要时对同一会话做锁状态恢复后缺少字段新代码 State 定义了新字段旧 checkpoint 里没有代码里用state.get(new_field, default)兼容旧数据checkpointer is not providedcompile()时忘了传saver编译图时显式传checkpointersaver4.2 序列化与并发问题细节先说序列化。LangGraph 默认会把 checkpoint 序列化成 JSON 再写入 BLOB 字段。Python 里很多对象是不能直接 JSON 序列化的比如datetime、set、自定义类实例、IO 对象。如果你在 State 里塞了这些写入时就会抛序列化错误。我的经验是State 里的字段尽量用基础类型字符串、数字、列表、字典。如果需要保存datetime先转成 ISO 格式字符串如果模型返回了特殊对象只保留对象里你关心的字段。这样做不仅是为了兼容序列化还能控制数据库体积避免把几十 MB 的中间结果都存进去。再说并发。同一个thread_id如果被多个请求同时调用可能会出现两个请求基于同一个旧 checkpoint 各写各的新状态导致状态覆盖。这在业务设计上是一大隐患。处理方式有两种。第一种是从业务上保证同一个会话串行执行比如给同一个thread_id的消息队列加锁或者只允许单实例处理指定会话。第二种是接受重叠发生在更新时利用parent_checkpoint_id做冲突检测如果发现要更新的父检查点不是最新版本就重新读取最新检查点再执行。我实际项目里用的是第一种简单可靠不用在 LangGraph 层做复杂合并逻辑。4.3 连接池与多实例注意事项生产环境基本都是多实例部署这时候数据库连接池的参数就很重要了。PyMySQLSaver 内部用的是 SQLAlchemy默认连接池配置不一定适合你的流量建议显式创建 engine 来控制参数。我常用的配置如下from sqlalchemy import create_engine engine create_engine( mysqlpymysql://langgraph:your_password127.0.0.1:3306/langgraph_state, pool_size10, max_overflow20, pool_pre_pingTrue, pool_recycle3600, )pool_pre_pingTrue会在从连接池取出连接时先做一个轻量 ping如果连接已经断开就丢弃并新建一个。这个参数能有效缓解 MySQL 空闲超时造成的Lost connection报错。pool_recycle也非常关键建议设置成小于 MySQL 的wait_timeout值。默认wait_timeout是 8 小时如果连接在池里闲置超过这个时间MySQL 会主动断开而应用不知道下一次查询时就出问题。设成 3600 秒基本能避开这个雷。多实例部署时还要注意不要给数据库无限扩容连接数。每个实例 10 到 20 个连接已经足够应付大部分场景。如果流量特别大优先考虑在应用层做限流而不是靠堆连接数解决问题。4.4 数据增长与清理策略有了持久化就必然面临数据增长。每轮对话都会产生新的 checkpoint如果用户量大这张表会涨得很快。我见过一个项目跑了两个月checkpoints 表膨胀到几十 GB查询恢复状态的延迟明显上升。清理策略不能一刀切。对话进行中的状态要保留历史会话的状态可以按业务留存周期来删。我常用的方案是两层第一层只保留每个thread_id最近 N 条 checkpoint。因为恢复状态只需要最新的那条历史 checkpoint 更多是调试用的普通人根本不会回看。如果确实需要回溯可以单独做归档。删除语句大致长这样DELETE c FROM checkpoints c JOIN ( SELECT thread_id, checkpoint_id, ROW_NUMBER() OVER (PARTITION BY thread_id ORDER BY checkpoint_id DESC) AS rn FROM checkpoints ) t ON c.thread_id t.thread_id AND c.checkpoint_id t.checkpoint_id WHERE t.rn 20;第二层按时间删除整个会话。如果业务规定会话数据保留 30 天那就定期清理超过 30 天的记录DELETE FROM checkpoints WHERE thread_id IN ( SELECT thread_id FROM ( SELECT thread_id, MAX(metadata-$.written_at) AS last_write FROM checkpoints GROUP BY thread_id HAVING last_write NOW() - INTERVAL 30 DAY ) old_sessions );注意MySQL 里不允许直接对同一张子查询做 DELETE所以多包了一层 SELECT。具体的清理频率和保留周期根据你自己的业务要求定。我建议在低峰期跑定时任务避免大事务锁表影响线上。4.5 连接不是越多越好数据不是越全越好这是我整体调试完 PyMySQLSaver 之后最大的体会。很多人看到“高可靠性的状态管理”第一反应是把所有状态都存进去连接池开到最大表里记录越多越好。实际跑下来你会发现真正决定可靠性的往往不是存储量而是会话标识是否稳定、连接池参数是否合理、序列化边界是否清楚。把 state 里可省的大字段省掉把thread_id当一等公民来设计把连接池参数调对再用定时任务管住数据膨胀这套组合下来LangGraph 的状态管理才能算真正可靠。MySQL 本身不是万能的但作为团队基础设施里已经成熟的一环用它来承载 LangGraph 的状态持久化性价比很高。