本文是「LangGraph 教程系列」第 6 篇。写作时基于 langgraph 1.2.10、langchain 1.3.14、langchain-openai 1.4.1、Python 3.12。配套代码仓库 https://github.com/wxj006007/deep-research-assistant 本篇对应 tagv2。上一篇 v1.1 让研究助手能真上网查了每次 invoke 都是从头跑到尾——plan → search → evaluate → (loop) → synthesize跑完就散场。你无法查看中途状态查了几轮、每轮的结果是什么暂停后继续用户说等一下我再回来回溯到历史状态重跑发现方向错了回滚多会话共享记忆thread 级短期记忆这些问题不是 bug是设计选择。v1 的图是无状态的invoke 只返回最终结果中间过程不留痕迹。有状态需要什么需要 checkpoint。这一篇就来接入 checkpointer给 agent 装上时间线。一、v1.1 撞的墙执行过程不留痕假设用户在问一个复杂问题跑了 5 轮检索才找到答案。过程中他想知道“刚才那两轮都查了什么怎么突然判定资料够了”——没有 checkpoint这问题没法回答。更极端的场景是用户看到第 3 轮查的方向好像不对想撤销到第 2 轮重新来。无状态图做不到这个它只有运行中和已结束两种状态中间态全部蒸发。还有一个真实需求多轮对话。比如用户先问LangGraph 是什么助手查了三轮给出答案紧接着用户追问它在生产环境怎么用同一个助手凭什么记住上一轮的上下文线程级短期记忆就是靠 checkpoint 实现的。二、checkpoint 机制的核心概念2.1 什么是 checkpointCheckpoint 在每个超级步superstep结束后自动保存 state 快照。LangGraph 的 superstep 是原子执行单元节点串行或并行执行→state 合并→reducer 累积→保存 checkpoint→进入下一轮。snapshot 包含什么values当前 state 的所有字段值config配置元数据最重要的是 thread_idmetadata附加信息如创建时间、备注等2.2 Thread 隔离thread_id 是 checkpoint 的主键不同 thread 之间 state 完全隔离# Thread A 的运行thread_a{configurable:{thread_id:thread-a}}graph.invoke({question:A 的问题},configthread_a)# Thread B 的运行thread_b{configurable:{thread_id:thread-b}}graph.invoke({question:B 的问题},configthread_b)两个线程的 checkpoint 独立存储互不干扰。这是多租户系统的基础同一张图可以服务多个用户每个用户有自己的记忆。2.3 Checkpoint 与 History单次 invoke 只返回最终 state但get_history(thread_config)能获取完整执行轨迹historylist(graph.get_history(thread_a))forsnapshotinhistory:print(fround{snapshot.values[round]}, verdict{snapshot.values[verdict]})每个 snapshot 代表一个超级步结束时的 state 快照按执行顺序排列。你可以用它做调试、审计、回放。三、v2 的图结构保存 snapshotStateGraph with MemorySavercontinuefinishSTARTplansearchevaluatesynthesizeENDCheckpointerMemory图结构没变唯一的差异是在 compile 时传入 checkpointerfromlanggraph.checkpoint.memoryimportMemorySaver checkpointerMemorySaver()graphbuilder.compile(checkpointercheckpointer)工具接入像这样对内部透明对外部有感知。这是 LangGraph 的设计哲学。四、动手实现4.1 环境准备pipinstalllanggraph1.2.10langchain1.3.14 langchain-openai1.4.1 python-dotenv内存版 checkpointer 不需要额外依赖直接导入即可。持久化版本需要安装相应驱动SQLite:pip install sqlite3(内置)PostgreSQL:pip install psycopg2-binary4.2 State保持不变classResearchState(TypedDict):question:strcurrent_query:strexecuted_queries:Annotated[list[str],operator.add]docs:Annotated[list[Doc],operator.add]round:intverdict:strgap:stranswer:strState 结构不需要改checkpoint 只管存不管语义。4.3 Node 实现和 v1.1 一模一样plan_node、search_node、evaluate_node、synthesize_node 都没变唯一的变化是图多了个 checkpointer 属性。这也是 checkpoint 设计的巧妙之处节点无需感知持久化。你不用在每个 node 里手动 save/checkpoint框架在 superstep 结束后自动处理。4.4 编译图时注入 checkpointer关键的一行代码defbuild_graph():builderStateGraph(ResearchState)# ... add nodes and edges ...checkpointerMemorySaver()# 内存版本returnbuilder.compile(checkpointercheckpointer)就这么简单。compare 到之前不带 checkpointer 的版本唯一新增的就是这 3 行。五、核心 API 用法5.1 带 config 的 invoke# 默认无 config每次 invoke 独立resultgraph.invoke({question:你好})# 指定 thread_id启用持久化thread_config{configurable:{thread_id:user-123}}resultgraph.invoke({question:你好},configthread_config)不提供 thread_idcheckpointer 不会生效或者说每次都新建一个 thread。必须显式指定 thread_id才能复用历史状态。5.2 查看历史记录thread_config{configurable:{thread_id:research-thread-1}}historylist(graph.get_history(thread_config))print(f共{len(history)}个历史状态)fori,snapshotinenumerate(history):statesnapshot.valuesprint(f状态{i1}: round{state[round]}, verdict{state[verdict]}, docs{len(state[docs])}条)get_history返回生成器每个元素是一个CheckpointTuple包含 values、config、metadata。5.3 从历史状态继续执行这是 checkpoint 最实用的功能之一中断后恢复。# 获取第 2 个状态的 snapshothistorylist(graph.get_history(thread_config))second_snapshothistory[1]# 从那个状态继续执行forstateingraph.stream(second_snapshot.values,stream_modevalues):print(f继续执行 → round{state[round]})注意这里是 stream second_snapshot.valuesstate 内容而不是用 second_snapshot.config。你想从历史状态重启但不想要历史记录累加到新 session 里。5.4 多会话隔离thread_a{configurable:{thread_id:thread-A}}thread_b{configurable:{thread_id:thread-B}}# Thread Aforstateingraph.stream({question:A 的问题},stream_modevalues,configthread_a):passprint(fThread A 最终 round:{state[round]})# Thread Bforstateingraph.stream({question:B 的问题},stream_modevalues,configthread_b):passprint(fThread B 最终 round:{state[round]})# 验证独立history_alen(list(graph.get_history(thread_a)))history_blen(list(graph.get_history(thread_b)))print(fThread A 历史状态数{history_a})print(fThread B 历史状态数{history_b})输出会显示两个线程各自完成了自己的执行流程彼此不影响。这就是多租户能力的基础。5.5 Delete 历史记录graph.delete(thread_config)# 删除指定 thread 的所有历史谨慎使用删除后无法恢复。适合清除用户隐私数据或测试环境重置场景。六、常用 Checkpointer 实现6.1 MemorySaver内存版fromlanggraph.checkpoint.memoryimportMemorySaver checkpointerMemorySaver()优点零依赖开箱即用响应快无 IO 开销适合开发调试、单元测试缺点进程重启后数据丢失单进程隔离不支持跨进程共享适用场景本地开发、教学演示、自动化测试。6.2 PostgresSaverPostgreSQL 版fromlanggraph.checkpoint.postgresimportPostgresSaver conn_strpostgresql://user:passlocalhost:5432/dbnamecheckpointerPostgresSaver(conn_str)优点持久化存储进程重启不丢数据支持多进程/多实例共享支持并发控制、事务隔离缺点依赖 PostgreSQL 数据库需要 DBA 运维备份、索引、扩缩容适用场景生产环境、多实例部署、需审计的场景。6.3 SQLiteSaverSQLite 版fromlanggraph.checkpoint.sqliteimportSqliteSaver conn_strsqlite:///checkpoints.dbcheckpointerSqliteSaver.from_conn_string(conn_str)优点文件型存储零依赖轻量级适合单机支持基本 CRUD缺点并发写性能一般不适合分布式部署适用场景单机测试、小规模部署、离线 Demo。七、实践建议7.1 Thread ID 设计策略thread_id 怎么选常见模式用户级thread_id user_id每个用户一个会话对话级thread_id f{user_id}:{conversation_id}每个对话一个线程消息级thread_id message_id每条消息独立失去记忆意义推荐对话级平衡记忆范围和复杂度。如果业务需要跨对话记忆考虑用 Store第 8 篇讲。7.2 Checkpoint 大小优化每次 superstep 都保存全量 state数据量大时会占空间。优化策略压缩对 doc.content 等大字段 gzip 压缩裁剪只保留关键字段过滤 debug info过期设置 TTL定期清理旧记录PostgresSaver 支持自定义 metadata可以加 created_at 字段配合定时任务清理。7.3 并发安全多线程同时写同一个 thread_id 可能冲突。LangGraph 内置乐观锁检测到 conflict 会抛异常调用方需重试。importtimeforattemptinrange(3):try:graph.invoke(input,configthread_config)breakexceptExceptionase:ifattempt2:raisetime.sleep(0.1*(2**attempt))# exponential backoff生产环境建议加 retry 逻辑避免瞬间失败。7.4 Debugging 技巧checkpoint 的最大好处是可调# 1. 查看完整历史historylist(graph.get_history(thread_config))forsnapinhistory:print(json.dumps(snap.values,indent2,defaultstr))# 2. 定位问题 stateproblem_snapnext(sforsinhistoryifs.values[round]2)# 修复逻辑problem_snap.values[gap]补充查询词# 3. 从问题点重跑forstateingraph.stream(problem_snap.values,stream_modeupdates):print(state)这种状态快照 - 修改 - 重放的能力是调试复杂 agent 工作流的利器。八、跑起来python-msrc.v2_checkpoint示例输出问题为什么 LangGraph 需要 checkpoint它和人在回路是什么关系 【第一轮运行】 第 1 轮 | 当前 verdict: insufficient 第 2 轮 | 当前 verdict: sufficient 最终回答长度523 字符 【查看 checkpoint 历史】 共 3 个历史状态 状态 1: round1, verdictinsufficient, docs2 条已完成False 状态 2: round2, verdictsufficient, docs4 条已完成False 状态 3: round2, verdictsufficient, docs4 条已完成True 【演示多会话隔离】 Thread A: Thread A 最终 round: 2 Thread B: Thread B 最终 round: 2 Thread A 历史状态数3 Thread B 历史状态数3 ✓ 两个线程的状态完全独立互不影响九、本篇小结回到持久化与 checkpoint这块v1.1 撞的墙是无状态导致无法追溯、无法恢复、无法共享记忆解法是接入 MemorySaver。我们强调了几个关键点Checkpoint 自动保存superstep 结束后框架自动存 snapshot节点无需感知Thread 隔离thread_id 主键多租户基础History APIget_history获取轨迹stream从历史重跑Checkpointer 选型MemorySaver开发、SQLiteSaver单机、PostgresSaver生产v2 有了时间线可以随时查看、回溯、回放。但它还有个问题——只能自己跑不能等人。用户说第 2 轮查得不行我手动改一下再跑怎么办能不能在关键节点暂停让人介入审批答案是能而且需要 interrupt。下一篇人在回路登场。赞或收藏 关注 我们下次再见
langgraph教程系列-06-让流程可暂停 - 持久化与 checkpoint
发布时间：2026/8/2 2:57:46
本文是「LangGraph 教程系列」第 6 篇。写作时基于 langgraph 1.2.10、langchain 1.3.14、langchain-openai 1.4.1、Python 3.12。配套代码仓库 https://github.com/wxj006007/deep-research-assistant 本篇对应 tagv2。上一篇 v1.1 让研究助手能真上网查了每次 invoke 都是从头跑到尾——plan → search → evaluate → (loop) → synthesize跑完就散场。你无法查看中途状态查了几轮、每轮的结果是什么暂停后继续用户说等一下我再回来回溯到历史状态重跑发现方向错了回滚多会话共享记忆thread 级短期记忆这些问题不是 bug是设计选择。v1 的图是无状态的invoke 只返回最终结果中间过程不留痕迹。有状态需要什么需要 checkpoint。这一篇就来接入 checkpointer给 agent 装上时间线。一、v1.1 撞的墙执行过程不留痕假设用户在问一个复杂问题跑了 5 轮检索才找到答案。过程中他想知道“刚才那两轮都查了什么怎么突然判定资料够了”——没有 checkpoint这问题没法回答。更极端的场景是用户看到第 3 轮查的方向好像不对想撤销到第 2 轮重新来。无状态图做不到这个它只有运行中和已结束两种状态中间态全部蒸发。还有一个真实需求多轮对话。比如用户先问LangGraph 是什么助手查了三轮给出答案紧接着用户追问它在生产环境怎么用同一个助手凭什么记住上一轮的上下文线程级短期记忆就是靠 checkpoint 实现的。二、checkpoint 机制的核心概念2.1 什么是 checkpointCheckpoint 在每个超级步superstep结束后自动保存 state 快照。LangGraph 的 superstep 是原子执行单元节点串行或并行执行→state 合并→reducer 累积→保存 checkpoint→进入下一轮。snapshot 包含什么values当前 state 的所有字段值config配置元数据最重要的是 thread_idmetadata附加信息如创建时间、备注等2.2 Thread 隔离thread_id 是 checkpoint 的主键不同 thread 之间 state 完全隔离# Thread A 的运行thread_a{configurable:{thread_id:thread-a}}graph.invoke({question:A 的问题},configthread_a)# Thread B 的运行thread_b{configurable:{thread_id:thread-b}}graph.invoke({question:B 的问题},configthread_b)两个线程的 checkpoint 独立存储互不干扰。这是多租户系统的基础同一张图可以服务多个用户每个用户有自己的记忆。2.3 Checkpoint 与 History单次 invoke 只返回最终 state但get_history(thread_config)能获取完整执行轨迹historylist(graph.get_history(thread_a))forsnapshotinhistory:print(fround{snapshot.values[round]}, verdict{snapshot.values[verdict]})每个 snapshot 代表一个超级步结束时的 state 快照按执行顺序排列。你可以用它做调试、审计、回放。三、v2 的图结构保存 snapshotStateGraph with MemorySavercontinuefinishSTARTplansearchevaluatesynthesizeENDCheckpointerMemory图结构没变唯一的差异是在 compile 时传入 checkpointerfromlanggraph.checkpoint.memoryimportMemorySaver checkpointerMemorySaver()graphbuilder.compile(checkpointercheckpointer)工具接入像这样对内部透明对外部有感知。这是 LangGraph 的设计哲学。四、动手实现4.1 环境准备pipinstalllanggraph1.2.10langchain1.3.14 langchain-openai1.4.1 python-dotenv内存版 checkpointer 不需要额外依赖直接导入即可。持久化版本需要安装相应驱动SQLite:pip install sqlite3(内置)PostgreSQL:pip install psycopg2-binary4.2 State保持不变classResearchState(TypedDict):question:strcurrent_query:strexecuted_queries:Annotated[list[str],operator.add]docs:Annotated[list[Doc],operator.add]round:intverdict:strgap:stranswer:strState 结构不需要改checkpoint 只管存不管语义。4.3 Node 实现和 v1.1 一模一样plan_node、search_node、evaluate_node、synthesize_node 都没变唯一的变化是图多了个 checkpointer 属性。这也是 checkpoint 设计的巧妙之处节点无需感知持久化。你不用在每个 node 里手动 save/checkpoint框架在 superstep 结束后自动处理。4.4 编译图时注入 checkpointer关键的一行代码defbuild_graph():builderStateGraph(ResearchState)# ... add nodes and edges ...checkpointerMemorySaver()# 内存版本returnbuilder.compile(checkpointercheckpointer)就这么简单。compare 到之前不带 checkpointer 的版本唯一新增的就是这 3 行。五、核心 API 用法5.1 带 config 的 invoke# 默认无 config每次 invoke 独立resultgraph.invoke({question:你好})# 指定 thread_id启用持久化thread_config{configurable:{thread_id:user-123}}resultgraph.invoke({question:你好},configthread_config)不提供 thread_idcheckpointer 不会生效或者说每次都新建一个 thread。必须显式指定 thread_id才能复用历史状态。5.2 查看历史记录thread_config{configurable:{thread_id:research-thread-1}}historylist(graph.get_history(thread_config))print(f共{len(history)}个历史状态)fori,snapshotinenumerate(history):statesnapshot.valuesprint(f状态{i1}: round{state[round]}, verdict{state[verdict]}, docs{len(state[docs])}条)get_history返回生成器每个元素是一个CheckpointTuple包含 values、config、metadata。5.3 从历史状态继续执行这是 checkpoint 最实用的功能之一中断后恢复。# 获取第 2 个状态的 snapshothistorylist(graph.get_history(thread_config))second_snapshothistory[1]# 从那个状态继续执行forstateingraph.stream(second_snapshot.values,stream_modevalues):print(f继续执行 → round{state[round]})注意这里是 stream second_snapshot.valuesstate 内容而不是用 second_snapshot.config。你想从历史状态重启但不想要历史记录累加到新 session 里。5.4 多会话隔离thread_a{configurable:{thread_id:thread-A}}thread_b{configurable:{thread_id:thread-B}}# Thread Aforstateingraph.stream({question:A 的问题},stream_modevalues,configthread_a):passprint(fThread A 最终 round:{state[round]})# Thread Bforstateingraph.stream({question:B 的问题},stream_modevalues,configthread_b):passprint(fThread B 最终 round:{state[round]})# 验证独立history_alen(list(graph.get_history(thread_a)))history_blen(list(graph.get_history(thread_b)))print(fThread A 历史状态数{history_a})print(fThread B 历史状态数{history_b})输出会显示两个线程各自完成了自己的执行流程彼此不影响。这就是多租户能力的基础。5.5 Delete 历史记录graph.delete(thread_config)# 删除指定 thread 的所有历史谨慎使用删除后无法恢复。适合清除用户隐私数据或测试环境重置场景。六、常用 Checkpointer 实现6.1 MemorySaver内存版fromlanggraph.checkpoint.memoryimportMemorySaver checkpointerMemorySaver()优点零依赖开箱即用响应快无 IO 开销适合开发调试、单元测试缺点进程重启后数据丢失单进程隔离不支持跨进程共享适用场景本地开发、教学演示、自动化测试。6.2 PostgresSaverPostgreSQL 版fromlanggraph.checkpoint.postgresimportPostgresSaver conn_strpostgresql://user:passlocalhost:5432/dbnamecheckpointerPostgresSaver(conn_str)优点持久化存储进程重启不丢数据支持多进程/多实例共享支持并发控制、事务隔离缺点依赖 PostgreSQL 数据库需要 DBA 运维备份、索引、扩缩容适用场景生产环境、多实例部署、需审计的场景。6.3 SQLiteSaverSQLite 版fromlanggraph.checkpoint.sqliteimportSqliteSaver conn_strsqlite:///checkpoints.dbcheckpointerSqliteSaver.from_conn_string(conn_str)优点文件型存储零依赖轻量级适合单机支持基本 CRUD缺点并发写性能一般不适合分布式部署适用场景单机测试、小规模部署、离线 Demo。七、实践建议7.1 Thread ID 设计策略thread_id 怎么选常见模式用户级thread_id user_id每个用户一个会话对话级thread_id f{user_id}:{conversation_id}每个对话一个线程消息级thread_id message_id每条消息独立失去记忆意义推荐对话级平衡记忆范围和复杂度。如果业务需要跨对话记忆考虑用 Store第 8 篇讲。7.2 Checkpoint 大小优化每次 superstep 都保存全量 state数据量大时会占空间。优化策略压缩对 doc.content 等大字段 gzip 压缩裁剪只保留关键字段过滤 debug info过期设置 TTL定期清理旧记录PostgresSaver 支持自定义 metadata可以加 created_at 字段配合定时任务清理。7.3 并发安全多线程同时写同一个 thread_id 可能冲突。LangGraph 内置乐观锁检测到 conflict 会抛异常调用方需重试。importtimeforattemptinrange(3):try:graph.invoke(input,configthread_config)breakexceptExceptionase:ifattempt2:raisetime.sleep(0.1*(2**attempt))# exponential backoff生产环境建议加 retry 逻辑避免瞬间失败。7.4 Debugging 技巧checkpoint 的最大好处是可调# 1. 查看完整历史historylist(graph.get_history(thread_config))forsnapinhistory:print(json.dumps(snap.values,indent2,defaultstr))# 2. 定位问题 stateproblem_snapnext(sforsinhistoryifs.values[round]2)# 修复逻辑problem_snap.values[gap]补充查询词# 3. 从问题点重跑forstateingraph.stream(problem_snap.values,stream_modeupdates):print(state)这种状态快照 - 修改 - 重放的能力是调试复杂 agent 工作流的利器。八、跑起来python-msrc.v2_checkpoint示例输出问题为什么 LangGraph 需要 checkpoint它和人在回路是什么关系 【第一轮运行】 第 1 轮 | 当前 verdict: insufficient 第 2 轮 | 当前 verdict: sufficient 最终回答长度523 字符 【查看 checkpoint 历史】 共 3 个历史状态 状态 1: round1, verdictinsufficient, docs2 条已完成False 状态 2: round2, verdictsufficient, docs4 条已完成False 状态 3: round2, verdictsufficient, docs4 条已完成True 【演示多会话隔离】 Thread A: Thread A 最终 round: 2 Thread B: Thread B 最终 round: 2 Thread A 历史状态数3 Thread B 历史状态数3 ✓ 两个线程的状态完全独立互不影响九、本篇小结回到持久化与 checkpoint这块v1.1 撞的墙是无状态导致无法追溯、无法恢复、无法共享记忆解法是接入 MemorySaver。我们强调了几个关键点Checkpoint 自动保存superstep 结束后框架自动存 snapshot节点无需感知Thread 隔离thread_id 主键多租户基础History APIget_history获取轨迹stream从历史重跑Checkpointer 选型MemorySaver开发、SQLiteSaver单机、PostgresSaver生产v2 有了时间线可以随时查看、回溯、回放。但它还有个问题——只能自己跑不能等人。用户说第 2 轮查得不行我手动改一下再跑怎么办能不能在关键节点暂停让人介入审批答案是能而且需要 interrupt。下一篇人在回路登场。赞或收藏 关注 我们下次再见
)
