从手写Loop到LangGraph Runtime：PostgreSQL Checkpoint与AG-UI中断恢复实战

1. 为什么我要从手写 Loop 切换到 LangGraph Runtime

最早做 AI Agent 编排的时候，我和大多数人一样，直接上手写while循环。逻辑很直白：调模型、解析输出、判断是否要调工具、执行工具、把结果塞回上下文、再调模型，直到模型不再请求工具为止。这套东西在 demo 阶段跑得飞快，几十行代码就能让一个 Agent 跑起来，调试也简单，打个断点就能看到每一步的状态。

但问题很快就来了。第一个真实需求是"用户中途关掉页面，回来还能接着聊"。手写 Loop 的状态全在内存里，进程一重启就全没了。第二个需求是"人工审核节点"——某些敏感操作需要人工确认后才能继续，这意味着执行流要能暂停、持久化、等外部信号再恢复。第三个需求是"多轮长任务"，一个任务可能跑十几分钟，中间涉及十几次工具调用，任何一次网络抖动都可能导致整个流程崩掉重来。

这三个需求叠加在一起，手写 Loop 就彻底撑不住了。你当然可以自己实现状态序列化、自己搞断点续传、自己设计恢复协议，但那就是在重新造一个 Runtime。LangGraph 的价值就在这里：它把"可中断、可恢复、可持久化"的 Agent 执行流抽象成了图结构，配合 PostgreSQL Checkpoint 做状态落盘，再用 AG-UI 把执行过程实时推给前端。整套组合下来，你得到的是一个真正意义上的 Agent Runtime，而不是一段跑完就丢的脚本。

这篇文章我会完整拆解这套方案：为什么选 LangGraph 而不是裸写 Loop，PostgreSQL Checkpoint 到底存了什么、怎么存，AG-UI 在前后端之间扮演什么角色，以及中断恢复这条链路从触发到恢复的完整实现。适合已经写过基础 Agent、正在被状态管理折磨、想上生产级编排的开发者。如果你还在纠结 LangChain 和 LangGraph 的区别，我也会在对应章节里说清楚。

2. 核心概念拆解：LangGraph、Checkpoint、AG-UI 各自解决什么问题

2.1 LangGraph 与 LangChain 的区别到底在哪

很多人第一次接触这两个名字会懵。简单说，LangChain 是"组件库"，LangGraph 是"编排引擎"。LangChain 提供的是 LLM 封装、Prompt 模板、工具定义、检索器这些积木；LangGraph 提供的是把这些积木按图结构串起来、并且管理执行状态的能力。

打个比方：LangChain 像是一箱乐高零件，LangGraph 是那张带轨道的底板，零件插在底板上才能跑起来。你完全可以用 LangChain 的组件配 LangGraph 的图，两者不是替代关系，而是互补关系。面试里常问的"langchain 和 langgraph 区别"，标准答案就是：LangChain 关注"单个组件怎么用"，LangGraph 关注"多个步骤怎么按状态流转、怎么中断、怎么恢复"。

LangGraph 的核心抽象只有三个：State（状态）、Node（节点）、Edge（边）。State 是一个共享的数据结构，所有节点读写它；Node 是一个执行单元，接收 State 返回 State 的增量更新；Edge 决定下一个走哪个 Node，可以是固定的，也可以是条件分支。整个图编译之后变成一个可执行对象，这个对象就是 Runtime 的载体。

2.2 Checkpoint 不是日志，是执行现场的快照

Checkpoint 这个词在不同领域含义差别很大。数据库里的 checkpoint 是 WAL 落盘的检查点，游戏里的 checkpoint 是存档点，而 LangGraph 里的 Checkpoint 是图执行状态在某一时刻的完整快照。

