1. 为什么“中断恢复”是 Agent 落地的分水岭

做过对话式 Agent 的人大概都有过这种体验：本地跑一个多轮工具调用的流程，前面几轮顺风顺水，结果到第五轮模型突然要调一个耗时很长的外部接口，或者用户中途关掉了页面，再回来的时候整个上下文全丢了，只能从头再来。更难受的是，如果这个流程里已经产生了副作用——比如已经写了一条数据库记录、已经发了一封邮件——你连“重跑一遍”都不敢，因为会重复执行。

这就是我一开始决定认真研究LangGraph + PostgreSQL Checkpoint + AG-UI这套组合的直接原因。标题里说的“从手写 Loop 到可恢复 Runtime”，其实描述的正是我自己的演进路径：最早我用一个while循环加messages数组手搓 Agent，后来发现状态管理、中断、恢复、并发这些事根本不该自己写，于是转向 LangGraph 的图结构，再往后发现内存里的状态一重启就没，于是接上 PostgreSQL 做持久化 Checkpoint，最后为了让前端能实时看到 Agent 的每一步、并且支持“人工介入后继续”，又引入了 AG-UI 这套面向 Agent 的前端协议。

这篇文章面向的是已经写过基础 Agent、但被“状态丢失”和“中断恢复”卡住的开发者。我会把整套方案的选型逻辑、核心机制、可复现的实操步骤，以及我踩过的坑全部摊开讲。读完你应该能自己搭出一个：进程重启后能接着跑、用户刷新页面不丢上下文、关键节点可以暂停等人确认再继续的 Agent Runtime。

先说结论性的判断：中断恢复不是一个“锦上添花”的功能，而是 Agent 从 Demo 走向可用的分水岭。因为真实业务里，长流程、人工审批、外部依赖超时、用户中途离开，全都是常态。你不可能要求用户“一次性把话说完、别刷新、别关页面”。所以 Checkpoint 机制本质上解决的是“Agent 的运行时状态该存在哪、以什么粒度存、怎么恢复”这三个问题。

2. 手写 Loop 的天花板到底在哪

2.1 一个典型的手写循环长什么样

我最早的那版代码大概是这样：维护一个messages列表，循环调用模型，如果返回里有tool_calls就执行工具、把结果 append 回去，再继续循环，直到模型不再要求调工具为止。逻辑上没毛病，跑简单场景也够用。但一旦需求变复杂，问题就成堆出现。

第一个问题是状态无处安放。messages只在内存里，进程一挂就没了。你可能会说那就存数据库呗，但存什么？存整个 messages 数组？那工具执行的中间状态呢？比如某个工具已经执行了一半，这时候恢复要怎么处理？手写循环里这些信息是散落在各个变量里的，根本没有一个统一的“状态快照”概念。

第二个问题是中断点无法表达。手写循环里，“中断”意味着你要在代码里插一堆if should_pause: return，然后下次进来还得想办法跳回原来的位置。这个“跳回”在命令式代码里极其难写，因为你没有显式的执行位置概念，只有调用栈，而调用栈是没法序列化的。

第三个问题是并发与重入。同一个会话如果两个请求同时进来，手写循环很容易把状态写乱。你需要自己加锁、自己做幂等，而这些在框架层面本来是可以统一处理的。

2.2 从“命令式循环”到“声明式图”的思维转变

LangGraph 给我的最大启发，是把 Agent 的执行过程从“一段会流动的代码”变成“一张有节点和边的图”。这个转变听起来抽象，但落到实操上非常具体：每个节点是一个函数，输入是当前 State，输出是对 State 的更新；边决定了下一个走哪个节点。整个执行过程就是在这张图上“走”，而每一次“走”都可以被记录成一个 Checkpoint。

为什么这个模型天然适合中断恢复？因为图上的位置是可以被命名的。你当前在哪个节点、State 长什么样，这两件事一旦能序列化，恢复就变成了“从某个 Checkpoint 重新加载 State，然后从那个节点继续走”。这比在命令式代码里试图恢复调用栈要干净太多。

提示：不要一上来就把所有逻辑塞进一个大节点。节点粒度越细，Checkpoint 的恢复点就越精确，中断恢复的体验越好。我一般会把“模型调用”“工具执行”“人工确认”拆成独立节点。

2.3 什么时候该果断放弃手写

