企业官网建设流程全解析

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

最早做 AI Agent 编排的时候，我和大多数人一样，直接写一个while True循环，把消息历史塞进列表，调一次模型，解析一次工具调用，执行完再塞回去。这套手写 Loop 在 Demo 阶段跑得飞快，代码不到一百行，逻辑一眼看穿。但只要业务稍微复杂一点，问题就全冒出来了：用户中途关掉页面，整个会话状态直接丢失；工具执行到一半报错，重试时不知道该从哪一步继续；多个分支并行跑的时候，状态互相覆盖，调试起来像在拆炸弹。

我踩过最典型的一个坑是：一个文档处理 Agent，跑到第三步调用外部接口时超时了，用户刷新页面后，Agent 从第一步重新开始，前面已经花掉的 token 和时间全部白费。那一刻我意识到，手写 Loop 缺的不是逻辑，而是"状态的可持久化"和"执行的可恢复"。这正是 LangGraph 的 Checkpoint 机制要解决的核心问题。

LangGraph 本质上是把 Agent 的执行过程建模成一张有向图，节点是执行单元，边是流转关系，而 Checkpoint 就是这张图在某个时刻的完整快照。配合 PostgreSQL 做持久化存储，再通过 AG-UI 把执行过程实时推给前端，就形成了一套"可中断、可恢复、可观测"的 Runtime。这篇文章我会把这套方案从设计思路到落地细节完整拆一遍，包括我实际踩过的坑和参数选择依据。适合已经写过基础 Agent、想把它做成生产级应用的同学，也适合正在纠结 LangChain 和 LangGraph 区别的人。

先说清楚一个常见误区：LangChain 和 LangGraph 不是替代关系。LangChain 提供的是模型、工具、检索这些组件，LangGraph 提供的是编排和状态管理。你可以把 LangChain 的组件塞进 LangGraph 的节点里用，但反过来不行。面试里经常问"langchain 和 langgraph 的区别"，标准答案就是：前者是工具箱，后者是流水线控制器。

2. 整体架构设计与方案选型思路

2.1 为什么是 LangGraph 而不是继续手写

手写 Loop 的本质问题是状态和执行耦合在一起。你的messages列表既是数据又是控制流，一旦要加"人工审批""条件分支""并行执行"这些能力，代码就会迅速膨胀成一团意大利面。LangGraph 把这些关注点拆开了：状态用 TypedDict 或 Pydantic 模型定义，控制流用图的边定义，持久化用 Checkpoint 定义。

我选 LangGraph 的三个硬理由：

• 状态显式化：所有需要在节点间传递的数据都必须在 State 里声明，不能靠闭包偷偷传。这强迫你把数据流想清楚，后期调试成本大幅下降。
• Checkpoint 原生支持：每个节点执行完自动落盘，中断后能从任意节点恢复，这是手写 Loop 要自己实现几百行还不一定对的功能。
• Human-in-the-loop 友好：interrupt机制让"暂停等人工确认"变成一行代码，而不是自己造一套信号系统。

2.2 PostgreSQL Checkpoint 的选型考量

LangGraph 官方支持多种 Checkpointer：内存版、SQLite 版、PostgreSQL 版。内存版重启即丢，只能用于测试；SQLite 适合单机小规模，但并发写入会锁表；PostgreSQL 是我在生产环境的默认选择，原因是它同时满足三个条件：支持高并发读写、支持 JSONB 存储复杂状态、支持事务保证快照一致性。

这里有个关键细节：Checkpoint 存的是序列化后的状态快照，不是增量。这意味着状态越大，每次落盘的开销越高。我实测下来，状态体积控制在 100KB 以内时，单次 Checkpoint 写入在 10ms 级别，完全可接受；超过 1MB 就要考虑拆分状态或者只存引用。所以设计 State 时，不要把大文件内容、完整文档正文塞进状态，存个 ID 或路径就够了。

2.3 AG-UI 在链路中的角色

AG-UI 是一套面向 Agent 前端的协议，负责把后端的执行事件（节点开始、节点结束、token 流、工具调用、中断信号）标准化地推给前端。没有它的时候，前端只能靠轮询或者自己定义 WebSocket 消息格式，每个项目都要重造一遍。有了 AG-UI，前端只需要实现一套事件处理器，就能对接任意符合协议的后端。

我把它放在架构里的位置是：LangGraph 负责"怎么执行"，PostgreSQL 负责"记住执行到哪"，AG-UI 负责"让用户看见执行"。三者职责清晰，任何一层替换都不影响其他两层。

