1. 为什么手写 Loop 撑不过第三个需求我最早做对话式 Agent 的时候和大多数人一样写了一个while True循环调模型、解析工具调用、执行工具、把结果塞回消息列表、再调模型直到模型不再要求调用工具为止。这套东西在 Demo 阶段非常好用二十行代码就能跑起来调试也直观打个断点就能看到每一步的消息数组长什么样。问题出在需求开始变复杂之后。第一个需求是用户中途想插话改方向第二个需求是工具执行到一半服务重启了不能丢状态第三个需求是前端要实时看到 Agent 现在到底在干嘛而不是转圈等到最后。这三个需求本质上指向同一件事执行过程必须是一个可以被外部观察、可以被暂停、可以被持久化、可以被恢复的对象而不是一个跑在函数栈里的黑盒循环。手写 Loop 的根本缺陷在于它的状态活在 Python 的调用栈和局部变量里。进程一挂栈没了状态就没了你想在中间插一脚只能靠回调或者全局变量打补丁你想让前端看到中间过程只能自己往消息队列里塞事件塞着塞着就发现自己在重新发明一套运行时。这就是从手写 Loop走向可恢复 Runtime的动机——不是赶时髦是被需求逼的。这篇内容我打算把 LangGraph 的图执行模型、PostgreSQL Checkpoint 的持久化机制、以及 AG-UI 这套前端事件协议串起来讲一遍重点放在中断与恢复这条主线上。适合已经写过基础 Agent 循环、现在被状态管理和前后端同步折磨的开发者。我会把每一步为什么这么做讲清楚也会把我在实际搭建中踩到的坑摊开说包括 checkpoint 的线程 ID 设计、恢复时消息重复、AG-UI 事件顺序错乱这些真实问题。先给一个整体认知LangGraph 负责图怎么走Checkpoint 负责走到哪了记下来AG-UI 负责把走到哪了告诉前端。三者拼起来才是一个能中断、能恢复、能实时反馈的 Runtime。下面逐层拆。2. LangGraph 的图执行模型到底解决了什么2.1 从函数调用栈到显式状态图手写 Loop 的状态是隐式的藏在局部变量里LangGraph 干的第一件事是把状态变成显式的、可序列化的字典在图的节点之间传递。你定义一个State通常是个 TypedDict每个节点接收当前 state返回一个增量更新框架负责把增量合并回 state。这个设计看起来只是换了个写法但它带来的连锁反应非常大。因为 state 是显式的所以它可以被序列化因为可以被序列化所以它可以被存进数据库因为可以存进数据库所以进程挂了之后可以重新加载因为可以重新加载所以恢复这件事才有了物理基础。整条链路的第一块砖就是这里。很多人学 LangGraph 只记住了节点和边忽略了 state 显式化才是它区别于手写循环的本质。我一般会把 state 设计成几块messages存对话历史current_step之类的字段记录流程位置tool_results存工具产出再加一些业务字段。关键原则是state 里只放需要跨节点、跨进程存活的数据临时变量不要往里塞否则 checkpoint 会越来越臃肿序列化开销也会上去。2.2 节点、边与条件跳转的执行语义LangGraph 的图由节点node和边edge组成。普通边是确定性的A 之后走 B条件边conditional edge是根据 state 决定下一步走哪个节点。这跟手写 Loop 里的if/else逻辑等价但区别在于跳转决策被显式建模成了图的一部分而不是散落在代码里的分支。这个显式化对中断恢复至关重要。假设你的 Agent 在调用工具节点执行到一半被中断恢复的时候框架需要知道下一步该去哪。如果跳转逻辑藏在函数里恢复时你得重新跑一遍判断逻辑还可能因为 state 不完整而判断错如果跳转是图上的边框架只要知道当前停在哪个节点就能顺着边找到后继节点。这就是为什么我强烈建议把所有影响流程走向的判断都放进条件边而不是塞在节点内部。还有一个容易忽略的点LangGraph 支持并行分支和 fan-in多个分支汇聚到一个节点。并行执行时多个节点的更新会合并到同一个 state这时候如果两个节点都写了同一个 key就会冲突。我的经验是给并行分支的产出用不同的 key或者用 reducer比如operator.add显式声明合并方式否则恢复时合并顺序不确定结果会飘。2.3 中断点是怎么被设计出来的中断不是随便在哪都能断的。LangGraph 的中断能力建立在图在节点边界上是可暂停的这个前提上。也就是说一个节点要么完整执行完要么没执行框架不会在节点函数执行到一半的时候把你冻住。这个约束很关键它意味着你的节点函数应该是幂等且相对原子的。那工具执行到一半要中断怎么办答案是把这个工具调用拆成一个独立的节点让中断发生在节点边界。比如发起工具调用和处理工具结果分成两个节点中间就可以插入中断。我见过有人把整个多步工具流程塞进一个节点然后抱怨没法中断这就是没理解节点边界的意义。LangGraph 提供了interrupt机制可以在节点执行前或执行后暂停把控制权交回给调用方等外部输入比如人工审批再继续。这个暂停-等待-继续的语义正是 human-in-the-loop 场景的基础。后面讲 checkpoint 的时候会看到暂停时框架会把当前 state 存下来恢复时从存下来的地方接着走。3. PostgreSQL Checkpoint把走到哪了变成可查询的数据3.1 Checkpoint 存的是什么Checkpoint 这个词在不同语境下含义不同在 LangGraph 里它指的是图执行状态在某个时间点的完整快照。每次图执行推进到一个新的超级步super-step可以粗略理解为一批节点执行完框架就会写一个 checkpoint。一个 checkpoint 至少包含当前 state 的值、当前停在哪个节点、下一步该走哪些节点、以及一些元数据时间戳、父 checkpoint 等。把这些存进 PostgreSQL好处是它变成了普通的关系型数据。你可以用 SQL 查这个会话最新的 checkpoint 是什么可以查这个会话一共经历了多少步可以做审计、可以做回放。相比把状态塞进内存或者 Redis 的简单 KVPostgreSQL 给你的是可查询、可事务、可备份的持久化。对于需要长期保存对话状态、需要排查历史问题的场景这个差别很大。LangGraph 官方提供了PostgresSaver这类 checkpointer 实现底层就是建几张表来存 checkpoint、写入记录和元数据。你不需要自己设计表结构但理解表里存了什么对排查问题非常有帮助。我遇到过一次恢复失败最后发现是 checkpoint 的父指针指向了一个已经被清理的记录导致恢复时找不到链路。3.2 thread_id恢复的钥匙Checkpoint 不是全局唯一的它按线程thread组织。每个 thread 用thread_id标识同一个 thread 下的 checkpoint 构成一条时间线。恢复的时候你提供thread_id框架就能找到这个线程最新的 checkpoint从那里接着跑。thread_id的设计是整个恢复机制里最容易被做错的地方。我的经验是一个逻辑会话对应一个 thread_id比如用会话 ID、用户 ID 加任务 ID 组合。千万不要用进程 ID 或者随机数否则重启之后你根本找不到原来的线程。也不要用一个全局固定的 thread_id那样所有会话的状态会互相覆盖。这里有个细节同一个 thread 下可以有多个 checkpoint恢复时默认取最新的。但有时候你想回到某个历史点重跑比如人工修正了某步输入就需要指定 checkpoint_id。LangGraph 支持这种时间旅行前提是你把 checkpoint 都留着。所以清理策略要谨慎别把还想回滚的历史删了。3.3 写入时机与事务边界Checkpoint 的写入时机直接决定了恢复的粒度。写得太频繁数据库压力大写得太稀疏恢复时丢的步骤多。LangGraph 默认在每个超级步之后写这个粒度对大多数场景够用。如果你的节点很重比如一次跑几十秒可以考虑在节点内部再手动触发写入但要注意别破坏原子性。事务边界是另一个坑。如果 checkpoint 写入和你的业务数据写入不在同一个事务里就可能出现业务数据提交了但 checkpoint 没写或者反过来导致恢复时状态不一致。我的做法是把 checkpoint 的写入和关键业务副作用尽量放在同一个事务语义下或者至少保证恢复逻辑能容忍这种不一致比如工具调用做成幂等的。还有一个实际经验PostgreSQL 的连接池要配好。Agent 执行过程中会频繁读写 checkpoint如果连接池太小高并发下会排队甚至超时。我一般会把 checkpointer 用的连接池和业务查询的连接池分开避免互相影响。4. AG-UI让前端实时看到 Runtime 在干什么4.1 为什么需要一套事件协议后端有了图执行和 checkpoint前端还是只能看到开始和结束两个状态中间过程是黑盒。用户会问它现在到底在干嘛是在调工具还是在思考要回答这个问题后端必须把执行过程中的事件实时推给前端。你可以自己定义一套 WebSocket 消息格式但很快就会遇到问题事件类型越来越多、顺序越来越乱、前后端对同一个事件的理解不一致。AG-UI 就是为解决这个问题而生的面向 Agent 前端的协议。它定义了一组标准事件类型比如文本增量、工具调用开始、工具调用结束、状态更新、执行完成等前端按这套协议消费事件就能统一渲染。它的价值不在于技术多高深而在于把Agent 执行过程这件事标准化了前后端不用再各写一套私有格式。我选它的理由是它和 LangGraph 的执行模型能对上。LangGraph 每个节点推进、每次工具调用都可以映射成 AG-UI 的事件。这样后端不用为了前端专门造一套中间层直接在图执行的过程中把事件发出去就行。4.2 事件流与 checkpoint 的对应关系这里有个很关键的认知AG-UI 的事件流是过程checkpoint 是状态。事件是瞬时的、可能丢的、不保证重放的checkpoint 是持久的、可查询的、可恢复的。两者不能互相替代。为什么强调这个因为我见过有人想用事件流来做恢复——把发出去的事件存下来恢复时重放。这条路很脆事件顺序、去重、幂等全靠自己保证而且事件里往往没有完整的 state。正确做法是恢复靠 checkpoint实时展示靠事件流。进程重启后从 checkpoint 加载 state然后继续产生新的事件推给前端前端如果断线重连可以先用一个当前状态快照事件把界面补齐再接着消费增量事件。这个分工想清楚之后架构就顺了checkpoint 是 source of truth事件流是它的实时投影。4.3 前端如何消费中断与恢复前端要处理的中断场景主要有两类一类是等待人工输入比如审批工具调用一类是连接断开后重连。第一类场景下后端发出一个需要输入的事件前端弹出交互界面用户操作后把结果回传后端从 checkpoint 恢复继续跑。第二类场景下前端重连后先拉一次当前状态把界面同步到最新再订阅后续事件。我踩过的一个坑是前端在等待人工输入时如果用户刷新了页面界面状态全丢了用户不知道刚才在等什么。解决办法是把待处理的中断也作为状态的一部分暴露给前端刷新后能重新拉取到。这又回到那个原则——凡是需要跨页面存活的信息都应该在 checkpoint 里而不是只活在事件流里。5. 中断恢复的完整链路从触发到续跑5.1 一次中断的完整生命周期把前面几块拼起来一次中断的完整流程是这样的图执行到某个节点触发中断条件比如工具需要人工确认框架在当前超级步之后写入一个 checkpoint标记状态为已中断然后把控制权交回调用方。调用方可能是 API 层把中断信息通过 AG-UI 事件推给前端。前端展示交互界面用户操作后把结果回传。后端拿到结果用同一个thread_id加载最新 checkpoint把用户输入合并进 state然后从断点继续执行。这条链路里每一步都有坑。比如把用户输入合并进 state这一步如果合并方式不对可能覆盖掉中断前的状态从断点继续这一步如果 checkpoint 的下一步节点记录错了可能重复执行已经跑过的节点。下面几节逐个拆。5.2 恢复时最容易出的三类问题第一类是消息重复。恢复时如果 state 里的 messages 被重新加载而你又把用户的新输入 append 进去很容易出现同一条消息出现两次。根因通常是恢复逻辑和正常执行逻辑走了两条不同的代码路径一条做了去重一条没做。我的做法是统一入口无论首次执行还是恢复都走同一个加载 state → 合并输入 → 继续执行的函数。第二类是副作用重复。如果中断发生在工具调用之后、结果写回之前恢复时可能重新调用一次工具。对于有副作用的工具发消息、写数据库这会造成重复。解决办法是给工具调用加幂等键或者把调用工具和记录结果拆成两个节点让中断只发生在两者之间。第三类是checkpoint 版本不匹配。如果你改了 state 的结构加了字段、改了类型旧的 checkpoint 加载进来可能反序列化失败。生产环境里 state 结构变更要有迁移方案或者至少做好版本标记加载旧 checkpoint 时能兼容处理。5.3 用 thread_id 和 checkpoint_id 精确控制恢复点默认恢复取最新 checkpoint但有些场景你需要精确控制。比如人工审批后想从审批前那个点重跑就需要指定 checkpoint_id。LangGraph 的 checkpointer 一般提供列出某个 thread 所有 checkpoint 的能力你可以按时间或按步骤选。我的经验是在 state 里显式记录一个业务层面的步骤标识比如step_name这样即使 checkpoint_id 是框架生成的、不好读你也能通过业务标识找到想恢复的点。排查问题时先按 thread_id 列出所有 checkpoint看每个 checkpoint 的 step_name 和 messages 长度基本就能定位到出问题的那一步。6. 我在实际搭建中踩过的坑6.1 checkpoint 表膨胀与清理策略跑了一段时间之后checkpoint 表会涨得很快尤其是长对话、多轮工具调用的场景。每个超级步一条记录一个复杂任务几十上百条很正常。如果不清理查询最新 checkpoint 的耗时会慢慢上去磁盘也会吃紧。我的清理策略是保留每个 thread 最近 N 条 checkpoint加上所有被标记为重要节点的 checkpoint。重要节点比如人工审批点、任务完成点这些可能需要回滚或审计不能删。清理任务用定时任务跑注意别删到正在执行的 thread 的 checkpoint否则恢复会失败。另外删除时要注意外键依赖父 checkpoint 被删了子 checkpoint 可能就悬空了。6.2 恢复后事件流错乱的处理前端重连后如果直接订阅新事件可能错过断线期间产生的事件导致界面状态和后端不一致。我的处理是重连时先发一个状态快照事件把当前 state 的关键信息当前步骤、待处理中断、最近几条消息推给前端前端用它重置界面然后再消费增量事件。这样即使中间丢了事件界面也能对齐。还有一个细节是事件的顺序。AG-UI 的事件如果经过消息队列可能乱序到达。前端要能容忍乱序或者后端在事件里带上序号前端按序号排序。我一般会在事件里加一个单调递增的 seq前端做一次缓冲排序再渲染。6.3 本地开发与生产环境的差异本地开发时我一般用内存 checkpointer 或者本地 PostgreSQL跑得快、好调试。但内存 checkpointer 重启就丢测不出恢复逻辑。所以测恢复一定要用持久化的 checkpointer而且要真的把进程杀掉再重启模拟真实故障。我见过有人只在代码里调了一下加载 checkpoint的函数就以为测过了结果真到线上进程崩溃发现恢复路径根本没走通。生产环境还要考虑 PostgreSQL 的高可用和备份。checkpoint 是恢复的唯一依据它丢了任务就真丢了。定期备份、主从复制这些该上的要上。另外生产环境的连接超时、重试策略要配好网络抖动导致 checkpoint 写入失败如果没有重试任务就卡住了。7. 几个能直接抄的配置与代码片段7.1 PostgresSaver 的初始化与连接池from langgraph.checkpoint.postgres import PostgresSaver from psycopg_pool import ConnectionPool # 单独给 checkpointer 一个连接池避免和业务查询抢连接 pool ConnectionPool( conninfopostgresql://user:passhost:5432/agent_db, min_size2, max_size10, timeout30, ) checkpointer PostgresSaver(pool) # 首次使用需要建表生产环境建议手动执行迁移而不是自动建 checkpointer.setup()这里min_size和max_size要根据并发量调。Agent 执行期间会频繁写 checkpoint池子太小会排队。timeout别设太短否则网络抖动时容易抛异常。7.2 带中断的图定义from langgraph.graph import StateGraph, END from langgraph.checkpoint.postgres import PostgresSaver def build_graph(checkpointer: PostgresSaver): graph StateGraph(AgentState) graph.add_node(plan, plan_node) graph.add_node(call_tool, call_tool_node) graph.add_node(handle_result, handle_result_node) graph.add_node(respond, respond_node) graph.set_entry_point(plan) graph.add_edge(plan, call_tool) graph.add_edge(call_tool, handle_result) graph.add_conditional_edges( handle_result, should_continue, {continue: call_tool, done: respond}, ) graph.add_edge(respond, END) # interrupt_before 让图在 call_tool 前暂停等待人工确认 return graph.compile( checkpointercheckpointer, interrupt_before[call_tool], )interrupt_before指定在哪些节点执行前暂停。这样每次要调工具前都会停下来等确认确认后再继续。注意暂停点选在节点边界别选在节点内部。7.3 恢复执行的调用方式config {configurable: {thread_id: session-123}} # 首次执行会在 call_tool 前中断 result app.invoke({messages: [user_msg]}, config) # 人工确认后用同一个 thread_id 恢复 # 传入 None 表示不修改 state直接继续 result app.invoke(None, config) # 如果要修改 state 再继续传入更新 result app.invoke( {messages: [approval_msg]}, config, )关键点恢复必须用同一个 thread_id否则框架会当成新线程从头开始。传None表示沿用已有 state 继续传字典表示合并更新。合并逻辑要小心别把不该覆盖的字段覆盖了。8. 关于这套组合的一些个人判断LangGraph 加 PostgreSQL Checkpoint 加 AG-UI 这套组合我的整体评价是它把 Agent 从一段代码变成了一个可运维的服务。手写 Loop 的时候Agent 是个函数用了这套之后Agent 是个有状态、可观测、可恢复的运行时对象。这个转变对 Demo 无所谓但对要上生产的系统是刚需。不过也别过度设计。如果你的 Agent 就是单轮问答、没有工具调用、不需要人工介入那手写 Loop 完全够用上这套反而是负担。这套东西的价值在长流程、多步骤、需要人工介入、需要故障恢复的场景。判断标准很简单如果你的用户会问它跑到哪了能不能暂停一下刚才崩了能不能接着来那你就需要它。最后分享一个我自己的习惯每次改 state 结构或者图结构都先跑一遍中断-杀进程-恢复的完整流程。这个流程能暴露绝大多数状态管理的问题比写单元测试还管用。跑通了这个流程再上生产心里才有底。