LangGraph Checkpoint 实战：PostgreSQL 持久化与中断恢复

1. 为什么手写 Loop 终究会撞上"恢复"这堵墙

我最早做对话式 Agent 的时候，和大多数人一样，写了一个while True循环：读用户输入、拼 prompt、调模型、解析工具调用、执行工具、把结果塞回消息列表、再调模型，直到模型不再请求工具为止。这套东西跑 demo 特别爽，几十行代码就能让模型查天气、算数学、读文件。但只要业务稍微真实一点，问题就全冒出来了。

最典型的一个场景：用户让 Agent 帮忙做一份包含十几个步骤的数据处理任务，跑到第七步的时候，用户突然关掉了浏览器，或者网络抖了一下，或者后端服务重启了。这时候你那个while True循环里的所有状态——已经执行到哪一步、中间产出了什么、下一步该干嘛——全部随着进程一起蒸发了。用户回来一看，得，从头再来。这种体验在真实产品里是致命的。

手写 Loop 的第二个死穴是人机协同（Human-in-the-loop）。很多任务需要中途让人确认一下，比如"我要给这个客户发邮件了，你确认下内容"，或者"这个数据库删除操作要不要执行"。在裸循环里，你只能靠input()阻塞，或者自己造一套状态机把"等待确认"这个状态存到某个地方。一旦要支持多个用户并发、要支持确认后从断点继续，你就会发现自己其实是在重新发明一个工作流引擎。

第三个问题是可观测性。手写循环里，每一步的输入输出都散落在各种变量和日志里，你很难回答"这个任务到底卡在哪一步了""上一步的模型输出是什么""工具调用参数对不对"这类问题。排查线上问题时，你只能靠翻日志猜。

所以当我第一次接触 LangGraph 的时候，最打动我的不是它能画流程图，而是它把Checkpoint（检查点）做成了框架的一等公民。它本质上是在告诉你：Agent 的执行状态不应该活在内存里，而应该被持久化下来，随时可以暂停、恢复、回放、甚至分叉。这篇文章我就想把这套东西讲透——从 LangGraph 的图结构，到 PostgreSQL 做 Checkpoint 存储，再到用 AG-UI 把中断恢复的能力暴露到前端，跑通一条完整的链路。

适合读这篇的人：已经用 LangChain 或裸 API 写过 Agent、现在被状态管理折磨、想上 LangGraph 但被 checkpoint 概念绕晕的开发者。如果你还没写过 Agent，建议先补一下基础，不然有些坑你体会不到。

2. LangGraph 的图模型到底和 LangChain 差在哪

2.1 从"链"到"图"：不是换个名字那么简单

很多人问 LangChain 和 LangGraph 的区别，网上答案一大堆，但大多停留在"一个是链一个是图"这种废话层面。我用一句话概括我的理解：LangChain 解决的是"怎么把组件串起来"，LangGraph 解决的是"怎么让一个有状态的流程可控地跑起来"。

LangChain 的 Chain 本质是线性的、无状态的（或者说状态是隐式传递的）。你写一个LLMChain，它的输入输出是明确的，跑完就结束了。就算你用AgentExecutor，它内部那个循环也是黑盒的，你没法在中途插手，也没法把中间状态存下来。

LangGraph 的核心抽象是StateGraph。你定义一个 State（通常是个 TypedDict 或者 Pydantic 模型），然后定义若干个 Node（节点，本质是函数，接收 State 返回 State 的更新），再用 Edge（边）把它们连起来。边可以是普通的，也可以是条件边（conditional edge），根据当前 State 决定下一步走哪个节点。

这个模型的关键在于：State 是显式声明的，每个 Node 对 State 的修改是可追踪的。这就为 Checkpoint 提供了基础——因为框架知道每一步之后 State 长什么样，它就能把 State 存下来。

2.2 State 与 Reducer：状态合并的隐藏规则

这里有个特别容易踩的坑，我必须单独拎出来讲。LangGraph 里每个 Node 返回的不是完整的 State，而是对 State 的部分更新。框架会把这些更新合并进当前 State。默认的合并策略是"覆盖"——你返回{"messages": [新消息]}，它就直接把messages字段覆盖掉。

但对话场景里我们想要的是追加，不是覆盖。这时候就要用到Reducer（归约函数）。LangGraph 提供了Annotated配合add_messages来声明某个字段用追加的方式合并：

