企业官网建设流程全解析

1. 为什么“中断恢复”是 Agent 工程绕不开的一道坎

做过对话式 Agent 的人大概都有过这种体验：用户跟机器人聊到一半，页面刷新了、网络抖了一下、后端服务滚动重启了一次，回来之后上下文全没了，机器人像失忆一样问“你好，请问有什么可以帮您”。更糟的是那种跑了十几步工具调用、眼看就要出结果的长任务，中途断掉只能从头再来，前面烧掉的 token 和时间全打了水漂。

这个项目要解决的就是这件事：让一个基于 LangGraph 构建的 Agent 具备“可恢复 Runtime”能力——状态持久化到 PostgreSQL，前端通过 AG-UI 协议对接，任何一步中断之后都能从断点接着跑，而不是从零开始。核心关键词就三个：LangGraph负责编排有状态的图执行，PostgreSQL Checkpoint负责把每一步的状态落盘，AG-UI负责把运行时的状态变化实时推给前端。

适合谁来参考？如果你已经写过简单的 LangChain Chain、跑通过一两个 Agent Demo，但一遇到“多轮 + 长任务 + 断线重连”就抓瞎，那这篇就是写给你的。我会从手写 Loop 的痛点讲起，一路拆到 Checkpoint 的表结构、恢复时的状态合并逻辑、AG-UI 的事件流设计，最后给一份能直接抄的排查清单。全程按我实际踩坑的顺序来，不绕弯子。

先说清楚一个前提：“可恢复”不等于“可重试”。重试是把整个任务重跑一遍，可恢复是从中断的那一步继续。前者浪费资源，后者才是工程上真正值钱的东西。LangGraph 的 Checkpoint 机制天生就是为后者设计的，但要用好它，得先理解它到底存了什么、什么时候存、恢复时怎么读。

2. 从手写 Loop 到 LangGraph：一次架构上的取舍

2.1 手写 Loop 到底卡在哪

最早我写 Agent 就是一个while循环：调模型、解析工具调用、执行工具、把结果塞回消息列表、再调模型，直到模型不再要求调工具为止。代码不到一百行，跑 Demo 特别爽。但一旦要上生产，问题一个接一个冒出来。

第一个问题是状态无处安放。整个对话历史、中间的工具返回值、当前的步数，全在内存里的一个 list 里。进程一重启，list 没了。你可能会说那就存 Redis 呗，但存进去容易，恢复的时候你得自己判断“上次跑到哪一步了、哪些工具已经执行过、哪些还没执行”，这套逻辑手写起来非常容易出错，尤其是工具调用有副作用的时候（比如已经发过一封邮件），重跑一遍就是灾难。

第二个问题是流程不可视、不可控。手写 Loop 里加一个“人工审核”环节，你得在循环里插if判断，加一个“条件分支”又得插一层嵌套。业务逻辑稍微复杂一点，这个循环就变成了一坨意大利面，改一处怕崩三处。

第三个问题是并发与流式。用户希望看到 Agent 一步步思考的过程，而不是等三十秒后一次性吐出来。手写 Loop 要做流式，得自己管理事件队列、自己处理背压，工作量陡增。

2.2 LangGraph 带来的三个关键变化

换成 LangGraph 之后，最直观的变化是执行模型从“循环”变成了“图”。节点是函数，边是流转条件，整个 Agent 的执行路径被显式地画了出来。这个转变带来的好处不是“代码更优雅”这种虚的，而是三个实打实的能力。

第一，状态是显式声明的。你用StateGraph定义一个 State 的 schema，所有节点读写这个 State，框架负责在节点之间传递和合并。这意味着状态不再散落在闭包里，而是有一个明确的、可序列化的结构。这一点是后面能做 Checkpoint 的前提——你没法给一个散装的内存变量做持久化。

第二，中断点是一等公民。LangGraph 支持在节点执行前后插入 Checkpoint，也支持interrupt主动暂停等待人工输入。这意味着“中断”不再是一个异常情况，而是流程里的正常状态。恢复的时候，框架知道该从哪个 Checkpoint 继续。

第三，流式是内建的。astream和astream_events能把每个节点的输入输出、每个 token 的生成实时吐出来，前端只要订阅就行。这为后面接 AG-UI 省了大量胶水代码。