我的经验判断标准很简单：当你的流程里出现“需要暂停等人”“需要跨请求保持状态”“需要重试某个中间步骤”这三个需求中的任意一个，就该上框架了。手写循环在这三种场景下要么写不出来，要么写出来一堆补丁，维护成本远高于学习 LangGraph 的成本。

LangGraph 和 LangChain 的区别这里也顺带说一句，因为这是热搜里高频出现的问题。LangChain 更像是一个“组件库”，提供模型、工具、检索器这些积木；LangGraph 则是“编排层”，负责把这些积木按图的方式组织起来并管理状态。你可以只用 LangGraph 不用 LangChain 的链，也可以两者混用。我现在的做法是：模型和工具用 LangChain 的封装，编排和状态用 LangGraph。

3. Checkpoint 机制：状态到底存在哪、怎么存

3.1 Checkpoint 的本质是一份可序列化的 State 快照

理解 Checkpoint，关键是理解它存的是什么。它不是存“代码执行到哪一行”，而是存“当前 State 的值 + 当前所在节点”。LangGraph 在每个节点执行前后都会写入一个 Checkpoint，这个 Checkpoint 包含：当前线程 ID（thread_id）、当前 Checkpoint ID、State 的完整值、下一个要执行的节点、以及一些元数据。

这里有个容易混淆的点：Checkpoint 不是增量日志，而是完整快照。也就是说每个 Checkpoint 里都有完整的 State，而不是只存变化量。这样做的好处是恢复时不需要回放，直接读最新快照即可；代价是存储会随 State 变大而变大。对于大多数 Agent 场景，State 主要是消息列表，体积可控，这个取舍是划算的。

3.2 为什么选 PostgreSQL 而不是内存或 Redis

LangGraph 默认提供一个内存版 Checkpointer，适合本地调试，但进程一重启就没了。生产环境必须换成持久化的。可选的有 SQLite、PostgreSQL、Redis 等。我最终选 PostgreSQL，理由有三条。

第一，事务保证。Checkpoint 的写入和业务数据的写入如果需要在同一个事务里，PostgreSQL 天然支持。比如“工具执行成功”和“Checkpoint 落库”要么都成功要么都失败，这种一致性用 Redis 做起来要费劲得多。

第二，查询能力。我需要按 thread_id 查历史、按时间范围做审计、甚至做一些状态分析，SQL 的表达能力是 KV 存储比不了的。热搜里那个“r checkpoint 替代”其实也反映了大家在找更灵活的持久化方案，PostgreSQL 就是那个更灵活的答案。

第三，运维成熟度。备份、主从、监控这些，PostgreSQL 的生态太成熟了，不需要我额外造轮子。

3.3 表结构与关键字段解读

LangGraph 的 PostgreSQL Checkpointer 会自动建表，核心是checkpoints和checkpoint_writes两张表。checkpoints存主快照，checkpoint_writes存节点写入的中间数据。关键字段包括thread_id（会话标识）、checkpoint_id（快照标识，通常是时间有序的）、parent_checkpoint_id（形成链式历史）、state（序列化后的状态）。

我建议你至少给thread_id建索引，因为按会话查询是最频繁的操作。如果 State 很大，可以考虑把state字段单独放到一张表或者用 TOAST 机制，PostgreSQL 对大字段本身有压缩和外部存储的处理，一般不用太担心。

-- 核心查询：取某个会话最新的 Checkpoint SELECT checkpoint_id, state FROM checkpoints WHERE thread_id = 'your-thread-id' ORDER BY checkpoint_id DESC LIMIT 1;

注意：thread_id 的生成策略很重要。我一般用业务侧的会话 ID 直接映射，这样前端刷新、换设备都能接上同一个会话。千万别用随机 UUID 每次新建，那样恢复就无从谈起了。

4. AG-UI：让前端真正“看见”并“介入”Agent

4.1 为什么需要一套面向 Agent 的前端协议

传统的前后端交互是“请求-响应”模型，前端发一个请求，后端返回一个完整结果。但 Agent 的执行是流式的、多步的、可能中断的。前端需要知道：现在走到哪一步了、正在调什么工具、需不需要用户确认、确认后怎么继续。这些用普通的 REST 接口表达起来很别扭。

