1. 为什么手写 Loop 撑不过一次进程重启如果你用 LangGraph 写过稍微复杂一点的 Agent大概率经历过这个阶段一开始觉得 StateGraph 挺香节点、边、条件跳转都清清楚楚跑起来也顺。但一旦业务要求用户关掉页面明天再回来接着聊或者服务发版重启后任务不能丢手写的那套 while 循环加内存字典就彻底歇菜了。我自己最早做的一个客服工单 Agent 就是这样。整个流程是识别意图 → 查订单 → 判断是否需要人工 → 生成回复。状态全塞在一个 Python dict 里循环靠while not done驱动。本地测试丝滑得不行上线第二天运维重启了一次容器所有进行中的会话全部归零用户回来发现机器人失忆了前面聊的全白聊。那次事故之后我才认真去啃 LangGraph 的 Checkpoint 机制也才有了后来这套可恢复 Runtime的完整方案。这篇文章要讲清楚三件事LangGraph 的 Checkpoint 到底把什么存进了 PostgreSQL、中断interrupt和恢复resume在运行时是怎么串起来的、以及怎么用 AG-UI 把暂停—人工介入—继续这套交互在前端跑通。适合已经写过基础 LangGraph、但被持久化和人工介入卡住的同学。如果你还在纠结 LangChain 和 LangGraph 的区别简单说LangChain 偏链式调用和组件编排LangGraph 偏有状态图和有环流程而 Checkpoint 是 LangGraph 区别于普通链式框架的核心能力之一。先说结论手写 Loop 的根本问题不是代码丑而是状态没有落盘锚点。你的循环变量、中间结果、下一步该走哪个分支全在进程内存里进程一没这些信息就没了。Checkpoint 的本质就是在每个超级步super-step结束时把整个图的状态快照写进一个外部存储并给它一个唯一的 thread_id 和 checkpoint_id。恢复时不是重跑而是从快照点续上。2. Checkpoint 存进 PostgreSQL 的到底是什么很多人以为 Checkpoint 就是存了个当前节点名字其实远不止。理解它存了什么你才能理解恢复时为什么能精确续上而不是从头再来。2.1 一次超级步的状态快照包含哪些字段LangGraph 的每个 checkpoint 本质上是一个状态对象落到 PostgreSQL 后大致包含这几类信息字段类别具体内容作用通道值channel_values图状态里所有 channel 的当前值比如 messages、order_info、intent恢复时直接还原业务数据待执行任务pending_writes当前超级步里已经产生但还没被下游消费的写入保证节点执行的原子性版本与父指针parent_checkpoint_id指向上一个 checkpoint支持时间旅行和分支回溯元数据metadatasource、step、writes 等区分是用户输入触发还是节点内部触发下一步next接下来要执行的节点列表恢复时知道从哪继续这里最关键的是parent_checkpoint_id 形成的链式结构。它让整个执行历史变成一条可回溯的链而不是一个孤立的当前状态。你可以把它想象成 Git 的 commit 历史每个 checkpoint 是一个 commitparent 指针指向上一个你随时可以 checkout 到任意历史点重新跑。2.2 为什么选 PostgreSQL 而不是内存或 RedisLangGraph 官方提供了多种 CheckpointerMemorySaver内存、SqliteSaver、PostgresSaver 等。选 PostgreSQL 的理由很实际持久性内存版进程一挂全没Sqlite 单文件在容器化部署下不好共享PostgreSQL 天然支持多实例共享同一份状态。并发控制多个 worker 同时处理不同 thread 时PostgreSQL 的行级锁和事务能保证状态写入不打架。查询能力想统计有多少会话卡在人工审核节点、想按时间范围清理旧 checkpointSQL 直接搞定不用自己写遍历逻辑。生态成熟备份、监控、扩容这些运维能力都是现成的不用为状态存储单独造轮子。提示PostgresSaver 需要单独建表官方提供了setup()方法自动创建 checkpoints、checkpoint_writes、checkpoint_blobs 等表。生产环境建议用独立的 schema 或数据库别和业务表混在一起。2.3 表结构背后的设计意图跑完setup()后你会看到几张表理解它们的分工很重要checkpoints主表存每个 checkpoint 的元信息和状态引用。checkpoint_writes存每个节点产生的中间写入配合 pending_writes 实现未完成任务的续跑。checkpoint_blobs存大的二进制或复杂对象避免主表膨胀。这种拆分的意图是把状态元信息和大块数据分离。元信息查询频繁但体积小大对象写入少但体积大分开存能让主表保持轻量查询性能更稳。我实测过一个存了几十万 checkpoint 的库主表查询依然是毫秒级就是因为大对象都被挪到了 blobs 表。3. 中断与恢复在 Runtime 里的完整链路光有 Checkpoint 还不够真正让可恢复落地的是 interrupt 机制和 Runtime 的配合。这部分是整套方案里最容易踩坑的地方。3.1 interrupt 不是抛异常而是受控暂停刚接触 LangGraph 的 interrupt 时我一度以为它就是个特殊的异常捕获了就暂停。实际不是。interrupt()的语义是在当前节点执行到这一行时把当前状态存成一个 checkpoint然后让整个图的执行干净地停下来并把 interrupt 携带的值返回给调用方。from langgraph.types import interrupt def human_review_node(state): # 到这里会暂停把待审核内容抛给前端 decision interrupt({ question: 这笔退款需要人工确认, amount: state[refund_amount], }) # 恢复后decision 就是前端传回来的值 return {approved: decision approve}关键点在于interrupt 之后这个节点的执行上下文是被完整保存的。恢复时不是重新进入这个节点从头跑而是从 interrupt 那一行之后继续decision直接拿到恢复时传入的值。这就是为什么它叫可恢复而不是可重试。3.2 恢复时 Runtime 怎么找到断点恢复的入口是Command(resume...)。当你用同一个 thread_id 再次调用图并传入 resume 值时Runtime 会做这几件事用 thread_id 查出最新的 checkpoint。检查这个 checkpoint 的 next 字段确认它停在哪个节点。把 resume 的值注入到对应 interrupt 的返回位置。从该节点继续往下执行而不是从 START 重跑。这里有个容易忽略的细节thread_id 是恢复的唯一钥匙。如果你恢复时用了不同的 thread_idRuntime 会认为这是一个全新的会话从 START 开始跑你的断点就找不回来了。我在项目里专门把 thread_id 和业务侧的会话 ID 做了强绑定避免这种低级错误。3.3 一个完整的中断恢复时序把上面的逻辑串起来一次暂停—人工介入—恢复的完整链路是这样的阶段触发方Runtime 行为存储变化执行到 interrupt图内部保存当前状态返回 interrupt 值新增一个 checkpointnext 指向当前节点前端展示待审核AG-UI渲染 interrupt 携带的数据无用户提交决定前端调用 resume 接口传 thread_id 和值无Runtime 恢复后端查最新 checkpoint注入 resume 值继续执行新增后续 checkpoint注意interrupt 的值必须是可序列化的。我踩过一次坑往 interrupt 里塞了一个自定义对象结果 PostgresSaver 序列化时报错。后来统一改成 dict 或基本类型问题就没了。4. 用 AG-UI 把暂停和恢复接到前端后端能暂停能恢复但如果前端不知道怎么展示正在等你确认、不知道怎么把用户的选择传回去这套机制对用户来说就是黑盒。AG-UI 在这里扮演的角色就是把 Runtime 的状态变化翻译成前端能消费的事件流。4.1 AG-UI 处理的是事件而不是响应传统 REST 是请求—响应模型你发一个请求等一个完整结果。但 Agent 的执行是流式的、可能中断的用 REST 表达很别扭。AG-UI 的思路是把 Agent 运行过程中的每个关键节点都变成事件文本增量、工具调用、状态更新、以及我们最关心的 interrupt 事件。前端订阅这些事件后就能实时知道现在 Agent 停下来了它在等一个决定。这比轮询任务完成了吗要自然得多也更省资源。4.2 interrupt 事件在前端怎么落地当后端触发 interruptAG-UI 会向前端推送一个中断事件里面带着 interrupt 的 payload。前端拿到后通常做两件事根据 payload 渲染一个交互组件比如批准/拒绝按钮、一个输入框、或者一个选项列表。把当前 thread_id 存好等用户操作完带着这个 thread_id 和用户的选择调用恢复接口。// 前端伪代码监听中断事件并渲染交互 onInterrupt((payload) { setPendingAction({ threadId: payload.threadId, question: payload.question, amount: payload.amount, }); }); // 用户点击批准后 async function handleApprove() { await resumeRun({ threadId: pendingAction.threadId, resume: approve, }); }这套流程跑通后用户体验就是Agent 说这笔退款需要你确认页面弹出确认框用户点一下Agent 接着往下走。中间哪怕用户去泡了杯咖啡、关了页面再回来只要 thread_id 还在状态就还在。4.3 状态同步里最容易出错的三个地方实际联调时我遇到最多的问题集中在这三处thread_id 丢失前端刷新页面后没持久化 thread_id导致恢复时找不到断点。解决办法是把它存进 localStorage 或 URL 参数。重复恢复用户手快点了两次批准触发两次 resume。后端需要做幂等或者在恢复后立即把该 interrupt 标记为已消费。事件乱序流式事件在网络抖动下可能乱序到达前端如果无脑按到达顺序渲染会出现先显示结果再显示问题的诡异现象。建议给事件带上序号前端按序号排序后再渲染。5. 从零搭一套可恢复 Runtime 的实操步骤前面讲的是原理这一节给一套可以直接抄的落地步骤。我用的是 LangGraph PostgresSaver AG-UI 的组合Python 侧负责图逻辑前端负责交互。5.1 环境准备与依赖安装先把依赖装齐。LangGraph 的版本迭代比较快建议锁定版本避免 API 变动导致跑不通。pip install langgraph langgraph-checkpoint-postgres psycopg[binary] pip install ag-ui-protocolPostgreSQL 建议用 14 以上版本低版本在并发写入时偶发锁等待。本地开发可以用 Docker 起一个docker run -d --name lg-pg \ -e POSTGRES_PASSWORDpostgres \ -e POSTGRES_DBlanggraph \ -p 5432:5432 postgres:16提示生产环境务必给 checkpoint 库配独立的连接池别和业务库共用。checkpoint 的写入频率可能很高共用连接池容易把业务查询拖慢。5.2 初始化 PostgresSaver 并建表from langgraph.checkpoint.postgres import PostgresSaver DB_URI postgresql://postgres:postgreslocalhost:5432/langgraph with PostgresSaver.from_conn_string(DB_URI) as checkpointer: checkpointer.setup() # 自动建表只需执行一次setup()是幂等的重复执行不会报错但生产环境建议只在初始化脚本里跑一次别每次启动都调。5.3 编译图时挂上 checkpointer这是让图具备持久化能力的关键一步。编译时不传 checkpointer图就是无状态的interrupt 也无法恢复。from langgraph.graph import StateGraph, START, END builder StateGraph(AgentState) builder.add_node(classify, classify_node) builder.add_node(human_review, human_review_node) builder.add_node(execute, execute_node) builder.add_edge(START, classify) builder.add_conditional_edges(classify, route_after_classify) builder.add_edge(human_review, execute) builder.add_edge(execute, END) graph builder.compile(checkpointercheckpointer)5.4 用 thread_id 驱动一次可中断的执行config {configurable: {thread_id: order-12345}} # 第一次执行会在 human_review 处暂停 result graph.invoke({messages: [...], refund_amount: 200}, config) # result 里会包含 __interrupt__ 信息 print(result[__interrupt__]) # 用户确认后恢复 from langgraph.types import Command resumed graph.invoke(Command(resumeapprove), config)注意恢复时必须传同一个 config也就是同一个 thread_id。这是整个机制的地基。5.5 把 AG-UI 事件流接上后端把图的执行包装成一个流式接口把 interrupt、文本增量、状态更新都转成 AG-UI 事件推给前端。前端订阅后按事件类型分发处理。这部分的具体协议实现各家略有差异核心是保证 interrupt 事件里带上 thread_id 和 payload恢复接口能接收 thread_id 和 resume 值。6. 那些文档里不会写的坑这套方案我前前后后调了两周踩的坑比想象中多。挑几个最有代表性的说说。6.1 checkpoint 无限增长的问题默认情况下每个超级步都会产生一个 checkpoint一个长会话跑下来可能积累几百上千条。时间一长checkpoints 表会膨胀得很快。我的做法是给 thread_id 加索引查询快。定期归档或清理超过 N 天的旧 checkpoint但保留每个 thread 的最新一条否则恢复会失败。如果业务需要时间旅行就保留完整链如果不需要可以只留最新快照。清理时一定要小心 parent 指针的完整性别把还在被引用的 checkpoint 删了。6.2 序列化失败的隐蔽性PostgresSaver 用 pickle 或 JSON 序列化状态。如果你的状态里塞了不可序列化的对象比如数据库连接、文件句柄、lambda写入时会报错而且报错信息往往不直观。我的经验是状态里只放纯数据复杂对象在节点内部临时创建用完即弃。6.3 恢复后节点重入的副作用这是最坑的一个。如果 interrupt 之前的节点有副作用比如发了一封邮件、扣了一次库存恢复时如果逻辑没设计好可能重复执行。LangGraph 的 checkpoint 机制能保证从断点续跑但不能自动帮你保证副作用幂等。我的做法是把有副作用的操作尽量放在 interrupt 之后或者在操作前先查一次是否已执行。6.4 多 worker 下的并发恢复当你有多个 worker 实例时同一个 thread 可能被两个请求同时恢复。PostgreSQL 的事务能保证写入不冲突但业务逻辑上可能出现两个 resume 都生效的情况。解决办法是在恢复入口加一层分布式锁按 thread_id 加锁保证同一时刻只有一个恢复在执行。7. 几个高频疑问的实测回答7.1 LangGraph 和 LangChain 到底怎么选这个问题被问太多次了。我的判断标准很简单如果你的流程是线性的、无环的、不需要人工介入和持久化LangChain 的链式编排够用。一旦出现根据中间结果决定下一步走哪、需要暂停等人工、进程重启后要接着跑就该上 LangGraph。两者不是替代关系LangGraph 里照样可以用 LangChain 的组件。7.2 Checkpoint 和传统数据库事务有什么区别有人会问这不就是把状态存数据库吗和普通事务有啥区别。区别在于粒度数据库事务保证的是一次操作的原子性而 checkpoint 保证的是整个图执行过程的原子性和可恢复性。它记录的不只是数据还有执行到哪了、下一步该干嘛这种控制流信息这是普通事务做不到的。7.3 中断恢复能不能跨版本实测下来跨 LangGraph 大版本恢复有风险。因为 checkpoint 的序列化格式可能变新版本读旧 checkpoint 可能失败。生产环境升级前建议先在测试库验证旧 checkpoint 能否正常恢复或者干脆在升级时清空历史 checkpoint只保留业务数据。7.4 前端刷新后怎么保证不丢状态核心就一句话thread_id 必须持久化。存 localStorage、存 URL、存后端会话表都行只要刷新后能拿回来。我见过有团队把 thread_id 只放在内存变量里用户一刷新就全丢了然后抱怨恢复功能不好用。这不是框架的问题是设计的问题。8. 写在最后的一点个人体会这套方案跑通之后我最大的感受是可恢复不是加个功能而是一种架构约束。它逼着你把状态管理、副作用控制、并发处理这些平时能糊弄过去的问题全部摆到台面上认真对待。手写 Loop 的时候你可以随便在内存里改状态但一旦上了 Checkpoint你就得想清楚这个状态该不该进快照这个操作重入会不会出问题。另一个体会是别一上来就追求完美的时间旅行和分支回溯。我一开始想做得特别全结果复杂度爆炸。后来退回到只保证最新断点能恢复先把核心链路跑稳再逐步加高级能力反而顺利得多。如果你也在做类似的东西建议先跑通暂停—恢复这一条最短路径剩下的慢慢来。