1. 为什么我要把手写 Loop 换成 LangGraph Runtime最早做 Agent 编排的时候我和大多数人一样直接写一个while True循环里面塞上模型调用、工具执行、状态判断跑通了就上线。简单场景下这套东西确实够用代码量少调试也直观。但一旦业务变复杂问题就全冒出来了用户中途关掉页面任务状态全丢工具执行到一半报错整个流程得从头再来想加个人工审核节点得在循环里塞一堆 if-else改到最后自己都看不懂。这套手写 Loop 的本质问题在于它把控制流和状态揉在了一起。循环本身是控制流变量是状态两者耦合在一个函数里既没法持久化也没法在任意节点暂停和恢复。而真实的生产环境里中断是常态而不是异常——网络抖动、用户主动取消、需要人工介入、服务重启任何一个环节都可能让一个跑了一半的任务变成孤儿。LangGraph 解决的正是这个问题。它把 Agent 的执行过程抽象成一张有向图节点是执行单元边是流转逻辑而整张图的状态由一套独立的 State 机制管理。更关键的是它内置了Checkpoint能力可以把每一步的状态快照持久化到外部存储里。我用 PostgreSQL 做 Checkpoint 的存储后端配合 AG-UI 做前端交互层最终跑通了一套真正意义上的可中断、可恢复 Runtime。这篇文章我会把整个改造过程拆开讲清楚为什么选 LangGraph 而不是继续手写、Checkpoint 到底存了什么、PostgreSQL 表结构怎么设计、AG-UI 怎么和 Runtime 对接、中断恢复时状态怎么对齐。如果你正在做 Agent 类产品或者被任务跑到一半挂了怎么办这个问题困扰过这篇内容应该能帮你少走不少弯路。2. LangGraph 与手写 Loop 的本质差异2.1 手写 Loop 的三个致命短板先把手写 Loop 的问题说透才能理解为什么要换。我总结下来主要是三点。第一是状态无处安放。手写循环里状态就是函数内的局部变量进程一挂全没了。你可能说可以存 Redis但存什么、什么时候存、怎么保证存的和内存里的一致这些都得自己实现而且极容易出 bug。我踩过的坑是工具执行完了但状态还没落盘这时候进程崩了恢复时工具会被重复执行产生副作用。第二是中断点无法定义。手写循环里中断意味着跳出循环但跳出之后从哪继续没有明确的边界。你只能靠一堆标志位来判断上次跑到哪了代码会迅速腐化成一团乱麻。第三是人工介入难以插入。Human-in-the-loop 是现在 Agent 产品的刚需比如让用户确认一个高风险操作再继续。手写循环里插入人工节点要么阻塞等待浪费资源要么把循环拆成两段状态传递又成问题。2.2 LangGraph 的图模型为什么更适合 RuntimeLangGraph 的核心抽象是StateGraph。你定义一张图图里每个节点是一个函数接收当前 State 返回 State 的更新。节点之间通过边连接边可以是固定的也可以是条件分支。这个模型的关键优势在于执行被显式地分解成了离散的步骤。每一步的输入和输出都是明确的 State这就为持久化提供了天然的切点。你不需要去猜现在跑到哪了因为图的结构本身就定义了所有可能的执行路径。再一个优势是状态与执行分离。State 是一个独立的数据结构由图的 schema 定义节点只负责读写 State不关心执行是怎么调度的。这种分离让 Checkpoint 变得可行——只要把 State 存下来就能在任何时候重建执行上下文。我实测下来从手写 Loop 迁移到 LangGraph代码量大概增加了 30%但可维护性和可观测性提升了一个量级。尤其是加了 Checkpoint 之后之前那些任务挂了要重跑的投诉基本消失了。2.3 什么时候该上 LangGraph什么时候别折腾不是所有场景都值得上 LangGraph。如果你的 Agent 就是一次性的问答没有多步工具调用没有中断需求那手写一个函数调用完全够用上 LangGraph 是过度设计。判断标准很简单任务是否需要跨请求保持状态。如果需要比如一个长流程的数据处理、一个多轮的工具编排、一个需要人工审核的工作流那 LangGraph 的 Checkpoint 就是刚需。如果不需要老老实实写函数别给自己找麻烦。还有一个判断维度是执行路径是否复杂。如果流程里有大量条件分支、循环、并行图模型能帮你把逻辑理清楚。如果就是一条直线图的优势体现不出来。3. Checkpoint 机制深度拆解到底存了什么3.1 Checkpoint 的数据结构很多人以为 Checkpoint 就是存个 State 的 JSON其实远不止。LangGraph 的 Checkpoint 包含几个关键部分channel_values当前所有 State channel 的值这是最核心的部分就是你的业务状态。channel_versions每个 channel 的版本号用于并发控制和增量更新。versions_seen记录每个节点看到了哪些 channel 的哪个版本这是实现只处理新数据的关键。pending_sends待处理的消息队列用于支持并行分支和消息传递。parent_config指向父 Checkpoint 的引用形成一条链支持时间旅行和回溯。这套结构的设计意图很明确让恢复时的状态重建是确定性的。给定一个 CheckpointRuntime 能精确知道每个节点应该看到什么数据从而保证恢复后的执行和中断前完全一致。我一开始没理解versions_seen的作用后来在调试一个并行分支的 bug 时才明白没有它恢复后节点会重复处理已经处理过的数据。这个字段是保证幂等性的关键。3.2 Checkpoint 的写入时机LangGraph 默认在每个节点执行完成后写入 Checkpoint。这个粒度是有讲究的。如果粒度太细比如每个工具调用都写那写入频率太高PostgreSQL 压力大。如果粒度太粗比如整个图跑完才写那中断恢复就失去了意义。节点级粒度是一个平衡点既保证了恢复时最多重跑一个节点又不会产生过多的写入。当然如果你的某个节点内部有很长的耗时操作可以考虑在节点内部手动触发 CheckpointLangGraph 提供了相应的 API。注意Checkpoint 的写入是同步的如果 PostgreSQL 响应慢会拖慢整个图的执行。生产环境建议给 Checkpoint 表做好索引并且考虑用连接池。3.3 为什么选 PostgreSQL 而不是内存或 RedisLangGraph 支持多种 Checkpoint 后端内存版适合开发调试Redis 版适合高性能场景但我最终选了 PostgreSQL原因有几个。持久性是第一位的。内存版进程一挂就没了Redis 虽然能持久化但配置复杂而且 Redis 的数据结构对复杂查询不友好。PostgreSQL 天然持久而且支持事务能保证 Checkpoint 写入的原子性。可查询性是第二位的。做运维的时候我经常需要查某个 thread 的历史 Checkpoint、某个时间段内失败的任务这些用 SQL 一句话就能搞定。Redis 里做同样的事得写一堆代码。成本是第三位的。PostgreSQL 几乎每个团队都有不用额外引入组件。Redis 虽然也常见但专门为 Checkpoint 维护一个 Redis 实例性价比不高。当然如果你的 QPS 特别高PostgreSQL 可能成为瓶颈那时候可以考虑 Redis 或者混合方案。但对大多数中小规模应用PostgreSQL 完全够用。4. PostgreSQL Checkpoint 表结构设计与实操4.1 官方表结构解析LangGraph 的 PostgreSQL Checkpointer 会创建几张表核心的是checkpoints和checkpoint_writes。checkpoints表存的是每个 Checkpoint 的元数据和主状态关键字段包括字段名类型作用thread_idTEXT会话/任务标识恢复时按这个查checkpoint_nsTEXT命名空间支持子图checkpoint_idTEXTCheckpoint 唯一标识parent_checkpoint_idTEXT父 Checkpoint形成链typeTEXT序列化类型checkpointJSONB主状态数据metadataJSONB元信息checkpoint_writes表存的是每个节点写入的具体数据用于增量恢复。这个设计的好处是主状态和增量写入分离恢复时先读主状态再按需读增量避免了一次性加载所有数据。4.2 建表与初始化实际部署时你不需要手动建表LangGraph 的 Checkpointer 会自动创建。但生产环境我建议手动建这样可以控制索引和分区。from langgraph.checkpoint.postgres import PostgresSaver DB_URI postgresql://user:passwordlocalhost:5432/agent_db with PostgresSaver.from_conn_string(DB_URI) as checkpointer: checkpointer.setup() # 自动建表setup()会创建所需的表结构。如果你用的是连接池需要传入 pool 而不是连接字符串。4.3 索引优化我踩过的性能坑默认建的表只有主键索引随着 Checkpoint 数量增长查询会越来越慢。我遇到过一次线上问题任务量上来后恢复一个 Checkpoint 要好几秒排查发现是全表扫描。手动加索引是必须的-- 按 thread_id 查询是最频繁的操作 CREATE INDEX idx_checkpoints_thread_id ON checkpoints(thread_id, checkpoint_id DESC); -- 按时间范围清理旧数据 CREATE INDEX idx_checkpoints_created_at ON checkpoints(created_at); -- checkpoint_writes 按 thread 和 checkpoint 查 CREATE INDEX idx_writes_thread_checkpoint ON checkpoint_writes(thread_id, checkpoint_id);加了索引之后恢复耗时从秒级降到了毫秒级。这个优化投入产出比极高强烈建议上线前就做好。4.4 数据清理策略Checkpoint 会无限增长必须定期清理。我的策略是已完成的任务保留 7 天的 Checkpoint用于问题追溯。失败的任务保留 30 天方便排查。正在进行的任务不清理。清理用定时任务跑 SQL 就行DELETE FROM checkpoints WHERE thread_id IN ( SELECT thread_id FROM checkpoints WHERE created_at NOW() - INTERVAL 7 days AND metadata-status completed );注意删除时要考虑外键关系先删 checkpoint_writes 再删 checkpoints或者用级联删除。我一开始没注意删主表时报了外键约束错误。5. AG-UI 对接 Runtime前端如何感知中断与恢复5.1 AG-UI 在架构中的位置AG-UI 是一个面向 Agent 应用的前端交互协议层它定义了前端和后端 Runtime 之间的通信规范。在我的架构里AG-UI 负责三件事把用户输入传给 Runtime、把 Runtime 的执行事件推给前端、处理中断和恢复的交互。为什么不用普通的 REST WebSocket因为 Agent 的执行是流式、多事件、可中断的普通 REST 表达不了这种语义。AG-UI 定义了一套事件模型包括run_started、node_started、node_finished、interrupt、resume等前端可以精确地知道当前执行到哪了。5.2 中断事件的传递当 Runtime 遇到需要人工介入的节点时会触发一个 interrupt。这个 interrupt 通过 AG-UI 传到前端前端弹出确认框用户操作后再通过 resume 接口把结果传回去。关键点在于interrupt 发生时Checkpoint 已经写入了。所以即使用户关掉页面下次打开时前端可以查询到有一个待处理的中断然后恢复上下文继续。# 后端定义中断节点 def human_review_node(state): # 触发中断等待前端输入 decision interrupt({ question: 是否批准这笔操作, context: state[operation_detail] }) return {approved: decision}前端收到 interrupt 事件后渲染确认 UI用户点击后调用 resume// 前端处理中断 onInterrupt((event) { showConfirmDialog(event.question, (result) { resumeRun({ threadId, decision: result }); }); });5.3 恢复时的状态对齐恢复时最容易出问题的是状态对齐。前端认为的状态和后端 Checkpoint 里的状态可能不一致比如用户在中断期间修改了某些数据。我的做法是恢复时以后端 Checkpoint 为准前端只负责传递用户的决策不负责传递状态。这样避免了状态冲突。具体实现上resume 接口只接收thread_id和用户的决策数据Runtime 根据 thread_id 加载最新的 Checkpoint把决策数据合并进去然后从断点继续执行。提示如果你的场景里前端确实需要修改状态建议在中断节点里显式地接收这些修改而不是让前端直接改 Checkpoint。这样状态的变更路径是清晰的便于审计。6. 完整实操从零跑通中断恢复6.1 环境准备与依赖安装先把依赖装好。核心是 langgraph、langgraph-checkpoint-postgres、psycopg。pip install langgraph langgraph-checkpoint-postgres psycopg[binary] fastapi uvicornPostgreSQL 用 Docker 起一个最方便docker run -d --name agent-pg \ -e POSTGRES_PASSWORDpassword \ -e POSTGRES_DBagent_db \ -p 5432:5432 postgres:166.2 定义 State 与图结构先定义 State schema。我用 TypedDict字段按业务需要来。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] task_status: str operation_detail: dict approved: booladd_messages是一个 reducer保证消息是追加而不是覆盖。这个细节很重要用错了会导致历史消息丢失。然后定义节点和边def plan_node(state): # 规划逻辑 return {task_status: planned} def execute_node(state): # 执行逻辑 return {task_status: executed, operation_detail: {...}} def review_node(state): decision interrupt({question: 批准, detail: state[operation_detail]}) return {approved: decision} builder StateGraph(AgentState) builder.add_node(plan, plan_node) builder.add_node(execute, execute_node) builder.add_node(review, review_node) builder.add_edge(START, plan) builder.add_edge(plan, execute) builder.add_edge(execute, review) builder.add_edge(review, END)6.3 接入 PostgreSQL Checkpointer编译图的时候传入 checkpointerfrom langgraph.checkpoint.postgres import PostgresSaver with PostgresSaver.from_conn_string(DB_URI) as checkpointer: checkpointer.setup() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: task-001}} result graph.invoke({messages: []}, config)thread_id是恢复的关键同一个 thread_id 的多次 invoke 会共享 Checkpoint 链。6.4 触发中断与恢复第一次 invoke 会在 review 节点中断返回一个包含 interrupt 信息的结果。此时 Checkpoint 已经写入。恢复时用同样的 thread_id 和Command对象from langgraph.types import Command # 恢复并传入用户决策 result graph.invoke( Command(resumeTrue), config{configurable: {thread_id: task-001}} )Command(resume...)会把值传给 interrupt 的返回值节点继续执行。6.5 用 FastAPI 暴露接口把上面封装成 HTTP 接口from fastapi import FastAPI app FastAPI() app.post(/run) def run_task(thread_id: str): config {configurable: {thread_id: thread_id}} result graph.invoke({messages: []}, config) return {status: interrupted, data: result} app.post(/resume) def resume_task(thread_id: str, decision: bool): config {configurable: {thread_id: thread_id}} result graph.invoke(Command(resumedecision), config) return {status: completed, data: result}前端调/run拿到中断信息用户决策后调/resume继续。整个链路就通了。7. 常见问题与排查技巧实录7.1 恢复后节点重复执行这是最常见的问题。原因通常是versions_seen没有正确更新或者 State 的 reducer 写错了。排查方法查checkpoints表里对应 thread 的channel_versions看恢复前后的版本号是否连续。如果版本号没变说明 Checkpoint 没写入成功。解决检查 checkpointer 是否正确传入 compile以及节点是否真的返回了 State 更新。如果节点返回空字典LangGraph 认为没有变化不会写 Checkpoint。7.2 PostgreSQL 连接耗尽高并发场景下每个请求都开一个连接很快就把连接数打满。解决用连接池。PostgresSaver支持传入ConnectionPoolfrom psycopg_pool import ConnectionPool pool ConnectionPool(DB_URI, max_size20) checkpointer PostgresSaver(pool)池大小根据你的并发量和 PostgreSQL 的 max_connections 来定一般 10-20 够用。7.3 中断后前端拿不到状态AG-UI 的事件是流式的如果前端没订阅对事件类型就会漏掉 interrupt。排查确认前端订阅了interrupt事件并且后端确实发出了这个事件。可以在后端加日志打印每次 emit 的事件类型。7.4 Checkpoint 数据过大如果 State 里存了大对象比如整个文档内容Checkpoint 会很大写入和读取都慢。解决State 里只存引用比如文档 ID实际内容存对象存储。需要时再查。这个原则叫State 轻量化是 LangGraph 的最佳实践之一。7.5 常见问题速查表问题现象可能原因排查方向恢复后从头执行Checkpoint 未写入查 checkpoints 表是否有记录节点重复执行versions_seen 异常检查 reducer 和 State 更新连接超时连接池未配置检查 pool 配置和 PG 连接数中断事件丢失前端未订阅检查 AG-UI 事件订阅写入慢缺索引或数据过大加索引、State 轻量化8. 我实际跑下来的一些体会整套东西跑通之后最大的感受是可恢复 Runtime 的价值不在于技术多炫而在于它把中断从一个异常变成了一个正常状态。以前用户关页面我们只能祈祷任务别挂现在可以坦然地说你随时回来任务还在。LangGraph 的图模型 PostgreSQL 的持久化 AG-UI 的交互层这三者组合起来构成了一个完整的可恢复 Runtime。每一层都有明确的职责替换其中任何一层都不影响其他层。这种解耦是我最满意的地方。如果让我给正在做类似改造的人一个建议那就是先把 State 设计清楚再动手写图。State 是整套机制的核心State 设计得好Checkpoint 和恢复都是水到渠成State 设计得烂后面全是补丁。我一开始图省事State 里塞了一堆临时变量结果恢复时各种状态不一致返工重设计了一遍才顺。另外一个小技巧开发阶段用内存 Checkpointer 快速迭代上线前再切 PostgreSQL。LangGraph 的 Checkpointer 接口是统一的切换成本几乎为零。这样能省下大量调试时间。