它存的东西包括：当前执行到哪个节点、State 的完整值、待处理的下一步、以及这次执行的线程 ID（thread_id）。有了这些，你就能在任意时刻把执行"冻住"，之后用同样的 thread_id 把状态读回来，从断点继续跑。这跟数据库 checkpoint 的思路其实是一致的——都是把易失的内存状态固化到持久化存储，区别只是 LangGraph 固化的是 Agent 的对话与工具调用上下文。

这里有个关键点：Checkpoint 是按 thread 组织的。一个 thread 代表一条独立的执行线，比如一个用户的会话。同一个 thread 下可以有多个 checkpoint，按时间顺序排列，最新的那个就是当前状态。这种设计让"回到历史某一步重跑"变得非常自然，你只要指定 checkpoint_id 就能从任意历史点恢复。

2.3 AG-UI 补上前后端之间缺失的那一环

Agent 跑在后端，用户在前端，中间怎么通信？传统做法是前端轮询或者后端推 SSE，但 Agent 的输出是流式的、结构化的、带工具调用事件的，普通 SSE 根本表达不了。AG-UI 就是为这个场景设计的协议，它定义了一套标准的事件类型：文本增量、工具调用开始、工具调用结束、状态更新、中断请求、恢复信号等等。

AG-UI 的核心价值在于把 Agent 的执行过程变成前端可消费的事件流。前端不需要知道 LangGraph 内部怎么跑，只需要按 AG-UI 协议接收事件、渲染 UI。当后端触发中断时，AG-UI 会推一个中断事件给前端，前端弹出确认框，用户点确认后前端发一个恢复请求，后端从 Checkpoint 恢复执行。整条链路是解耦的，前端换框架、后端换模型都不影响协议层。

3. 整体架构设计：三层结构怎么搭

3.1 分层思路与数据流向

整套系统我分成三层：编排层（LangGraph）、持久层（PostgreSQL Checkpoint）、交互层（AG-UI + 前端）。

数据流是这样的：用户在前端发消息，前端通过 AG-UI 协议把请求发给后端；后端把消息塞进 LangGraph 的 State，用 thread_id 启动或恢复图执行；图执行过程中每个节点跑完都会写一次 Checkpoint 到 PostgreSQL；同时节点产生的事件通过 AG-UI 推给前端；如果遇到需要人工介入的节点，图主动中断，Checkpoint 记录中断位置，AG-UI 推中断事件；用户确认后，后端用同一个 thread_id 恢复执行，从 Checkpoint 读回状态继续跑。

这个设计的精髓在于状态与执行分离。执行是瞬时的、可能崩的，状态是持久的、可靠的。只要 Checkpoint 写成功了，执行崩了也能恢复。这跟传统后端"无状态服务 + 外部存储"的思路一脉相承，只不过这里的外部存储存的是 Agent 的执行现场。

3.2 为什么持久层选 PostgreSQL 而不是 Redis

LangGraph 官方支持多种 Checkpointer：内存版、SQLite 版、PostgreSQL 版。内存版重启就丢，只能做测试；SQLite 版适合单机小规模，但并发写会锁；PostgreSQL 版是生产首选。

选 PostgreSQL 的理由很实在。第一，Checkpoint 本质是结构化数据，用关系型数据库存天然合适，查询、索引、事务都有保障。第二，Agent 场景经常需要按用户、按会话、按时间查历史，SQL 表达力足够。第三，PostgreSQL 的 JSONB 类型能直接存 State 这种半结构化数据，既保留了 schema 的严谨性，又有文档数据库的灵活性。第四，运维成熟，备份、主从、监控一整套都是现成的。

Redis 不是不行，但它的持久化是"尽力而为"的，RDB 有丢数据窗口，AOF 性能又打折扣。Checkpoint 这种"丢了就要重跑整个任务"的数据，还是交给 PostgreSQL 更稳妥。

3.3 中断恢复的触发时机设计

中断不是随便触发的，得想清楚在哪些节点中断。我的经验是分三类：