提示：LangChain 和 LangGraph 的区别经常被问到。简单说，LangChain 是“组件库”，提供模型、工具、检索器这些积木；LangGraph 是“编排层”，负责把这些积木按图的方式串起来并管理状态。做简单 Chain 用 LangChain 够了，做有状态、可中断、可恢复的 Agent，LangGraph 才是对的工具。

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

LangGraph 官方提供了多种 Checkpointer：内存版、SQLite 版、PostgreSQL 版。内存版只能用于测试，进程一挂全没。SQLite 版适合单机小项目，但并发写会锁库，多实例部署直接歇菜。

选 PostgreSQL 的理由很实在：它本来就是大多数团队已经在用的数据库，不用为了 Checkpoint 单独引入一套存储；它的事务和并发控制足够强，多个 Agent 实例同时写 Checkpoint 不会互相踩；它的 JSONB 类型天然适合存状态快照，查询和索引都方便。Redis 虽然快，但持久化保证弱一些，而且状态数据往往需要长期保留做审计，放 Redis 里成本反而更高。

这里有个容易忽略的点：Checkpoint 的数据量会随着对话轮次增长。一个跑了五十步的 Agent，状态快照可能有几百 KB。如果每步都全量存一份，存储会膨胀得很快。所以后面讲表结构的时候，我会说清楚哪些字段是增量、哪些是全量。

3. PostgreSQL Checkpoint 的表结构与落盘时机

3.1 Checkpoint 到底存了什么

很多人以为 Checkpoint 存的是“对话历史”，其实不止。一份完整的 Checkpoint 至少包含四部分：channel_values（各个状态通道的当前值，比如 messages、tool_results）、channel_versions（每个通道的版本号，用于判断哪些通道自上次以来变了）、pending_writes（还没提交的写入，用于崩溃恢复时重放）、metadata（步数、来源、时间戳等）。

LangGraph 的 PostgresSaver 会把这些拆成几张表。核心是checkpoints表存主快照，checkpoint_writes表存待提交的写入，checkpoint_blobs表存大的二进制或 JSON 数据。这种拆分的目的是让小的元数据查询快、大的状态数据按需加载，而不是每次恢复都把几百 KB 全读出来。

3.2 thread_id 是整个恢复机制的钥匙

恢复能不能成功，八成取决于thread_id设计得对不对。thread_id是 LangGraph 用来标识“一条独立会话线”的标识。同一个 thread_id 下的所有 Checkpoint 构成一条链，通过parent_id串起来，恢复时框架会沿着这条链找到最新的那个 Checkpoint。

我的经验是：thread_id 一定要和业务实体绑定，而不是随机生成。比如客服场景用user_id + session_id，订单处理场景用order_id。这样即使用户换了设备、清了缓存，只要业务标识还在，就能找回之前的进度。反过来，如果你用 UUID 随机生成 thread_id 又不落库到业务表，那用户一刷新就彻底找不回来了，Checkpoint 存了也白存。

# thread_id 的推荐构造方式 thread_id = f"order-{order_id}-user-{user_id}" config = {"configurable": {"thread_id": thread_id}}

3.3 落盘时机：不是每一步都值得存

LangGraph 默认在每个节点执行后落一次 Checkpoint。听起来很安全，但实际跑起来你会发现两个问题：一是写库频率太高，一个十步的任务写十次，QPS 上去了；二是很多中间步骤其实没有恢复价值，存了纯属浪费。

我的做法是分级落盘：关键节点（工具调用前后、人工审核点、外部 API 调用后）强制落盘，纯计算节点（格式化、拼接字符串）可以跳过。LangGraph 允许你在编译图的时候配置 checkpointer 的行为，也可以通过自定义节点来控制。判断标准很简单——这一步如果重跑会不会产生副作用或浪费大量资源，会的话就必须存。

注意：工具调用是最需要落盘的地方。因为工具往往有副作用（发消息、写数据库、调第三方接口），一旦执行了但没落盘，恢复时就会重复执行。LangGraph 的 pending_writes 机制就是为此设计的，它记录“这个工具已经执行但结果还没提交”，恢复时优先重放而不是重跑。

4. AG-UI 协议：把 Runtime 状态实时喂给前端

4.1 为什么需要 AG-UI 这一层

