1. 为什么手写 Loop 终究会撞上恢复这堵墙做过对话式 AI 应用的人大概都经历过这样一个阶段一开始觉得 Agent 不就是个循环嘛把用户输入丢给模型模型返回工具调用就执行工具把结果再塞回去直到模型不再调用工具为止。于是花一个下午手写一个while循环几十行代码跑得挺欢Demo 演示效果也不错。但只要你把这个东西往真实业务里推问题就会一个接一个冒出来。用户点了发送之后网络断了页面刷新了服务端进程因为发版重启了或者用户自己中途点了取消又想继续——这时候你那个手写的while循环里所有的中间状态全在内存里进程一没上下文全丢。用户看到的就是刚才聊到一半的东西没了体验直接崩掉。这就是我这次要聊的核心问题从手写 Loop 到可恢复 Runtime 的演进。手写循环解决的是能不能跑通而可恢复 Runtime 解决的是跑挂了能不能接着跑。这两者之间的差距不是加几行 try-catch 能补上的它涉及到状态持久化、执行图建模、前后端事件协议三个层面的系统性设计。我这次用的技术组合是LangGraph PostgreSQL Checkpoint AG-UI。LangGraph 负责把 Agent 的执行逻辑从命令式循环变成声明式状态图PostgreSQL Checkpoint 负责把每一步的状态快照落库AG-UI 负责把后端的执行事件实时推给前端并支持前端发起恢复指令。三者串起来就能实现一个真正意义上的中断可恢复运行时。这篇文章适合谁看如果你已经用 LangChain 或 LangGraph 写过一些 Agent但一直被状态丢失无法断点续跑前后端事件对不齐这些问题困扰那这篇就是写给你的。如果你还没接触过 LangGraph也没关系我会把关键概念用生活化的方式讲清楚你至少能理解这套架构为什么这么设计。全文我会按照整体设计思路 → 核心机制拆解 → 实操落地 → 问题排查的顺序展开每一步都尽量给到可直接抄作业的配置和代码。先说一个我踩过的坑作为引子我最早的手写 Loop 版本为了图省事把对话历史存在一个全局字典里key 是 session_id。本地测试一切正常上线第二天就出事了——服务端做了多副本部署用户的两次请求被负载均衡打到了不同副本上第二个副本的字典里根本没有这个 session 的历史模型直接失忆。这个坑让我彻底明白状态不能放在进程内存里必须外置到共享存储。而 LangGraph 的 Checkpoint 机制本质上就是把这个外置做成了框架级能力。2. 整体架构设计与技术选型思路2.1 手写 Loop 的三个致命缺陷在讲新架构之前得先把旧方案的问题说透不然你不知道新方案到底在解决什么。第一个缺陷是状态与执行耦合。手写循环里当前执行到哪一步和已经产生了哪些中间结果是混在同一个函数栈里的。函数一旦返回或抛异常这些信息就没了。你想恢复就得自己设计一套序列化方案把循环里的局部变量一个个存下来恢复时再一个个还原。这个工作量随着 Agent 复杂度上升是指数级增长的。第二个缺陷是无法表达复杂控制流。真实业务里的 Agent 往往不是一条直线可能需要根据模型输出决定走哪条分支可能需要并行调用多个工具再汇总可能需要人工审核介入后再继续。手写循环里这些全靠 if-else 堆堆到后面自己都看不懂更别说维护和调试了。第三个缺陷是前后端事件割裂。手写方案通常是后端跑完一次性返回或者用 SSE 推一些零散的 token。前端根本不知道后端现在处于哪个阶段也就无法在中断后告诉后端从哪个阶段继续。恢复这件事必须是前后端协同的单靠后端自己做不到。2.2 为什么是 LangGraph 而不是继续用 LangChain这里要澄清一个高频疑问LangGraph 和 LangChain 到底什么区别。简单说LangChain 的核心抽象是链Chain它擅长表达线性的、单向的数据流输入经过 A 再到 B 再到 C。而 LangGraph 的核心抽象是图Graph节点是执行单元边是流转条件它天生支持循环、分支、并行和中断。Agent 的本质恰恰是一个带循环的图模型思考 → 调用工具 → 观察结果 → 再思考这个环要能转起来还要能在任意节点停下来、存下来、再恢复。用 Chain 去硬凑循环就像用直尺画圆能画但很别扭。LangGraph 把状态提升为一等公民每个节点读写的是同一个 State 对象框架负责在节点切换时做 Checkpoint这才是可恢复 Runtime 的地基。所以选型逻辑很清晰只要你的 Agent 需要循环、需要分支、需要中断恢复就应该上 LangGraph。如果只是简单的输入→模型→输出LangChain 的 Chain 或者直接调 API 就够了没必要上重型框架。2.3 PostgreSQL Checkpoint 的角色定位LangGraph 本身提供了 Checkpoint 的抽象接口官方也给了内存版和 SQLite 版。但生产环境我强烈建议用PostgreSQL原因有三。一是并发与事务。PostgreSQL 的行级锁和 MVCC 能保证多个请求同时读写 Checkpoint 时不出乱子SQLite 在并发写场景下容易锁库。二是持久化可靠性。PostgreSQL 有 WAL 日志进程崩了、机器重启了数据不丢这是可恢复三个字的物理基础。三是可查询性。Checkpoint 落在 PG 里你可以直接用 SQL 查某个 thread 的历史快照做审计、做回放、做数据分析都很方便这在排查线上问题时价值巨大。需要说明的是LangGraph 的 Checkpoint 不是每一步都存全量状态而是存增量 引用配合它的序列化机制存储开销是可控的。这一点我在实操部分会详细讲。2.4 AG-UI 解决的是最后一公里后端有了可恢复能力还得让前端能感知、能触发。AG-UI是一套面向 Agent 前端的交互协议它定义了后端如何把 Agent 的执行事件节点开始、节点结束、工具调用、状态更新、中断信号标准化地推给前端以及前端如何把用户的操作发送、取消、恢复标准化地传回后端。没有这层协议你就得自己定义一套事件格式前端自己解析恢复指令自己约定做起来又累又容易出错。AG-UI 把这些约定固化了前后端各司其职恢复流程变成前端发一个 resume 事件后端从 Checkpoint 加载状态继续跑干净利落。技术组件核心职责替代方案选它的理由LangGraph执行图建模与状态管理手写 Loop、LangChain Chain原生支持循环、分支、中断PostgreSQL Checkpoint状态快照持久化内存、SQLite、Redis事务可靠、可查询、并发稳AG-UI前后端事件协议自定义 SSE 格式标准化、恢复语义清晰3. 核心机制拆解状态、Checkpoint 与恢复语义3.1 State 到底存了什么理解可恢复 Runtime第一步是搞清楚 LangGraph 里的 State 是什么。你可以把它想象成一张共享工作台所有节点都在这张台子上干活谁需要什么就从台子上拿干完把结果放回台子上。台子上的东西就是 State。一个典型的 Agent State 大概长这样消息列表对话历史、当前任务计划、工具调用结果缓存、一些业务字段比如用户 ID、订单号、以及控制字段比如是否需要人工介入。关键在于State 必须是可序列化的因为 Checkpoint 要把它写进数据库。这就意味着你不能往 State 里塞数据库连接、文件句柄这类东西只能塞纯数据。我见过有人把 ORM 对象直接塞进 State本地跑没事一开 Checkpoint 就报序列化错误。正确做法是只存 ID 和必要字段需要完整对象时在节点里现查。这个原则叫State 存引用不存实体跟数据库设计里的外键思路是一样的。3.2 Checkpoint 的触发时机与粒度LangGraph 的 Checkpoint 默认在每个节点执行完成后触发一次。也就是说如果你的图有 5 个节点一次完整执行会产生 5 个 Checkpoint。每个 Checkpoint 记录了执行完这个节点后的完整 State以及下一步该去哪个节点。这个粒度设计很讲究。太粗比如整条图跑完才存一次中断后就得从头再来太细每个 token 都存存储和性能都扛不住。按节点存是个平衡点节点通常对应一个有意义的业务步骤恢复时从节点边界继续语义清晰开销也可接受。需要提醒的是Checkpoint 的写入是同步阻塞的也就是说节点执行完必须等 Checkpoint 落库成功才会流转到下一个节点。这是为了保证存了才继续避免出现状态没存住但流程已经往下走了的不一致。如果你的节点执行很快、Checkpoint 写入成了瓶颈可以考虑用异步写入但要接受极端情况下丢最后一个 Checkpoint 的风险。生产环境我建议保持同步可靠性优先。3.3 恢复是怎么发生的恢复的本质是给定一个 thread_id从数据库里找到这个 thread 最新的 Checkpoint把 State 加载回内存然后从 Checkpoint 记录的下一步节点继续执行。这里有个关键概念叫thread_id它是会话的唯一标识。同一个 thread_id 的所有 Checkpoint 构成一条时间线恢复时默认取最新的那个。你也可以指定取某个历史 Checkpoint实现回滚到某个时间点重跑这在调试和 A/B 测试时非常有用。恢复的触发方式有两种。一种是自动恢复进程重启后扫描数据库里所有未完成的 thread主动把它们捡起来继续跑。另一种是手动恢复前端用户点击继续发一个 resume 事件后端按 thread_id 加载并继续。AG-UI 主要服务于后者因为用户主动恢复的场景更常见也更需要前端参与。注意恢复不是重放。重放是把所有节点重新执行一遍恢复是从断点继续已经执行过的节点不会重复执行。这个区别决定了恢复的性能远优于重放但也要求 Checkpoint 必须准确记录执行到哪了。3.4 中断的两种类型中断分两种处理方式完全不同。被动中断是意外进程崩溃、网络断开、超时。这种中断发生时最后一个 Checkpoint 之后的工作丢了恢复时从最后一个 Checkpoint 继续用户可能会感知到少了一小段。主动中断是设计比如 Agent 执行到需要人工审核的节点主动调用中断把控制权交还给用户等用户确认后再恢复。LangGraph 提供了interrupt机制专门处理这种场景它会在 Checkpoint 里标记这里中断了等待外部输入恢复时把用户的输入注入 State 再继续。主动中断是 Human-in-the-loop 的基础。比如一个自动下单的 Agent在真正提交订单前中断让用户确认金额和收货地址确认后再恢复执行提交。这种设计既保证了自动化效率又保留了人工兜底是生产级 Agent 的标配。4. 实操落地从零搭一个可恢复 Runtime4.1 环境准备与依赖安装先把地基打好。Python 环境建议 3.10 以上LangGraph 对类型注解用得比较多低版本容易出幺蛾子。pip install langgraph langgraph-checkpoint-postgres psycopg[binary] psycopg-pool pip install ag-ui-protocolPostgreSQL 建议 14 以上低版本在 JSONB 和并发处理上有些限制。本地开发可以用 Docker 起一个省得污染本机环境。docker run -d --name lg-pg \ -e POSTGRES_PASSWORDyourpass \ -e POSTGRES_DBlanggraph \ -p 5432:5432 \ postgres:16起来之后连上去建个库LangGraph 的 PostgresSaver 会自动建表你不用手动写 DDL但库得先存在。4.2 定义 State 与构建执行图先定义 State。我用 TypedDict 来声明字段尽量精简只放真正需要在节点间传递的数据。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] task_plan: str tool_results: list need_human_review: bool review_decision: str这里add_messages是个 reducer它告诉 LangGraph当多个节点都往 messages 里写时用追加而不是覆盖的方式合并。这个细节很重要不写 reducer 的话后一个节点会把前一个节点的消息覆盖掉对话历史就断了。接着建图。我以一个研究助手为例先规划任务再执行搜索然后判断是否需要人工审核最后生成报告。from langgraph.graph import StateGraph, START, END def plan_node(state: AgentState): # 调用模型生成任务计划 plan llm.invoke(f为以下问题制定研究计划{state[messages][-1].content}) return {task_plan: plan.content} def search_node(state: AgentState): # 根据计划执行搜索 results search_tool.run(state[task_plan]) return {tool_results: results} def review_node(state: AgentState): # 主动中断等待人工审核 decision interrupt({question: 是否批准继续生成报告, plan: state[task_plan]}) return {review_decision: decision, need_human_review: False} def report_node(state: AgentState): report llm.invoke(f基于{state[tool_results]}生成报告) return {messages: [report]} builder StateGraph(AgentState) builder.add_node(plan, plan_node) builder.add_node(search, search_node) builder.add_node(review, review_node) builder.add_node(report, report_node) builder.add_edge(START, plan) builder.add_edge(plan, search) builder.add_edge(search, review) builder.add_edge(review, report) builder.add_edge(report, END) graph builder.compile()注意review_node里的interrupt这就是主动中断的入口。执行到这里图会停下来Checkpoint 里记录停在 review 节点等待输入。4.3 接入 PostgreSQL Checkpoint编译图的时候把 checkpointer 传进去这是让图具备持久化能力的关键一步。from langgraph.checkpoint.postgres import PostgresSaver from psycopg_pool import ConnectionPool DB_URI postgresql://postgres:yourpasslocalhost:5432/langgraph pool ConnectionPool(conninfoDB_URI, max_size10) checkpointer PostgresSaver(pool) checkpointer.setup() # 首次运行建表 graph builder.compile(checkpointercheckpointer)setup()只需要跑一次它会创建checkpoints、checkpoint_writes等表。生产环境建议把建表脚本单独抽出来用迁移工具管理别每次启动都跑 setup。调用的时候必须传thread_id否则 Checkpoint 不知道往哪个会话存。config {configurable: {thread_id: user-123-session-1}} result graph.invoke( {messages: [HumanMessage(content帮我研究一下向量数据库的选型)]}, configconfig )执行到 review 节点时invoke会返回一个包含中断信息的对象而不是跑到底。这时候 Checkpoint 已经落库了。4.4 恢复执行与注入人工决策恢复的代码非常简洁核心就是Command(resume...)。from langgraph.types import Command resumed graph.invoke( Command(resume批准), configconfig )LangGraph 会根据 thread_id 找到最新的 Checkpoint发现停在 review 节点等待输入于是把 批准 注入从 review 节点继续往下跑 report 节点。整个过程 plan 和 search 节点不会重复执行因为它们的结果已经在 Checkpoint 里了。这里有个实测心得resume 的值类型要和 interrupt 时约定的类型一致。我在 interrupt 里传的是 dictresume 时如果传字符串虽然不报错但节点里取值会拿到意料之外的结构。建议在 interrupt 里明确约定好输入格式前端也按这个格式传。4.5 用 AG-UI 打通前后端事件流后端能力有了现在让前端能感知和触发。AG-UI 的核心是把 Agent 的执行过程抽象成一系列标准事件通过 SSE 推给前端。from ag_ui.core import EventType, RunAgentInput from ag_ui.encoder import EventEncoder async def run_agent(input_data: RunAgentInput): encoder EventEncoder() thread_id input_data.thread_id config {configurable: {thread_id: thread_id}} async for event in graph.astream_events(input_data.messages, configconfig): # 把 LangGraph 事件映射成 AG-UI 事件 ag_ui_event map_to_ag_ui(event) yield encoder.encode(ag_ui_event)前端收到RUN_STARTED、NODE_STARTED、NODE_FINISHED、INTERRUPT这些事件后就能准确知道后端在干什么。当收到INTERRUPT事件时前端弹出审核界面用户点批准后前端发一个带resume字段的请求后端走上面那段恢复逻辑。这套协议的价值在于语义清晰。前端不需要理解 LangGraph 的内部结构只需要按 AG-UI 的事件类型做响应后端也不需要关心前端怎么渲染只管把标准事件推出去。两边解耦各自演进。AG-UI 事件触发时机前端典型处理RUN_STARTED一次执行开始显示加载态NODE_STARTED某节点开始更新进度提示NODE_FINISHED某节点结束渲染中间结果INTERRUPT主动中断弹出人工审核界面RUN_FINISHED执行结束收起加载态展示终态5. 常见问题与排查技巧实录5.1 Checkpoint 相关的高频坑问题一恢复后状态不对像是从更早的地方开始的。这通常是因为 Checkpoint 写入失败但流程继续了。排查方法是直接查数据库SELECT * FROM checkpoints WHERE thread_id xxx ORDER BY created_at DESC LIMIT 5;看最新的 Checkpoint 对应哪个节点。如果发现节点和预期不符检查 checkpointer 是否真的传进了 compile以及数据库连接是否正常。问题二并发请求同一个 thread_id 导致状态错乱。同一个会话不应该被并发执行。解决办法是在业务层加锁或者用 LangGraph 的checkpoint_id做乐观锁。我一般会在 Redis 里对 thread_id 加一个短租约锁拿到锁才能执行执行完释放。问题三State 太大导致 Checkpoint 写入慢。如果 State 里塞了几十 MB 的搜索结果每次 Checkpoint 都要序列化写库性能会很差。解决办法是把大对象存到对象存储State 里只留 URL 或 ID。记住那条原则State 存引用不存实体。5.2 中断恢复的典型故障故障一resume 之后图没有继续直接结束了。大概率是 thread_id 对不上。恢复时必须用和中断时完全相同的 thread_id否则 LangGraph 找不到对应的 Checkpoint会当成一次全新的执行。前端传 thread_id 时要确保一致性别在刷新页面后重新生成了。故障二interrupt 的值在恢复后拿不到。检查 interrupt 和 resume 的类型是否匹配。另外注意interrupt 的返回值是在节点内部通过interrupt()函数的返回值拿到的不是从 State 里读的。这个设计容易搞混我第一次用的时候也绕了半天。故障三恢复后重复执行了已经完成的节点。这通常是因为图的边定义有问题或者 Checkpoint 记录的下一步被覆盖了。检查你的边是不是有环以及节点是否有副作用比如发邮件、扣款。有副作用的节点一定要做幂等因为恢复机制不能百分百保证节点只执行一次。提示所有有副作用的操作支付、发消息、写外部系统都要做幂等设计用业务唯一键去重。这是可恢复 Runtime 的必备配套不是可选项。5.3 性能与运维注意事项Checkpoint 表会随着使用不断增长一个活跃会话可能产生成百上千条记录。建议定期归档把超过一定时间的 Checkpoint 导出到冷存储主表只保留近期数据。归档前确认这些会话确实不会再恢复了。数据库连接池大小要合理设置。PostgresSaver 每个操作都要拿连接池子太小会排队太大又浪费资源。我一般按峰值 QPS × 平均操作耗时来估算再留 30% 余量。10 到 20 的池子对中小规模应用足够了。监控方面重点盯三个指标Checkpoint 写入延迟、恢复成功率、中断到恢复的平均时长。前两个反映系统健康度第三个直接关系用户体验。如果恢复时长超过几秒用户就会觉得卡需要考虑优化 State 大小或数据库索引。5.4 常见问题速查表现象可能原因排查方向解决手段恢复后从头开始thread_id 不一致对比前后端 thread_id统一会话标识生成逻辑状态字段丢失未定义 reducer检查 State 注解给列表字段加 add_messages写入超时State 过大查 Checkpoint 记录大小大对象外置State 存引用并发状态错乱同 thread 并发执行查执行日志时间线加会话级分布式锁恢复后重复副作用节点非幂等查外部系统调用记录业务唯一键去重6. 几个我踩过之后才明白的经验先说一个关于 State 设计的体会。我一开始图省事把整个对话历史、所有工具返回的原始数据全塞进 State觉得这样恢复时信息最全。结果跑了几天发现 Checkpoint 表膨胀得飞快单条记录动辄几百 KB恢复时加载慢得让人抓狂。后来改成State 只存必要字段 大对象存外部单条 Checkpoint 降到几 KB恢复速度肉眼可见地变快。这个教训是State 是工作台不是仓库台面上只放当前要用的东西。再说一个关于中断粒度的经验。不是所有节点都值得中断。我早期设计时恨不得每个节点都加人工确认结果用户被弹窗烦得不行直接弃用。后来我把中断收敛到真正需要人决策的关键节点比如提交订单前发送对外邮件前其他环节全自动。中断是给用户掌控感的不是给用户添堵的这个度要把握好。最后分享一个调试技巧。排查恢复问题时别光看日志直接去数据库里把某个 thread 的所有 Checkpoint 按时间顺序拉出来一条条看 State 的变化和节点的流转。你会非常直观地看到在哪一步状态变了在哪一步中断了恢复时从哪条记录开始的。这个方法帮我定位过好几个靠日志根本看不出来的问题比如某个 reducer 悄悄把字段覆盖了。这套架构后续还能往几个方向扩展。一是加多级 Checkpoint把高频小状态和低频大状态分开存进一步优化性能。二是把恢复逻辑做成独立的服务专门负责扫描和重启未完成的会话实现真正的无人值守恢复。三是结合 AG-UI 的事件流做执行回放把一次完整的 Agent 执行录下来用于调试和演示。这些我都还在摸索有新的心得再单独写一篇。
从手写Loop到可恢复Runtime：LangGraph+PostgreSQL+AG-UI实战
1. 为什么手写 Loop 终究会撞上恢复这堵墙做过对话式 AI 应用的人大概都经历过这样一个阶段一开始觉得 Agent 不就是个循环嘛把用户输入丢给模型模型返回工具调用就执行工具把结果再塞回去直到模型不再调用工具为止。于是花一个下午手写一个while循环几十行代码跑得挺欢Demo 演示效果也不错。但只要你把这个东西往真实业务里推问题就会一个接一个冒出来。用户点了发送之后网络断了页面刷新了服务端进程因为发版重启了或者用户自己中途点了取消又想继续——这时候你那个手写的while循环里所有的中间状态全在内存里进程一没上下文全丢。用户看到的就是刚才聊到一半的东西没了体验直接崩掉。这就是我这次要聊的核心问题从手写 Loop 到可恢复 Runtime 的演进。手写循环解决的是能不能跑通而可恢复 Runtime 解决的是跑挂了能不能接着跑。这两者之间的差距不是加几行 try-catch 能补上的它涉及到状态持久化、执行图建模、前后端事件协议三个层面的系统性设计。我这次用的技术组合是LangGraph PostgreSQL Checkpoint AG-UI。LangGraph 负责把 Agent 的执行逻辑从命令式循环变成声明式状态图PostgreSQL Checkpoint 负责把每一步的状态快照落库AG-UI 负责把后端的执行事件实时推给前端并支持前端发起恢复指令。三者串起来就能实现一个真正意义上的中断可恢复运行时。这篇文章适合谁看如果你已经用 LangChain 或 LangGraph 写过一些 Agent但一直被状态丢失无法断点续跑前后端事件对不齐这些问题困扰那这篇就是写给你的。如果你还没接触过 LangGraph也没关系我会把关键概念用生活化的方式讲清楚你至少能理解这套架构为什么这么设计。全文我会按照整体设计思路 → 核心机制拆解 → 实操落地 → 问题排查的顺序展开每一步都尽量给到可直接抄作业的配置和代码。先说一个我踩过的坑作为引子我最早的手写 Loop 版本为了图省事把对话历史存在一个全局字典里key 是 session_id。本地测试一切正常上线第二天就出事了——服务端做了多副本部署用户的两次请求被负载均衡打到了不同副本上第二个副本的字典里根本没有这个 session 的历史模型直接失忆。这个坑让我彻底明白状态不能放在进程内存里必须外置到共享存储。而 LangGraph 的 Checkpoint 机制本质上就是把这个外置做成了框架级能力。2. 整体架构设计与技术选型思路2.1 手写 Loop 的三个致命缺陷在讲新架构之前得先把旧方案的问题说透不然你不知道新方案到底在解决什么。第一个缺陷是状态与执行耦合。手写循环里当前执行到哪一步和已经产生了哪些中间结果是混在同一个函数栈里的。函数一旦返回或抛异常这些信息就没了。你想恢复就得自己设计一套序列化方案把循环里的局部变量一个个存下来恢复时再一个个还原。这个工作量随着 Agent 复杂度上升是指数级增长的。第二个缺陷是无法表达复杂控制流。真实业务里的 Agent 往往不是一条直线可能需要根据模型输出决定走哪条分支可能需要并行调用多个工具再汇总可能需要人工审核介入后再继续。手写循环里这些全靠 if-else 堆堆到后面自己都看不懂更别说维护和调试了。第三个缺陷是前后端事件割裂。手写方案通常是后端跑完一次性返回或者用 SSE 推一些零散的 token。前端根本不知道后端现在处于哪个阶段也就无法在中断后告诉后端从哪个阶段继续。恢复这件事必须是前后端协同的单靠后端自己做不到。2.2 为什么是 LangGraph 而不是继续用 LangChain这里要澄清一个高频疑问LangGraph 和 LangChain 到底什么区别。简单说LangChain 的核心抽象是链Chain它擅长表达线性的、单向的数据流输入经过 A 再到 B 再到 C。而 LangGraph 的核心抽象是图Graph节点是执行单元边是流转条件它天生支持循环、分支、并行和中断。Agent 的本质恰恰是一个带循环的图模型思考 → 调用工具 → 观察结果 → 再思考这个环要能转起来还要能在任意节点停下来、存下来、再恢复。用 Chain 去硬凑循环就像用直尺画圆能画但很别扭。LangGraph 把状态提升为一等公民每个节点读写的是同一个 State 对象框架负责在节点切换时做 Checkpoint这才是可恢复 Runtime 的地基。所以选型逻辑很清晰只要你的 Agent 需要循环、需要分支、需要中断恢复就应该上 LangGraph。如果只是简单的输入→模型→输出LangChain 的 Chain 或者直接调 API 就够了没必要上重型框架。2.3 PostgreSQL Checkpoint 的角色定位LangGraph 本身提供了 Checkpoint 的抽象接口官方也给了内存版和 SQLite 版。但生产环境我强烈建议用PostgreSQL原因有三。一是并发与事务。PostgreSQL 的行级锁和 MVCC 能保证多个请求同时读写 Checkpoint 时不出乱子SQLite 在并发写场景下容易锁库。二是持久化可靠性。PostgreSQL 有 WAL 日志进程崩了、机器重启了数据不丢这是可恢复三个字的物理基础。三是可查询性。Checkpoint 落在 PG 里你可以直接用 SQL 查某个 thread 的历史快照做审计、做回放、做数据分析都很方便这在排查线上问题时价值巨大。需要说明的是LangGraph 的 Checkpoint 不是每一步都存全量状态而是存增量 引用配合它的序列化机制存储开销是可控的。这一点我在实操部分会详细讲。2.4 AG-UI 解决的是最后一公里后端有了可恢复能力还得让前端能感知、能触发。AG-UI是一套面向 Agent 前端的交互协议它定义了后端如何把 Agent 的执行事件节点开始、节点结束、工具调用、状态更新、中断信号标准化地推给前端以及前端如何把用户的操作发送、取消、恢复标准化地传回后端。没有这层协议你就得自己定义一套事件格式前端自己解析恢复指令自己约定做起来又累又容易出错。AG-UI 把这些约定固化了前后端各司其职恢复流程变成前端发一个 resume 事件后端从 Checkpoint 加载状态继续跑干净利落。技术组件核心职责替代方案选它的理由LangGraph执行图建模与状态管理手写 Loop、LangChain Chain原生支持循环、分支、中断PostgreSQL Checkpoint状态快照持久化内存、SQLite、Redis事务可靠、可查询、并发稳AG-UI前后端事件协议自定义 SSE 格式标准化、恢复语义清晰3. 核心机制拆解状态、Checkpoint 与恢复语义3.1 State 到底存了什么理解可恢复 Runtime第一步是搞清楚 LangGraph 里的 State 是什么。你可以把它想象成一张共享工作台所有节点都在这张台子上干活谁需要什么就从台子上拿干完把结果放回台子上。台子上的东西就是 State。一个典型的 Agent State 大概长这样消息列表对话历史、当前任务计划、工具调用结果缓存、一些业务字段比如用户 ID、订单号、以及控制字段比如是否需要人工介入。关键在于State 必须是可序列化的因为 Checkpoint 要把它写进数据库。这就意味着你不能往 State 里塞数据库连接、文件句柄这类东西只能塞纯数据。我见过有人把 ORM 对象直接塞进 State本地跑没事一开 Checkpoint 就报序列化错误。正确做法是只存 ID 和必要字段需要完整对象时在节点里现查。这个原则叫State 存引用不存实体跟数据库设计里的外键思路是一样的。3.2 Checkpoint 的触发时机与粒度LangGraph 的 Checkpoint 默认在每个节点执行完成后触发一次。也就是说如果你的图有 5 个节点一次完整执行会产生 5 个 Checkpoint。每个 Checkpoint 记录了执行完这个节点后的完整 State以及下一步该去哪个节点。这个粒度设计很讲究。太粗比如整条图跑完才存一次中断后就得从头再来太细每个 token 都存存储和性能都扛不住。按节点存是个平衡点节点通常对应一个有意义的业务步骤恢复时从节点边界继续语义清晰开销也可接受。需要提醒的是Checkpoint 的写入是同步阻塞的也就是说节点执行完必须等 Checkpoint 落库成功才会流转到下一个节点。这是为了保证存了才继续避免出现状态没存住但流程已经往下走了的不一致。如果你的节点执行很快、Checkpoint 写入成了瓶颈可以考虑用异步写入但要接受极端情况下丢最后一个 Checkpoint 的风险。生产环境我建议保持同步可靠性优先。3.3 恢复是怎么发生的恢复的本质是给定一个 thread_id从数据库里找到这个 thread 最新的 Checkpoint把 State 加载回内存然后从 Checkpoint 记录的下一步节点继续执行。这里有个关键概念叫thread_id它是会话的唯一标识。同一个 thread_id 的所有 Checkpoint 构成一条时间线恢复时默认取最新的那个。你也可以指定取某个历史 Checkpoint实现回滚到某个时间点重跑这在调试和 A/B 测试时非常有用。恢复的触发方式有两种。一种是自动恢复进程重启后扫描数据库里所有未完成的 thread主动把它们捡起来继续跑。另一种是手动恢复前端用户点击继续发一个 resume 事件后端按 thread_id 加载并继续。AG-UI 主要服务于后者因为用户主动恢复的场景更常见也更需要前端参与。注意恢复不是重放。重放是把所有节点重新执行一遍恢复是从断点继续已经执行过的节点不会重复执行。这个区别决定了恢复的性能远优于重放但也要求 Checkpoint 必须准确记录执行到哪了。3.4 中断的两种类型中断分两种处理方式完全不同。被动中断是意外进程崩溃、网络断开、超时。这种中断发生时最后一个 Checkpoint 之后的工作丢了恢复时从最后一个 Checkpoint 继续用户可能会感知到少了一小段。主动中断是设计比如 Agent 执行到需要人工审核的节点主动调用中断把控制权交还给用户等用户确认后再恢复。LangGraph 提供了interrupt机制专门处理这种场景它会在 Checkpoint 里标记这里中断了等待外部输入恢复时把用户的输入注入 State 再继续。主动中断是 Human-in-the-loop 的基础。比如一个自动下单的 Agent在真正提交订单前中断让用户确认金额和收货地址确认后再恢复执行提交。这种设计既保证了自动化效率又保留了人工兜底是生产级 Agent 的标配。4. 实操落地从零搭一个可恢复 Runtime4.1 环境准备与依赖安装先把地基打好。Python 环境建议 3.10 以上LangGraph 对类型注解用得比较多低版本容易出幺蛾子。pip install langgraph langgraph-checkpoint-postgres psycopg[binary] psycopg-pool pip install ag-ui-protocolPostgreSQL 建议 14 以上低版本在 JSONB 和并发处理上有些限制。本地开发可以用 Docker 起一个省得污染本机环境。docker run -d --name lg-pg \ -e POSTGRES_PASSWORDyourpass \ -e POSTGRES_DBlanggraph \ -p 5432:5432 \ postgres:16起来之后连上去建个库LangGraph 的 PostgresSaver 会自动建表你不用手动写 DDL但库得先存在。4.2 定义 State 与构建执行图先定义 State。我用 TypedDict 来声明字段尽量精简只放真正需要在节点间传递的数据。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] task_plan: str tool_results: list need_human_review: bool review_decision: str这里add_messages是个 reducer它告诉 LangGraph当多个节点都往 messages 里写时用追加而不是覆盖的方式合并。这个细节很重要不写 reducer 的话后一个节点会把前一个节点的消息覆盖掉对话历史就断了。接着建图。我以一个研究助手为例先规划任务再执行搜索然后判断是否需要人工审核最后生成报告。from langgraph.graph import StateGraph, START, END def plan_node(state: AgentState): # 调用模型生成任务计划 plan llm.invoke(f为以下问题制定研究计划{state[messages][-1].content}) return {task_plan: plan.content} def search_node(state: AgentState): # 根据计划执行搜索 results search_tool.run(state[task_plan]) return {tool_results: results} def review_node(state: AgentState): # 主动中断等待人工审核 decision interrupt({question: 是否批准继续生成报告, plan: state[task_plan]}) return {review_decision: decision, need_human_review: False} def report_node(state: AgentState): report llm.invoke(f基于{state[tool_results]}生成报告) return {messages: [report]} builder StateGraph(AgentState) builder.add_node(plan, plan_node) builder.add_node(search, search_node) builder.add_node(review, review_node) builder.add_node(report, report_node) builder.add_edge(START, plan) builder.add_edge(plan, search) builder.add_edge(search, review) builder.add_edge(review, report) builder.add_edge(report, END) graph builder.compile()注意review_node里的interrupt这就是主动中断的入口。执行到这里图会停下来Checkpoint 里记录停在 review 节点等待输入。4.3 接入 PostgreSQL Checkpoint编译图的时候把 checkpointer 传进去这是让图具备持久化能力的关键一步。from langgraph.checkpoint.postgres import PostgresSaver from psycopg_pool import ConnectionPool DB_URI postgresql://postgres:yourpasslocalhost:5432/langgraph pool ConnectionPool(conninfoDB_URI, max_size10) checkpointer PostgresSaver(pool) checkpointer.setup() # 首次运行建表 graph builder.compile(checkpointercheckpointer)setup()只需要跑一次它会创建checkpoints、checkpoint_writes等表。生产环境建议把建表脚本单独抽出来用迁移工具管理别每次启动都跑 setup。调用的时候必须传thread_id否则 Checkpoint 不知道往哪个会话存。config {configurable: {thread_id: user-123-session-1}} result graph.invoke( {messages: [HumanMessage(content帮我研究一下向量数据库的选型)]}, configconfig )执行到 review 节点时invoke会返回一个包含中断信息的对象而不是跑到底。这时候 Checkpoint 已经落库了。4.4 恢复执行与注入人工决策恢复的代码非常简洁核心就是Command(resume...)。from langgraph.types import Command resumed graph.invoke( Command(resume批准), configconfig )LangGraph 会根据 thread_id 找到最新的 Checkpoint发现停在 review 节点等待输入于是把 批准 注入从 review 节点继续往下跑 report 节点。整个过程 plan 和 search 节点不会重复执行因为它们的结果已经在 Checkpoint 里了。这里有个实测心得resume 的值类型要和 interrupt 时约定的类型一致。我在 interrupt 里传的是 dictresume 时如果传字符串虽然不报错但节点里取值会拿到意料之外的结构。建议在 interrupt 里明确约定好输入格式前端也按这个格式传。4.5 用 AG-UI 打通前后端事件流后端能力有了现在让前端能感知和触发。AG-UI 的核心是把 Agent 的执行过程抽象成一系列标准事件通过 SSE 推给前端。from ag_ui.core import EventType, RunAgentInput from ag_ui.encoder import EventEncoder async def run_agent(input_data: RunAgentInput): encoder EventEncoder() thread_id input_data.thread_id config {configurable: {thread_id: thread_id}} async for event in graph.astream_events(input_data.messages, configconfig): # 把 LangGraph 事件映射成 AG-UI 事件 ag_ui_event map_to_ag_ui(event) yield encoder.encode(ag_ui_event)前端收到RUN_STARTED、NODE_STARTED、NODE_FINISHED、INTERRUPT这些事件后就能准确知道后端在干什么。当收到INTERRUPT事件时前端弹出审核界面用户点批准后前端发一个带resume字段的请求后端走上面那段恢复逻辑。这套协议的价值在于语义清晰。前端不需要理解 LangGraph 的内部结构只需要按 AG-UI 的事件类型做响应后端也不需要关心前端怎么渲染只管把标准事件推出去。两边解耦各自演进。AG-UI 事件触发时机前端典型处理RUN_STARTED一次执行开始显示加载态NODE_STARTED某节点开始更新进度提示NODE_FINISHED某节点结束渲染中间结果INTERRUPT主动中断弹出人工审核界面RUN_FINISHED执行结束收起加载态展示终态5. 常见问题与排查技巧实录5.1 Checkpoint 相关的高频坑问题一恢复后状态不对像是从更早的地方开始的。这通常是因为 Checkpoint 写入失败但流程继续了。排查方法是直接查数据库SELECT * FROM checkpoints WHERE thread_id xxx ORDER BY created_at DESC LIMIT 5;看最新的 Checkpoint 对应哪个节点。如果发现节点和预期不符检查 checkpointer 是否真的传进了 compile以及数据库连接是否正常。问题二并发请求同一个 thread_id 导致状态错乱。同一个会话不应该被并发执行。解决办法是在业务层加锁或者用 LangGraph 的checkpoint_id做乐观锁。我一般会在 Redis 里对 thread_id 加一个短租约锁拿到锁才能执行执行完释放。问题三State 太大导致 Checkpoint 写入慢。如果 State 里塞了几十 MB 的搜索结果每次 Checkpoint 都要序列化写库性能会很差。解决办法是把大对象存到对象存储State 里只留 URL 或 ID。记住那条原则State 存引用不存实体。5.2 中断恢复的典型故障故障一resume 之后图没有继续直接结束了。大概率是 thread_id 对不上。恢复时必须用和中断时完全相同的 thread_id否则 LangGraph 找不到对应的 Checkpoint会当成一次全新的执行。前端传 thread_id 时要确保一致性别在刷新页面后重新生成了。故障二interrupt 的值在恢复后拿不到。检查 interrupt 和 resume 的类型是否匹配。另外注意interrupt 的返回值是在节点内部通过interrupt()函数的返回值拿到的不是从 State 里读的。这个设计容易搞混我第一次用的时候也绕了半天。故障三恢复后重复执行了已经完成的节点。这通常是因为图的边定义有问题或者 Checkpoint 记录的下一步被覆盖了。检查你的边是不是有环以及节点是否有副作用比如发邮件、扣款。有副作用的节点一定要做幂等因为恢复机制不能百分百保证节点只执行一次。提示所有有副作用的操作支付、发消息、写外部系统都要做幂等设计用业务唯一键去重。这是可恢复 Runtime 的必备配套不是可选项。5.3 性能与运维注意事项Checkpoint 表会随着使用不断增长一个活跃会话可能产生成百上千条记录。建议定期归档把超过一定时间的 Checkpoint 导出到冷存储主表只保留近期数据。归档前确认这些会话确实不会再恢复了。数据库连接池大小要合理设置。PostgresSaver 每个操作都要拿连接池子太小会排队太大又浪费资源。我一般按峰值 QPS × 平均操作耗时来估算再留 30% 余量。10 到 20 的池子对中小规模应用足够了。监控方面重点盯三个指标Checkpoint 写入延迟、恢复成功率、中断到恢复的平均时长。前两个反映系统健康度第三个直接关系用户体验。如果恢复时长超过几秒用户就会觉得卡需要考虑优化 State 大小或数据库索引。5.4 常见问题速查表现象可能原因排查方向解决手段恢复后从头开始thread_id 不一致对比前后端 thread_id统一会话标识生成逻辑状态字段丢失未定义 reducer检查 State 注解给列表字段加 add_messages写入超时State 过大查 Checkpoint 记录大小大对象外置State 存引用并发状态错乱同 thread 并发执行查执行日志时间线加会话级分布式锁恢复后重复副作用节点非幂等查外部系统调用记录业务唯一键去重6. 几个我踩过之后才明白的经验先说一个关于 State 设计的体会。我一开始图省事把整个对话历史、所有工具返回的原始数据全塞进 State觉得这样恢复时信息最全。结果跑了几天发现 Checkpoint 表膨胀得飞快单条记录动辄几百 KB恢复时加载慢得让人抓狂。后来改成State 只存必要字段 大对象存外部单条 Checkpoint 降到几 KB恢复速度肉眼可见地变快。这个教训是State 是工作台不是仓库台面上只放当前要用的东西。再说一个关于中断粒度的经验。不是所有节点都值得中断。我早期设计时恨不得每个节点都加人工确认结果用户被弹窗烦得不行直接弃用。后来我把中断收敛到真正需要人决策的关键节点比如提交订单前发送对外邮件前其他环节全自动。中断是给用户掌控感的不是给用户添堵的这个度要把握好。最后分享一个调试技巧。排查恢复问题时别光看日志直接去数据库里把某个 thread 的所有 Checkpoint 按时间顺序拉出来一条条看 State 的变化和节点的流转。你会非常直观地看到在哪一步状态变了在哪一步中断了恢复时从哪条记录开始的。这个方法帮我定位过好几个靠日志根本看不出来的问题比如某个 reducer 悄悄把字段覆盖了。这套架构后续还能往几个方向扩展。一是加多级 Checkpoint把高频小状态和低频大状态分开存进一步优化性能。二是把恢复逻辑做成独立的服务专门负责扫描和重启未完成的会话实现真正的无人值守恢复。三是结合 AG-UI 的事件流做执行回放把一次完整的 Agent 执行录下来用于调试和演示。这些我都还在摸索有新的心得再单独写一篇。