from typing import Annotated from typing_extensions import TypedDict from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] current_step: str task_result: dict

add_messages这个 reducer 不只是简单 append，它还会处理消息 ID 去重、处理RemoveMessage这类特殊指令。我见过太多人因为没加Annotated[list, add_messages]，导致每轮对话历史都被覆盖，模型永远"失忆"，然后花半天时间怀疑是不是模型的问题。

提示：自定义 reducer 时，函数签名是(current_value, new_update) -> merged_value。如果你有复杂的合并逻辑（比如字典深合并），完全可以自己写一个 reducer 传进去。

2.3 条件边与循环：Agent 的"思考-行动"骨架

一个 ReAct 风格的 Agent，在 LangGraph 里通常长这样：一个agent节点负责调模型，一个tools节点负责执行工具，然后用条件边判断——如果模型输出了工具调用，就去tools；如果模型直接给了最终答案，就END。

from langgraph.graph import StateGraph, END def should_continue(state: State): last_message = state["messages"][-1] if last_message.tool_calls: return "tools" return END builder = StateGraph(State) builder.add_node("agent", call_model) builder.add_node("tools", tool_node) builder.set_entry_point("agent") builder.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END}) builder.add_edge("tools", "agent") graph = builder.compile()

注意tools -> agent这条边，它构成了一个循环。这就是为什么 LangGraph 能表达 Agent 的迭代行为，而普通 DAG 不行。这个循环也是 Checkpoint 最有价值的地方——因为循环意味着"执行到一半"是常态，你必须能随时把循环的当前状态存下来。

3. Checkpoint 机制：中断恢复的地基

3.1 Checkpoint 到底存了什么

很多人以为 Checkpoint 就是存了个"消息列表"，其实远不止。一个完整的 Checkpoint 包含：

• channel_values：当前所有 State 字段的值，也就是你的messages、current_step这些。
• channel_versions：每个字段的版本号，用于并发控制和增量更新。
• versions_seen：每个节点"看到"的字段版本，用来判断哪些节点需要重新执行。
• pending_sends：待处理的消息发送（用于 map-reduce 之类的并行场景）。

除此之外，还有一个CheckpointTuple的概念，它把 checkpoint 本身、它的 ID、父 checkpoint ID、以及 metadata（比如step、source、writes）打包在一起。这个父子链结构非常关键，它让整个执行历史变成了一棵树，你可以从任意一个历史节点"分叉"出去重新跑。

3.2 thread_id：会话隔离的钥匙

Checkpoint 是按thread_id隔离的。你可以把它理解成"会话 ID"或者"任务 ID"。同一个 thread_id 下的所有 checkpoint 构成一条时间线，graph.get_state(config)拿到的永远是这条时间线上最新的状态。

config = {"configurable": {"thread_id": "user-123-task-456"}} result = graph.invoke({"messages": [("user", "帮我分析这份数据")]}, config)

这里有个经验：thread_id 的生成策略要提前想清楚。如果你用用户 ID 做 thread_id，那同一个用户的所有对话会共享一条时间线，历史会无限增长；如果你用"用户ID + 会话ID"，那每个会话独立，但跨会话的上下文就断了。我的做法是：会话级 thread_id 用 UUID，同时在 State 里存一个 user_id 字段做业务关联，这样既隔离了会话，又能按用户维度查询。

3.3 内存 Saver 与持久化 Saver 的分水岭

LangGraph 内置了MemorySaver，把 checkpoint 存在内存里。开发调试够用，但进程一重启就没了，而且多实例部署时各存各的，完全不能用。

生产环境必须换成持久化 Saver。官方支持的有SqliteSaver、PostgresSaver，社区还有 Redis、MongoDB 的实现。选哪个？我的判断标准很简单：

我选 PostgreSQL 的原因很实在：Agent 的状态数据天然需要事务和一致性。你不想看到"工具执行了但状态没存上"这种脏数据吧？Postgres 的 ACID 能兜住这个底。而且 checkpoint 表会随着对话增长变得很大，Postgres 的分区、索引、归档能力都是现成的。

4. 用 PostgreSQL 落地 Checkpoint 存储

4.1 环境准备与依赖安装

先把依赖装齐。注意版本，LangGraph 迭代很快，不同版本 API 有差异：

pip install langgraph langgraph-checkpoint-postgres psycopg[binary] psycopg-pool