• 强制中断：涉及资金、删除、对外发送等不可逆操作前，必须人工确认。
• 条件中断：当模型置信度低于阈值、或者工具返回异常时中断，让人来判断。
• 主动中断：用户自己点"暂停"，或者系统检测到长时间无响应。

这三类中断在 LangGraph 里的实现方式略有不同。强制中断用interrupt_before在节点执行前打断；条件中断在节点内部判断后主动抛中断；主动中断则通过外部信号触发。不管哪种，最终都会落到 Checkpoint 上，恢复逻辑是统一的。

4. 环境准备与依赖安装

4.1 Python 环境与核心依赖

我用的 Python 3.11，太老的版本有些异步特性支持不好。核心依赖就三个：

pip install langgraph langgraph-checkpoint-postgres ag-ui-protocol

langgraph是核心库，langgraph-checkpoint-postgres是 PostgreSQL 的 Checkpointer 实现，ag-ui-protocol是 AG-UI 的 Python SDK。如果你还要接具体模型，再装对应的 SDK，比如langchain-openai或langchain-anthropic。

注意：langgraph-checkpoint-postgres依赖psycopg（psycopg3），不是老的psycopg2。如果你项目里已经有 psycopg2，两个可以共存，但别搞混连接字符串的格式。

4.2 PostgreSQL 准备与表结构初始化

PostgreSQL 建议 14 以上，JSONB 的性能和索引支持更完善。建库建用户这些常规操作就不展开了，重点说 Checkpointer 的表。

LangGraph 的 PostgresSaver 提供了setup()方法，会自动建表。你只需要在应用启动时调一次：

from langgraph.checkpoint.postgres import PostgresSaver DB_URI = "postgresql://user:pass@localhost:5432/agent_db" with PostgresSaver.from_conn_string(DB_URI) as checkpointer: checkpointer.setup()

它会建三张表：checkpoints存主状态，checkpoint_blobs存大的二进制数据（比如消息历史），checkpoint_writes存节点写入的中间结果。这三张表的分工要理解清楚，后面排查问题全靠它们。

4.3 连接池配置与生产注意事项

生产环境千万别每次请求都新建连接。用ConnectionPool管理连接，池大小根据并发量定。我的经验值是：每个并发执行线程占一个连接，池大小设为"预期并发数 × 1.5"留点余量。

from psycopg_pool import ConnectionPool pool = ConnectionPool( conninfo=DB_URI, min_size=5, max_size=20, timeout=30, ) checkpointer = PostgresSaver(pool)

提示：checkpoint_blobs表会随着对话轮次增长而膨胀，一定要配定期清理策略。我一般按 thread 的最后活跃时间，超过 30 天没动的 thread 归档或删除。别等到表几个 G 了才想起来。

5. 用 LangGraph 构建可中断的图

5.1 State 设计：存什么、不存什么

State 是整个 Runtime 的核心，设计得好后面省一半事。我的原则是：存业务状态，不存临时变量。

from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_id: str task_status: str pending_action: dict | None approval_result: str | None

messages用add_messages注解，这是 LangGraph 提供的 reducer，新消息会追加而不是覆盖。task_status记录任务阶段，pending_action存待审批的操作，approval_result存审批结果。

不存什么？不存数据库连接、不存 HTTP client、不存任何不可序列化的对象。State 最终要序列化进 PostgreSQL，塞个连接对象进去直接报错。需要这些资源就在节点函数里现取，或者用依赖注入。

5.2 节点划分与职责边界

节点划分我遵循"单一职责 + 可独立恢复"原则。一个节点只做一件事，做完就写 Checkpoint。这样中断恢复的粒度最细，恢复时重跑的成本最低。

典型的节点划分：

• agent_node：调模型，决定下一步是回复还是调工具
• tool_node：执行工具调用
• approval_node：人工审批检查点
• finalize_node：收尾，生成最终回复

