LangGraph+PostgreSQL Checkpoint+AG-UI：构建可中断恢复的Agent运行时

接手过 AI Agent 项目的朋友应该都有同感一开始写执行循环特别爽一个while True加几个 if 分支就能跑起来但一旦业务要求做到一半可以暂停、断电能续跑、Web 页面上能看到实时进度手写 Loop 就彻底露馅了。这篇博文我会用 LangGraph 替换手写编排用 PostgreSQL Checkpoint 把执行现场持久化再用 AG-UI 把中断恢复这条链路完整跑通——不是抄官方文档而是把我自己从睡着被线上告警叫醒到半夜停服还能无损恢复的全过程梳理成实操经验适合正在做 Agent 服务、被无状态循环折磨过的后端开发。1. 手写 Loop 的瓶颈从能用到不可恢复1.1 我最初的手写执行循环长什么样最早的项目其实不复杂。一个订单处理 Agent用户提交工单后后端起一个线程跑循环调大模型判断意图、调工具查库存、再调大模型生成回复最后更新工单状态。代码大概长这样def run_agent(order_id: str): state load_order(order_id) while state[status] ! done: step decide_next(state) # 调一次大模型 if step need_refund: result refund_tool(state) # 调工具 elif step need_human: send_to_human(state) # 挂着等人工 break else: result chat_completion(state) apply_state(state, result) save_order(state) # 只存业务数据不存执行现场在低并发、演示用的阶段这个循环完全没问题。所有状态就是订单表单本身反正跑挂了重新人工处理。但一旦有几十个会话同时在跑就会暴露一个本质矛盾循环里流动的执行现场根本没有持久化保存的只是业务结果。大模型返回了一半、工具已经扣了积分、下一步刚要调用另一个接口——这时候进程一崩现场全丢。1.2 三个让我决定重构的现场事故第一个事故是容器被 OOM Kill。K8s 直接杀掉 pod正在处理的 6 个会话全部停留在已支付但未发货的中间态没有一个能自己恢复。因为decide_next的中间结果保存在内存里重启后state又从数据库里老的那份读起等于白跑。第二个事故是人工审核中断。系统调用send_to_human后把工单挂起等客服在后台点了通过代码又在另一个服务实例上while循环早就退出了没人继续往下走。逻辑上审核完就要自动执行退款实际却变成了客服每天手工补单。第三个事故最烦用户在前端看到正在处理退款刷新一下就变成处理失败请重试。原因是后端进程还在跑但前端没有订阅事件流的能力也没有恢复会话的接口用户一刷新就和执行中的 runtime 失联了。这三个事故叠加让我下决心把架构从手写业务循环改成可恢复的 Agent Runtime核心就三件事执行编排交给图、执行现场存进 PostgreSQL、事件通道用 AG-UI 统一。2. 换引擎LangGraph 的 StateGraph 与检查点机制2.1 状态图怎么定义节点、边、ReducerLangGraph 的核心模型是状态图。你不需要写while和if了而是声明一批节点和节点之间的转移规则由 Runtime 自己去决定每一步走哪条边。我重构后的订单 Agent 大概是这个结构from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): order_id: str messages: Annotated[list, add_messages] decisions: dict refund_status: str def parse_order(state: AgentState): # 解析工单返回 {decisions: {need_refund: True}} return {decisions: extract_entities(state[messages])} def call_llm(state: AgentState): # 调用模型生成下一步动作返回消息 reply llm.invoke(str(state[decisions])) return {messages: [reply]} def refund_tool(state: AgentState): # 调用退款工具幂等设计 refund_id do_refund(state[order_id]) return {refund_status: refund_id} def route_after_parse(state: AgentState): if state[decisions].get(need_human): return ask_human return call_llm builder StateGraph(AgentState) builder.add_node(parse_order, parse_order) builder.add_node(call_llm, call_llm) builder.add_node(refund_tool, refund_tool) builder.add_edge(START, parse_order) builder.add_conditional_edges(parse_order, route_after_parse, {call_llm: call_llm, ask_human: ask_human}) # ask_human 节点里用 interrupt() 挂起等待人工输入 builder.add_edge(call_llm, refund_tool) builder.add_edge(refund_tool, END)这里messages: Annotated[list, add_messages]是 Reducer作用是告诉 LangGraph节点返回的新消息不是覆盖旧状态而是追加。这块很容易被忽略但它正是过程可恢复的地基——每次节点结果都会按 Reducer 规则被合并进状态而不是粗暴覆盖。为什么我不再手写循环因为图定义把流程拓扑和实际执行解耦了。Runtime 知道上一步执行到了哪个节点、输入是什么、输出是什么手写循环只知道当前局部变量里那个statedictionary。2.2 Checkpoint 到底存了什么以及什么时候存LangGraph 的检查点机制简单说就是每执行完一个节点官方叫 super-stepRuntime 就把整个 AgentState 快照存到 Checkpointer 里。快照不只是 messages还包括图的位置信息、节点的输入输出、配置信息。看起来像是在给每一步拍照。我最初以为持久化就是自己定义save_order存几个字段后来发现完全不够。自研保存通常会忘掉流程位置比如刚执行完 parse_order下一步应该走 call_llm而 LangGraph 的 checkpoint 会把流程位置和状态一起存恢复时直接从断点继续。用带检查点的编译方式from langgraph.checkpoint.postgres import PostgresSaver conn_str postgresql://agent_user:passlocalhost:5432/agent_runtime with PostgresSaver.from_conn_string(conn_str) as checkpointer: checkpointer.setup() # 创建 checkpoint 相关表幂等 agent_app builder.compile(checkpointercheckpointer)运行时要显式携带一个线程 ID。这个 ID 就是业务流程 ID相当于这份执行现场归谁config {configurable: {thread_id: order-1024}} result agent_app.invoke({order_id: 1024, messages: [{role: user, content: 帮我退款}]}, config)同一个订单再次调用同一个thread_idLangGraph 会去 PostgreSQL 查上次的 checkpoint而不是从零开始# 客服在后台点了通过以后业务服务只需要再次调用 resume_result agent_app.invoke( {messages: [{role: human, content: 确认通过}]}, {configurable: {thread_id: order-1024}} )这一步就把之前人工审核后没人继续跑的事故根除了没有while循环在等是 Runtime 通过 checkpoint 知道流程卡在ask_human收到新消息后自然往下走。2.3 从App到可恢复 Runtime把编译产物叫 App 很容易让人误以为只是一个函数。实际它应该被当作一个微型 Runtime它有生命周期、有状态存储、有恢复语义。在这个思维转变下我建议项目里单独封装一个runtime.py不要在每个接口处随手创建图实例。我会把 Runtime 封装成三层第一层graph_builder()只负责定义节点、边、Reducer不持有任何数据库连接。第二层build_runtime()加载 Checkpointer编译图返回可调用对象。第三层execute()/resume()接收thread_id和输入消息统一走同一条 invoke 路径。这样的好处是恢复和首次执行共用一套逻辑不会出现新订单走新流程、老订单走另一套流程的屎山代码。3. PostgreSQL 作为 Checkpoint 存储选型理由与接入细节3.1 为什么选 PostgreSQL 而不是 Redis 或 SQLiteLangGraph 官方支持内存、SQLite、PostgreSQL 和 Redis 等检查点实现。我最终选了 PostgreSQL主要因为三点事务边界和状态一致性。Checkpoint 的写入和业务数据的更新如果放在同一个 PostgreSQL 事务里就能做到要么都生效、要么都回滚。这对退款、物流这类场景太重要了。Redis 也能存但跨存储的事务一致性要自己折腾。运维熟快照恢复容易。业务数据库本来就是 PostgreSQL 的人加一张 checkpoint 表不会引入新组件。备份策略直接覆盖。并发访问可控。多实例部署时多个 worker 会同时读写同一个 thread 的 checkpointPostgreSQL 的行锁和乐观锁语义比 SQLite 可靠得多也比 Redis 的方案更好排查。内存检查点只适合单元测试进程一重启就全没了和手写 Loop 没有本质区别。SQLite 适合单机演示不适合多副本部署——它支持文件锁但跨机共享一个 SQLite 文件是灾难。3.2 建表和连接参数的实际细节PostgresSaver.setup()会自动建表唯一的额外工作就是准备一个独立的数据库或者 schema。我建议单独建库agent_runtime不要和业务主库混在一起因为 checkpoint 写入非常频繁容易把主库的缓存和 IO 打乱。连接串上建议考虑连接池。langgraph-checkpoint-postgres底层用的是 psycopg连接串里可以带池化参数或者在应用层用一个连接池对象传递进去。我第一次直接裸连接结果并发一上来Postgres 的连接数直接被打满。一个稳定配置的例子from langgraph.checkpoint.postgres import PostgresSaver from psycopg_pool import ConnectionPool pool ConnectionPool( conninfopostgresql://agent_user:passlocalhost:5432/agent_runtime, max_size20, kwargs{autocommit: True}, ) checkpointer PostgresSaver(pool)注意PostgresSaver的同步 API 会占用一个连接直到操作完成。如果你每个请求都从池里拿连接而不归还很快会耗尽。我当时写了一个 FastAPI 中间件在请求结束时统一checkpointer.close()或者干脆把 Runtime 做成单例连接池只初始化一次。3.3 Checkpoint 存储结构需要了解的三张表LangGraph 的 PostgreSQL Checkpointer 会自动维护一组表。了解它们的大致用途排查问题会快很多表用途我的实际使用感受checkpoints保存每个 thread 每个 step 的状态快照生产里增长最快需要定期清理checkpoint_blobs保存大对象消息内容等和 checkpoints 一起膨胀checkpoint_writes保存节点写入的中间结果并发排查时最关心这张表我印象最深的是checkpoint_writes它记录了每一个节点写入的键值。做 AG-UI 事件回放时如果前端要显示某一步的原始输出直接查这张表就能精确到节点级。清理策略我放在第五节讲这里先记住一个原则不要想着删 checkpoint 来释放空间而丢掉恢复能力应该用 LangGraph 自带的按时间清理方法只保留最近 N 天。比如checkpointer.delete_older_than(days7)4. AG-UI 事件流与前端恢复跑通中断恢复的最后一公里4.1 AG-UI 在架构里的位置Runtime 能恢复不等于前端能恢复。用户刷新页面后前端必须知道当前 thread 停在哪个节点、下一步要等什么。这块我引入了 AG-UI目标只有一个把 Runtime 的节点级状态变成前端可以订阅、可以续传的事件流。我理解的 AG-UI 是一种轻量的 Agent UI 协议约定定义一组标准事件比如run_started、node_started、node_completed、token、interrupted、run_completed。它不像业务 API 那样返回一个 JSON 结果而是把执行过程的每一个阶段都推给前端。这样用户看到的不是转圈等结果而是当前正在解析订单 → 正在调用模型 → 正在执行退款。如果你不想引入完整协议也可以只实现其中三个端点。我把它们命名为POST /agent/run发起一次执行返回 SSE 事件流。GET /agent/events订阅/重连事件流。POST /agent/resume对中断的 thread 发起恢复。4.2 三个端点的职责与伪代码run端点的核心逻辑是拿到thread_id和用户消息调用 Runtime 的 invoke同时把事件塞进流式响应。这里不要用同步方式等完整结果再返回必须边执行边推事件。from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse app FastAPI() app.post(/agent/run) async def agent_run(req: Request): body await req.json() thread_id body[thread_id] message body[message] async def event_stream(): async for event in agent_runtime.astream_events( {messages: [{role: user, content: message}]}, {configurable: {thread_id: thread_id}}, versionv2, ): # 把 LangGraph 事件映射到 AG-UI 事件 ui_event map_runtime_event_to_agui(event) if ui_event: yield fevent: {ui_event[type]}\ndata: {json.dumps(ui_event)}\n\n return StreamingResponse(event_stream(), media_typetext/event-stream)astream_events会产出节点开始、节点结束、令牌增量等事件。映射时要特别注意不是每个 LangGraph 内部事件都该推到前端。我只推送节点级事件和 tokenon_chain_start这种内部调试信息全部过滤否则前端会被噪音冲爆。resume端点的实现比较有意思它本质还是 invoke只是不带新消息或者带一条人工审核指令。LangGraph 会从 checkpoint 恢复继续走没有执行完的边app.post(/agent/resume) async def agent_resume(req: Request): body await req.json() thread_id body[thread_id] action body.get(action, {type: continue}) async def event_stream(): async for event in agent_runtime.astream_events( None, # 没有新消息只是继续 {configurable: {thread_id: thread_id}}, versionv2, ): yield format_agui_event(event) return StreamingResponse(event_stream(), media_typetext/event-stream)4.3 前端断线重连与消息游标SSE 本身有断线重连机制浏览器使用EventSource时断线后会自动重连。但重连之后服务端怎么知道客户端已经收到哪些事件我的方案是维护一个以thread_id为维度的游标——每个事件除了类型还要带上序号或时间戳cursor_store {} # 生产里放 Redis 或一张表 def format_agui_event(event, thread_id): seq cursor_store.get(thread_id, 0) 1 cursor_store[thread_id] seq return { seq: seq, type: event[type], node: event.get(name), data: event.get(data), }前端重连时可以携带Last-Event-ID我后端根据这个 ID 把没发出去的事件补推。这样刷新页面丢进度的问题就被消灭了前端刷新后拿到同一thread_id既能重建历史事件也能继续接收新事件。4.4 跑通中断恢复的完整链路演示我搭建的最小验证环境分四步读者可以照着复现启动 PostgreSQL创建agent_runtime库。启动后端服务初始化PostgresSaver并编译图。用一个测试脚本给order-1024发首条消息中途在ask_human节点用interrupt()挂起。打开前端页面调用POST /agent/resume确认事件流从ask_human继续而不是从头执行。实测下来断点位置、节点输入输出、前端历史消息三项都能完美恢复。这一步成功之后我立刻把以前手写循环里的人工审核后继续跑逻辑全部删掉因为它已经变成 Runtime 的内置能力了。5. 落地过程中的坑与心得5.1 checkpoint 表膨胀得像日志表LangGraph 的 checkpoint 是 append-only 风格每次节点执行都追加记录。订单量一上来checkpoints和checkpoint_blobs增长很快。这不是 bug是设计使然——你要支持时间线回放就得保留历史。我踩的坑是直到磁盘报警才想起来清理。后来设了一个定时任务每天凌晨调用checkpointer.delete_older_than(days3)只保留最近 3 天的完整检查点。如果业务上有长期审计需求可以把慢速存储和快速存储分离但大多数 Agent 场景三天足够了。5.2 状态里放了不可序列化对象LangGraph 的状态最终要写进 PostgreSQL所以状态里的字段必须能被序列化。一次我为了图方便把某个工具的 pandas DataFrame 塞进了状态结果节点执行完就抛cannot serialize异常。解决方式很朴素状态里只保留序列化友好的数据比如 dict、list、str。大对象如果需要跟踪可以在节点内部把它转成 JSON 字符串或者存一个引用 ID等真正需要时再从外部存储读取。这个原则也适用于 AG-UI 事件推送推到前端的数据必须可控不能把整个 DataFrame 直接 JSON dump。5.3 多个 worker 并发恢复同一个 thread上线后遇到一个隐蔽问题用户在前端点了两次继续或者客服系统和用户系统同时调用了resume两个请求同时读同一个 checkpoint然后都去执行下一步产生重复退款操作。LangGraph 的 checkpointer 有并发保护但我的业务节点不是天然幂等的。最后我用两层方案第一层业务节点全部实现幂等比如退款工具用refund_id去重第二层resume接口加了一个 Redis 分布式锁同一个thread_id同时只能有一个执行中的请求。实践下来还是幂等更可靠锁只能减少并发不能根治重复。5.4 AG-UI 事件重复与乱序SSE 重连之后浏览器会自动补发Last-Event-ID之后的事件但这会导致已经收到的事件在后端日志里被再次发送。前端如果不做去重UI 上就会显示两次正在调用模型。我的处理是在前端维护一个received_seq集合遇到重复seq直接丢弃后端的游标任务则要保证同一个 thread 的事件序号单调递增。乱序问题主要出在多节点并行执行时LangGraph 的 fan-out 分支可能会让两个子节点同时输出。这种场景下前端不要依赖事件到达顺序来渲染而应该以node_completed的state为准。5.5 关于这套方案的最终选择如果你现在的 Agent 项目也是手写循环我的建议是先别急着把全部代码重构成 LangGraph。可以先挑一个流程状态最多、恢复要求最高的业务做试点把 PostgreSQL Checkpointer 和 AG-UI 事件通道跑通。跑通之后你会发现中间态丢失、断线重试、人工介入恢复这些问题不再是运维事故而是架构本身就能兜住的事情。我个人在实际操作中的体会是手写 Loop 省下的时间最后都会在线上以更贵的成本还回去。状态持久化、流程可恢复、事件可续传这三件事在 Agent 服务里不是加分项是基本盘。下一次如果再有人跟我说“等挂了再人工处理也行”我会直接把这次重构前后的线上报警对比甩给他看。