3. 核心细节解析与实操要点

3.1 State 设计：决定成败的第一步

State 是整个 Runtime 的地基，设计错了后面全是坑。我用的是TypedDict加Annotated的方式，核心是给需要累积的字段指定 reducer。

from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] task_id: str current_step: str tool_results: dict retry_count: int

add_messages这个 reducer 是关键，它保证多个节点往messages里追加消息时是合并而不是覆盖。我一开始没加 reducer，结果并行分支各自写 messages，后写的把先写的全冲掉了，排查了半天才发现是状态合并策略的问题。

注意：retry_count这类计数器字段不要用 reducer，它需要的是覆盖语义。只有列表类、需要累积的字段才加 reducer。搞反了会导致计数无限增长。

另一个经验是状态字段要尽量扁平。我见过有人把状态设计成三层嵌套字典，结果 Checkpoint 序列化后体积暴涨，而且前端解析困难。扁平化之后，不仅体积小了，AG-UI 推送事件时也能直接映射字段。

3.2 Checkpoint 落盘时机与性能权衡

LangGraph 默认在每个节点执行后落盘一次。这个默认值在大多数场景下是对的，但有两种情况需要调整：

第一种是节点内部有长耗时操作，比如调用一个跑 30 秒的外部服务。如果这个节点中途失败，默认配置下你要从节点开头重跑。解决办法是在节点内部手动调用checkpoint或者把长操作拆成多个子节点。

第二种是高频小节点，比如一个循环里连续跑十几个轻量节点。每次都落盘会让数据库压力陡增。我的做法是给这类节点配置checkpoint_during=False，只在关键节点落盘。

from langgraph.checkpoint.postgres import PostgresSaver with PostgresSaver.from_conn_string(DB_URI) as checkpointer: checkpointer.setup() graph = builder.compile(checkpointer=checkpointer)

setup()这一步千万别漏，它会自动建表。我第一次跑的时候忘了调，报了个表不存在的错，还以为是连接串写错了。表结构里最关键的是checkpoints和checkpoint_writes两张表，前者存快照，后者存待写入的中间状态。

3.3 interrupt 实现真正的中断恢复

interrupt是这套方案里我最喜欢的功能。它的语义是：执行到某个节点时暂停，把当前状态存下来，等外部输入后再继续。

from langgraph.types import interrupt def approval_node(state: AgentState): decision = interrupt({"question": "是否继续执行下一步？"}) return {"current_step": "approved" if decision else "rejected"}

调用interrupt后，图的执行会抛出一个特殊信号，Checkpoint 记录下暂停位置。前端通过 AG-UI 收到中断事件，展示确认按钮，用户点击后前端把决定回传，后端用同一个thread_id恢复执行，图会从interrupt那一行继续往下走。

这里有个必须注意的点：恢复执行时用的thread_id必须和中断时一致，否则 LangGraph 会当成一个全新的会话，从起点重跑。我踩过这个坑，前端传参时把 thread_id 拼错了，结果用户点了"继续"却从头开始，体验极差。

4. 实操过程与核心环节实现

4.1 环境准备与依赖安装

先把依赖装齐。LangGraph 的版本迭代很快，我建议锁定版本，避免线上和本地行为不一致。

pip install langgraph==0.2.60 pip install langgraph-checkpoint-postgres==2.0.9 pip install psycopg[binary,pool] pip install ag-ui-protocol

PostgreSQL 我用的是 15 版本，JSONB 的性能和索引支持都比较成熟。建库的时候记得把max_connections调大一点，因为 Checkpointer 会维护连接池，默认 100 在高并发下可能不够。

CREATE DATABASE agent_runtime; CREATE USER agent_user WITH PASSWORD 'your_password'; GRANT ALL PRIVILEGES ON DATABASE agent_runtime TO agent_user;

4.2 构建可恢复的图

下面是一个完整的图定义，包含三个节点：规划、执行、审批。审批节点用interrupt实现暂停。

from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.postgres import PostgresSaver def build_graph(checkpointer): builder = StateGraph(AgentState) builder.add_node("plan", plan_node) builder.add_node("execute", execute_node) builder.add_node("approval", approval_node) builder.add_edge(START, "plan") builder.add_edge("plan", "execute") builder.add_edge("execute", "approval") builder.add_conditional_edges( "approval", lambda s: "execute" if s["current_step"] == "approved" else END ) return builder.compile(checkpointer=checkpointer)

