企业官网建设流程全解析

1. 为什么手写 Loop 撑不过一次进程重启

如果你用 LangGraph 写过稍微复杂一点的 Agent，大概率经历过这个阶段：一开始觉得 StateGraph 挺香，节点、边、条件跳转都清清楚楚，跑起来也顺。但一旦业务要求"用户关掉页面明天再回来接着聊"，或者"服务发版重启后任务不能丢"，手写的那套 while 循环加内存字典就彻底歇菜了。

我自己最早做的一个客服工单 Agent 就是这样。整个流程是：识别意图 → 查订单 → 判断是否需要人工 → 生成回复。状态全塞在一个 Python dict 里，循环靠while not done驱动。本地测试丝滑得不行，上线第二天运维重启了一次容器，所有进行中的会话全部归零，用户回来发现机器人"失忆"了，前面聊的全白聊。那次事故之后我才认真去啃 LangGraph 的 Checkpoint 机制，也才有了后来这套"可恢复 Runtime"的完整方案。

这篇文章要讲清楚三件事：LangGraph 的 Checkpoint 到底把什么存进了 PostgreSQL、中断（interrupt）和恢复（resume）在运行时是怎么串起来的、以及怎么用 AG-UI 把"暂停—人工介入—继续"这套交互在前端跑通。适合已经写过基础 LangGraph、但被持久化和人工介入卡住的同学。如果你还在纠结 LangChain 和 LangGraph 的区别，简单说：LangChain 偏链式调用和组件编排，LangGraph 偏有状态图和有环流程，而 Checkpoint 是 LangGraph 区别于普通链式框架的核心能力之一。

先说结论：手写 Loop 的根本问题不是代码丑，而是状态没有"落盘锚点"。你的循环变量、中间结果、下一步该走哪个分支，全在进程内存里，进程一没，这些信息就没了。Checkpoint 的本质，就是在每个"超级步"（super-step）结束时，把整个图的状态快照写进一个外部存储，并给它一个唯一的 thread_id 和 checkpoint_id。恢复时不是重跑，而是从快照点"续上"。

2. Checkpoint 存进 PostgreSQL 的到底是什么

很多人以为 Checkpoint 就是存了个"当前节点名字"，其实远不止。理解它存了什么，你才能理解恢复时为什么能精确续上，而不是从头再来。

2.1 一次超级步的状态快照包含哪些字段

LangGraph 的每个 checkpoint 本质上是一个状态对象，落到 PostgreSQL 后大致包含这几类信息：

这里最关键的是parent_checkpoint_id 形成的链式结构。它让整个执行历史变成一条可回溯的链，而不是一个孤立的当前状态。你可以把它想象成 Git 的 commit 历史：每个 checkpoint 是一个 commit，parent 指针指向上一个，你随时可以 checkout 到任意历史点重新跑。

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

LangGraph 官方提供了多种 Checkpointer：MemorySaver（内存）、SqliteSaver、PostgresSaver 等。选 PostgreSQL 的理由很实际：

• 持久性：内存版进程一挂全没，Sqlite 单文件在容器化部署下不好共享，PostgreSQL 天然支持多实例共享同一份状态。
• 并发控制：多个 worker 同时处理不同 thread 时，PostgreSQL 的行级锁和事务能保证状态写入不打架。
• 查询能力：想统计"有多少会话卡在人工审核节点"、想按时间范围清理旧 checkpoint，SQL 直接搞定，不用自己写遍历逻辑。
• 生态成熟：备份、监控、扩容这些运维能力都是现成的，不用为状态存储单独造轮子。

提示：PostgresSaver 需要单独建表，官方提供了setup()方法自动创建 checkpoints、checkpoint_writes、checkpoint_blobs 等表。生产环境建议用独立的 schema 或数据库，别和业务表混在一起。

2.3 表结构背后的设计意图

跑完setup()后你会看到几张表，理解它们的分工很重要：

• checkpoints：主表，存每个 checkpoint 的元信息和状态引用。
• checkpoint_writes：存每个节点产生的中间写入，配合 pending_writes 实现"未完成任务的续跑"。
• checkpoint_blobs：存大的二进制或复杂对象，避免主表膨胀。