后端把状态存好了、能恢复了，但用户看不到过程，体验还是差。用户需要知道“Agent 现在在干嘛、卡在哪一步、还要多久”。这就需要一套前后端之间的事件协议，把 Runtime 的状态变化实时推过去。AG-UI 就是干这个的——它定义了一组标准事件类型，让前端能以统一的方式消费 Agent 的运行状态。

AG-UI 的核心事件大致分几类：生命周期事件（run started、run finished、run error）、文本事件（text message start、content、end）、工具事件（tool call start、args、end）、状态事件（state snapshot、state delta）。前端订阅这些事件，就能拼出一个完整的执行视图。

4.2 事件流设计的关键：状态快照 + 增量

这里有个设计上的坑我踩过：一开始我只推增量事件（delta），前端自己维护状态。结果一旦丢了一个事件，前端的状态就和后端对不上了，而且很难发现。后来改成周期性推全量快照 + 中间推增量，前端每次收到快照就整体覆盖，收到增量就合并，容错性好很多。

具体节奏是：run 开始时推一次全量快照，之后每个节点结束推增量，每 N 个节点或每 M 秒再推一次全量快照做校准。这样即使中间丢了几个增量，下一个快照也能把状态拉回来。

# 伪代码：AG-UI 事件推送的节奏控制 async def stream_with_snapshot(graph, input_state, config): step = 0 async for event in graph.astream_events(input_state, config): step += 1 yield to_agui_delta(event) # 每个事件推增量 if step % 5 == 0: snapshot = graph.get_state(config) yield to_agui_snapshot(snapshot) # 每5步推全量校准

4.3 恢复时前端怎么接上

恢复场景下，前端重新连接后第一件事是拉一次当前状态快照，把界面渲染成中断前的样子，然后再订阅后续事件。这里要注意的是，快照里要包含“当前正在执行哪一步”的信息，前端才能显示“正在恢复中”而不是“已完成”。

我遇到过一个细节问题：恢复时如果 Agent 正好卡在一个工具调用中间，快照里会显示这个工具是 pending 状态。前端需要能渲染这种“半完成”状态，否则用户会以为卡死了。AG-UI 的 tool call 事件里带 status 字段，前端根据 status 显示不同的 UI，这个设计很实用。

5. 完整实操：从零跑通一个可恢复的 Agent

5.1 环境准备与依赖安装

先把依赖装齐。核心是 langgraph、langgraph-checkpoint-postgres、psycopg，前端侧如果用 React 就装对应的 AG-UI 客户端库。

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

PostgreSQL 建议用 14 以上版本，因为要用到 JSONB 的一些新特性。本地开发用 Docker 起一个最省事：

docker run -d --name agent-pg -e POSTGRES_PASSWORD=secret -p 5432:5432 postgres:16

5.2 初始化 Checkpointer 与建表

LangGraph 的 PostgresSaver 提供了setup()方法自动建表，第一次跑的时候调一次就行。

from langgraph.checkpoint.postgres import PostgresSaver DB_URI = "postgresql://postgres:secret@localhost:5432/postgres" with PostgresSaver.from_conn_string(DB_URI) as checkpointer: checkpointer.setup() # 自动创建 checkpoints 等表

提示：setup()是幂等的，重复调用不会报错，但生产环境建议用迁移工具管理表结构，而不是每次启动都调 setup，避免权限问题和并发建表冲突。

5.3 定义 State 与构建图

State 的设计直接决定了 Checkpoint 存什么。我的原则是只放需要跨节点传递和恢复的数据，临时变量不要塞进去。

from typing import Annotated from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] order_id: str tool_results: dict step_count: int def call_model(state: AgentState): # 调模型，返回新的 messages ... def call_tool(state: AgentState): # 执行工具，返回 tool_results ... builder = StateGraph(AgentState) builder.add_node("model", call_model) builder.add_node("tool", call_tool) builder.add_edge(START, "model") builder.add_conditional_edges("model", should_continue, {"tool": "tool", "end": END}) builder.add_edge("tool", "model") graph = builder.compile(checkpointer=checkpointer)

5.4 跑一次带中断的执行

用同一个 thread_id 跑两次，第二次就能看到恢复效果。第一次跑到工具调用前主动中断，第二次从断点继续。