psycopg是 psycopg3，PostgresSaver 用的是它，不是老的psycopg2。这点很多人会搞混，装错了会报连接相关的奇怪错误。

数据库这边，你需要一个可用的 PostgreSQL 实例（本地 Docker 起一个最省事）：

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

4.2 初始化 checkpoint 表结构

PostgresSaver 需要几张表来存 checkpoint、writes、migrations。官方提供了setup()方法自动建表：

from langgraph.checkpoint.postgres import PostgresSaver from psycopg_pool import ConnectionPool DB_URI = "postgresql://postgres:yourpassword@localhost:5432/langgraph?sslmode=disable" pool = ConnectionPool(conninfo=DB_URI, max_size=20, kwargs={"autocommit": True}) checkpointer = PostgresSaver(pool) checkpointer.setup() # 建表，只需执行一次

注意：setup()会创建checkpoints、checkpoint_writes、checkpoint_migrations等表。生产环境建议用独立的 migration 流程管理，而不是每次启动都调setup()，避免权限和并发问题。

建完之后你可以连上去看看表结构，理解一下数据是怎么落的：

SELECT thread_id, checkpoint_id, parent_checkpoint_id, metadata->>'step' AS step, type FROM checkpoints WHERE thread_id = 'user-123-task-456' ORDER BY checkpoint_id;

你会看到每个 checkpoint 都有parent_checkpoint_id，这就是那条时间线的链。metadata里的step告诉你这是第几步，source告诉你是loop（节点执行）还是input（用户输入）触发的。

4.3 把 checkpointer 挂到图上

编译图的时候把 checkpointer 传进去，整个图就具备了持久化能力：

graph = builder.compile(checkpointer=checkpointer)

之后每次invoke或stream，只要 config 里带了thread_id，状态就会自动落库。这里有个细节：checkpointer 是编译期绑定的，不是调用期。也就是说你不能一个图实例用两个不同的 checkpointer，需要不同存储就编译两个图。

4.4 连接池与并发：别让数据库成为瓶颈

生产环境一定要用连接池。我上面用了ConnectionPool，max_size根据你的并发量调。有个坑：PostgresSaver 的写入是同步的，如果你的图节点里有大量并发，checkpoint 写入可能成为瓶颈。

我的优化经验是：

• 连接池max_size至少设为预期并发数的 1.5 倍。
• 给checkpoints表的thread_id建索引（setup()默认会建，但你要确认）。
• 定期归档老 checkpoint。对话历史无限增长的话，单表几个 G 是迟早的事。可以按created_at分区，或者写个定时任务把超过 N 天的 thread 数据迁到冷存储。

CREATE INDEX IF NOT EXISTS idx_checkpoints_thread_id ON checkpoints (thread_id, checkpoint_id DESC);

5. 中断恢复的完整实现链路

5.1 用 interrupt 实现"暂停等人确认"

LangGraph 提供了interrupt()函数，在节点里调用它就能暂停执行，把控制权交还给调用方。这是实现 Human-in-the-loop 最优雅的方式：

from langgraph.types import interrupt, Command def send_email_node(state: State): draft = state["email_draft"] # 暂停，把草稿抛给前端确认 decision = interrupt({ "action": "confirm_email", "draft": draft }) if decision["approved"]: # 真正发送 send_email(draft) return {"status": "sent"} else: return {"status": "rejected", "feedback": decision.get("feedback")}

当图执行到这个节点时，interrupt()会抛出一个特殊信号，LangGraph 把当前状态存成 checkpoint，然后invoke返回。此时你可以通过graph.get_state(config)看到state.next指向send_email_node，说明它卡在这里等人。

5.2 恢复执行：Command 的正确用法

用户确认之后，怎么让它继续跑？用Command(resume=...)：

from langgraph.types import Command # 用户点了"批准" resumed = graph.invoke( Command(resume={"approved": True}), config={"configurable": {"thread_id": "user-123-task-456"}} )

resume的值会作为interrupt()的返回值，注入到节点里，节点从暂停的那一行继续往下执行。这里有个极其容易踩的坑：恢复时必须用同一个 thread_id，而且图实例必须是同一个编译产物（节点定义一致）。如果你改了节点代码再恢复，LangGraph 会尝试从 checkpoint 恢复，但节点逻辑变了，行为可能不符合预期。