这种拆分的意图是把"状态元信息"和"大块数据"分离。元信息查询频繁但体积小，大对象写入少但体积大，分开存能让主表保持轻量，查询性能更稳。我实测过一个存了几十万 checkpoint 的库，主表查询依然是毫秒级，就是因为大对象都被挪到了 blobs 表。

3. 中断与恢复在 Runtime 里的完整链路

光有 Checkpoint 还不够，真正让"可恢复"落地的是 interrupt 机制和 Runtime 的配合。这部分是整套方案里最容易踩坑的地方。

3.1 interrupt 不是抛异常，而是"受控暂停"

刚接触 LangGraph 的 interrupt 时，我一度以为它就是个特殊的异常，捕获了就暂停。实际不是。interrupt()的语义是：在当前节点执行到这一行时，把当前状态存成一个 checkpoint，然后让整个图的执行"干净地停下来"，并把 interrupt 携带的值返回给调用方。

from langgraph.types import interrupt def human_review_node(state): # 到这里会暂停，把待审核内容抛给前端 decision = interrupt({ "question": "这笔退款需要人工确认", "amount": state["refund_amount"], }) # 恢复后，decision 就是前端传回来的值 return {"approved": decision == "approve"}

关键点在于：interrupt 之后，这个节点的执行上下文是被完整保存的。恢复时不是重新进入这个节点从头跑，而是从 interrupt 那一行之后继续，decision直接拿到恢复时传入的值。这就是为什么它叫"可恢复"，而不是"可重试"。

3.2 恢复时 Runtime 怎么找到断点

恢复的入口是Command(resume=...)。当你用同一个 thread_id 再次调用图，并传入 resume 值时，Runtime 会做这几件事：

1. 用 thread_id 查出最新的 checkpoint。
2. 检查这个 checkpoint 的 next 字段，确认它停在哪个节点。
3. 把 resume 的值注入到对应 interrupt 的返回位置。
4. 从该节点继续往下执行，而不是从 START 重跑。

这里有个容易忽略的细节：thread_id 是恢复的唯一钥匙。如果你恢复时用了不同的 thread_id，Runtime 会认为这是一个全新的会话，从 START 开始跑，你的断点就找不回来了。我在项目里专门把 thread_id 和业务侧的会话 ID 做了强绑定，避免这种低级错误。

3.3 一个完整的中断恢复时序

把上面的逻辑串起来，一次"暂停—人工介入—恢复"的完整链路是这样的：

注意：interrupt 的值必须是可序列化的。我踩过一次坑，往 interrupt 里塞了一个自定义对象，结果 PostgresSaver 序列化时报错。后来统一改成 dict 或基本类型，问题就没了。

4. 用 AG-UI 把暂停和恢复接到前端

后端能暂停能恢复，但如果前端不知道怎么展示"正在等你确认"、不知道怎么把用户的选择传回去，这套机制对用户来说就是黑盒。AG-UI 在这里扮演的角色，就是把 Runtime 的状态变化翻译成前端能消费的事件流。

4.1 AG-UI 处理的是"事件"而不是"响应"

传统 REST 是请求—响应模型，你发一个请求，等一个完整结果。但 Agent 的执行是流式的、可能中断的，用 REST 表达很别扭。AG-UI 的思路是把 Agent 运行过程中的每个关键节点都变成事件：文本增量、工具调用、状态更新、以及我们最关心的 interrupt 事件。

前端订阅这些事件后，就能实时知道"现在 Agent 停下来了，它在等一个决定"。这比轮询"任务完成了吗"要自然得多，也更省资源。

4.2 interrupt 事件在前端怎么落地

当后端触发 interrupt，AG-UI 会向前端推送一个中断事件，里面带着 interrupt 的 payload。前端拿到后，通常做两件事：

• 根据 payload 渲染一个交互组件，比如"批准/拒绝"按钮、一个输入框、或者一个选项列表。
• 把当前 thread_id 存好，等用户操作完，带着这个 thread_id 和用户的选择调用恢复接口。