add_conditional_edges是控制流的核心，它根据状态决定下一步走哪。注意条件函数的返回值必须是节点名或END，返回一个不存在的节点名会直接报错。

4.3 中断与恢复的完整调用链

第一次调用时，图会跑到approval节点暂停：

config = {"configurable": {"thread_id": "task-001"}} result = graph.invoke({"messages": [("user", "帮我处理这份文档")]}, config) # 此时 result 里会包含 __interrupt__ 字段

前端通过 AG-UI 收到中断事件后，用户确认，后端恢复：

from langgraph.types import Command resume_result = graph.invoke( Command(resume=True), config # 同一个 thread_id )

Command(resume=True)这个值会作为interrupt的返回值传回节点内部。我实测下来，从暂停到恢复，即使中间隔了几个小时，只要数据库还在，状态就能完整还原。

4.4 AG-UI 事件对接

AG-UI 的核心是把 LangGraph 的执行事件转成标准事件流。我用的是 SSE 方式推送：

async def stream_events(thread_id: str): config = {"configurable": {"thread_id": thread_id}} async for event in graph.astream_events(input_data, config, version="v2"): yield format_ag_ui_event(event)

astream_events会吐出节点开始、节点结束、token 流等事件，format_ag_ui_event负责把它们映射成 AG-UI 的标准格式。前端拿到后，节点开始就显示 loading，token 流就逐字渲染，中断事件就弹确认框。

提示：astream_events的version参数一定要显式指定，不同版本事件结构差异很大。我升级 LangGraph 后没改这个参数，前端直接白屏，查了半天。

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

5.1 Checkpoint 相关的高频故障

5.2 我踩过的三个真实坑

第一个坑是状态序列化失败。我在 State 里塞了一个自定义类的实例，本地跑没问题，因为内存 Checkpointer 直接存对象引用。换成 PostgreSQL 后直接报序列化错误。教训是：State 里只放可 JSON 序列化的基础类型，复杂对象要么转 dict，要么存 ID。

第二个坑是 interrupt 恢复后重复执行。原因是我的节点里有副作用操作（写数据库），恢复时从节点开头重跑，副作用执行了两次。解决办法是把副作用操作设计成幂等的，或者把副作用放在 interrupt 之后的独立节点里。

第三个坑是 AG-UI 事件乱序。高并发下 SSE 推送的事件偶尔会乱序，前端渲染出错。后来我在事件里加了单调递增的序号，前端按序号排序后再渲染，问题解决。

5.3 性能调优的几个参数

Checkpoint 的写入频率和状态体积是性能的两个关键变量。我的经验值：

• 状态体积控制在 100KB 以内，超过就拆分
• 关键节点落盘，非关键节点用checkpoint_during=False
• PostgreSQL 的shared_buffers调到内存的 25%
• 连接池pool_size设为并发数的 1.5 倍

实测下来，一个中等复杂度的 Agent，单次完整执行（含 5 个节点、2 次 Checkpoint）的额外开销在 50ms 以内，相比手写 Loop 几乎可以忽略，但换来的是完整的中断恢复能力。

6. 从 Demo 到生产的几个关键决策

把 Demo 跑通只是第一步，真正上线还要考虑几件事。第一是状态清理策略，Checkpoint 会无限增长，我一般保留最近 30 天的记录，用定时任务清理。第二是 thread_id 的生成规则，我用的是业务 ID 加时间戳的组合，既保证唯一又方便排查。第三是降级方案，PostgreSQL 挂了怎么办？我的做法是降级到内存 Checkpointer，功能受限但不至于整个服务不可用。

关于 LangGraph 和 LangChain 的面试题，我补充一个实战视角的答案：LangChain 的 Chain 是线性的、无状态的，适合简单的一次性调用；LangGraph 的 Graph 是有环的、有状态的，适合需要循环、分支、中断的复杂流程。判断标准很简单——如果你的流程需要"记住上次执行到哪"，就该用 LangGraph。

最后分享一个我在实际使用中的小技巧：调试 Checkpoint 问题时，直接查数据库比看日志快得多。SELECT * FROM checkpoints WHERE thread_id = 'xxx' ORDER BY created_at DESC一眼就能看出状态是在哪一步、以什么内容落盘的，比在代码里打断点高效太多。这套方案我跑了半年多，中断恢复的成功率稳定在 99% 以上，唯一几次失败都是数据库连接抖动导致的，加个重试就解决了。