LangGraph + PostgreSQL Checkpoint + AG-UI：LLM应用状态管理与中断恢复实践

算下来我踩这个坑踩了整整一个季度。当时手头有个客服 Agent核心逻辑就是教科书式的while True收消息、拼 Prompt、调 LLM、执行工具、再调 LLM……本地 demo 怎么跑怎么顺结果一上生产用户聊到一半网络断了一下进程重启上下文全部清空用户回来看到的是一个失忆的机器人。更抓狂的是并发一高session 状态存在内存里互相串A 用户的 history 跑到 B 用户的对话里去了。后来我把整套架构重写了一遍用 LangGraph 替代手写循环用 PostgreSQL 做 Checkpoint 持久化再通过 AG-UI 协议把 Runtime 暴露给前端。这套方案上线后中断恢复才真正变成了可预期的能力。这篇文章就围绕这三个核心——LangGraph 状态图、PostgreSQL Checkpoint、AG-UI 协议——记录我完整的改造路径和实测数据给正在从 demo 走向生产环境、被状态管理折磨的 LLM 应用开发者一个可参考的范本。1. 从 while True 到图执行器手写 Loop 在生产环境暴露的问题1.1 手写循环的典型结构与隐藏债务先看一下最传统的手写 Loop 长什么样。大部分人在小项目里都会写出类似的代码async def run_agent(user_id: str, user_message: str): # 从内存字典里拿历史会话 messages session_store.get(user_id, []) messages.append({role: user, content: user_message}) while True: response await llm.ainvoke(messages, toolsTOOLS) messages.append(response) tool_calls response.additional_kwargs.get(tool_calls, []) if not tool_calls: break for tool_call in tool_calls: result await execute_tool(tool_call[function][name], tool_call[function][arguments]) messages.append({ role: tool, tool_call_id: tool_call[id], content: result, }) session_store[user_id] messages return response.content这段代码在功能上是完整的能聊、能调工具、能多轮。但它有三个在生产环境必现的坑而且这三个坑是架构性的不是靠调参能解决的。第一会话状态全部活在内存里。session_store是一个普通的 Python dict进程一重启所有用户的上下文灰飞烟灭。真实场景里你不光要面对主动重启还要面对 OOM、宿主机迁移、发布时滚动更新。任何一个情况发生用户对话就断了。第二LLM 调用和工具调用没有检查点。假设 LLM 已经返回了一个工具调用Agent 开始执行工具执行到一半网络超时抛异常。此时 LLM 的响应已经生成了但你的messages还没 append 进去整个执行栈回滚下一次重试会把同一个工具调用再跑一遍。对于查数据库、发通知这类工具重复执行就是脏数据就是线上事故。第三并发场景下共享状态互相污染。上面这段代码如果直接塞进 FastAPI 里跑多个用户同时进来session_store的读写没有锁轻则读到别人半截的上下文重则两个协程同时往同一个列表 append直接RuntimeError: dictionary changed size during iteration。1.2 Agent 执行模型为什么循环不适合做可恢复流程更深层的问题是while True这种命令式结构天然不适合做可恢复流程。你想想看一个循环体里当前执行到了哪一轮、上一轮的 LLM 响应是什么、工具结果有没有落库——这些信息全藏在调用栈和局部变量里。你想在任何一步中断它然后从断点继续必须手动把这些东西全部序列化出来。短期可以硬编码一旦逻辑复杂到有分支、有并行节点、有条件跳转手写序列化逻辑的复杂度会直接爆炸。这也是我从手写循环转向 LangGraph 的根本原因。LangGraph 的本质就是把Agent 跑一轮对话这件事从一个隐式的循环变成一个显式的有向图。图的每个节点都是可独立执行、可单独保存状态的单元图的边决定了数据流向图的运行过程会被一个叫 Checkpointer 的组件持续记录。程序崩了没关系图的执行状态已经持久化了重新加载之后可以直接从最近一个完成的节点续跑。2. 把对话循环重构为 LangGraph StateGraph状态显式化是恢复的前提2.1 先定义状态再定义节点LangGraph 里最重要的一步不是写代码而是想清楚 State 结构。State 就是贯穿整个图的数据载体它的类型定义直接决定了你能恢复什么、不能恢复什么。我当时定义的 State 长这样from typing import Annotated, TypedDict from langgraph.graph import add_messages from typing import Literal class AgentState(TypedDict): messages: Annotated[list, add_messages] # 当前正在执行的工具名断点续跑用 current_tool: str # 工具执行原始结果重新跑的时候避免副作用 tool_result: str # 是否已确认来自用户 user_confirmed: bool # 执行状态机idle / running / waiting_human / done phase: Literal[idle, running, waiting_human, done]关键在messages字段。我不推荐用普通list而是用Annotated[list, add_messages]这是因为add_messages是 LangGraph 内置的 reducer。它能在多个节点并发返回消息时自动按消息 ID 去重合并而不是简单粗暴地覆盖。这对恢复尤其重要当图从 checkpoint 恢复时历史 messages 和新的消息要能正确合并而不是互相覆盖。然后是节点。我把原来的 while 循环拆成四个节点from langgraph.graph import StateGraph, END from langgraph.checkpoint.postgres import PostgresSaver async def call_model_node(state: AgentState) - dict: 调用 LLM 生成响应可能是普通文本也可能是工具调用意图。 response await llm.ainvoke(state[messages], toolsTOOLS) # 这里只返回消息不执行工具 return {messages: [response]} async def execute_tool_node(state: AgentState) - dict: 执行 model 节点决定要调用的工具。 last_message state[messages][-1] tool_call last_message.additional_kwargs[tool_calls][0] result await execute_tool(tool_call[function][name], tool_call[function][arguments]) # 把结果写回 statecheckpoint 会把它一并保存 return { messages: [{ role: tool, tool_call_id: tool_call[id], content: result, }], current_tool: tool_call[function][name], tool_result: result, phase: done, } async def human_review_node(state: AgentState) - dict: 人工审批节点常用于工具执行前的二次确认。 return {phase: waiting_human} async def route_after_model(state: AgentState): last_message state[messages][-1] if last_message.additional_kwargs.get(tool_calls): return execute_tool return end2.2 compile 时的关键参数checkpointer 决定了你能不能恢复节点和边都定义好之后最关键的一步是compilecheckpointer PostgresSaver.from_conn_string( postgresql://myuser:mypasslocalhost:5432/langgraph ) graph builder.compile( checkpointercheckpointer, interrupt_before[human_review_node], # 进入人工审批前暂停 )compile(checkpointer...)这个参数是整套可恢复 Runtime 的基石。编译后的对象不再是一个简单的函数而是一个运行时可暂停、可续跑的执行器。LangGraph 每执行完一个节点就会把整个 state 的增量变化写进 checkpointer 指定的存储里。中途进程死了也没关系state 已经落库了。interrupt_before[human_review_node]是第二层保障当图执行到该节点之前会主动暂停并等待外部输入而不是直接往下跑。这被 LangGraph 称为 Human-in-the-loop。用途很广工具执行前让用户确认、生成内容前让运营审核、敏感操作前二次鉴权。暂停状态下图会一直挂着等待resume信号。2.3 Thread ID恢复会话的唯一钥匙LangGraph 通过config里的thread_id来区分不同会话。你必须显式传入 thread_id否则 checkpointer 不知道把状态存在哪个槽位。config {configurable: {thread_id: user_id}} # 第一次调用 result await graph.ainvoke({messages: [{role: user, content: 帮我订下周去北京的票}]}, configconfig) # 后续轮次同样是这个 config result await graph.ainvoke({messages: [{role: user, content: 换成高铁}]}, configconfig) # 获取当前会话完整状态 snapshot await graph.aget_state(config) print(snapshot.values)这个thread_id的设计让我印象很深。它把恢复的语义从底层抽离了上层业务只需要记住一个用户 IDLangGraph 自动从数据库里把对应会话的最新状态捞出来。不需要自己拼接 messages不需要记录上次跑到第几个节点一切都由 checkpointer 管理。对比我手写 Loop 时维护的那套 session_store这套机制的可靠性完全是另一个量级的。3. PostgreSQL 持久化 Checkpoint让 Runtime 状态睡进数据库3.1 为什么选 PostgreSQL 而不是 SQLite 或内存存储LangGraph 官方支持多种 checkpointer包括内存版、SQLite 版和 Postgres 版还有 Redis、Mongo 等。我在选型时直接跳过了前两个锁定 PostgreSQL原因有三个。第一生产环境普遍已经部署了 PostgreSQL。公司内部的用户表、订单表都在 PG 里多一套 checkpointer 不需要额外引入新的中间件运维成本最低。第二PostgreSQL 支持行级锁和事务。Agent 进程在生产环境通常不止一个实例多个 worker 同时处理同一个 thread_id 的请求时必须有可靠的并发控制。SQLite 的锁粒度是整个库高并发下分分钟锁等待超时Postgres 的SELECT ... FOR UPDATE可以锁住特定 checkpointer 行保证同一个会话同一时间只有一个 writer。第三checkpoint 数据需要可查询、可清理。存内存里的数据一旦数量上来就没法查了存 PG 里可以直接用 SQL 查询某个 thread 的历史、清理过期数据、做容量统计。对运维排查帮助极大。3.2 初始化连接池与表结构我用的是psycopg_pool的AsyncConnectionPool配合 LangGraph 官方提供的PostgresSaverfrom psycopg_pool import AsyncConnectionPool from langgraph.checkpoint.postgres import PostgresSaver pool AsyncConnectionPool( conninfopostgresql://myuser:mypasslocalhost:5432/langgraph, max_size20, kwargs{autocommit: True}, ) await pool.open() checkpointer PostgresSaver(pool) # 首次运行时建表 await checkpointer.setup()setup()会创建三张核心表checkpoints主干状态存每个 thread 最新的 checkpoint 元数据包括 thread_id、checkpoint_id、parent_checkpoint_id 以及序列化后的 state 主体。checkpoint_blobs大对象存储存具体的状态数据 blob比如 messages 数组。checkpoint_writes增量写入每个节点执行后产生的增量数据。这里要特别说一下checkpoint_blobs和checkpoint_writes的设计逻辑。LangGraph 不是每次把全量 state 写入一个字段而是把 state 拆成多个 blob按写入操作维度增量记录。这样做的好处是恢复时只需读取目标 checkpoint 关联的增量数据即可不需要反序列化整个历史排查问题时也能清楚看到每一步谁写了什么。3.3 连接串与性能调优的注意点连接串有几个参数非常影响实际表现。一开始我图省事直接用同步方式连接结果 API 一启动就被 IO 阻塞卡死。后来改用异步连接 autocommitconninfo ( postgresql://myuser:mypasslocalhost:5432/langgraph ?connect_timeout10application_nameagent_runtime )connect_timeout10是防止 PG 短暂不可用时导致 Agent 进程无限等待application_name纯粹是为了 PG 侧查看当前活跃连接来源排查问题很有用。另外一个容易被忽视的点是autocommitTrue。checkpointer 的写入操作依赖独立事务如果外层套一个大事务会导致 checkpointer 与业务逻辑耦合出问题很难定位。PostgresSaver 的官方示例都是要求 autocommit 的我建议照做。3.4 如何验证 Checkpoint 真的写进去了写完后我习惯跑一条 SQL 确认SELECT c.thread_id, c.checkpoint_id, c.created_at, length(cb.blob) AS blob_size FROM checkpoints c JOIN checkpoint_blobs cb ON cb.thread_id c.thread_id ORDER BY c.created_at DESC LIMIT 5;正常情况应该看到每次调用图之后checkpoints表里就多出一条记录blob_size随对话轮次增长。如果表里一直没有新记录多半是 checkpointer 没正确传给compile()或者传入的config里缺thread_id。我踩过的一个坑在独立测试时 graph 是能起 checkpointer 的但部署到 FastAPI 里后我图省事用了全局单例的 graph 对象config里却忘传 thread_id。结果 LangGraph 会报一个thread_id is required when using a checkpointer的运行时错误。这个问题排查了快两个小时最后是看了异常堆栈才发现的。4. 用 AG-UI 把 Runtime 暴露给前端中断恢复的协议层设计4.1 AG-UI 解决的是前端怎么和 Agent Runtime 对话把 LangGraph Runtime 和 PostgreSQL Checkpoint 接通之后后端的能力已经很强了但前端还面临一个现实问题怎么把 Runtime 暴露给 Web 页面、小程序、甚至别的后端服务AG-UI 协议解决的正是在这个问题——它定义了一套 Agent 与 UI/客户端之间的标准事件流格式。所谓事件流就是后端把 Agent 运行过程中的各种状态变化以事件形式逐个推给前端而不是等整个 Agent 全部跑完再一次性返回一个 final result。这样用户能实时看到Agent 正在调用工具Agent 正在等待确认而不是盯着一个转圈的 loading 等几十秒。事件类型我实际用到的有agent_messageAgent 产生给用户看的文本消息。agent_actionAgent 调用某个工具的意图包含工具名和参数。agent_observation工具执行后的结果返回。agent_error运行中出现错误前端据此展示失败态。agent_finish一次 Agent 执行完成。agent_interruptAgent 主动暂停等待用户输入或确认。4.2 把 LangGraph 的执行过程流式翻译成事件LangGraph 支持astream_events可以拿到图中每个节点的运行事件。把它翻译成 AG-UI 事件流核心代码大致如下import json def to_agui_event(event_type: str, payload: dict) - dict: return {type: event_type, payload: payload} async def run_agent_with_agui(thread_id: str, user_message: str): config {configurable: {thread_id: thread_id}} # 先回放当前线程的最新状态判断是不是恢复场景 snapshot await graph.aget_state(config) if snapshot.next: # 有未完成的节点说明是中断后的恢复 yield to_agui_event(agent_interrupt, { resume_required: True, pending_nodes: list(snapshot.next), }) # 等待外部 resume 消息 return async for event in graph.astream_events( {messages: [{role: user, content: user_message}]}, configconfig, versionv2, ): kind event[event] if kind on_chat_model_stream: # 累积 token按 agent_message 推给前端 token event[data][chunk].content yield to_agui_event(agent_message, {delta: token}) elif kind on_tool_start: yield to_agui_event(agent_action, { tool: event[name], args: event[data].get(input), }) elif kind on_tool_end: yield to_agui_event(agent_observation, { tool: event[name], output: event[data].get(output), })这里最关键的一步是graph.aget_state(config)。在 AG-UI 的resume场景里前端会在用户重连时先调用一次类似/agent/resume的接口后端拿到 thread_id 后先去 checkpointer 里查询当前状态如果snapshot.next为空说明没有中断恢复正常跑新的 Agent 流程。如果snapshot.next不为空说明该线程之前卡在某个节点比如等待人工确认此时不能从头跑而是应该触发graph.ainvoke(None, config)续跑。4.3 前端恢复会话的交互流程前端配合 AG-UI 的恢复交互我推荐这样设计用户重新打开页面前端调用GET /agent/state?thread_idxxx。后端返回{ has_pending: true, pending_nodes: [human_review_node] }。前端根据 pending 状态显示你有一次工具调用待确认之类的横幅并提供继续执行、撤销按钮。用户点击后前端发送POST /agent/resumebody 里带{thread_id: xxx, action: continue}。后端graph.ainvoke(None, config)恢复执行前端通过 EventSource 或 WebSocket 实时接收事件流。这套模式的好处是前端完全不关心后端用的是 LangGraph 还是别的框架只关心 AG-UI 定义的事件类型。协议层一抽离前后端就可以独立演进。5. 完整故障演练kill 进程后恢复对话的实测过程5.1 演练场景设定为了验证整套方案是不是真的能扛住生产事故我做了一次灾难测试环境Docker 里跑 PostgreSQL 16本机跑 FastAPI LangGraph。剧本用户发起一个查询订单的工具调用Agent 已执行到human_review_node等待用户确认此时直接对 Python 进程执行kill -9模拟强杀重启服务尝试恢复对话。之所以选在human_review_node这个位置杀进程是因为这是最复杂的中断场景图还有未完成的节点LLM 已经产生了一轮工具调用结果状态里既有消息历史又有工具结果。如果这个场景能恢复过来其他场景基本都没问题。5.2 恢复步骤与日志验证服务重启后我写了一个简单的恢复脚本config {configurable: {thread_id: user_10086}} snapshot await graph.aget_state(config) print(next nodes:, snapshot.next) # 期望看到 (human_review_node,) print(phase:, snapshot.values.get(phase)) # 期望看到 waiting_human # 恢复执行 result await graph.ainvoke(None, config) print(final answer:, result[messages][-1].content)日志输出next nodes: (human_review_node,) phase: waiting_human final answer: 已为您查询订单订单号 AB20241234 状态为已发货。注意到snapshot.next精确地告诉我们中断发生在哪个节点。然后graph.ainvoke(None, config)从该节点继续执行完全不需要重放整个对话历史。5.3 对比验证如果用手写 Loop会怎样为了让自己更确信这套方案的价值我还做了一个对照组同样场景用最初的while True手写循环。流程是用户发起查询LLM 返回工具调用。工具执行完成结果暂存在 Python 局部变量tool_result里。kill -9。重启内存里的session_store已经空了。用户再发消息Agent 完全不记得工具结果甚至不记得自己在查订单。你可能会说那我可以在工具执行完就把结果写进数据库啊。但问题是手写循环里什么时候写、写哪些数据、下次从哪一步恢复、怎么避免重复执行工具全都是你自己造的轮子每造一个轮子就多一个出错的点。LangGraph Checkpoint 把这条链路标准化了你只需要明确哪些节点需要检查点哪些节点之间可能中断。5.4 关于重复执行工具的隐患在实测中我还专门验证了一个隐患——恢复时会不会重复执行已经完成的工具调用。答案是不会。LangGraph 的 checkpointer 保存的是已完成节点之后的 state恢复时只会执行snapshot.next里记录的未完成节点已完成的execute_tool_node不会重新跑。这一点对生产环境极其重要。假设你的工具是扣减用户余额如果不小心重复执行了资金就扣了两笔。LangGraph 的 checkpoint 能规避这个问题底层实现是每个节点执行完都会产生一个 checkpoint里面带了当前图执行到哪条边的信息恢复时从最近的 checkpoint 出发只执行未完成的节点。6. 上线 ChecklistCheckpoint 膨胀、并发冲突与数据安全6.1 checkpoint 膨胀对话轮次多了一定要清理PostgreSQL 存储的 checkpoint 是按状态增量记录的。虽然每条记录不会太大但当某个 thread_id 持续跑了几百轮对话后checkpoint_writes表里可能积累几千条增量记录占用空间膨胀查询性能也会逐步下滑。我的处理方案分两层应用层在compile()时配置store为可选的空实现或者定期把messages裁剪成最近 N 轮再写入。LangGraph 支持在节点里对 state 做截断比如只保留最近 20 条消息。数据库层写一个定时清理任务删除超过 7 天或超过一定条数的 checkpoint。一个干净的清理 SQL 示例DELETE FROM checkpoint_writes WHERE thread_id IN ( SELECT thread_id FROM checkpoints WHERE created_at now() - interval 7 days ); DELETE FROM checkpoint_blobs WHERE thread_id IN ( SELECT thread_id FROM checkpoints WHERE created_at now() - interval 7 days ); DELETE FROM checkpoints WHERE created_at now() - interval 7 days;注意删除顺序先删子表writes、blobs再删主表。别问我是怎么知道的第一次清理时忘了顺序直接外键报错。6.2 并发写同一个 thread_id连接池与锁生产环境经常出现同一个用户连发两条消息两个 worker 同时拿到同一个thread_id同时对 PG 里的同一行 checkpoint 做写入。如果没有锁控制后写的会覆盖先写的用户消息丢失。我用SELECT ... FOR UPDATE在业务入口做了一层串行化async with pool.connection() as conn: await conn.execute( SELECT 1 FROM checkpoints WHERE thread_id %s FOR UPDATE, (thread_id,) ) # 锁住后再调 graph.ainvoke result await graph.ainvoke(...)当然这只是兜底方案。更好的做法是在应用网关层对同一个 thread_id 做路由保证同一个用户的请求始终打到同一个 worker 上。但如果你的架构做不到这一点数据库锁至少能保证不丢消息。6.3 敏感信息进入 checkpoint 的安全风险checkpoint 会把 state 里的所有字段包括 messages、tool_result原样序列化存进 PG。这就带来一个安全风险如果用户消息里包含身份证号、地址、Token 等敏感信息它们会被明文存在数据库里。这个问题在演示环境无所谓但生产环境就得处理。我现在的做法是在 write checkpoint 之前对 state 中敏感字段做脱敏比如把消息里的手机号替换成138****1234。对checkpoints表所在的库做磁盘加密。checkpoint 表只授权给后端应用账号绝不直接开放给数据分析团队用 SQL 查询。从产品层面也要跟用户说明对话数据会被自动保存用于中断恢复。这既是交互设计问题也是合规要求。6.4 AG-UI 流式输出的背压与重连恢复AG-UI 的流式事件推送通常走 SSE 或 WebSocket。生产环境一旦接入网关可能会有超时断开、客户端中途刷新等问题。我建议把 AG-UI 事件流做成可重放的后端把每个 thread 的事件流存一份短期日志比如 Redis 里保留 5 分钟。前端如果断开重连先通过GET /agent/events?thread_idxxxfrom_event_idyyy补拉缺失事件。如果前端重连时后端 Agent 已经执行完了就直接从graph.aget_state拿最新状态把最终结果一次性推给前端。这样即使事件流在传输层丢失前端也不会卡在中间状态。最后补一句实在话整套方案跑通之后我最深的体会是Agent 应用能不能上生产不看你在 demo 里跑得多顺而看你在进程被杀死、网络断掉、用户中途离开这些脏场景下能不能兜住底。手写 Loop 不是不能用但它把恢复这件事的复杂度全部推给了开发者LangGraph PostgreSQL Checkpoint AG-UI 的组合本质上是把恢复从业务逻辑里抽离出来变成一个基础设施层面的能力。如果你也正在被 Agent 状态管理折磨我建议不要纠结要不要迁移直接拿一个小场景试水先把checkpointer接上、跑通一次kill -9之后的恢复你就能感受到这两套方案之间的代差。最后再分享一个小技巧上线前准备一个专门压测中断恢复的脚本随机在任意节点杀进程再恢复别只测完美路径。只有把这种故障场景变成日常测试的一部分你的 Runtime 才能真正配得上可恢复三个字。