AG-UI 就是为解决这个问题设计的。它定义了一套事件流协议，把 Agent 的执行过程拆成一系列标准事件：文本增量、工具调用开始、工具调用结束、状态更新、中断请求等。前端订阅这些事件，就能实时渲染出 Agent 的“思考过程”，并且在收到中断事件时弹出确认框，用户点确认后再把结果发回去继续执行。

4.2 事件流与状态同步的核心设计

AG-UI 的事件大致分几类：生命周期事件（run 开始/结束）、文本事件（token 增量）、工具事件（调用参数、执行结果）、状态事件（State 的 patch）、以及中断事件。前端拿到这些事件后，不是简单地拼接文本，而是维护一份和后台一致的 State 视图。

这里的关键设计是状态同步的方向。后台是 State 的唯一真相来源，前端通过事件流接收 State 的变更，本地做一份镜像用于渲染。当用户在前端做出操作（比如确认、修改参数），这个操作会作为一次新的输入发回后台，触发后台 State 更新并产生新的 Checkpoint。这样就形成了一个闭环。

4.3 中断与恢复在前端如何呈现

举个具体场景：Agent 要执行一个“删除数据”的工具，这是危险操作，需要人工确认。后台在执行到这个节点前触发一个中断事件，前端收到后弹出确认框，同时把当前 State 渲染出来让用户看清楚要删什么。用户点“确认”，前端把确认信号发回后台，后台从 Checkpoint 恢复，继续执行删除节点。

这个流程里，中断点本身也是一个 Checkpoint。所以即使用户在确认框弹出时关掉了页面，下次回来依然能看到这个待确认的中断，因为状态已经落库了。这就是 Checkpoint 和 AG-UI 配合的威力：前端负责呈现和收集输入，后台负责状态的持久化和恢复。

5. 完整实操：从零搭一个可恢复的 Runtime

5.1 环境准备与依赖安装

先把依赖装好。我用的核心包是langgraph、langgraph-checkpoint-postgres、psycopg（PostgreSQL 驱动），前端侧用 AG-UI 的客户端 SDK。

pip install langgraph langgraph-checkpoint-postgres psycopg[binary] langchain-openai

PostgreSQL 我用 Docker 起一个本地实例，方便测试：

docker run -d --name agent-pg \ -e POSTGRES_PASSWORD=yourpassword \ -e POSTGRES_DB=agentdb \ -p 5432:5432 postgres:16

提示：生产环境记得把密码、连接串放到环境变量里，别硬编码。另外 PostgreSQL 版本建议 14 以上，对 JSONB 的支持更完善。

5.2 定义 State 与图结构

State 我建议用 TypedDict 定义，字段尽量扁平。消息列表用 LangGraph 提供的add_messagesreducer，这样多个节点往消息列表里追加时能自动合并而不是覆盖。

from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] pending_action: dict | None confirmed: bool

图结构我拆成四个节点：think（模型推理）、act（工具执行）、confirm（人工确认）、finalize（收尾）。边用条件边，根据模型是否要求调工具、是否需要确认来决定走向。

builder = StateGraph(AgentState) builder.add_node("think", think_node) builder.add_node("act", act_node) builder.add_node("confirm", confirm_node) builder.add_node("finalize", finalize_node) builder.add_edge(START, "think") builder.add_conditional_edges("think", route_after_think, { "act": "act", "confirm": "confirm", "finalize": "finalize", }) builder.add_edge("act", "think") builder.add_edge("confirm", "act") builder.add_edge("finalize", END)

5.3 接入 PostgreSQL Checkpointer

这是让整个 Runtime 可恢复的关键一步。用PostgresSaver作为 checkpointer，编译图的时候传进去。

from langgraph.checkpoint.postgres import PostgresSaver DB_URI = "postgresql://postgres:yourpassword@localhost:5432/agentdb" with PostgresSaver.from_conn_string(DB_URI) as checkpointer: checkpointer.setup() # 首次运行建表 graph = builder.compile(checkpointer=checkpointer)

setup()只需要跑一次，它会创建必要的表。之后每次调用图的时候，通过 config 传入 thread_id，Checkpoint 就会自动按这个线程存。

config = {"configurable": {"thread_id": "user-123-session-1"}} result = graph.invoke({"messages": [("user", "帮我查一下订单状态")]}, config)

5.4 实现中断与恢复的完整链路

中断用 LangGraph 的interrupt机制。在confirm节点里调用interrupt()，图会在这里暂停，并把当前 State 存成 Checkpoint。

