1. 手写 Loop 的崩溃现场状态全丢的那一晚1.1 一个能跑的Agent是怎么迭代出来的先说实话——绝大多数人在搭Agent时第一步都不是画流程图而是写一个while循环。我自己就是这么起步的messages [{role: user, content: 帮我查一下上海最近三天的天气}] while True: response model.invoke(messages) if response.tool_calls: for tool_call in response.tool_calls: result dispatch_tool(tool_call) messages.append({role: tool, tool_call_id: tool_call.id, content: result}) continue messages.append(response) break这个版本跑起来没问题demo效果也不错。但问题出在跑起来和扛住生产之间那层东西状态。当Agent需要调用三个工具才能回答一个问题时每个工具结果都会追加到messages里。如果某个工具调用很慢用户关掉页面或者服务在等待期间重启这个循环就断了。再打开时所有已经执行过的步骤全都不见了。用户必须从头再问一遍Agent也必须重新跑一遍之前做过的工具调用。这不是体验问题是架构问题。手写Loop默认了进程是永久存在的、内存是永远不会丢的。但真实环境恰好相反进程一定会被重新部署连接一定会断网络一定会超时。只要这些场景出现手写循环就会把用户晾在查无此对话的尽头。1.2 手写循环的三个硬伤无状态、无断点、无法交接我后来把踩过的坑归了一下类手写循环的硬伤其实就三个。第一是无状态。messages只存在于当前进程的内存里任何一次重启都会让它蒸发。你以为你写的是对话记录实际上只是一个随时可能清空的数据结构。真正需要落地的是状态而不是循环本身。第二是无断点。手写循环只有继续跑和整体失败两种状态。工具调用中途挂掉整个Loop直接抛异常就算你做了重试也只能从头执行到当前位置之前的所有步骤。注意工具调用是有副作用的。天气查询还好如果其中一步是提交订单或发送邮件重跑一遍就等于重复执行了一次。有的工具幂等性做得差重复执行直接引起业务事故。第三是无法交接。Loop跑在哪个进程里状态就绑在哪个进程里。如果进程A挂了进程B想起来继续B没有任何线索知道这个Session进展到哪一步。所谓可恢复Runtime核心就是解决进程死了、状态别死这两个问题。多实例部署、滚动发布、断线重连这些都不是锦上添花而是能不能上生产的硬门槛。我当时被逼着做中断恢复触发点其实很蠢某次模型在工具调用之前输出了一长段最终回答导致前端解析出错用户刷新页面后所有上下文消失成了一个真实的事故工单。从那一刻我明白了一个道理Loop只是代码结构Runtime才是产品。2. LangGraph 的运行时模型把 Loop 变成可暂停的状态机2.1 LangChain和LangGraph到底差在哪很多人在学LangGraph之前会先接触LangChain我也是。LangChain解决的是LLM调用链的问题把提示词、模型、输出解析串成链。但链是线性的串完之后就完事了它没有一个真正的运行时概念。LangGraph把问题抽象成了一个有向状态机节点是执行单元边是流转规则状态是全局共享的数据结构。它在LangChain之上补了三个关键能力状态持久化图的每一个超级步节点执行完一轮都可以触发状态保存。暂停与恢复节点可以主动暂停把控制权交给外部系统等条件满足后再恢复。时间旅行与分支可以从任意历史checkpoint分支出去重新执行适合调试和回滚。这三个能力组合起来就是一个可恢复Runtime的全部要素。LangChain关注的是这一条链怎么跑通LangGraph关注的是整张图怎么活着。2.2 StateGraph、节点、边五步搭出一个图LangGraph里没有messages.append这种命令式的循环写法而是一个一个节点构建图结构。最基础的实现逻辑是这样的from langgraph.graph import StateGraph, START, END from typing import TypedDict, Annotated, List class AgentState(TypedDict): messages: Annotated[list, lambda left, right: left right] tool_results: dict def call_model(state: AgentState): response model.invoke(state[messages]) return {messages: [response]} def execute_tools(state: AgentState): results {} for tool_call in get_tool_calls(state[messages][-1]): results[tool_call.name] dispatch_tool(tool_call) return {messages: [ToolMessage(...)], tool_results: results} graph StateGraph(AgentState) graph.add_node(model, call_model) graph.add_node(tools, execute_tools) graph.add_edge(START, model) graph.add_conditional_edge(model, route_by_action, { call_tools: tools, end: END }) graph.add_edge(tools, model) app graph.compile()这个图做的事情固然简单但它的重点不是逻辑更复杂而是每一步的发生节点都明确了。手写Loop里循环变量和跳转逻辑埋在代码里在StateGraph里每个节点都成为状态机中的一个位置状态机本身才有了被记录和恢复的前提。2.3 数据流深层状态更新和Reducer很多人刚开始用LangGraph时会犯一个错return返回的 dict 是直接覆盖字段的。如果你的 state 里只有一个messages字段图跑两轮每轮都覆盖上一轮的消息列表上下文就剩最后一步了数据越跑越少。所以我在定义state时用了Annotated[list, lambda left, right: left right]这种Reducer写法。这个注解的意思是新的返回值不是覆盖state[messages]而是和旧值合并成一个新列表。这就是LangGraph的add_messages做的事情它还会自动判断消息的类型如果是同一id的消息会原地替换而不是重复追加。这个机制看起来只是数据合并的小技巧但它和中断恢复密切相关**只有状态是增量累积的checkpoint里保存的那份完整状态才有意义。**如果覆盖了恢复出来的状态就变了如果只追加不判断恢复出来的状态就有重复消息。Reducer是保证状态收敛的。Deep学习即可。3. PostgreSQL Checkpoint让每一刻状态都落盘3.1 Checkpoint与Thread数据在数据库里长什么样LangGraph自带一个内存版CheckpointSaver用字典存状态只适合本地调试。生产环境里我选了PostgreSQL作为存储端——你可以理解成给整个状态机装了行车记录仪每一个步骤跑完当前状态包括消息列表、工具结果、将要跳转的下一条边会被序列化保存下来。进程崩了记录还在。具体落到数据库核心是thread_id。每个对话Session对应一个ThreadPostgres里每条Checkpoint记录都带一个thread_id字段。恢复时不是找回整个服务而是找到那一个线程的最后一个Checkpoint把它还原成状态机继续往下走。表结构大致涉及这些字段LangGraph会自定义建表不用手动创建thread_id会话唯一标识。checkpoint_id每次保存的编号同一个Thread下多条记录按顺序排列。parent_checkpoint_id前一个checkpoint的ID用来串起历史链。checkpoint序列化后的完整图状态常用JSON格式。metadata图的配置元数据。有一点值得注意默认情况下LangGraph不是每一步都立刻落库批量写入和定时写入的使用要看你自己的图逻辑。对需要严格断点恢复的场景我建议按超级步结束即落盘来配。3.2 连接与写入基于 psycopg 的配置过程实际用起来需要装的是langgraph-checkpoint-postgres这个包。它底层用的是psycopg或者asyncpg配置非常直接我在项目里是这么接的from langgraph.checkpoint.postgres import PostgresSaver # 同步方式 with PostgresSaver.from_conn_string( postgresql://user:passlocalhost:5432/langgraph ) as saver: saver.setup() # 确认/创建表结构 graph app.compile(checkpointersaver)如果想走异步LangGraph同一个包也提供AsyncPostgresSaver用法一样只是from_conn_string换成异步版本。我前期跑Demo用的同步版后来接口流量上来之后切换到了异步版配合asyncpg连接池差异主要在并发写场景单线程调试看不出区别。这里有个细节值得注意Graph在compile的时候传入了checkpointer图就有了持久化能力。但如果没有传入thread_idcheckpoint不会真正生效。thread_id是每个会话的身份ID我用它来区分不同用户的对话流。调用方式config {configurable: {thread_id: user_conversation_id}} result app.invoke({messages: [user_message]}, configconfig)恢复时只需用同一个thread_id再次invokeLangGraph会自动加载最后一次Checkpoint并接着上次结束的位置继续执行。3.3 为什么选 PostgreSQL 而不是 SQLite 或 Redis我实测对比过三种存储SQLite单文件、零运维但并发写锁很严重。Agent图的状态更新频率不低多个线程同时写同一张表时容易因database is locked报错。本地个人项目可以面向多用户服务不行。Redis读写快、内存访问延迟极低但默认没有持久化就算开 AOF/RDB 也有丢数据窗口。Agent的Checkpoint不仅要求快更要求崩了之后一定能恢复Redis在这点上是弱项。PostgreSQL支持完整事务、崩溃恢复连接池成熟能支撑多实例同时读写。LangGraph官方就提供了PostgresSaver实现适配成本也很低。一句话总结PostgreSQL 解决的是可靠地写与随时能读两个问题。对Agent服务来说状态就是业务数据应该放在数据库里而不是放在缓存里。还有一个实际感受Postgres的JSON字段能力对LangGraph很友好。检查点的状态本身就是嵌套结构调试时直接SELECT checkpoint FROM checkpoints WHERE thread_id...能直观看到当前对话的完整快照这对定位为什么恢复后行为不对特别有用。我在排障时经常这么干。4. AG-UI 与 Runtime 的对接前端怎么知道该恢复哪一步4.1 AG-UI 消息协议解决了什么后端有了可恢复Runtime接下来是前端如何感知并在中断后无缝续上的问题。这里就要引入AG-UI。AG-UI是一个面向Agent交互的标准消息协议基础概念非常像ChatML但比它多了一层工具调用和中断消息的结构化表达。没有AG-UI之前前后端之间传消息全靠自己定JSON格式。前端需要手工区分区分普通文本和工具结果中断时前端不知道后端卡在哪一步、需要什么输入。AG-UI做的事情就是统一消息结构让Runtime和任何前端之间有一套共同语言。常用消息类型包括user用户输入。assistant模型回复。function_call模型请求执行工具带name和arguments字段。function_response工具执行结果带call_id关联到对应的function_call。interruptRuntime在某个暂停点发出的卡住了需要用户确认/输入事件。这套协议解决了谁先说话、谁在等谁的语义问题。后端通过AG-UI格式推送事件前端按类型渲染、按类型决定下一步交互整个链路清晰很多。4.2 中断消息与手动恢复消息如何映射在LangGraph中路由节点里可以显式设置一个人工审核点。例如一个生成内容后需要人工确认的节点在模型输出完内容后执行到这里不会继续流向自动发布节点而是停止并等待外部确认。LangGraph的处理方式是让节点抛出一个interrupt信号让Graph处于等待恢复的状态。此时通过AG-UI向Runtime发送消息的意义就体现出来了。LangGraph恢复时有两种方式Command(resume...)直接传恢复数据或者再次调用invoke并指定同一个thread_id运行时读取Checkpoint发现图正处于中断点把新收到的消息作为继续执行的外部输入。用AG-UI来描述的话中断恢复的链路是这样的Runtime发出interrupt消息附带上需要人工审核的说明。前端收到interrupt给用户渲染确认按钮。用户点击确认前端发送一个user类型消息内容为CONFIRM或reject。Runtime收到消息后以相同的thread_id恢复Checkpoint把用户确认结果注入到中断节点的resume参数。图继续往下走工具发布节点拿到确认结果正常执行。这套流程里前端也不需要关心后端到底是重新加载了模型还是从磁盘还原本地状态——协议层保证了两者都表现为同一语义中断事件、用户输入、继续执行。4.3 Runtime 层到底封装了什么最初看到AG-UI相关概念时我也很困惑这不又是一个API格式吗为什么要叫Runtime后来我把整体架构重新梳理了一遍发现AG-UI加上LangGraph的Checkpoint机制刚好补上了手写Loop三个硬伤中的一个无法交接。后端本身是语言无关的状态机运行时加上AG-UI之后整个Session的前后端交互也变成了一套可恢复的协议流。每一段消息都有类型和状态前端不是一次性把整个对话拉走”而是按步骤消费消息流”。这就让用户刷新页面后回到原会话变成一件可期待的事前端重新连接后端带上thread_idRuntime从数据库里把最后的Checkpoint捞回来通过AG-UI把最后一条interrupt或function_call状态原样推给前端前端按消息类型重新绘制就像什么都没发生过一样。在我的实际项目里还额外加了一层会话映射表用户ID 浏览器端会话名 →thread_id。这层映射让服务端可以随时根据业务维度找到某个线程的Checkpoint不需要让前端硬存数据库主键。5. 中断恢复完整演练从卡在工具调用到继续执行5.1 一个贴近业务的场景下面用一个真实可复现的场景把整条链路串起来。场景是一个写周报并发送的Agent模型先生成一份周报草稿。草稿内容需要用户确认后才能通过邮件工具发送。用户确认时回复一条消息Agent继续执行发送动作。发送完成后Agent生成最终总结。这是一个典型的需要人工介入才能继续的场景也天然需要中断恢复。5.2 带 Checkpoint 的图实现代码from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.postgres import PostgresSaver from langgraph.types import Command class ReportState(TypedDict): messages: Annotated[list, add] # 简化写法 draft: str confirmed: bool def gen_draft(state: ReportState): content model.invoke(写周报基于以下要点... state[messages][-1].content) return {draft: content, messages: [AIMessage(contentcontent)]} def human_confirm(state: ReportState): raise interrupt(请确认周报内容回复CONFIRM后发送) def send_report(state: ReportState): call_send_email(state[draft]) return {messages: [ToolMessage(content已发送, ...)]} graph StateGraph(ReportState) graph.add_node(gen_draft, gen_draft) graph.add_node(confirm, human_confirm) graph.add_node(send, send_report) graph.add_edge(START, gen_draft) graph.add_edge(gen_draft, confirm) # 关键中断后如何走取决于外部传入的resume graph.add_conditional_edge(confirm, lambda state: send if state.get(confirmed) else END) graph.add_edge(send, END) with PostgresSaver.from_conn_string(DATABASE_URI) as saver: app graph.compile(checkpointersaver)这里human_confirm节点是中断点。运行时执行到这里app.invoke不会返回最终的输出消息而是停在断点处等待继续输入。5.3 完整执行与恢复过程第一次调用config {configurable: {thread_id: ticket-10086}} result app.invoke( {messages: [human_message_batch]}, configconfig ) # result[messages] 中包含了draft但没有发送成功消息用户看到图停在了确认步骤然后回复CONFIRMapp.invoke( Command(resumeCONFIRM), configconfig ) # 此时Graph从上次中断点恢复confirmedTrue图继续走向send节点第二次invoke后系统从PostgreSQL里读到了当前线程的Checkpoint恢复了draft和中断点状态把resumeCONFIRM注入然后继续往下执行。整个过程没有重新调用gen_draft周报草稿也没有再生一遍——这正是恢复而不是重跑的意义。我还验证了一条**如果第一次调用后进程直接kill掉再重新启动一个新的Python进程只要PostgreSQL同一张表、同一个thread_id整个恢复效果是完全一致的。**这也是我判定中断恢复方案能上线的重要依据。5.4 时间旅行调试带来的额外价值LangGraph的Checkpoint机制还带来了一个额外好处可以回溯到之前的任意一次Checkpoint分支出去调试开发。比如发送邮件工具出错了检查结果显示是草稿生成环节的某段Prompt问题。此时可以取出草稿生成前的那次Checkpoint改一下Prompt重新从旧分支走下去。这在传统开发流程里很难做到低成本因为历史状态早就丢了。我就用这个能力做过一次修改工具调用参数后重新模拟发送流程的回归测试省了非常多时间。6. 我在实战中踩过的坑与优化细节6.1 线程ID不隔离串数据第一次接入PostgreSQL Checkpoint时我图省事所有会话用同一个thread_id。结果A用户的中断恢复状态直接出现在B用户的请求里生产问题立刻爆炸。这个坑的关键点在于LangGraph的Checkpoint是按thread_id隔离的同一个ID就等于同一个会话。要建立业务身份到线程ID的映射每新建一个会话就生成一个UUID并通过数据库表维护映射关系。千万不要用用户ID直接当thread_id因为一个用户可以有多个会话。6.2 恢复时重复注入历史消息有一次遇到恢复后模型把之前已经执行过的工具调用又执行了一遍的问题。排查了很久最后发现是因为我在输出到模型的Prompt里加了一段历史消息快照而这一段快照在human_confirm恢复后又被拼接了一次。Checkpoint本身保留了完整历史不需要在Prompt里额外携带旧消息否则会造成重复注入。这个坑的教训是**Checkpoint恢复时先把state[messages]洗一遍确认里面已经包含所有必要的上下文再组装新的调用输入。**必要的话可以用RemoveMessage清理掉已过时的时间线条目避免上下文越积越乱。6.3 连接池、锁与提交时机PostgreSQL的连接池我没有特别高级的配置但有一个基本底线的经验长跑任务别挂在一条连接上。刚开始我把PostgresSaver的对象在进程里全局复用时偶尔会遇到connection is closed的报错。后来看代码发现它内部持有连接多线程并发时如果不同时处理事务会互相干扰。建议改成每次会话开始时从连接池获取连接用完还回去或者直接用官方提供的AsyncPostgresSaver加asyncpg连接池并发场景会更稳。另外关于提交时机如果整张图很长在中断点之前已经执行了多个节点尽量让Checkpoint在每个超级步结束的时候就落盘。默认情况下LangGraph在每个步骤保存时内部会开事务但有些批量策略说等攒一批再写这样如果中途崩了最后几步状态没有保存。宁可多落几次也别为了省IO丢状态。6.4 AG-UI 中断事件与前端状态同步接入AG-UI时最容易踩的坑是前端只根据消息内容渲染不根据消息类型决定行为。具体表现是当Runtime发来interrupt时前端把它当成普通assistant消息渲染成文本按钮不出现用户完全不知道需要确认。我的做法是前端加一个事件分类器function_call渲染成工具执行卡片。function_response静默处理或显示结果摘要。interrupt渲染确认弹窗并把恢复后要发送的消息预填好。assistant正常聊天显示。这层分类不复杂但缺了它整条链路在前端体验上就断了。6.5 幂等设计与工具副作用防护上面提到过手写Loop在恢复时会重放全部工具调用LangGraph的Checkpoint机制虽然不让模型重新生成工具调用但如果你在开发阶段频繁调试同一个thread_id反复 invoke 是会发生恢复点之后的节点重新执行的。这提醒我们工具调用本身需要做幂等保护。最直接的做法是在工具调用前加一个这次调用是否已经执行过的检查。比如发邮件工具可以在业务表里保存tool_call_id 活动ID的唯一约束工具侧查询到相同记录就直接返回缓存结果而不是真发第二遍邮件。这个防护对外部系统更友好也是生产级Runtime必须具备的自觉。整套方案跑通后我最大的体会是Loop和Runtime之间的差距恰恰就体现在对状态的掌控力上。状态能持久化、能暂停、能恢复Agent才像一个真正有记忆的助手而不是一个每次对话都从零开始的提词器。如果哪一天你也要做类似的方案建议按照先加Checkpoint、再加中断点、最后补AG-UI协议这个顺序逐步推进每完成一步都做一次崩溃恢复测试。状态服务这种事早暴露问题比晚上线好得多。