提示：interrupt()之前的代码在恢复时会重新执行一遍。所以interrupt()之前不要放有副作用的操作（比如已经发了邮件、已经扣了款）。把副作用放在interrupt()之后，或者用幂等设计兜底。

5.3 从任意历史点分叉：时间旅行

Checkpoint 的父子链结构让你可以做"时间旅行"——拿到某个历史 checkpoint 的 ID，从那里重新跑：

history = list(graph.get_state_history(config)) # history[0] 是最新，history[-1] 是最早 target = history[3] # 回到第 4 个状态 forked = graph.invoke( None, # 不传新输入，从该点继续 config={"configurable": {"thread_id": "user-123-task-456", "checkpoint_id": target.config["configurable"]["checkpoint_id"]}} )

这个能力在调试时简直是神器。线上出问题，你可以把那个 thread 的历史拉出来，从出错前的状态重放，看看到底哪一步的模型输出跑偏了。也可以做"如果当时用户选了另一个选项会怎样"的假设分析。

5.4 状态查询与前端对接的接口设计

前端要展示"当前卡在哪一步、等什么确认"，需要后端暴露查询接口。核心就是get_state：

state = graph.get_state(config) # state.values: 当前 State # state.next: 下一步要执行的节点元组 # state.tasks: 当前待处理的任务（含 interrupt 信息）

state.tasks里会包含 interrupt 抛出的 payload，前端拿到就能渲染确认弹窗。恢复时前端把用户选择 POST 回来，后端调Command(resume=...)。这套接口设计的关键是：把 thread_id 和 checkpoint_id 都透传给前端，前端每次请求带上，后端就能精确定位到该恢复哪个状态。

6. AG-UI 把中断恢复搬到前端

6.1 AG-UI 解决的是什么问题

前面讲的都是后端能力。但用户是在浏览器里操作的，后端"暂停了等人确认"这件事，怎么实时、结构化地传到前端？传统做法是前端轮询或者自己定一套 WebSocket 协议，每个项目重复造轮子。

AG-UI 是一套面向 Agent 前端的协议，它定义了 Agent 和后端之间的事件流格式——文本增量、工具调用开始/结束、状态更新、中断请求等等。它和 LangGraph 的配合点在于：LangGraph 的 stream 事件可以被适配成 AG-UI 事件，前端用统一的协议消费。

6.2 事件流：从 graph.stream 到前端

LangGraph 的stream支持多种模式，updates模式会在每个节点执行后吐出更新，messages模式会流式吐 token，custom模式可以吐你自定义的事件。做 AG-UI 对接时，通常组合使用：

async for event in graph.astream(input, config, stream_mode=["updates", "messages", "custom"]): mode, payload = event if mode == "messages": # 适配成 AG-UI 的 TEXT_MESSAGE_CONTENT yield to_agui_text(payload) elif mode == "updates": # 节点级更新，适配成 STATE_DELTA 或 STEP 事件 yield to_agui_step(payload) elif mode == "custom": # 自定义事件，比如工具调用进度 yield to_agui_custom(payload)

当图执行到interrupt()时，stream 会结束当前轮次，前端收到一个"中断"信号，弹出确认 UI。用户操作后，前端发一个新的请求，后端用Command(resume=...)继续 stream。

6.3 前端如何渲染"等待确认"状态

前端拿到 interrupt payload 后，需要根据action类型渲染不同的 UI。我的做法是约定一套 action 到组件的映射：

这套映射让后端可以自由地插入人工确认点，前端只需要按 action 渲染。恢复时把用户的选择作为resume值传回，节点里interrupt()的返回值就是它。

6.4 断线重连：前端刷新后怎么找回状态

这是实际产品里绕不开的问题。用户确认弹窗弹出来，结果他刷新了页面，弹窗没了，任务卡在那。解决办法是：前端启动时用 thread_id 调一次状态查询接口，后端返回state.tasks，如果里面有 pending 的 interrupt，前端就重新渲染确认 UI。

@app.get("/threads/{thread_id}/state") async def get_thread_state(thread_id: str): config = {"configurable": {"thread_id": thread_id}} state = graph.get_state(config) return { "values": state.values, "next": state.next, "interrupts": [ task.interrupts for task in state.tasks if task.interrupts ] }

这样无论用户怎么刷新、换设备，只要 thread_id 还在（通常存在 URL 或 localStorage），状态就能恢复。这也是 Checkpoint 持久化最直接的价值体现。