// 前端伪代码：监听中断事件并渲染交互 onInterrupt((payload) => { setPendingAction({ threadId: payload.threadId, question: payload.question, amount: payload.amount, }); }); // 用户点击"批准"后 async function handleApprove() { await resumeRun({ threadId: pendingAction.threadId, resume: "approve", }); }

这套流程跑通后，用户体验就是：Agent 说"这笔退款需要你确认"，页面弹出确认框，用户点一下，Agent 接着往下走。中间哪怕用户去泡了杯咖啡、关了页面再回来，只要 thread_id 还在，状态就还在。

4.3 状态同步里最容易出错的三个地方

实际联调时，我遇到最多的问题集中在这三处：

• thread_id 丢失：前端刷新页面后没持久化 thread_id，导致恢复时找不到断点。解决办法是把它存进 localStorage 或 URL 参数。
• 重复恢复：用户手快点了两次"批准"，触发两次 resume。后端需要做幂等，或者在恢复后立即把该 interrupt 标记为已消费。
• 事件乱序：流式事件在网络抖动下可能乱序到达，前端如果无脑按到达顺序渲染，会出现"先显示结果再显示问题"的诡异现象。建议给事件带上序号，前端按序号排序后再渲染。

5. 从零搭一套可恢复 Runtime 的实操步骤

前面讲的是原理，这一节给一套可以直接抄的落地步骤。我用的是 LangGraph + PostgresSaver + AG-UI 的组合，Python 侧负责图逻辑，前端负责交互。

5.1 环境准备与依赖安装

先把依赖装齐。LangGraph 的版本迭代比较快，建议锁定版本，避免 API 变动导致跑不通。

pip install langgraph langgraph-checkpoint-postgres psycopg[binary] pip install ag-ui-protocol

PostgreSQL 建议用 14 以上版本，低版本在并发写入时偶发锁等待。本地开发可以用 Docker 起一个：

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

提示：生产环境务必给 checkpoint 库配独立的连接池，别和业务库共用。checkpoint 的写入频率可能很高，共用连接池容易把业务查询拖慢。

5.2 初始化 PostgresSaver 并建表

from langgraph.checkpoint.postgres import PostgresSaver DB_URI = "postgresql://postgres:postgres@localhost:5432/langgraph" with PostgresSaver.from_conn_string(DB_URI) as checkpointer: checkpointer.setup() # 自动建表，只需执行一次

setup()是幂等的，重复执行不会报错，但生产环境建议只在初始化脚本里跑一次，别每次启动都调。

5.3 编译图时挂上 checkpointer

这是让图具备持久化能力的关键一步。编译时不传 checkpointer，图就是无状态的，interrupt 也无法恢复。

from langgraph.graph import StateGraph, START, END builder = StateGraph(AgentState) builder.add_node("classify", classify_node) builder.add_node("human_review", human_review_node) builder.add_node("execute", execute_node) builder.add_edge(START, "classify") builder.add_conditional_edges("classify", route_after_classify) builder.add_edge("human_review", "execute") builder.add_edge("execute", END) graph = builder.compile(checkpointer=checkpointer)

5.4 用 thread_id 驱动一次可中断的执行

config = {"configurable": {"thread_id": "order-12345"}} # 第一次执行，会在 human_review 处暂停 result = graph.invoke({"messages": [...], "refund_amount": 200}, config) # result 里会包含 __interrupt__ 信息 print(result["__interrupt__"]) # 用户确认后恢复 from langgraph.types import Command resumed = graph.invoke(Command(resume="approve"), config)

注意恢复时必须传同一个 config，也就是同一个 thread_id。这是整个机制的地基。

5.5 把 AG-UI 事件流接上

后端把图的执行包装成一个流式接口，把 interrupt、文本增量、状态更新都转成 AG-UI 事件推给前端。前端订阅后按事件类型分发处理。这部分的具体协议实现各家略有差异，核心是保证 interrupt 事件里带上 thread_id 和 payload，恢复接口能接收 thread_id 和 resume 值。

6. 那些文档里不会写的坑

这套方案我前前后后调了两周，踩的坑比想象中多。挑几个最有代表性的说说。