from langgraph.graph import StateGraph, START, END def agent_node(state: AgentState): response = llm.invoke(state["messages"]) return {"messages": [response]} def tool_node(state: AgentState): last = state["messages"][-1] results = execute_tools(last.tool_calls) return {"messages": results} def approval_node(state: AgentState): action = state.get("pending_action") if action and action["risk"] == "high": decision = interrupt({"action": action, "reason": "需要人工确认"}) return {"approval_result": decision} return {"approval_result": "auto_approved"}

interrupt()是 LangGraph 提供的中断原语，调用它会立刻暂停图执行，把当前 State 写进 Checkpoint，然后等外部恢复信号。

5.3 条件边与中断点的设置

条件边决定流程走向，中断点决定在哪停。这两个要配合设计。

def route_after_agent(state: AgentState): last = state["messages"][-1] if hasattr(last, "tool_calls") and last.tool_calls: return "tool" return "end" builder = StateGraph(AgentState) builder.add_node("agent", agent_node) builder.add_node("tool", tool_node) builder.add_node("approval", approval_node) builder.add_edge(START, "agent") builder.add_conditional_edges("agent", route_after_agent, { "tool": "approval", "end": END, }) builder.add_edge("approval", "tool") builder.add_edge("tool", "agent") graph = builder.compile( checkpointer=checkpointer, interrupt_before=["tool"], )

interrupt_before=["tool"]表示每次执行 tool 节点前都中断。这是最粗暴的做法，实际项目里我会更精细：只在特定条件下中断，比如工具是"删除"或"转账"时才打断。

实操心得：interrupt_before是静态的，编译时就定了。如果你需要动态决定是否中断，用节点内的interrupt()函数更灵活。我踩过的坑是：一开始全用interrupt_before，结果所有工具调用都要人工确认，用户体验极差。后来改成节点内判断，只有高风险操作才中断。

6. PostgreSQL Checkpoint 的落盘与恢复机制

6.1 Checkpoint 的写入时机与内容

LangGraph 的 Checkpointer 是每个超级步骤（super-step）写一次。超级步骤可以理解为"一批可以并行执行的节点"。每批节点跑完，Checkpointer 就把当前 State 的完整快照写进checkpoints表，把消息等大对象写进checkpoint_blobs。

写入的内容包括：thread_id、checkpoint_id（本次快照的唯一 ID）、parent_checkpoint_id（上一个快照）、state 的序列化值、以及元数据（时间戳、来源节点等）。这个链式结构让历史回溯成为可能——顺着 parent 指针能一路回到起点。

6.2 恢复时怎么读回状态

恢复的核心 API 是graph.invoke(None, config)。注意第一个参数传None，表示"不注入新输入，从 Checkpoint 恢复"。config 里带 thread_id：

config = {"configurable": {"thread_id": "user-123-session-1"}} # 首次执行 result = graph.invoke({"messages": [user_msg]}, config) # 中断后恢复 result = graph.invoke(None, config)

LangGraph 会根据 thread_id 找到最新的 checkpoint，把 State 读回来，从中断点继续。如果中断是interrupt()触发的，恢复时interrupt()会返回外部传入的值：

from langgraph.types import Command # 恢复并传入审批结果 result = graph.invoke( Command(resume="approved"), config, )

Command(resume=...)是恢复中断的标准方式，resume的值会成为interrupt()的返回值。

6.3 多线程并发下的隔离

thread_id 是隔离的关键。不同用户的会话用不同 thread_id，状态天然隔离。但要注意：同一个 thread_id 的并发执行会冲突。如果用户快速点两次发送，两个执行流会同时读写同一个 thread，导致状态错乱。

我的处理方式是在应用层加锁：同一个 thread_id 同时只允许一个执行流。用 Redis 分布式锁或者数据库行锁都行。锁的粒度是 thread 级别，不同 thread 之间不影响。