from langgraph.types import interrupt def confirm_node(state: AgentState): action = state["pending_action"] decision = interrupt({"action": action, "question": "确认执行该操作吗？"}) return {"confirmed": decision == "yes"}

恢复的时候，用Command(resume=...)把用户的决定传回去，图会从断点继续。

from langgraph.types import Command # 用户确认后 result = graph.invoke(Command(resume="yes"), config)

这里有个细节值得强调：恢复时用的 config 必须和中断时是同一个 thread_id，否则找不到对应的 Checkpoint。我一开始就是因为 thread_id 生成逻辑不一致，导致恢复时总是从头开始，排查了半天。

5.5 用 AG-UI 把过程推到前端

后台跑起来后，通过 AG-UI 的事件流把执行过程推给前端。核心是把图的执行包装成一个异步生成器，每产生一个事件就 yield 出去。

async def run_agent(input_data, thread_id): config = {"configurable": {"thread_id": thread_id}} async for event in graph.astream_events(input_data, config, version="v2"): # 转换成 AG-UI 标准事件 yield to_agui_event(event)

前端订阅这个流，就能实时看到 token 增量、工具调用、中断请求。中断事件到达时，前端渲染确认 UI，用户操作后调用恢复接口。

6. 踩坑实录与排查速查表

6.1 我遇到过的典型问题

问题一：恢复后重复执行工具。原因是工具执行节点没有做幂等，恢复时从上一个 Checkpoint 重放，把已经执行过的工具又跑了一遍。解决办法是在工具执行前检查 State 里是否已有该工具的执行结果，有就跳过。

问题二：State 太大导致 Checkpoint 写入慢。消息列表无限增长，每个 Checkpoint 都存全量，很快就拖慢了写入。解决办法是定期做消息裁剪，或者把历史消息归档到单独的表，State 里只保留最近 N 条。

问题三：并发请求把状态写乱。同一个 thread_id 两个请求同时进来，Checkpoint 互相覆盖。解决办法是在业务层对 thread_id 加锁，或者用 LangGraph 的并发控制机制。

问题四：中断后前端刷新，确认框消失。这是因为前端没有从后台拉取“当前是否有待处理的中断”。解决办法是前端启动时先查一次当前 State，如果发现有 pending 的 interrupt，就重新渲染确认 UI。

6.2 常见问题速查表

6.3 几条压箱底的经验

第一，Checkpoint 的粒度要权衡。太粗（比如整个流程只存一次）恢复不精确，太细（每个 token 都存）写入压力大。我一般按“节点级”存，配合关键节点的手动 Checkpoint。

第二，中断点要设计得可解释。用户看到确认框时，得清楚知道自己在确认什么。所以interrupt里传的信息要包含足够的上下文，别只传一个 ID。

第三，恢复逻辑要能处理“状态已过期”。比如用户三天后才回来确认，这时候外部数据可能已经变了。我的做法是在恢复时重新校验一次前置条件，不满足就提示用户重新发起。

第四，本地调试用内存 Checkpointer，生产用 PostgreSQL。两者接口一致，切换成本很低，但调试时用内存能省去很多数据库操作的干扰。

7. 关于这套方案后续还能怎么扩展

跑通基础的中断恢复之后，我最近在尝试几个方向。一个是多 Agent 协作下的 Checkpoint 隔离，不同 Agent 用不同的 thread_id 命名空间，避免状态串味。另一个是Checkpoint 的版本迁移，当 State 结构变化时，老 Checkpoint 怎么兼容读取，这个在长期运行的系统里迟早会遇到。

还有一个我觉得很有价值的方向是基于 Checkpoint 的回放调试。因为每个 Checkpoint 都是完整快照，理论上你可以从任意一个历史 Checkpoint 重新执行，用来复现 bug 或者做 A/B 测试。我现在排查线上问题时，经常就是把某个出问题的 Checkpoint 拉出来，在本地重放一遍，比看日志高效得多。

这套组合拳打下来，最大的感受是：Agent 的可靠性不来自模型本身，而来自 Runtime 的设计。模型会犯错、会超时、会被打断，但只要 Runtime 能把状态管好、把中断处理好、把恢复做对，整个系统就是可用的。LangGraph 提供图结构和 Checkpoint，PostgreSQL 提供持久化，AG-UI 提供前端交互，三者各司其职，拼起来就是一个能扛住真实业务折腾的 Agent Runtime。