LangGraph + PostgreSQL Checkpoint + AG-UI：构建可恢复的 Agent Runtime

1. 为什么“中断恢复”是 Agent 工程绕不开的一道坎做过对话式 Agent 的人大概都有过这种体验用户跟机器人聊到一半页面刷新了、网络抖了一下、后端服务滚动重启了一次回来之后上下文全没了机器人像失忆一样问“你好请问有什么可以帮您”。更糟的是那种跑了十几步工具调用、眼看就要出结果的长任务中途断掉只能从头再来前面烧掉的 token 和时间全打了水漂。这个项目要解决的就是这件事让一个基于 LangGraph 构建的 Agent 具备“可恢复 Runtime”能力——状态持久化到 PostgreSQL前端通过 AG-UI 协议对接任何一步中断之后都能从断点接着跑而不是从零开始。核心关键词就三个LangGraph负责编排有状态的图执行PostgreSQL Checkpoint负责把每一步的状态落盘AG-UI负责把运行时的状态变化实时推给前端。适合谁来参考如果你已经写过简单的 LangChain Chain、跑通过一两个 Agent Demo但一遇到“多轮 长任务 断线重连”就抓瞎那这篇就是写给你的。我会从手写 Loop 的痛点讲起一路拆到 Checkpoint 的表结构、恢复时的状态合并逻辑、AG-UI 的事件流设计最后给一份能直接抄的排查清单。全程按我实际踩坑的顺序来不绕弯子。先说清楚一个前提“可恢复”不等于“可重试”。重试是把整个任务重跑一遍可恢复是从中断的那一步继续。前者浪费资源后者才是工程上真正值钱的东西。LangGraph 的 Checkpoint 机制天生就是为后者设计的但要用好它得先理解它到底存了什么、什么时候存、恢复时怎么读。2. 从手写 Loop 到 LangGraph一次架构上的取舍2.1 手写 Loop 到底卡在哪最早我写 Agent 就是一个while循环调模型、解析工具调用、执行工具、把结果塞回消息列表、再调模型直到模型不再要求调工具为止。代码不到一百行跑 Demo 特别爽。但一旦要上生产问题一个接一个冒出来。第一个问题是状态无处安放。整个对话历史、中间的工具返回值、当前的步数全在内存里的一个 list 里。进程一重启list 没了。你可能会说那就存 Redis 呗但存进去容易恢复的时候你得自己判断“上次跑到哪一步了、哪些工具已经执行过、哪些还没执行”这套逻辑手写起来非常容易出错尤其是工具调用有副作用的时候比如已经发过一封邮件重跑一遍就是灾难。第二个问题是流程不可视、不可控。手写 Loop 里加一个“人工审核”环节你得在循环里插if判断加一个“条件分支”又得插一层嵌套。业务逻辑稍微复杂一点这个循环就变成了一坨意大利面改一处怕崩三处。第三个问题是并发与流式。用户希望看到 Agent 一步步思考的过程而不是等三十秒后一次性吐出来。手写 Loop 要做流式得自己管理事件队列、自己处理背压工作量陡增。2.2 LangGraph 带来的三个关键变化换成 LangGraph 之后最直观的变化是执行模型从“循环”变成了“图”。节点是函数边是流转条件整个 Agent 的执行路径被显式地画了出来。这个转变带来的好处不是“代码更优雅”这种虚的而是三个实打实的能力。第一状态是显式声明的。你用StateGraph定义一个 State 的 schema所有节点读写这个 State框架负责在节点之间传递和合并。这意味着状态不再散落在闭包里而是有一个明确的、可序列化的结构。这一点是后面能做 Checkpoint 的前提——你没法给一个散装的内存变量做持久化。第二中断点是一等公民。LangGraph 支持在节点执行前后插入 Checkpoint也支持interrupt主动暂停等待人工输入。这意味着“中断”不再是一个异常情况而是流程里的正常状态。恢复的时候框架知道该从哪个 Checkpoint 继续。第三流式是内建的。astream和astream_events能把每个节点的输入输出、每个 token 的生成实时吐出来前端只要订阅就行。这为后面接 AG-UI 省了大量胶水代码。提示LangChain 和 LangGraph 的区别经常被问到。简单说LangChain 是“组件库”提供模型、工具、检索器这些积木LangGraph 是“编排层”负责把这些积木按图的方式串起来并管理状态。做简单 Chain 用 LangChain 够了做有状态、可中断、可恢复的 AgentLangGraph 才是对的工具。2.3 为什么 Checkpoint 选 PostgreSQL 而不是内存或 RedisLangGraph 官方提供了多种 Checkpointer内存版、SQLite 版、PostgreSQL 版。内存版只能用于测试进程一挂全没。SQLite 版适合单机小项目但并发写会锁库多实例部署直接歇菜。选 PostgreSQL 的理由很实在它本来就是大多数团队已经在用的数据库不用为了 Checkpoint 单独引入一套存储它的事务和并发控制足够强多个 Agent 实例同时写 Checkpoint 不会互相踩它的 JSONB 类型天然适合存状态快照查询和索引都方便。Redis 虽然快但持久化保证弱一些而且状态数据往往需要长期保留做审计放 Redis 里成本反而更高。这里有个容易忽略的点Checkpoint 的数据量会随着对话轮次增长。一个跑了五十步的 Agent状态快照可能有几百 KB。如果每步都全量存一份存储会膨胀得很快。所以后面讲表结构的时候我会说清楚哪些字段是增量、哪些是全量。3. PostgreSQL Checkpoint 的表结构与落盘时机3.1 Checkpoint 到底存了什么很多人以为 Checkpoint 存的是“对话历史”其实不止。一份完整的 Checkpoint 至少包含四部分channel_values各个状态通道的当前值比如 messages、tool_results、channel_versions每个通道的版本号用于判断哪些通道自上次以来变了、pending_writes还没提交的写入用于崩溃恢复时重放、metadata步数、来源、时间戳等。LangGraph 的 PostgresSaver 会把这些拆成几张表。核心是checkpoints表存主快照checkpoint_writes表存待提交的写入checkpoint_blobs表存大的二进制或 JSON 数据。这种拆分的目的是让小的元数据查询快、大的状态数据按需加载而不是每次恢复都把几百 KB 全读出来。表名存什么关键字段用途checkpoints主快照元信息thread_id, checkpoint_id, parent_id定位某次运行的断点checkpoint_blobs通道的实际值channel, version, type, blob恢复时重建 Statecheckpoint_writes未提交的写入task_id, idx, channel崩溃后重放未完成步骤checkpoint_migrations版本管理v表结构升级3.2 thread_id 是整个恢复机制的钥匙恢复能不能成功八成取决于thread_id设计得对不对。thread_id是 LangGraph 用来标识“一条独立会话线”的标识。同一个 thread_id 下的所有 Checkpoint 构成一条链通过parent_id串起来恢复时框架会沿着这条链找到最新的那个 Checkpoint。我的经验是thread_id 一定要和业务实体绑定而不是随机生成。比如客服场景用user_id session_id订单处理场景用order_id。这样即使用户换了设备、清了缓存只要业务标识还在就能找回之前的进度。反过来如果你用 UUID 随机生成 thread_id 又不落库到业务表那用户一刷新就彻底找不回来了Checkpoint 存了也白存。# thread_id 的推荐构造方式 thread_id forder-{order_id}-user-{user_id} config {configurable: {thread_id: thread_id}}3.3 落盘时机不是每一步都值得存LangGraph 默认在每个节点执行后落一次 Checkpoint。听起来很安全但实际跑起来你会发现两个问题一是写库频率太高一个十步的任务写十次QPS 上去了二是很多中间步骤其实没有恢复价值存了纯属浪费。我的做法是分级落盘关键节点工具调用前后、人工审核点、外部 API 调用后强制落盘纯计算节点格式化、拼接字符串可以跳过。LangGraph 允许你在编译图的时候配置 checkpointer 的行为也可以通过自定义节点来控制。判断标准很简单——这一步如果重跑会不会产生副作用或浪费大量资源会的话就必须存。注意工具调用是最需要落盘的地方。因为工具往往有副作用发消息、写数据库、调第三方接口一旦执行了但没落盘恢复时就会重复执行。LangGraph 的 pending_writes 机制就是为此设计的它记录“这个工具已经执行但结果还没提交”恢复时优先重放而不是重跑。4. AG-UI 协议把 Runtime 状态实时喂给前端4.1 为什么需要 AG-UI 这一层后端把状态存好了、能恢复了但用户看不到过程体验还是差。用户需要知道“Agent 现在在干嘛、卡在哪一步、还要多久”。这就需要一套前后端之间的事件协议把 Runtime 的状态变化实时推过去。AG-UI 就是干这个的——它定义了一组标准事件类型让前端能以统一的方式消费 Agent 的运行状态。AG-UI 的核心事件大致分几类生命周期事件run started、run finished、run error、文本事件text message start、content、end、工具事件tool call start、args、end、状态事件state snapshot、state delta。前端订阅这些事件就能拼出一个完整的执行视图。4.2 事件流设计的关键状态快照 增量这里有个设计上的坑我踩过一开始我只推增量事件delta前端自己维护状态。结果一旦丢了一个事件前端的状态就和后端对不上了而且很难发现。后来改成周期性推全量快照 中间推增量前端每次收到快照就整体覆盖收到增量就合并容错性好很多。具体节奏是run 开始时推一次全量快照之后每个节点结束推增量每 N 个节点或每 M 秒再推一次全量快照做校准。这样即使中间丢了几个增量下一个快照也能把状态拉回来。# 伪代码AG-UI 事件推送的节奏控制 async def stream_with_snapshot(graph, input_state, config): step 0 async for event in graph.astream_events(input_state, config): step 1 yield to_agui_delta(event) # 每个事件推增量 if step % 5 0: snapshot graph.get_state(config) yield to_agui_snapshot(snapshot) # 每5步推全量校准4.3 恢复时前端怎么接上恢复场景下前端重新连接后第一件事是拉一次当前状态快照把界面渲染成中断前的样子然后再订阅后续事件。这里要注意的是快照里要包含“当前正在执行哪一步”的信息前端才能显示“正在恢复中”而不是“已完成”。我遇到过一个细节问题恢复时如果 Agent 正好卡在一个工具调用中间快照里会显示这个工具是 pending 状态。前端需要能渲染这种“半完成”状态否则用户会以为卡死了。AG-UI 的 tool call 事件里带 status 字段前端根据 status 显示不同的 UI这个设计很实用。5. 完整实操从零跑通一个可恢复的 Agent5.1 环境准备与依赖安装先把依赖装齐。核心是 langgraph、langgraph-checkpoint-postgres、psycopg前端侧如果用 React 就装对应的 AG-UI 客户端库。pip install langgraph langgraph-checkpoint-postgres psycopg[binary] langchain-openaiPostgreSQL 建议用 14 以上版本因为要用到 JSONB 的一些新特性。本地开发用 Docker 起一个最省事docker run -d --name agent-pg -e POSTGRES_PASSWORDsecret -p 5432:5432 postgres:165.2 初始化 Checkpointer 与建表LangGraph 的 PostgresSaver 提供了setup()方法自动建表第一次跑的时候调一次就行。from langgraph.checkpoint.postgres import PostgresSaver DB_URI postgresql://postgres:secretlocalhost:5432/postgres with PostgresSaver.from_conn_string(DB_URI) as checkpointer: checkpointer.setup() # 自动创建 checkpoints 等表提示setup()是幂等的重复调用不会报错但生产环境建议用迁移工具管理表结构而不是每次启动都调 setup避免权限问题和并发建表冲突。5.3 定义 State 与构建图State 的设计直接决定了 Checkpoint 存什么。我的原则是只放需要跨节点传递和恢复的数据临时变量不要塞进去。from typing import Annotated from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] order_id: str tool_results: dict step_count: int def call_model(state: AgentState): # 调模型返回新的 messages ... def call_tool(state: AgentState): # 执行工具返回 tool_results ... builder StateGraph(AgentState) builder.add_node(model, call_model) builder.add_node(tool, call_tool) builder.add_edge(START, model) builder.add_conditional_edges(model, should_continue, {tool: tool, end: END}) builder.add_edge(tool, model) graph builder.compile(checkpointercheckpointer)5.4 跑一次带中断的执行用同一个 thread_id 跑两次第二次就能看到恢复效果。第一次跑到工具调用前主动中断第二次从断点继续。config {configurable: {thread_id: order-123-user-456}} # 第一次执行在 tool 节点前中断 for event in graph.stream( {messages: [(user, 帮我查一下订单状态)], order_id: 123}, config, interrupt_before[tool], ): print(event) # 模拟进程重启后从断点恢复 for event in graph.stream(None, config): print(event) # 会从 tool 节点继续第二次传None作为输入LangGraph 会自动从该 thread_id 最新的 Checkpoint 恢复。这就是可恢复 Runtime 的核心用法。5.5 接上 AG-UI 推事件把graph.stream换成graph.astream_events把事件转成 AG-UI 格式推给前端。这一步的转换逻辑建议单独抽一个 adapter方便测试和替换。async def run_agent(input_state, config): async for event in graph.astream_events(input_state, config, versionv2): agui_event adapt_to_agui(event) if agui_event: yield agui_event前端侧订阅这个流按事件类型分发到不同的 UI 组件。文本事件渲染成气泡工具事件渲染成卡片状态事件更新顶部的进度条。6. 常见问题与排查技巧实录6.1 恢复后状态对不上怎么办最常见的症状是恢复之后 Agent 重复执行了已经做过的工具或者丢失了部分上下文。原因通常是 thread_id 不一致或者 Checkpoint 没落盘成功。排查顺序先确认两次执行用的是不是同一个 thread_id可以直接查checkpoints表看有没有对应记录再确认 checkpointer 是不是真的挂到了 graph 上compile时有没有传最后看是不是在工具执行后、落盘前就崩了这种情况 pending_writes 里会有记录恢复时应该重放而不是重跑。症状可能原因排查方法恢复后从头开始thread_id 不一致查 checkpoints 表工具重复执行落盘时机晚于执行检查 pending_writes状态字段丢失State schema 不匹配对比两次的 channel_values恢复报错找不到 Checkpoint表未初始化确认 setup() 已执行6.2 Checkpoint 表膨胀太快跑了一段时间发现checkpoint_blobs表几个 G 了。这是因为每步都全量存了 messages而 messages 是累积的。解决办法有两个一是配置只保留最近 N 条消息在 State 里历史消息归档到业务表二是定期清理旧的 Checkpoint只保留每个 thread 最新的几个。-- 清理 30 天前的 Checkpoint按需调整 DELETE FROM checkpoints WHERE created_at NOW() - INTERVAL 30 days;注意清理前一定要确认这些 Checkpoint 对应的任务已经终态否则删了就没法恢复了。建议加一个业务状态字段做联合判断而不是单纯按时间删。6.3 AG-UI 事件乱序或丢失前端偶尔出现状态错乱多半是事件乱序或丢失导致的。WebSocket 本身不保证顺序SSE 好一些但也会断。我的做法是给每个事件加一个单调递增的 seq 号前端发现 seq 跳号就主动拉一次全量快照校准。这个机制加上前面说的周期性快照基本能覆盖绝大多数异常。6.4 多实例部署时的并发问题多个 Agent 实例同时处理同一个 thread_id 会出问题。LangGraph 的 PostgresSaver 有基本的并发控制但业务上还是要避免同一 thread 被并发处理。我的做法是在业务层加一个分布式锁用 thread_id 做 key拿到锁才能执行。锁的粒度要细别一把大锁锁住所有会话。7. 几个我踩过之后才明白的经验第一个经验是别急着上 Checkpoint先把 State 设计干净。我一开始 State 里塞了一堆临时变量结果 Checkpoint 又大又乱恢复时还经常因为某个字段类型不对报错。后来把 State 精简到只留必要字段问题少了一大半。State 是契约设计的时候要像设计数据库表一样认真。第二个经验是恢复逻辑一定要写测试。正常路径跑通不代表恢复路径没问题。我专门写了一套测试模拟在任意节点中断然后恢复断言最终状态和不断的情况一致。这套测试帮我抓出了好几个隐藏的 bug比如某个工具在恢复时被重复调用。第三个经验是AG-UI 的事件设计要留扩展位。一开始我只推了文本和工具事件后来想加“思考过程”展示发现事件类型不够用只能改协议。现在我会在事件里预留一个meta字段放一些非核心但可能有用的信息扩展的时候不用动主结构。最后分享一个小技巧调试恢复问题时直接查数据库比看日志快得多。checkpoints表里的parent_id链能让你一眼看出执行路径checkpoint_writes能告诉你哪些步骤没提交。把这两张表的查询语句存成 snippet排查效率能翻倍。