import redis lock_key = f"thread_lock:{thread_id}" lock = redis_client.lock(lock_key, timeout=300) if not lock.acquire(blocking=False): raise RuntimeError("该会话正在执行中，请稍后") try: result = graph.invoke(input_data, config) finally: lock.release()

注意：锁的超时时间要大于任务最长执行时间，否则任务还没跑完锁就释放了，并发问题照样出现。我一般设 5 分钟，超长任务另做处理。

7. AG-UI 打通前后端事件流

7.1 AG-UI 的事件模型

AG-UI 定义了一套事件类型，核心的有这么几类：

这套事件模型的好处是前端只依赖协议，不依赖后端实现。后端从 LangGraph 换成别的编排引擎，只要事件按 AG-UI 格式推，前端一行不用改。

7.2 后端事件推送实现

LangGraph 支持astream_events，能拿到执行过程中的细粒度事件。我把它转成 AG-UI 格式推给前端：

from ag_ui.core import EventType, TextMessageContentEvent async def stream_agent(input_data, config): async for event in graph.astream_events(input_data, config, version="v2"): kind = event["event"] if kind == "on_chat_model_stream": chunk = event["data"]["chunk"] if chunk.content: yield TextMessageContentEvent( type=EventType.TEXT_MESSAGE_CONTENT, delta=chunk.content, ) elif kind == "on_tool_start": yield ToolCallStartEvent( type=EventType.TOOL_CALL_START, tool_name=event["name"], ) elif kind == "on_tool_end": yield ToolCallEndEvent( type=EventType.TOOL_CALL_END, tool_name=event["name"], result=str(event["data"]["output"]), )

中断事件要单独处理。当图因为interrupt()暂停时，astream_events会结束，你需要检查最终状态里有没有待处理的中断：

state = graph.get_state(config) if state.next: yield InterruptEvent( type=EventType.INTERRUPT, reason="需要人工确认", payload=state.tasks[0].interrupts[0].value if state.tasks else None, )

7.3 前端接收与恢复请求

前端用 SSE 或 WebSocket 接收事件流，按类型分发处理。中断事件到达时，弹出确认 UI，用户操作后发恢复请求：

async function resumeAgent(threadId, decision) { const response = await fetch('/api/agent/resume', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ thread_id: threadId, resume: decision }), }); // 继续消费 SSE 流 consumeStream(response.body); }

后端收到恢复请求后，用Command(resume=decision)恢复图执行，继续推事件流。整条链路闭环。

实操心得：SSE 连接容易断，尤其是移动端切后台再回来。我的做法是前端记录最后收到的事件序号，重连时带上序号，后端从 Checkpoint 里找到对应位置重放。这样用户不会丢消息，体验好很多。

8. 完整实操：从零跑通一次中断恢复

8.1 初始化与首次执行

先把所有组件串起来：

from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.postgres import PostgresSaver from psycopg_pool import ConnectionPool pool = ConnectionPool(DB_URI, min_size=2, max_size=10) checkpointer = PostgresSaver(pool) checkpointer.setup() builder = StateGraph(AgentState) builder.add_node("agent", agent_node) builder.add_node("approval", approval_node) builder.add_node("tool", tool_node) builder.add_edge(START, "agent") builder.add_conditional_edges("agent", route_after_agent, {"tool": "approval", "end": END}) builder.add_edge("approval", "tool") builder.add_edge("tool", "agent") graph = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "demo-thread-001"}} result = graph.invoke( {"messages": [{"role": "user", "content": "帮我删除订单 #12345"}]}, config, )

执行到approval_node时，因为操作是"删除"，interrupt()触发，图暂停，Checkpoint 落盘。

8.2 中断状态检查

state = graph.get_state(config) print("下一步:", state.next) print("待处理中断:", state.tasks[0].interrupts if state.tasks else None)

输出会显示next=('approval',)和中断的 payload。这时候去数据库查checkpoints表，能看到最新一条记录，checkpoint_id就是当前快照。