6.1 checkpoint 无限增长的问题

默认情况下，每个超级步都会产生一个 checkpoint，一个长会话跑下来可能积累几百上千条。时间一长，checkpoints 表会膨胀得很快。我的做法是：

• 给 thread_id 加索引，查询快。
• 定期归档或清理超过 N 天的旧 checkpoint，但保留每个 thread 的最新一条，否则恢复会失败。
• 如果业务需要时间旅行，就保留完整链；如果不需要，可以只留最新快照。

清理时一定要小心 parent 指针的完整性，别把还在被引用的 checkpoint 删了。

6.2 序列化失败的隐蔽性

PostgresSaver 用 pickle 或 JSON 序列化状态。如果你的状态里塞了不可序列化的对象（比如数据库连接、文件句柄、lambda），写入时会报错，而且报错信息往往不直观。我的经验是：状态里只放纯数据，复杂对象在节点内部临时创建，用完即弃。

6.3 恢复后节点重入的副作用

这是最坑的一个。如果 interrupt 之前的节点有副作用（比如发了一封邮件、扣了一次库存），恢复时如果逻辑没设计好，可能重复执行。LangGraph 的 checkpoint 机制能保证"从断点续跑"，但不能自动帮你保证副作用幂等。我的做法是：把有副作用的操作尽量放在 interrupt 之后，或者在操作前先查一次"是否已执行"。

6.4 多 worker 下的并发恢复

当你有多个 worker 实例时，同一个 thread 可能被两个请求同时恢复。PostgreSQL 的事务能保证写入不冲突，但业务逻辑上可能出现"两个 resume 都生效"的情况。解决办法是在恢复入口加一层分布式锁，按 thread_id 加锁，保证同一时刻只有一个恢复在执行。

7. 几个高频疑问的实测回答

7.1 LangGraph 和 LangChain 到底怎么选

这个问题被问太多次了。我的判断标准很简单：如果你的流程是线性的、无环的、不需要人工介入和持久化，LangChain 的链式编排够用。一旦出现"根据中间结果决定下一步走哪"、"需要暂停等人工"、"进程重启后要接着跑"，就该上 LangGraph。两者不是替代关系，LangGraph 里照样可以用 LangChain 的组件。

7.2 Checkpoint 和传统数据库事务有什么区别

有人会问，这不就是把状态存数据库吗，和普通事务有啥区别。区别在于粒度：数据库事务保证的是一次操作的原子性，而 checkpoint 保证的是整个图执行过程的原子性和可恢复性。它记录的不只是数据，还有"执行到哪了、下一步该干嘛"这种控制流信息，这是普通事务做不到的。

7.3 中断恢复能不能跨版本

实测下来，跨 LangGraph 大版本恢复有风险。因为 checkpoint 的序列化格式可能变，新版本读旧 checkpoint 可能失败。生产环境升级前，建议先在测试库验证旧 checkpoint 能否正常恢复，或者干脆在升级时清空历史 checkpoint，只保留业务数据。

7.4 前端刷新后怎么保证不丢状态

核心就一句话：thread_id 必须持久化。存 localStorage、存 URL、存后端会话表都行，只要刷新后能拿回来。我见过有团队把 thread_id 只放在内存变量里，用户一刷新就全丢了，然后抱怨"恢复功能不好用"。这不是框架的问题，是设计的问题。

8. 写在最后的一点个人体会

这套方案跑通之后，我最大的感受是："可恢复"不是加个功能，而是一种架构约束。它逼着你把状态管理、副作用控制、并发处理这些平时能糊弄过去的问题，全部摆到台面上认真对待。手写 Loop 的时候你可以随便在内存里改状态，但一旦上了 Checkpoint，你就得想清楚"这个状态该不该进快照""这个操作重入会不会出问题"。

另一个体会是，别一上来就追求完美的时间旅行和分支回溯。我一开始想做得特别全，结果复杂度爆炸。后来退回到"只保证最新断点能恢复"，先把核心链路跑稳，再逐步加高级能力，反而顺利得多。如果你也在做类似的东西，建议先跑通"暂停—恢复"这一条最短路径，剩下的慢慢来。