7. 实测中那些文档不会告诉你的坑

7.1 checkpoint 表膨胀与清理策略

跑了一段时间后你会发现checkpoints表涨得飞快。每执行一个节点就写一条，一个 20 步的任务就是 20 条记录，加上checkpoint_writes表，数据量是翻倍的。我的处理方式：

• 对不再活跃的 thread，保留最近 N 个 checkpoint，老的删掉。但注意不能随便删中间的，因为父子链断了会导致时间旅行失效。要删就删整条 thread 的早期记录，或者按时间归档。
• 用checkpointer.delete_thread(thread_id)删整个 thread，这是官方提供的干净方式。
• 定期跑归档任务，把超过 30 天的 thread 数据导出到对象存储，然后从主表删除。

7.2 恢复时节点重复执行的副作用

前面提过interrupt()之前的代码会重跑。我踩过一个具体的坑：在interrupt()之前调了一次外部 API 记录"待确认"日志，结果恢复时又记了一次，日志里出现两条。解决办法是把这类副作用移到interrupt()之后，或者用幂等键去重。

还有一个更隐蔽的：如果你的节点里有time.sleep()或者耗时计算，恢复时会重新执行，导致响应变慢。所以节点设计要尽量"轻"，重活拆到独立节点，让 checkpoint 的粒度更细。

7.3 thread_id 设计不当导致的串话

我见过一个事故：开发时图省事，用固定的"default"做 thread_id，结果所有用户的对话全串在一起，A 用户看到了 B 用户的历史。这在生产环境是灾难级的。

正确做法是 thread_id 必须全局唯一且和用户绑定。我的生成规则是f"{user_id}:{session_uuid}"，既保证唯一，又方便按用户查询。同时在后端做一层校验：请求里的 user_id 必须和 thread_id 里的 user_id 匹配，防止越权访问别人的会话。

7.4 版本升级导致的 checkpoint 不兼容

LangGraph 的 checkpoint 格式在不同版本间可能变化。我遇到过一次升级后老 checkpoint 反序列化失败的情况。经验是：

• 升级 LangGraph 前，先在测试环境用生产数据的副本验证老 checkpoint 能否正常读取。
• 大版本升级时，考虑把活跃 thread 跑完再升级，或者写迁移脚本。
• 生产环境锁定版本，别用latest。

7.5 并发写同一 thread 的竞态

如果同一个 thread_id 同时有两个请求进来（比如用户手快点了两次），两个请求都会尝试写 checkpoint，可能产生冲突。PostgresSaver 内部有版本控制，但业务层面最好加锁——用 Redis 分布式锁按 thread_id 加锁，保证同一时刻只有一个执行流在跑。

lock_key = f"langgraph:lock:{thread_id}" if not acquire_lock(lock_key, timeout=30): raise HTTPException(409, "该任务正在处理中，请稍候")

8. 从裸 Loop 迁移到 LangGraph 的实操建议

如果你手上已经有一个跑得还行的手写 Loop，想迁移到 LangGraph，我的建议是别一次性重写。分三步走：

第一步，先把 Loop 里的"状态"抽出来，定义成显式的 State 结构。这一步不改逻辑，只是把散落的变量收拢。你会发现很多隐式状态之前根本没意识到。

第二步，把 Loop 的每一轮拆成节点。调模型是一个节点，执行工具是一个节点，判断是否继续是一个条件边。这时候先不接 checkpointer，用 MemorySaver 跑通，确认行为和原来一致。

第三步，接上 PostgresSaver，加上 thread_id，测试中断恢复。这一步最容易暴露问题——你会发现有些节点有副作用、有些状态没被正确 reducer 合并。逐个修。

迁移完之后，你会获得几个原来没有的能力：任务可以暂停恢复、可以时间旅行调试、可以多人协同确认、可以水平扩展（因为状态在数据库不在内存）。这些能力在裸 Loop 里要么做不了，要么得自己造一套，成本远高于直接用框架。

最后分享一个我自己的判断标准：当一个 Agent 任务的执行时间超过 30 秒，或者需要人工介入，或者需要跨请求保持上下文，就该上 LangGraph + Checkpoint 了。低于这个复杂度，裸 Loop 反而更直接。工具是拿来解决问题的，不是拿来炫技的。