第一次把 LangGraph 应用从“能跑”推到“生产可用”的时候我碰到的第一道硬墙就是持久化执行。光把图跑通谁都会但 Agent 跑着跑着进程一重启或者用户会话一断开整个对话状态连渣都不剩那前面做的一切交互设计都等于白做。这篇我想把 LangGraph 的持久化执行从原理到实践完整拆一遍它到底在持久化什么、checkpointer 是怎么介入图执行的、Thread 状态怎么管理以及三种落地方案的选型和避坑。如果你正在用 LangGraph 做有状态的 Agent、客服对话、多轮工具调用这类应用这篇内容就是按我实际踩坑的顺序整理的建议收藏了慢慢对着做。1. 先弄明白LangGraph 里到底什么需要持久化1.1 图执行状态到底是什么LangGraph 的核心模型是 StateGraph。你定义一堆节点这些节点之间用边连接起来然后图的每个节点在执行时读一个共享的 State执行完把结果写回这个 State。节点之间的数据传递不靠返回值靠的是这个全局可见的 State。所以这里的“执行状态”不是某个局部变量而是整张图在某个时刻所有节点共用的数据结构。它包含了消息历史、工具调用结果、用户输入、中间计算值甚至一些自定义的标记位。你可以把 State 想象成一张团队共用的白板每个节点在白板上写自己的产出下一个节点读白板继续干。白板上写了什么就是这张图的全部“记忆”。在 LangGraph 里State 的每个字段都由一个 Channel 管理Channel 定义了数据怎么被写入、合并和读取。比如消息列表这个字段新增消息时是追加还是覆盖由 Reducer 决定。Reducer 本质上就是两个值的合并函数比如add_messages会把新消息追加到已有消息列表后面。搞清楚 Reducer 的语义特别重要否则你恢复状态时看到的字段结果会莫名其妙。1.2 持久化解决的是“进程之外”的状态保存很多人的误区是我把 State 存在内存里应用不退出这不就持久化了吗其实这不叫持久化这只能算“暂存”。真正的持久化执行指的是 LangGraph 把图的执行状态写入外部存储让图可以在进程重启、机器宕机、网络断开之后从某个时间点继续执行。我用一个更直白的类比你写文档写了一半停电了。内存就好比你屏幕上没保存的文字断电就没了持久化是你在停电前点了“保存”下次开机打开文件还能从上次停下的位置接着写。LangGraph 的 checkpoint 机制就是这个“保存”按钮。那为什么 LangGraph 需要专门设计一套持久化机制因为图的执行并不总是线性的。它可以有条件分支、有循环、有中断等待人工输入、有多个节点并行。你恢复时不能只恢复最后的消息列表你得恢复“这个图执行到哪一步了”“哪些分支执行了”“哪些节点正在等待”。这些信息是普通的数据库存个 JSON 解决不了的它需要一套和运行时绑定的 checkpoint 格式。1.3 CheckpointerLangGraph 的“存档点”设计LangGraph 把持久化的核心抽象成了 Checkpointer。每个 Checkpointer 需要实现一个读写接口LangGraph 引擎在图执行的关键节点自动调用这个接口把当前图的完整状态写入存储。关键写入时机有三个图开始执行前每个节点执行完成后更准确地说是每个超步即 super-step 完成后图执行结束时。这里的超步我要稍微解释一下。LangGraph 会把图执行切分成多个步骤每个步骤里可以并行执行若干节点一个节点输出可能继续触发后续节点。每一个这样的步骤就是一个超步。Checkpointer 每个超步写一次存档所以你可以精确恢复到每个步骤结束时的状态而不是只能恢复到整个图跑完。这个设计有什么好处假设你的图里有一个调用第三方 API 的节点API 超时了图挂了。因为 checkpoint 在超时时已经写过一版存档你可以从这版存档继续而不是从用户第一次输入重新跑一遍。这在多轮 Agent 场景里收益极大用户感知就是“刚才问题没处理完换个说法再问一次也能接着上下文”。2. 核心概念拆解Thread、Checkpointer、State 三者如何协作2.1 Thread持久化的作用域Thread 是 LangGraph 里给状态划分的唯一标识。每个 Thread 有一个thread_id你可以把它理解为会话 ID、对话 ID甚至一个业务流程 ID。所有 checkpoint 都归属于某个 thread_id不同 thread 之间的状态完全隔离。这个设计在工程上非常好用。你在同一个 thread_id 下连续执行多次图LangGraph 会自动找到该 thread 最近的 checkpoint把新一次执行作为上一次的续跑如果你用一个新的 thread_id那相当于从零开始一个全新会话。实际开发里我习惯这样映射一个终端用户对应一个 thread_id或者一次多轮任务对应一个 thread_id。比如客服机器人每个用户手机号或用户 ID 就相当于是 thread_id 的制造来源固定下来用户回访时带上次没聊完的上下文。Thread 隔离还有一个好处并发安全。多个 thread 之间互不影响可以并行调度。这也是为什么 LangGraph 能做多会话并发的根基——每个会话都有自己的存档线。2.2 Checkpointer 的存储格式Checkpointer 持久化的内容比你想的多。它不仅要保存 State 本身还要保存图的执行位置和元信息。我用一张表拆解一下典型的 checkpoint 记录里有什么组成部分作用示例内容state图的全局状态快照messages、自定义字段、中间结果metadata描述该 checkpoint 的信息step 编号、写入时间、来源节点parent_checkpoint_id指向上一个 checkpoint用于构建状态链、回溯历史pending_writes尚未提交的节点输出中断或并行时的待写入数据versions各 channel 的版本号用于并发控制、增量恢复我在排查问题时发现理解parent_checkpoint_id很有用。它把同一 thread 下的所有 checkpoint 串成了一条链LangGraph 恢复时会沿着这条链找到最近的一个完整存档必要时还会重放若干 pending writes。也就是说你每次调用图生成的 checkpoint 不只是一个快照它是一份结构化的执行履历。这在调试“这轮对话为什么状态不对”的时候是救命神器。2.3 State 的序列化与 Reducer 的恢复逻辑持久化执行对 State 最大的挑战不是存不存得下而是怎么把 State 语义原样还原。LangGraph 默认会对 State 做序列化把 Python 对象转成可存储的字节流。保存的时候能序列化加载的时候能反序列化回来这一点任何人做半天都能想到。但恢复的时候有个隐藏问题State 字段是用 Reducer 合并的恢复时也需要执行同样的合并逻辑。比如 messages 字段的 Reducer 是add_messages新消息追加到尾部。当你恢复 checkpoint 时LangGraph 会把已保存的消息列表直接作为 State 的初始值然后重放之后发生的节点写入。如果这个过程对相同消息重复执行了两遍可能造成消息重复追加。所以正确理解 Reducer 的幂等性很重要。实际推荐的方案是让你的 Reducer 尽量做到“可重入”同一事件被重新应用一次最终得到的结果不应该变化。比如用消息的 ID 去重而不是盲目追加。如果你自己定义了 Reducer一定要想清楚重放场景下它是否安全。3. 动手实操三种持久化方案从配置到跑通理论说得再多不如直接把代码跑一遍。这一节我会按“开发调试 → 单机落盘 → 生产级”三个梯度给出可复制的配置方案。所有代码基于常见的 LangGraph 版本我用的是 Python 环境版本差异不大核心 API 基本一致。3.1 方案一MemorySaver五分钟先跑通持久化MemorySaver 是 LangGraph 内置的最简单 Checkpointer它把 checkpoint 存在内存字典里。注意这只适合本地调试进程一结束存档就没了。但它足够轻量能让你先把“持久化执行”的全流程跑通。from langgraph.graph import StateGraph, State from langgraph.checkpoint.memory import MemorySaver class MyState(State): messages: list step_count: int def step_one(state: MyState): return {step_count: state.step_count 1} def step_two(state: MyState): return {step_count: state.step_count * 2} checkpointer MemorySaver() builder StateGraph(MyState) builder.add_node(one, step_one) builder.add_node(two, step_two) builder.add_edge(one, two) builder.set_entry_point(one) builder.set_finish_point(two) graph builder.compile(checkpointercheckpointer) # 第一次执行指定 thread_id config {configurable: {thread_id: demo-thread-1}} result graph.invoke({messages: [], step_count: 1}, configconfig) print(result) # {messages: [], step_count: 4} # 第二次执行同一个 thread_id状态从上次结尾继续 result2 graph.invoke({messages: [], step_count: 100}, configconfig) print(result2) # 你传入的 step_count 会作为输入覆盖但 messages 等历史字段仍在注意这里的step_count是新输入覆盖了旧值因为我没有给step_count定义 reducer默认就是覆盖写。而 messages 字段如果定义了追加 reducer那你第二次调用时传入的新 messages 会和历史消息合并而不是替换。这一点我刚开始总是搞混放这里提醒你。3.2 方案二SqliteSaver单机落盘的入门首选MemorySaver 用法学会之后换到 SqliteSaver 几乎是无痛的。SqliteSaver 把 checkpoint 写到本地 SQLite 文件进程重启之后存档还在。对于单机应用、个人项目、内部工具这个方案已经够用。from langgraph.checkpoint.sqlite import SqliteSaver # 注意必须用 check_same_threadFalse因为 LangGraph 可能在多线程里访问存档 with SqliteSaver.from_conn_string(checkpoints.db, check_same_threadFalse) as checkpointer: graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: sqlite-thread-1}} result graph.invoke({messages: [], step_count: 1}, configconfig) print(result)这里我踩过一个坑如果在连接 SQLite 时不开check_same_threadFalseLangGraph 内部的多线程执行会直接抛异常提示 SQLite 对象只能被创建它的线程使用。加了参数之后瞬时就好。SqliteSaver 还有一个异步版本AsyncSqliteSaver用法几乎一样只是调用时用await graph.ainvoke(...)。如果你的项目用了 FastAPI 这类异步框架建议直接用异步版避免把事件循环堵住。3.3 方案三PostgresSaver生产环境的主力方案到了生产环境我推荐用 PostgresSaver。理由很简单支持真正的多实例并发、支持云端部署、有成熟的备份恢复生态。如果你的团队已经在用 PostgreSQL那就没有额外引入新存储的成本。from langgraph.checkpoint.postgres import PostgresSaver # 连接串按你的实际环境调整 conn_string postgresql://user:passwordlocalhost:5432/langgraph_db with PostgresSaver.from_conn_string(conn_string) as checkpointer: # 首次使用需要建表 checkpointer.setup() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: pg-thread-1}} result graph.invoke({messages: [], step_count: 1}, configconfig) print(result)setup()方法会自动创建 LangGraph 需要的表。如果表已存在它不会重复执行破坏性操作所以可以放心在生产部署脚本里调用。使用 PostgresSaver 时有个实际的好处你可以从另一台机器、另一个进程连接到同一个数据库读取同一个 thread_id 的状态。这意味着你的 Agent 服务可以水平扩展多个实例处理不同请求但共享同一份会话记忆。这是 MemorySaver 和 SqliteSaver 都做不到的。3.4 结合真实业务API 层如何传 thread_id框架层的配置说完最重要的问题是你的 HTTP 接口怎么和 LangGraph 对接。我推荐的做法是客户端请求里携带会话标识服务端把它映射成thread_id然后传给 LangGraph 的invoke。别让客户端直接传任意thread_id一定要做一层校验或映射否则用户传别人会话的 thread_id 就能读取他人状态了。# 一个简化的示例 def chat_endpoint(user_id: str, message: str): # 从自己的用户表映射会话标识不要直接用 user_id 当 thread_id 也不行 # 因为一个用户可能有多个会话。建议用 session_id。 thread_id f{user_id}:{session_id} config {configurable: {thread_id: thread_id}} result graph.invoke({messages: [{role: user, content: message}]}, configconfig) return {reply: result[messages][-1]}这一步看起来简单实则是生产事故高发区。我见过有人直接把内部 thread_id 原样暴露给前端结果用户抓包改了一下 thread_id读走了别人的会话记录。正确的做法是让服务端维护一份“用户 ↔ thread_id”的对照表永远不要让客户端指定任意值。4. 排查实录我踩过的坑和速查表这一节我把自己实际遇到的高频问题按症状整理出来了每个问题都给了排查思路和解决办法。建议直接对照表格排查。4.1 状态“丢”了的三种情况我排第一的问题是明明配置了 checkpointer但第二次调用时状态像没保存一样。检查顺序如下编译图的时候有没有传 checkpointer。有人用的是builder.compile()忘了把 checkpointer 参数传进去LangGraph 完全不会报错只是静默不持久化。config有没有传对。必须是{configurable: {thread_id: xxx}}如果少了configurable这一层LangGraph 找不到 thread_id每次都当新会话处理。数据库连接对不对。如果你换库了、换表了、连接串写错了都会导致读不到旧 checkpoint。排查时直接查数据库里的表看看有没有该 thread_id 的记录。现象可能原因解决办法第二次调用没有历史上下文编译时没传 checkpointer在compile(checkpointer...)中传入实例每次都是全新会话config 格式错误确保{configurable: {thread_id: ...}}层级正确换机器后状态丢失MemorySaver 只存内存换成 Sqlite/Postgres 持久化方案读不到历史但数据库有记录thread_id 拼写不一致打印日志确认请求传入的 thread_id与库中对比4.2 并发写同一个 thread_id 的冲突第二个高发问题是同一时刻对同一个 thread_id 发起多次ainvoke。LangGraph 的 checkpoint 机制对同一线程的并发写入处理并不像分布式数据库那么健壮两个并发的执行可能读到同一个旧 checkpoint然后各自写入新 checkpoint导致后写入的覆盖前一个。规避手段我在生产里用了两种应用层加线程锁或分布式锁保证同一 thread_id 串行执行。对于单机进程用 Python 的字典锁就行多机部署用 Redis 锁或者数据库行锁。一个用户一个活跃会话前端在发起新请求前取消旧请求从交互层面避免并发。如果你实在无法避免并发可以考虑为每次用户请求生成一个新的子 thread再通过parent_checkpoint_id关联回主 thread。但这个玩法比较复杂我建议优先从架构上规避不要硬杠。4.3 Checkpoint 越存越大Long-running 会话跑久了checkpoint 数量会线性增长。同一个 thread 下每次执行都产生若干个 checkpoint每个 checkpoint 里还存着完整消息历史。跑一周以后光 checkpoint 就能占掉几个 GB。这时候几个处理手段可以组合用定期清理旧的 checkpoint只保留最近 N 个或最近 N 天的归档已结束会话的 thread把它们从活跃表中移到归档表调整写入频率如果你不需要精确到每个超步的状态可以在部分子图上不配持久化。LangGraph 在不同版本里对清理接口的支持程度不一样我建议你直接写一个定时任务按thread_id和created_at删除旧记录配套一个数据保留策略文档。4.4 恢复后状态对不上这个问题比较阴它通常在自定义 Reducer 时爆发。症状是恢复后某些列表字段重复了或者某些字段值比预期多。根本原因是重放过程中同一段逻辑被执行了两次且 Reducer 不具备幂等性。排查方法是把同一条输入在同一 thread 上手动执行两次对比两次最终状态。如果两次结果不一致那问题基本锁定在自定义 Reducer 上。整改方案就是我前面说的为每条消息或记录增加唯一 IDReducer 合并前先按 ID 去重。这样即使重放也不会产生重复。5. 经验沉淀怎么设计一套适合自己的持久化方案5.1 选型四个维度决定了你用哪种 Checkpointer我在项目里总结了一套选型判断逻辑按优先级排列数据安全要求丢了会话状态会不会造成资损、客诉、业务中断会就必须用数据库方案部署架构单机进程还是多实例多实例必须用 Postgres 或等效的共享存储方案性能与规模每秒并发多少个 thread 上千级并发建议直接上 PostgresSqlite 写入锁会拖后腿运维成本团队有没有专职 DBA没有就老老实实 Sqlite配定时备份。方案数据保存位置跨进程共享适用场景运维成本MemorySaver进程内存不支持本地调试、单元测试零SqliteSaver本地文件单机多进程弱个人项目、内网工具低PostgresSaverPostgreSQL 数据库支持生产环境、多实例部署中5.2 进阶玩法人机交互与多会话编排持久化执行不只是防宕机它让一类特殊场景变得简单Human-in-the-loop也就是让 Agent 在关键节点停下来等人确认确认完再继续。LangGraph 的interrupt机制和 checkpoint 是绝配。图执行到某个节点时调用interrupt引擎会把当前状态写入 checkpoint然后暂停执行。外部系统接收人工反馈后用同一个 thread_id 再次调用图LangGraph 从暂停位置恢复执行。整个过程跨任意长时间因为状态已经在存储里了。我做过一个审批流程 AgentAgent 生成方案之后中断等待主管在界面点“批准”批准后同一 thread 继续执行后续动作。这个模式用在企业内部工具里价值非常高而且代码并不复杂——本质就是持久化执行加一个事件入口。5.3 容易被忽略的三个细节第一个是 metadata。Checkpointer 存的不只是 State还有 metadata。你可以往 metadata 里写用户 ID、业务单号、版本号等业务字段排查问题时直接查库就能定位是哪条业务的存档不用打开每条记录看内容。这个习惯建议从第一天就养成。第二个是测试环境要隔离。开发、测试、生产千万不要共用一个 checkpoint 数据库。我踩过一次坑测试脚本用随机 thread_id 跑并发直接把生产环境的 checkpoint 表打满了。隔离不仅是数据库实例隔离连 thread_id 前缀都建议做区分比如dev-、prod-。第三个是备份策略。Postgres 方案可以依赖数据库自身的备份这个没问题。但如果你用 Sqlite就得自己想方案定时把 .db 文件复制走或者用 dump 命令导出。别以为本地文件不需要备份磁盘坏了照样丢。5.4 我的实际体会持久化执行是所有有状态 Agent 应用的基石但它不是配完一个 checkpointer 就万事大吉的。LangGraph 给了你很好的“存档”能力但怎么组织 thread、怎么设计 Reducer、怎么清理 checkpoint、怎么并发隔离这些都得靠应用层自己拿捏。我个人的落地顺序建议是先用 MemorySaver 把逻辑跑通再换 Sqlite 做单机验证最后一步挪到 Postgres 并进行并发压测。每一步都验证一下状态恢复是否符合预期再往前推进。这样既不会在初期被存储问题拖慢进度也不会在后期被数据问题打个措手不及。最后补一个小技巧你可以在每次图执行结束后打印checkpointer.get(thread_id)返回的元数据确认写入确实发生了。这个习惯帮我快速发现过好几次生产环境连接串配错的问题。
第一次把 LangGraph 应用从“能跑”推到“生产可用”的时候我碰到的第一道硬墙就是持久化执行。光把图跑通谁都会但 Agent 跑着跑着进程一重启或者用户会话一断开整个对话状态连渣都不剩那前面做的一切交互设计都等于白做。这篇我想把 LangGraph 的持久化执行从原理到实践完整拆一遍它到底在持久化什么、checkpointer 是怎么介入图执行的、Thread 状态怎么管理以及三种落地方案的选型和避坑。如果你正在用 LangGraph 做有状态的 Agent、客服对话、多轮工具调用这类应用这篇内容就是按我实际踩坑的顺序整理的建议收藏了慢慢对着做。1. 先弄明白LangGraph 里到底什么需要持久化1.1 图执行状态到底是什么LangGraph 的核心模型是 StateGraph。你定义一堆节点这些节点之间用边连接起来然后图的每个节点在执行时读一个共享的 State执行完把结果写回这个 State。节点之间的数据传递不靠返回值靠的是这个全局可见的 State。所以这里的“执行状态”不是某个局部变量而是整张图在某个时刻所有节点共用的数据结构。它包含了消息历史、工具调用结果、用户输入、中间计算值甚至一些自定义的标记位。你可以把 State 想象成一张团队共用的白板每个节点在白板上写自己的产出下一个节点读白板继续干。白板上写了什么就是这张图的全部“记忆”。在 LangGraph 里State 的每个字段都由一个 Channel 管理Channel 定义了数据怎么被写入、合并和读取。比如消息列表这个字段新增消息时是追加还是覆盖由 Reducer 决定。Reducer 本质上就是两个值的合并函数比如add_messages会把新消息追加到已有消息列表后面。搞清楚 Reducer 的语义特别重要否则你恢复状态时看到的字段结果会莫名其妙。1.2 持久化解决的是“进程之外”的状态保存很多人的误区是我把 State 存在内存里应用不退出这不就持久化了吗其实这不叫持久化这只能算“暂存”。真正的持久化执行指的是 LangGraph 把图的执行状态写入外部存储让图可以在进程重启、机器宕机、网络断开之后从某个时间点继续执行。我用一个更直白的类比你写文档写了一半停电了。内存就好比你屏幕上没保存的文字断电就没了持久化是你在停电前点了“保存”下次开机打开文件还能从上次停下的位置接着写。LangGraph 的 checkpoint 机制就是这个“保存”按钮。那为什么 LangGraph 需要专门设计一套持久化机制因为图的执行并不总是线性的。它可以有条件分支、有循环、有中断等待人工输入、有多个节点并行。你恢复时不能只恢复最后的消息列表你得恢复“这个图执行到哪一步了”“哪些分支执行了”“哪些节点正在等待”。这些信息是普通的数据库存个 JSON 解决不了的它需要一套和运行时绑定的 checkpoint 格式。1.3 CheckpointerLangGraph 的“存档点”设计LangGraph 把持久化的核心抽象成了 Checkpointer。每个 Checkpointer 需要实现一个读写接口LangGraph 引擎在图执行的关键节点自动调用这个接口把当前图的完整状态写入存储。关键写入时机有三个图开始执行前每个节点执行完成后更准确地说是每个超步即 super-step 完成后图执行结束时。这里的超步我要稍微解释一下。LangGraph 会把图执行切分成多个步骤每个步骤里可以并行执行若干节点一个节点输出可能继续触发后续节点。每一个这样的步骤就是一个超步。Checkpointer 每个超步写一次存档所以你可以精确恢复到每个步骤结束时的状态而不是只能恢复到整个图跑完。这个设计有什么好处假设你的图里有一个调用第三方 API 的节点API 超时了图挂了。因为 checkpoint 在超时时已经写过一版存档你可以从这版存档继续而不是从用户第一次输入重新跑一遍。这在多轮 Agent 场景里收益极大用户感知就是“刚才问题没处理完换个说法再问一次也能接着上下文”。2. 核心概念拆解Thread、Checkpointer、State 三者如何协作2.1 Thread持久化的作用域Thread 是 LangGraph 里给状态划分的唯一标识。每个 Thread 有一个thread_id你可以把它理解为会话 ID、对话 ID甚至一个业务流程 ID。所有 checkpoint 都归属于某个 thread_id不同 thread 之间的状态完全隔离。这个设计在工程上非常好用。你在同一个 thread_id 下连续执行多次图LangGraph 会自动找到该 thread 最近的 checkpoint把新一次执行作为上一次的续跑如果你用一个新的 thread_id那相当于从零开始一个全新会话。实际开发里我习惯这样映射一个终端用户对应一个 thread_id或者一次多轮任务对应一个 thread_id。比如客服机器人每个用户手机号或用户 ID 就相当于是 thread_id 的制造来源固定下来用户回访时带上次没聊完的上下文。Thread 隔离还有一个好处并发安全。多个 thread 之间互不影响可以并行调度。这也是为什么 LangGraph 能做多会话并发的根基——每个会话都有自己的存档线。2.2 Checkpointer 的存储格式Checkpointer 持久化的内容比你想的多。它不仅要保存 State 本身还要保存图的执行位置和元信息。我用一张表拆解一下典型的 checkpoint 记录里有什么组成部分作用示例内容state图的全局状态快照messages、自定义字段、中间结果metadata描述该 checkpoint 的信息step 编号、写入时间、来源节点parent_checkpoint_id指向上一个 checkpoint用于构建状态链、回溯历史pending_writes尚未提交的节点输出中断或并行时的待写入数据versions各 channel 的版本号用于并发控制、增量恢复我在排查问题时发现理解parent_checkpoint_id很有用。它把同一 thread 下的所有 checkpoint 串成了一条链LangGraph 恢复时会沿着这条链找到最近的一个完整存档必要时还会重放若干 pending writes。也就是说你每次调用图生成的 checkpoint 不只是一个快照它是一份结构化的执行履历。这在调试“这轮对话为什么状态不对”的时候是救命神器。2.3 State 的序列化与 Reducer 的恢复逻辑持久化执行对 State 最大的挑战不是存不存得下而是怎么把 State 语义原样还原。LangGraph 默认会对 State 做序列化把 Python 对象转成可存储的字节流。保存的时候能序列化加载的时候能反序列化回来这一点任何人做半天都能想到。但恢复的时候有个隐藏问题State 字段是用 Reducer 合并的恢复时也需要执行同样的合并逻辑。比如 messages 字段的 Reducer 是add_messages新消息追加到尾部。当你恢复 checkpoint 时LangGraph 会把已保存的消息列表直接作为 State 的初始值然后重放之后发生的节点写入。如果这个过程对相同消息重复执行了两遍可能造成消息重复追加。所以正确理解 Reducer 的幂等性很重要。实际推荐的方案是让你的 Reducer 尽量做到“可重入”同一事件被重新应用一次最终得到的结果不应该变化。比如用消息的 ID 去重而不是盲目追加。如果你自己定义了 Reducer一定要想清楚重放场景下它是否安全。3. 动手实操三种持久化方案从配置到跑通理论说得再多不如直接把代码跑一遍。这一节我会按“开发调试 → 单机落盘 → 生产级”三个梯度给出可复制的配置方案。所有代码基于常见的 LangGraph 版本我用的是 Python 环境版本差异不大核心 API 基本一致。3.1 方案一MemorySaver五分钟先跑通持久化MemorySaver 是 LangGraph 内置的最简单 Checkpointer它把 checkpoint 存在内存字典里。注意这只适合本地调试进程一结束存档就没了。但它足够轻量能让你先把“持久化执行”的全流程跑通。from langgraph.graph import StateGraph, State from langgraph.checkpoint.memory import MemorySaver class MyState(State): messages: list step_count: int def step_one(state: MyState): return {step_count: state.step_count 1} def step_two(state: MyState): return {step_count: state.step_count * 2} checkpointer MemorySaver() builder StateGraph(MyState) builder.add_node(one, step_one) builder.add_node(two, step_two) builder.add_edge(one, two) builder.set_entry_point(one) builder.set_finish_point(two) graph builder.compile(checkpointercheckpointer) # 第一次执行指定 thread_id config {configurable: {thread_id: demo-thread-1}} result graph.invoke({messages: [], step_count: 1}, configconfig) print(result) # {messages: [], step_count: 4} # 第二次执行同一个 thread_id状态从上次结尾继续 result2 graph.invoke({messages: [], step_count: 100}, configconfig) print(result2) # 你传入的 step_count 会作为输入覆盖但 messages 等历史字段仍在注意这里的step_count是新输入覆盖了旧值因为我没有给step_count定义 reducer默认就是覆盖写。而 messages 字段如果定义了追加 reducer那你第二次调用时传入的新 messages 会和历史消息合并而不是替换。这一点我刚开始总是搞混放这里提醒你。3.2 方案二SqliteSaver单机落盘的入门首选MemorySaver 用法学会之后换到 SqliteSaver 几乎是无痛的。SqliteSaver 把 checkpoint 写到本地 SQLite 文件进程重启之后存档还在。对于单机应用、个人项目、内部工具这个方案已经够用。from langgraph.checkpoint.sqlite import SqliteSaver # 注意必须用 check_same_threadFalse因为 LangGraph 可能在多线程里访问存档 with SqliteSaver.from_conn_string(checkpoints.db, check_same_threadFalse) as checkpointer: graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: sqlite-thread-1}} result graph.invoke({messages: [], step_count: 1}, configconfig) print(result)这里我踩过一个坑如果在连接 SQLite 时不开check_same_threadFalseLangGraph 内部的多线程执行会直接抛异常提示 SQLite 对象只能被创建它的线程使用。加了参数之后瞬时就好。SqliteSaver 还有一个异步版本AsyncSqliteSaver用法几乎一样只是调用时用await graph.ainvoke(...)。如果你的项目用了 FastAPI 这类异步框架建议直接用异步版避免把事件循环堵住。3.3 方案三PostgresSaver生产环境的主力方案到了生产环境我推荐用 PostgresSaver。理由很简单支持真正的多实例并发、支持云端部署、有成熟的备份恢复生态。如果你的团队已经在用 PostgreSQL那就没有额外引入新存储的成本。from langgraph.checkpoint.postgres import PostgresSaver # 连接串按你的实际环境调整 conn_string postgresql://user:passwordlocalhost:5432/langgraph_db with PostgresSaver.from_conn_string(conn_string) as checkpointer: # 首次使用需要建表 checkpointer.setup() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: pg-thread-1}} result graph.invoke({messages: [], step_count: 1}, configconfig) print(result)setup()方法会自动创建 LangGraph 需要的表。如果表已存在它不会重复执行破坏性操作所以可以放心在生产部署脚本里调用。使用 PostgresSaver 时有个实际的好处你可以从另一台机器、另一个进程连接到同一个数据库读取同一个 thread_id 的状态。这意味着你的 Agent 服务可以水平扩展多个实例处理不同请求但共享同一份会话记忆。这是 MemorySaver 和 SqliteSaver 都做不到的。3.4 结合真实业务API 层如何传 thread_id框架层的配置说完最重要的问题是你的 HTTP 接口怎么和 LangGraph 对接。我推荐的做法是客户端请求里携带会话标识服务端把它映射成thread_id然后传给 LangGraph 的invoke。别让客户端直接传任意thread_id一定要做一层校验或映射否则用户传别人会话的 thread_id 就能读取他人状态了。# 一个简化的示例 def chat_endpoint(user_id: str, message: str): # 从自己的用户表映射会话标识不要直接用 user_id 当 thread_id 也不行 # 因为一个用户可能有多个会话。建议用 session_id。 thread_id f{user_id}:{session_id} config {configurable: {thread_id: thread_id}} result graph.invoke({messages: [{role: user, content: message}]}, configconfig) return {reply: result[messages][-1]}这一步看起来简单实则是生产事故高发区。我见过有人直接把内部 thread_id 原样暴露给前端结果用户抓包改了一下 thread_id读走了别人的会话记录。正确的做法是让服务端维护一份“用户 ↔ thread_id”的对照表永远不要让客户端指定任意值。4. 排查实录我踩过的坑和速查表这一节我把自己实际遇到的高频问题按症状整理出来了每个问题都给了排查思路和解决办法。建议直接对照表格排查。4.1 状态“丢”了的三种情况我排第一的问题是明明配置了 checkpointer但第二次调用时状态像没保存一样。检查顺序如下编译图的时候有没有传 checkpointer。有人用的是builder.compile()忘了把 checkpointer 参数传进去LangGraph 完全不会报错只是静默不持久化。config有没有传对。必须是{configurable: {thread_id: xxx}}如果少了configurable这一层LangGraph 找不到 thread_id每次都当新会话处理。数据库连接对不对。如果你换库了、换表了、连接串写错了都会导致读不到旧 checkpoint。排查时直接查数据库里的表看看有没有该 thread_id 的记录。现象可能原因解决办法第二次调用没有历史上下文编译时没传 checkpointer在compile(checkpointer...)中传入实例每次都是全新会话config 格式错误确保{configurable: {thread_id: ...}}层级正确换机器后状态丢失MemorySaver 只存内存换成 Sqlite/Postgres 持久化方案读不到历史但数据库有记录thread_id 拼写不一致打印日志确认请求传入的 thread_id与库中对比4.2 并发写同一个 thread_id 的冲突第二个高发问题是同一时刻对同一个 thread_id 发起多次ainvoke。LangGraph 的 checkpoint 机制对同一线程的并发写入处理并不像分布式数据库那么健壮两个并发的执行可能读到同一个旧 checkpoint然后各自写入新 checkpoint导致后写入的覆盖前一个。规避手段我在生产里用了两种应用层加线程锁或分布式锁保证同一 thread_id 串行执行。对于单机进程用 Python 的字典锁就行多机部署用 Redis 锁或者数据库行锁。一个用户一个活跃会话前端在发起新请求前取消旧请求从交互层面避免并发。如果你实在无法避免并发可以考虑为每次用户请求生成一个新的子 thread再通过parent_checkpoint_id关联回主 thread。但这个玩法比较复杂我建议优先从架构上规避不要硬杠。4.3 Checkpoint 越存越大Long-running 会话跑久了checkpoint 数量会线性增长。同一个 thread 下每次执行都产生若干个 checkpoint每个 checkpoint 里还存着完整消息历史。跑一周以后光 checkpoint 就能占掉几个 GB。这时候几个处理手段可以组合用定期清理旧的 checkpoint只保留最近 N 个或最近 N 天的归档已结束会话的 thread把它们从活跃表中移到归档表调整写入频率如果你不需要精确到每个超步的状态可以在部分子图上不配持久化。LangGraph 在不同版本里对清理接口的支持程度不一样我建议你直接写一个定时任务按thread_id和created_at删除旧记录配套一个数据保留策略文档。4.4 恢复后状态对不上这个问题比较阴它通常在自定义 Reducer 时爆发。症状是恢复后某些列表字段重复了或者某些字段值比预期多。根本原因是重放过程中同一段逻辑被执行了两次且 Reducer 不具备幂等性。排查方法是把同一条输入在同一 thread 上手动执行两次对比两次最终状态。如果两次结果不一致那问题基本锁定在自定义 Reducer 上。整改方案就是我前面说的为每条消息或记录增加唯一 IDReducer 合并前先按 ID 去重。这样即使重放也不会产生重复。5. 经验沉淀怎么设计一套适合自己的持久化方案5.1 选型四个维度决定了你用哪种 Checkpointer我在项目里总结了一套选型判断逻辑按优先级排列数据安全要求丢了会话状态会不会造成资损、客诉、业务中断会就必须用数据库方案部署架构单机进程还是多实例多实例必须用 Postgres 或等效的共享存储方案性能与规模每秒并发多少个 thread 上千级并发建议直接上 PostgresSqlite 写入锁会拖后腿运维成本团队有没有专职 DBA没有就老老实实 Sqlite配定时备份。方案数据保存位置跨进程共享适用场景运维成本MemorySaver进程内存不支持本地调试、单元测试零SqliteSaver本地文件单机多进程弱个人项目、内网工具低PostgresSaverPostgreSQL 数据库支持生产环境、多实例部署中5.2 进阶玩法人机交互与多会话编排持久化执行不只是防宕机它让一类特殊场景变得简单Human-in-the-loop也就是让 Agent 在关键节点停下来等人确认确认完再继续。LangGraph 的interrupt机制和 checkpoint 是绝配。图执行到某个节点时调用interrupt引擎会把当前状态写入 checkpoint然后暂停执行。外部系统接收人工反馈后用同一个 thread_id 再次调用图LangGraph 从暂停位置恢复执行。整个过程跨任意长时间因为状态已经在存储里了。我做过一个审批流程 AgentAgent 生成方案之后中断等待主管在界面点“批准”批准后同一 thread 继续执行后续动作。这个模式用在企业内部工具里价值非常高而且代码并不复杂——本质就是持久化执行加一个事件入口。5.3 容易被忽略的三个细节第一个是 metadata。Checkpointer 存的不只是 State还有 metadata。你可以往 metadata 里写用户 ID、业务单号、版本号等业务字段排查问题时直接查库就能定位是哪条业务的存档不用打开每条记录看内容。这个习惯建议从第一天就养成。第二个是测试环境要隔离。开发、测试、生产千万不要共用一个 checkpoint 数据库。我踩过一次坑测试脚本用随机 thread_id 跑并发直接把生产环境的 checkpoint 表打满了。隔离不仅是数据库实例隔离连 thread_id 前缀都建议做区分比如dev-、prod-。第三个是备份策略。Postgres 方案可以依赖数据库自身的备份这个没问题。但如果你用 Sqlite就得自己想方案定时把 .db 文件复制走或者用 dump 命令导出。别以为本地文件不需要备份磁盘坏了照样丢。5.4 我的实际体会持久化执行是所有有状态 Agent 应用的基石但它不是配完一个 checkpointer 就万事大吉的。LangGraph 给了你很好的“存档”能力但怎么组织 thread、怎么设计 Reducer、怎么清理 checkpoint、怎么并发隔离这些都得靠应用层自己拿捏。我个人的落地顺序建议是先用 MemorySaver 把逻辑跑通再换 Sqlite 做单机验证最后一步挪到 Postgres 并进行并发压测。每一步都验证一下状态恢复是否符合预期再往前推进。这样既不会在初期被存储问题拖慢进度也不会在后期被数据问题打个措手不及。最后补一个小技巧你可以在每次图执行结束后打印checkpointer.get(thread_id)返回的元数据确认写入确实发生了。这个习惯帮我快速发现过好几次生产环境连接串配错的问题。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