config = {"configurable": {"thread_id": "order-123-user-456"}} # 第一次执行，在 tool 节点前中断 for event in graph.stream( {"messages": [("user", "帮我查一下订单状态")], "order_id": "123"}, config, interrupt_before=["tool"], ): print(event) # 模拟进程重启后，从断点恢复 for event in graph.stream(None, config): print(event) # 会从 tool 节点继续

第二次传None作为输入，LangGraph 会自动从该 thread_id 最新的 Checkpoint 恢复。这就是可恢复 Runtime 的核心用法。

5.5 接上 AG-UI 推事件

把graph.stream换成graph.astream_events，把事件转成 AG-UI 格式推给前端。这一步的转换逻辑建议单独抽一个 adapter，方便测试和替换。

async def run_agent(input_state, config): async for event in graph.astream_events(input_state, config, version="v2"): agui_event = adapt_to_agui(event) if agui_event: yield agui_event

前端侧订阅这个流，按事件类型分发到不同的 UI 组件。文本事件渲染成气泡，工具事件渲染成卡片，状态事件更新顶部的进度条。

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

6.1 恢复后状态对不上怎么办

最常见的症状是：恢复之后 Agent 重复执行了已经做过的工具，或者丢失了部分上下文。原因通常是 thread_id 不一致，或者 Checkpoint 没落盘成功。

排查顺序：先确认两次执行用的是不是同一个 thread_id，可以直接查checkpoints表看有没有对应记录；再确认 checkpointer 是不是真的挂到了 graph 上（compile时有没有传）；最后看是不是在工具执行后、落盘前就崩了，这种情况 pending_writes 里会有记录，恢复时应该重放而不是重跑。

6.2 Checkpoint 表膨胀太快

跑了一段时间发现checkpoint_blobs表几个 G 了。这是因为每步都全量存了 messages，而 messages 是累积的。解决办法有两个：一是配置只保留最近 N 条消息在 State 里，历史消息归档到业务表；二是定期清理旧的 Checkpoint，只保留每个 thread 最新的几个。

-- 清理 30 天前的 Checkpoint（按需调整） DELETE FROM checkpoints WHERE created_at < NOW() - INTERVAL '30 days';

注意：清理前一定要确认这些 Checkpoint 对应的任务已经终态，否则删了就没法恢复了。建议加一个业务状态字段做联合判断，而不是单纯按时间删。

6.3 AG-UI 事件乱序或丢失

前端偶尔出现状态错乱，多半是事件乱序或丢失导致的。WebSocket 本身不保证顺序，SSE 好一些但也会断。我的做法是给每个事件加一个单调递增的 seq 号，前端发现 seq 跳号就主动拉一次全量快照校准。这个机制加上前面说的周期性快照，基本能覆盖绝大多数异常。

6.4 多实例部署时的并发问题

多个 Agent 实例同时处理同一个 thread_id 会出问题。LangGraph 的 PostgresSaver 有基本的并发控制，但业务上还是要避免同一 thread 被并发处理。我的做法是在业务层加一个分布式锁，用 thread_id 做 key，拿到锁才能执行。锁的粒度要细，别一把大锁锁住所有会话。

7. 几个我踩过之后才明白的经验

第一个经验是别急着上 Checkpoint，先把 State 设计干净。我一开始 State 里塞了一堆临时变量，结果 Checkpoint 又大又乱，恢复时还经常因为某个字段类型不对报错。后来把 State 精简到只留必要字段，问题少了一大半。State 是契约，设计的时候要像设计数据库表一样认真。

第二个经验是恢复逻辑一定要写测试。正常路径跑通不代表恢复路径没问题。我专门写了一套测试：模拟在任意节点中断，然后恢复，断言最终状态和不断的情况一致。这套测试帮我抓出了好几个隐藏的 bug，比如某个工具在恢复时被重复调用。

第三个经验是AG-UI 的事件设计要留扩展位。一开始我只推了文本和工具事件，后来想加“思考过程”展示，发现事件类型不够用，只能改协议。现在我会在事件里预留一个meta字段，放一些非核心但可能有用的信息，扩展的时候不用动主结构。

最后分享一个小技巧：调试恢复问题时，直接查数据库比看日志快得多。checkpoints表里的parent_id链能让你一眼看出执行路径，checkpoint_writes能告诉你哪些步骤没提交。把这两张表的查询语句存成 snippet，排查效率能翻倍。