8.3 恢复执行与结果验证

from langgraph.types import Command result = graph.invoke(Command(resume="approved"), config) print(result["messages"][-1].content)

恢复后，interrupt()返回"approved"，approval_node继续执行，返回approval_result="approved"，然后走tool节点执行删除，最后回到agent生成最终回复。

验证恢复是否真的从断点续跑，而不是从头重跑：看agent_node有没有被重复调用。如果 Checkpoint 生效，agent只会跑一次，恢复后直接从approval继续。

8.4 数据库侧验证

SELECT thread_id, checkpoint_id, parent_checkpoint_id, metadata->>'step' AS step, created_at FROM checkpoints WHERE thread_id = 'demo-thread-001' ORDER BY created_at;

你会看到多条记录，parent_checkpoint_id串成一条链。中断点那条的metadata里会标记中断信息。恢复后新增的记录 parent 指向中断点，证明是续跑而非重跑。

9. 常见问题与排查技巧实录

9.1 中断恢复典型问题速查

9.2 我踩过的三个坑

第一个坑：State 里塞了 datetime 对象。本地测试没问题，因为内存 Checkpointer 不序列化。换 PostgreSQL 后直接报错，因为 JSONB 不认 datetime。解决办法是统一用 ISO 格式字符串，或者自定义序列化器。

第二个坑：thread_id 用了随机 UUID。每次请求都生成新 thread_id，结果永远恢复不了，因为找不到历史 Checkpoint。thread_id 必须由业务逻辑生成，比如user_id + session_id，保证同一会话用同一个。

第三个坑：中断后没检查 state.next。我以为astream_events结束就是执行完了，结果中断时它也结束。前端一直显示 loading，用户以为卡死了。后来加了状态检查，发现state.next非空就推中断事件。

9.3 性能优化建议

Checkpoint 写入是同步的，高频写入会成为瓶颈。我的优化手段：

• 批量写入：如果一批节点可以并行，让它们跑完一起写，而不是每个节点写一次。
• 异步 Checkpointer：用AsyncPostgresSaver配合异步图执行，写入不阻塞主流程。
• 冷热分离：活跃 thread 的 Checkpoint 放主库，历史 thread 归档到冷存储。
• 索引优化：checkpoints表的thread_id和created_at建联合索引，查询快很多。

提示：checkpoint_blobs表存的是消息历史，增长最快。如果消息里带大文件或长文本，考虑把大对象抽出来单独存，Checkpoint 里只存引用。

10. 生产部署的几点经验

10.1 数据库连接与事务

生产环境用 PgBouncer 做连接池前置，应用侧的连接池大小可以调小。事务隔离级别用默认的 Read Committed 就够，Checkpoint 写入本身是单条 INSERT，不需要更严格的隔离。

10.2 监控指标

必须监控的指标：Checkpoint 写入延迟、恢复成功率、中断到恢复的平均时长、活跃 thread 数、表增长速度。我用 Prometheus + Grafana 搭的看板，异常时告警。

10.3 灰度与回滚

新版本图结构上线前，先用小流量验证。LangGraph 的图是编译时确定的，改结构要重新编译。回滚时注意：老版本的 Checkpoint 可能和新版本 State schema 不兼容，要么做 schema 迁移，要么让老 thread 用老版本图跑完。

我在实际项目里跑这套组合已经大半年了，最深的体会是：中断恢复不是加个功能，而是整个架构的地基。一旦你决定要支持中断恢复，State 设计、节点划分、事件推送、前端交互全都要围绕它来。LangGraph + PostgreSQL Checkpoint + AG-UI 这套组合的好处是每一层职责清晰，出问题能快速定位到是哪一层。最后分享一个小技巧：调试恢复逻辑时，直接在数据库里手动改 Checkpoint 的 state 值，能模拟各种边界情况，比写测试用例快得多。