去年我做内部一个自动化客服原型时最让我头疼的不是模型回答质量而是一个看起来很“基础”的问题任务跑到一半被打断了怎么让它恢复之后还记得自己刚才在干什么。前期原型用的是最朴素的手写 Loop模型调用、工具结果全堆在一个 Python 列表里。后来我把它整体改造成基于 LangGraph 的可恢复 Runtime用 PostgreSQL 保存 Checkpoint并通过 AG-UI 把交互断点暴露给前端。这篇就聊聊这个从“Loop”到“Runtime”的完整迁移过程。这套方案能解决的核心问题有三个一是执行中断后的上下文恢复二是多个服务实例共享同一段任务状态三是把“暂停等你确认”这类交互完完整整地传到网页端。如果你也正在从“while True 调模型”往“图编排 状态持久化”过渡这篇文章应该能帮你少踩几个我在写代码和排障时踩过的坑。我会把选型逻辑、关键实现、协议对接方式以及真实联调中遇到的意外都摊开讲。1. 为什么我放弃了手写 Loop状态丢失只是最轻的一个问题1.1 手写循环的“快乐时光”与第一根稻草先还原一下最原始的写法。一个典型的 agent 手写循环长这样messages: list[dict] [] while True: resp llm.chat(messages) tool_calls extract_tool_calls(resp) if not tool_calls: break messages.append(resp) for call in tool_calls: messages.append(execute_tool(call))这个模式非常简单直观本地做 Demo 的时候完全够用。当时我觉得所谓“agent runtime”不过就是循环外面套一个 HTTP 服务前端把用户消息传进来循环跑完再把最终结果传回去完事。第一根稻草出现在部署之后。服务发布新版本需要重启进程重启之后所有会话历史从内存里蒸发得一干二净。用户上一秒还在跟助手确认收货地址下一秒系统就“失忆”了。更尴尬的是一旦某个环节需要人工审批——比如“确认是否执行这笔退款”——循环根本不知道该停在哪个位置等用户回来。你想等就得自己写一个半吊子状态机把所有挂起中的请求塞进数据库再用一轮轮轮询去唤醒它们。这套东西写完就是一座屎山你不仅要处理并发写冲突还要自己设计恢复协议。1.2 四个致命短板状态、暂停、并发、审计手写循环的问题远不只是一个“丢了状态”那么简单。用一段时间之后我把它的问题归类成四个致命短板状态只活在内存里。进程一死连带着所有进行中的任务一起死没有任何持久化兜底。没有暂停原语。你无法让一个 agent 在某个节点停下来等用户确认后再回到同一个位置继续执行。没有并发隔离。所有会话共享同一个 Python 进程的内存多个用户同时触发任务时要么加锁要么互相污染。没有审计与重放能力。任务跑完之后没人能回答“它当时到底经过了哪些步骤”更别说从历史状态重新分支出一个新的走向。我做了个对比把“手写 Loop 的目标状态”和“我期望的 Runtime 目标状态”放在一起差距一下就很明显能力项手写 Loop可恢复 Runtime执行状态存储进程内列表外部持久化存储中断恢复无法恢复从最近 Checkpoint 继续人工确认自造状态机原生暂停原语多实例并发互斥或加锁共享状态天然支持过程审计无历史 Checkpoint 可追溯前端协议自造轮询标准事件流与交互协议这个表格不是想做纸上谈兵的架构对比而是我踩坑之后真正想明白的一件事Runtime 最核心的价值不在“循环写得有多快”而在“挂起之后能不能无损地回来”。如果你只需要一个在后台跑完就结束的任务手写循环其实够用只要一旦涉及长时间运行、中间等待、多端共享状态你就不再需要“更好的循环”而是需要一个“可挂起、可恢复、可审计的作业执行器”。1.3 想清楚要什么我的 Runtime 目标清单在动手换架构之前我给自己列了一张目标清单后面所有技术选型都是这张清单驱动的任务执行到任意节点进程被杀后都能从持久化状态里恢复。支持暂停并等待用户输入恢复后能回到同一个节点继续执行。多个服务实例可以共享同一个会话状态互相不打架。前端能实时收到任务进度也能在暂停发生时收到“交互请求”。提供可查询的历史执行记录方便排查问题。清单写完之后选型就变得清楚了。我需要一个编排框架负责“节点 暂停 恢复”一个数据库负责“状态落地”一个通信协议负责“前端交互”。最后落地的组合就是 LangGraph PostgreSQL Checkpoint AG-UI。2. 选型复盘LangGraph 负责编排PostgreSQL 负责记忆AG-UI 负责交互2.1 先把“Runtime”这个词掰扯清楚很多人一提 agent Runtime第一反应是“调模型的引擎”。但在我这个场景里Runtime 其实包含三层含义编排层决定任务按照什么顺序、什么条件执行遇到中断如何暂停恢复后怎么走。状态层把每一步执行结果持久化让 Runtime 的“记忆”不依赖某个具体进程。通信层把执行进度、暂停原因、用户交互请求以标准格式推送给前端。手写 Loop 只解决了第一层里最浅的部分顺序调用 LLM 和工具。状态层和通信层几乎为零。所以选型时我不是选一个框架而是选三个组件各管一段。LangGraph 管编排PostgreSQL 管状态AG-UI 管交互。这也是为什么我不太认同“LangChain 和 LangGraph 哪个好”这种问法前者偏模型和工具调用封装后者偏图编排与持久化解决的问题不一样。2.2 为什么是 LangGraph图就是一条可挂载的“断点时间线”LangGraph 让我最心动的一点是它把 agent 流程显式建模为图和状态并且在图的执行引擎里内置了 Checkpoint 与中断机制。你可以把 LangGraph 的状态管理理解成 Git每次节点执行产生一个新的“版本”版本之间通过检查点连接。对一个执行中的任务调用中断就像在代码里打了一个断点所有现场都保留恢复执行则像从断点继续跑不会从头再来。具体到代码层面它提供了几个我非常需要的能力StateGraph定义节点和边支持条件分支、循环、并行。checkpointer注入到编译后的图里自动保存每一步状态。interrupt()原语让节点执行在某个位置暂停并返回一个可以被前端捕获的事件对象。Command()用来带着用户输入恢复一个已经暂停的执行。这套组合拳正是我在 1.1 节里列出的“暂停原语”和“恢复能力”。它不是靠我自己拿 while 循环加数据库模拟出来的而是框架层面的第一公民特性可靠性和一致性都有保障。2.3 为什么是 PostgreSQL共享状态的老实人状态层我几乎没有犹豫就选了 PostgreSQL。核心原因有三点事务与一致性。检查点写入涉及多个表的更新PostgreSQL 的事务机制能保证要么全部成功要么全部失败不会出现写了一半的脏状态。多实例共享。LangGraph 的 Checkpoint 只需要一个数据库连接多个服务实例读取同一个会话状态天然安全省去了我自己实现分布式锁的麻烦。可观测性。我可以用 SQL 直接查checkpoints表看一个任务到底执行到了哪里这在调试中断恢复时简直救命。我也考虑过 Redis 和本地文件。Redis 的问题是持久化策略和内存容量限制文件存储则无法在多实例间共享。PostgreSQL 在这些选项里最“老实”数据落地稳定支持并发访问运维经验和工具链也最成熟。实际项目中我还用 Docker 起了本地开发库生产环境用的是云数据库迁移成本几乎为零。2.4 AG-UI 补上最后一块UI 与 Agent 的标准对话有了编排和状态还差最后一公里前端怎么知道任务“卡”在哪儿怎么把用户的选择传回去我最早的做法是自己定义 WebSocket 消息格式结果前端同学一天问我八次“这个字段是啥意思”。AG-UI 的价值就在这时体现出来了。它是一套面向 agent 与界面交互的开放协议核心是让 Runtime 和前端之间的事件交互有统一格式。你不需要自己发明一套“任务状态 暂停事件 恢复指令”的轮子协议已经把这类语义定义好了。在我理解里AG-UI 最关心三件事Task 状态任务从创建、执行到完成的状态变化。Agent 更新执行过程中产生的进度信息、中间消息。Interaction 交互Runtime 需要用户输入或确认时发给前端的“暂停请求”以及前端回传的“用户响应”。这跟 LangGraph 的interrupt()天然互补。LangGraph 负责“在代码层面停下来”AG-UI 负责“让界面知道为什么停、如何继续”。二者通过适配层对接前端不需要关心 LangGraph 内部 schema只要遵循 AG-UI 的消息规范就能驱动整个 agent 任务。3. 中断恢复的关键链路Checkpoint 机制与线程隔离3.1 先理解 Checkpoint每隔一个 super-step 就给状态拍快照LangGraph 的持久化单位是“超级步骤”简单说就是两个检查点之间的节点执行区间。每执行完一个节点引擎都会把当前图的通道状态、节点执行结果、待处理事件保存到检查点里。用数据库的视角看就是把一整份任务状态快照写入 PostgreSQL。这个快照不是草率地在内存里 dump 一下而是有结构地拆分存储检查点主记录保存状态标识和元数据blob 保存大块的状态数据writes 保存每个节点产生的结果。这样设计的好处是恢复时不需要加载整个历史只需要读取最近一个检查点加上必要时从 writes 里继续推进。在我实际使用中LangGraph 的后端在这里采用 PostgreSQL 后thread_id就是一切恢复逻辑的锚点。你可以把thread_id理解成“这条会话的业务主键”它对应一个订单、一个工单或者一次客服对话。同一个thread_id下每次执行都会叠加新的检查点形成一条可回溯的时间线。3.2 准备环境PostgreSQL 与连接参数先说我用的开发环境。本地用 Docker 启动一个 PostgreSQL 实例最省事docker run -d --name langgraph-pg \ -e POSTGRES_USERagent \ -e POSTGRES_PASSWORDagent \ -e POSTGRES_DBruntime \ -p 5432:5432 \ postgres:16连接字符串就是这样的格式DB_URL postgresql://agent:agentlocalhost:5432/runtime需要注意LangGraph 的 PostgreSQL Checkpoint 会自己创建checkpoints、checkpoint_blobs、checkpoint_writes这些表。你不需要手工建表但必须在首次使用前调用setup()完成初始化否则会撞上“relation does not exist”的报错。这个坑我后文具体讲。3.3 一个带 interrupt 的最小可用图下面是一个带人工确认中断的最小图。我用它来做所有恢复实验的基础from typing import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.postgres import PostgresSaver from langgraph.types import interrupt, Command class AgentState(TypedDict): messages: list pending: dict def draft_reply(state: AgentState): # 模拟 LLM 生成一个需要用户确认的回复草稿 draft {content: 我准备给您退款 128 元请确认} decision interrupt({ type: confirm_refund, payload: draft, }) if decision.get(approved): return {messages: [draft[content]], pending: {}} else: return {messages: [用户拒绝], pending: {}} builder StateGraph(AgentState) builder.add_node(draft, draft_reply) builder.add_edge(START, draft) builder.add_edge(draft, END) with PostgresSaver.from_conn_string(DB_URL) as checkpointer: checkpointer.setup() graph builder.compile(checkpointercheckpointer)这里最关键的是interrupt()。当节点执行到这一行图不会继续往下走而是把执行现场保存到检查点里然后“挂起”。调用方拿到的返回结果里会包含中断事件。整个过程不需要我写任何持久化代码框架自动完成。3.4 用 thread_id 恢复Command 包装的恢复输入恢复执行同样简洁。假设我第一次调用时用的线程 ID 是order-123config {configurable: {thread_id: order-123}} # 首次执行会停在 interrupt 处返回中断信息 result graph.invoke({messages: []}, config) # 用户在前端点了“确认”我把决定回传 resume_result graph.invoke( Command(resume{approved: True}), config, )这里有个重要细节恢复执行时我没有重新传入{messages: []}而是用Command(resume...)携带用户输入。原因在于检查点已经保存了之前的完整状态恢复时只需要告诉引擎“用户对上次暂停的问题给了什么回答”。如果我在恢复时又传一遍普通输入LangGraph 会把新旧状态做合并很容易造成消息重复或状态错乱。这个我在实际项目中踩过一次后面会展开讲。线程隔离也要提一句不同业务任务用不同的thread_id检查点天然隔离互不影响。同一个人在处理多个工单时也可以打开多个线程每个线程是一个独立的时间线。这就解决了我 1.2 节说的并发隔离问题。4. AG-UI 协议集成让 Runtime 能力真正暴露给前端4.1 AG-UI 到底解决什么问题在引入 AG-UI 之前我的前端只能做两件事发消息给后端、轮询后端拿结果。这种方式面对“任务可能随时暂停”的状态机时很笨拙因为前端根本不知道什么时候该停下来等用户。AG-UI 的定位就是解决这个信息不对称它定义了 agent 在执行过程中和前端之间传递的标准消息流包括任务进度、代理更新、交互请求等。前端拿到这些消息之后不需要理解 LangGraph 内部实现只需要理解“这个事件是让我显示进度”“这个事件是让我弹确认框”“这个事件是任务完成了”。我把它理解为一种“协议层翻译”后端 Runtime 说“我在draft_reply节点停了有一个confirm_refund请求等待用户响应”。AG-UI 把它翻译成前端认识的交互事件“这里有一个人工确认请求展示给用户吧”。用户点击确认后前端把事件回传后端再把结果翻译成 LangGraph 的Command(resume...)。这样前后端就解耦了。以后如果我把 LangGraph 换成别的 agent 引擎只要保证继续输出 AG-UI 事件前端一行代码都不用改。4.2 事件与交互协议两侧的消息流我实际实现里AG-UI 的 WebSocket 连接上的消息大致是这样的节奏{type: run_started, task_id: order-123, thread_id: order-123} {type: agent_update, step: 正在生成退款方案} {type: interaction, id: it_001, request: confirm_refund, payload: {content: 确认退款 128 元}}前端收到interaction事件后渲染一个确认框。用户点击“同意”前端发送一条交互响应消息{type: interaction_response, id: it_001, response: {approved: true}}后端拿到这条消息后把response打包成Command(resume...)使用同一个thread_id恢复 LangGraph 的执行。这套消息格式比我自己之前乱写的字段规范清晰得多。你不需要从零设计协议只需要把 LangGraph 的中断点映射成 AG-UI 的交互事件把事件响应映射回Command。映射关系可以固化成一张表LangGraph 内部事件AG-UI 事件语义前端行为节点开始执行agent_update展示进度节点执行中断interaction渲染确认框/输入框用户提交响应interaction_response发送响应执行完成run_completed展示最终结果4.3 把 LangGraph 暂停翻译成 AG-UI 交互事件翻译的核心不复杂。你在调用图时如果遇到中断返回不要直接把它丢掉而是先构造一个 AG-UI 风格的interaction事件再挂起等待用户响应。为了把这个逻辑封装起来我写了一个很薄的桥接层class RuntimeBridge: def __init__(self, graph): self.graph graph async def on_action(self, action, thread_id): config {configurable: {thread_id: thread_id}} return self.graph.invoke(action, config) async def on_user_response(self, interaction_value, thread_id): config {configurable: {thread_id: thread_id}} return self.graph.invoke( Command(resumeinteraction_value), config )on_action是前端发起任务时走的入口on_user_response是用户处理完交互之后走的入口。两个方法都用同一个thread_idLangGraph 就会自动把两次调用衔接在同一个时间线上。Bridge 层不负责业务逻辑只负责协议翻译这让后面的维护和排查都变得简单。4.4 WebSocket 适配器骨架有了 Bridge剩下的 WebSocket 适配层就是把它暴露给前端。我用了最简单的 FastAPI WebSocket 做演示性质实现from fastapi import FastAPI, WebSocket app FastAPI() bridge RuntimeBridge(graph) app.websocket(/ws/agent) async def agent_socket(ws: WebSocket): await ws.accept() while True: payload await ws.receive_json() action payload[type] if action action: result bridge.on_action( payload[input], payload[thread_id], ) await ws.send_json(result) elif action interaction_response: result bridge.on_user_response( payload[response], payload[thread_id], ) await ws.send_json(result)注意我把thread_id放在客户端发送的 payload 里这是为了让前端可以随时恢复一个旧的会话。用户刷新页面之后只要在 URL 上带上thread_id后端马上就能把之前挂起的任务重新接上。5. 实测恢复流程与踩坑记录5.1 一个完整的断线恢复剧本理论说完看一个完整的实测剧本。场景是电商售后工单agent 需要草拟一个退款方案等客服主管确认后才能提交。流程是这样跑的用户在工单系统创建售后单后端以thread_idaftersale-2024-001启动 LangGraph 执行。Agent 节点生成退款草稿执行到interrupt()暂停后端通过 AG-UI 向前端发送interaction事件。此时服务因为发布新版本被 kill进程退出。发布完成后服务重启。客服主管打开页面前端从 URL 参数里读到thread_id重新建立 WebSocket 连接。后端在数据库里查到这个thread_id的检查点确认任务处于暂停状态。前端发一条interaction_response内容为{approved: true}。后端用Command(resume...)恢复执行LangGraph 从上次暂停的节点继续生成最终退款指令。整个过程里前端只感知到了任务开始、暂停请求、恢复后完成。它不需要知道服务曾经重启过也不需要关心状态存在哪个表里。这就是“可恢复 Runtime”在最终用户体验上的价值。步骤和时间线整理成表格更直观阶段动作状态存储用户创建任务graph.invoke检查点写入 PostgreSQL模型草拟完成interrupt()暂停检查点保存现场服务重启进程退出数据库保留全部检查点用户回复前端发送交互响应恢复执行恢复执行Command(resume...)追加新的检查点5.2 坑一恢复时重新传了整轮输入Graph 从“起点”再跑一遍这是我遇到的第一个“看似合理但完全错误”的恢复方式。起初我恢复时是这样写的# 错误示范 graph.invoke({messages: [user_message]}, config)结果任务不仅没有从暂停处继续反而把新消息合并到了历史消息里导致前面的节点全部重跑了一遍。暂停前的状态被覆盖还会出现消息重复。原因在于 LangGraph 的检查点是对通道数据的合并不是覆盖。你用普通输入去恢复它会把输入追加到已有的状态通道里而不是回到中断处等待Command。正确做法就一条恢复执行时用Command(resume...)传用户响应不要直接把任务输入重新塞进去。5.3 坑二AsyncPostgresSaver 与事件循环的爱恨情仇因为我整套后端是异步 WebSocket 服务一开始我直接用同步版本的PostgresSaver请求一多就出现连接卡顿。后来换成AsyncPostgresSaver才解决。但这又带出一个新问题异步检查点器必须跟它所属的事件循环生命周期绑定。我遇到的具体表现是在不同请求里反复新建 saver 实例导致数据库连接池被反复重建偶发connection already closed报错。排查链路是这样的先在 Python 脚本里单独跑图确认恢复逻辑没有错。再用一个简单的 FastAPI 接口包一层发现单请求正常多请求就报错。打开 PostgreSQL 的pg_stat_activity发现大量闲置连接判断是 saver 生命周期管理不当。最后改成统一的连接生命周期管理把 AsyncPostgresSaver 实例和 ASGI 应用一起初始化整个进程只维护一套连接池。如果你用的是 FastAPI一个可参考的做法是在应用启动时创建 saver然后注入到图编译中。不要在每个请求里现建现拆。5.4 坑三忘了初始化 Schema这个问题藏在最容易忽略的细节里。搭建新环境时如果你直接编译图然后执行会看到类似这样的报错psycopg.errors.UndefinedTable: relation checkpoints does not exist原因很简单LangGraph 的 PostgreSQL 检查点需要checkpoints、checkpoint_blobs、checkpoint_writes这三张表但框架不会在每次运行时自动建表。初始化动作是显式调用checkpointer.setup()。我在 3.3 节的示例里已经写了这一行但实际项目中如果代码路径太多很容易漏掉。建议在服务启动阶段做一次整体初始化并且把建表逻辑放到 CI/CD 的数据库迁移任务里而不是等运行时再去碰运气。5.5 坑四状态里混进了不可序列化对象检查点数据最终要落到 PostgreSQL 的 JSONB 字段里所以状态通道里所有内容必须是可 JSON 序列化的。我在早期版本里往状态里塞了一个 Pydantic 模型对象结果执行到检查点保存时就抛序列化错误。排查时我先用json.dumps手动序列化状态发现确实有对象转不了 JSON才定位到是自定义模型没有被序列化。解决办法有两种一是状态里只存 dict 或 list 等基础类型二是给自定义对象注册序列化器。我最后选了前者因为从架构上更干净状态层就是纯数据不掺任何业务对象。5.6 一个关于幂等性的提醒Checkpoint 不会替你做账这是我在上线前反复跟自己确认的问题。Checkpoint 能恢复执行现场但它不会帮你消除副作用。什么意思如果你的某个节点已经调用了一个外部支付接口然后任务在下一个节点暂停恢复执行时会从暂停处继续不会重放之前已经执行过的步骤。这看起来是好事但反过来也意味着如果外部副作用发生在“检查点保存之前”而进程在保存前就挂了那恢复时可能什么都不做外部系统却已经收到过请求。所以在设计外部副作用时务必要做幂等控制比如在订单上携带唯一的操作键或者把外部调用放到靠近检查点保存之后的位置。恢复不是魔法它只管把状态带回来不负责替你回滚世界。6. 把可恢复 Runtime 当作基础能力再往深想一步6.1 进一步从恢复走向审计和时间旅行可恢复性给我带来的额外惊喜是它顺带解决了审计问题。只要任务跑过数据库里就有一串检查点记录着每一步的关键状态。出问题的时候我可以直接查checkpoints表按thread_id拉出完整执行历史甚至能看到当时的中断事件是什么。更进一步的用法是时间旅行LangGraph 提供了get_state_history和update_state可以在历史检查点之间切换甚至从某个旧状态分支出新的执行路径。这在做“假设分析”和“问题复现”时非常有用。比如用户说“上次你执行到第三步就错了”我可以精确恢复第三步时的完整状态重新调试。6.2 性能和部署建议PostgreSQL 写检查点的频率取决于你的图有多少节点和边。大多数场景下一个检查点只有几 KB 到几十 KB 的 JSON 数据不会给数据库造成压力。如果你的图在高频循环节点上跑可以考虑把状态通道拆小只保存必要数据避免一个节点状态动辄几百 KB。部署层面我现在的做法是数据库使用 PostgreSQL 14 以上版本开启合理的 WAL 和备份策略。应用服务多实例部署共享同一个数据库靠thread_id隔离会话。每个实例的图定义必须保持一致否则检查点对应的状态结构会不兼容。这里最容易被忽略的就是“多实例共享状态”反而在大并发场景下成了优势因为不需要做会话粘滞。任何实例只要拿到thread_id都能无缝接管任务。6.3 最后聊聊我自己的体会从手写 Loop 迁到这套可恢复 Runtime代码量并没有减少多少但系统的心智模型完全变了。以前我每天担心的是“状态别丢”现在担心的是“状态怎么用好”。LangGraph 把“中断”做成了第一公民PostgreSQL 把“记忆”做成了基础设施AG-UI 把“交互”做成了标准协议这三者组合在一起才真正让我觉得 agent 项目有了可长期演进的底座。如果你也在做类似的项目我的建议是第一优先把中断恢复跑通第二再考虑协议标准化。状态恢复是地基AG-UI 是地面上的路标地基稳了路标才有意义。等你把中断恢复做成默认能力之后会发现很多以前设计起来很痛苦的功能——人工审核、分步确认、暂停后再启动——都会突然变得非常自然。到那时候你就会明白“Runtime”这个词的重量。