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 负责让用户看见执行。三者职责清晰任何一层替换都不影响其他两层。组件核心职责替换成本我的选型理由LangGraph图编排与状态流转高Checkpoint 与 interrupt 原生支持PostgreSQL状态快照持久化中并发、事务、JSONB 三合一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: intadd_messages这个 reducer 是关键它保证多个节点往messages里追加消息时是合并而不是覆盖。我一开始没加 reducer结果并行分支各自写 messages后写的把先写的全冲掉了排查了半天才发现是状态合并策略的问题。注意retry_count这类计数器字段不要用 reducer它需要的是覆盖语义。只有列表类、需要累积的字段才加 reducer。搞反了会导致计数无限增长。另一个经验是状态字段要尽量扁平。我见过有人把状态设计成三层嵌套字典结果 Checkpoint 序列化后体积暴涨而且前端解析困难。扁平化之后不仅体积小了AG-UI 推送事件时也能直接映射字段。3.2 Checkpoint 落盘时机与性能权衡LangGraph 默认在每个节点执行后落盘一次。这个默认值在大多数场景下是对的但有两种情况需要调整第一种是节点内部有长耗时操作比如调用一个跑 30 秒的外部服务。如果这个节点中途失败默认配置下你要从节点开头重跑。解决办法是在节点内部手动调用checkpoint或者把长操作拆成多个子节点。第二种是高频小节点比如一个循环里连续跑十几个轻量节点。每次都落盘会让数据库压力陡增。我的做法是给这类节点配置checkpoint_duringFalse只在关键节点落盘。from langgraph.checkpoint.postgres import PostgresSaver with PostgresSaver.from_conn_string(DB_URI) as checkpointer: checkpointer.setup() graph builder.compile(checkpointercheckpointer)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 langgraph0.2.60 pip install langgraph-checkpoint-postgres2.0.9 pip install psycopg[binary,pool] pip install ag-ui-protocolPostgreSQL 我用的是 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(checkpointercheckpointer)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(resumeTrue), config # 同一个 thread_id )Command(resumeTrue)这个值会作为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, versionv2): yield format_ag_ui_event(event)astream_events会吐出节点开始、节点结束、token 流等事件format_ag_ui_event负责把它们映射成 AG-UI 的标准格式。前端拿到后节点开始就显示 loadingtoken 流就逐字渲染中断事件就弹确认框。提示astream_events的version参数一定要显式指定不同版本事件结构差异很大。我升级 LangGraph 后没改这个参数前端直接白屏查了半天。5. 常见问题与排查技巧实录5.1 Checkpoint 相关的高频故障现象根因解决方式恢复后从头执行thread_id 不一致前后端统一用业务 ID 作为 thread_id表不存在报错没调 setup()初始化时调用 checkpointer.setup()状态字段丢失没加 reducer累积型字段加 Annotated reducer写入超时状态体积过大大对象存引用不存内容并发冲突连接池太小调大 max_connections 和 pool_size5.2 我踩过的三个真实坑第一个坑是状态序列化失败。我在 State 里塞了一个自定义类的实例本地跑没问题因为内存 Checkpointer 直接存对象引用。换成 PostgreSQL 后直接报序列化错误。教训是State 里只放可 JSON 序列化的基础类型复杂对象要么转 dict要么存 ID。第二个坑是 interrupt 恢复后重复执行。原因是我的节点里有副作用操作写数据库恢复时从节点开头重跑副作用执行了两次。解决办法是把副作用操作设计成幂等的或者把副作用放在 interrupt 之后的独立节点里。第三个坑是 AG-UI 事件乱序。高并发下 SSE 推送的事件偶尔会乱序前端渲染出错。后来我在事件里加了单调递增的序号前端按序号排序后再渲染问题解决。5.3 性能调优的几个参数Checkpoint 的写入频率和状态体积是性能的两个关键变量。我的经验值状态体积控制在 100KB 以内超过就拆分关键节点落盘非关键节点用checkpoint_duringFalsePostgreSQL 的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% 以上唯一几次失败都是数据库连接抖动导致的加个重试就解决了。