开篇从一个让你抓狂的“断点”说起我现在的日常已经离不开 LangGraph但真正让我下定决心重构整个 Agent 项目架构的是一次印象极深的线上事故。当时我用一个最朴素的 while 循环在代码里让大模型反复调用工具、把结果塞回上下文再继续跑。业务方管这个叫“Agent”我也没多想直到某个凌晨任务跑到一半进程崩了。重启之后前面十几轮的工具调用记录、中间结果、已经写入数据库的半成品订单全部付之东流。那一刻我才意识到这个所谓的 Agent 运行时本质上就是一个会丢状态的内存循环。只要进程一死所有干活到一半的步骤全没了而且没有任何“接着干”的可能。那次事故之后我开始认真研究怎么把 Loop 从“手写的 while True”升级成一个具备恢复能力的可控 Runtime。整个改造的核心就是三样东西LangGraph 的状态图、PostgreSQL 的 Checkpoint 持久化以及 AG-UI 的协议化事件输出。如果你现在还在用普通 Python 代码硬写 Agent 循环或者刚接触 LangGraph 但不知道它比 LangChain 好在哪又或者已经被“human in the loop”这个词绕晕了——这篇文章就是给你准备的。我会完整拆解这套方案的选型逻辑、落地步骤和踩坑记录文末还有我私藏的排查清单希望能帮你跳过我在凌晨三点踩过的那些坑。1. 整体设计与思路拆解为什么不再手写 Loop1.1 手写 Loop 的真相它不是不能用是扛不住事先说结论手写循环在 demo 和单机脚本里完全没问题但一旦进入生产环境它会暴露三个致命弱点。第一个弱点是状态全在内存。所有中间变量、模型回复、工具返回值都堆在一个 Python dict 或 SQLite 临时表里进程一退出什么都没了。第二个弱点是没有“暂停”的概念。你想在某个节点停下来等用户确认只能靠 sleep 轮询或者硬生生拆成两段代码中间状态靠数据库手工同步。第三个弱点最隐蔽循环跑飞了没人知道。大模型偶尔会陷入死循环比如反复调用同一个工具十几次返回结果都是一样的错你的手写循环根本来不及拦截只会把账单和错误一起滚雪球。那 LangGraph 是怎么解决这三个问题的它把整个 Agent 流程建模成一个显式的状态图每个节点接收一个共享状态State处理后返回一个状态增量。这种“单状态流转 节点纯函数化”的设计天然就给持久化、恢复、人为干预留好了位置。你不需要自己设计消息队列或状态快照框架层面已经提供了。1.2 LangGraph 与 LangChain别再纠结它们谁代替谁网上经常有人问 langchain 和 langgraph 的区别。我说一句可能得罪人的话LangChain 的定位是工具集LangGraph 的定位是运行时。LangChain 提供了一大堆封装好的组件——模型调用、检索器、Tool 接口、Prompt 模板——但它的链式调用Chain模式更像是顺序脚本并不强调状态管理和循环控制。你可以在 LangGraph 的节点里放心使用 LangChain 的模型和工具这完全不冲突。从工程实践的角度看LangGraph 真正吸引我的是它对“状态”做了显式建模。你可以定义一个 TypedDict 作为全局状态也可以定义节点之间的消息传递协议。LangChain 的 AgentExecutor 本质上也是循环但它的循环是封装死的你对它的控制只有几个参数。而 LangGraph 允许你把循环画出来甚至允许你在某个条件边conditional edge上做任意分支。这个自由度在生产环境里非常值钱。1.3 为什么选 PostgreSQL Checkpoint 而不是 Redis 或本地文件LangGraph 官方提供了多种 Checkpoint 存储后端包括内存、SQLite、PostgreSQL、Redis 等。我在选型时的判断标准很简单生产环境要保证两点——持久化可靠以及并发恢复时不丢状态。Redis 当然快但如果不做 AOF 持久化宕机一样丢数据。SQLite 适合单机但多实例部署时锁竞争很头疼。PostgreSQL 在行业里几乎是“默认的可靠数据库”而且 LangGraph 的langgraph-checkpoint-postgres库写得很成熟基于它做状态恢复基本是开箱即用。我们团队的生产环境已经有一套自建的 PostgreSQL 集群所以选它还有一个现实理由不用额外引入基础设施。Checkpoint 本质上就是把图的执行快照存下来包括当前的节点位置、状态数据和待处理的任务队列。后端的区别只在于序列化格式和存储介质而这个存储介质的可靠性直接决定了整个 Runtime 能不能扛住宕机。接下来的内容我会围绕三个重点展开LangGraph 图的搭建、PostgreSQL Checkpoint 的接入与恢复、AG-UI 协议化输出的设计与事件流规范。这三块拼在一起才构成一个真正“可恢复 Runtime”的完整闭环缺一个都不行。2. 核心细节解析与实操要点LangGraph 图、Checkpoint 与 AG-UI 三件套2.1 LangGraph 状态图的核心概念不扯术语直接讲LangGraph 的核心就四个东西State、Node、Edge 和 Conditional Edge。State 是全局共享的数据结构我一般用 TypedDict 定义这样 IDE 提示和运行时校验都比较舒服。Node 是处理逻辑的单元它接收整个 State返回一个部分 State 的 dict框架会自动合并回去。Edge 把 Node 串起来表示“这一步执行完下一步固定走哪个节点”。Conditional Edge 则是一个函数输入是当前的 State输出是下一跳节点的名称。整个 Agent 主循环在我这里就是一条带条件边的图入口是agent节点生成一次模型回复如果回复里有工具调用就跳tools节点把工具结果写回 State再跳回agent节点继续生成如果没有工具调用条件边直接指向end节点。看一段伪代码应该比读十篇文档都有用from typing import TypedDict, Literal from langgraph.graph import StateGraph, END class AgentState(TypedDict): messages: list tool_results: dict def agent_node(state: AgentState) - dict: # 调用 LLM返回新的消息列表 ... def tools_node(state: AgentState) - dict: # 执行工具调用把结果写入 state ... def route_after_agent(state: AgentState) - Literal[tools, end]: if state[messages][-1].tool_calls: return tools return end g StateGraph(AgentState) g.add_node(agent, agent_node) g.add_node(tools, tools_node) g.add_edge(tools, agent) g.add_conditional_edges(agent, route_after_agent) g.set_entry_point(agent) g.add_edge(end, END) app g.compile()这段代码跑起来就是一个标准的 Agent Loop。但它和手写循环最大的区别是这个 Loop 的每一步都是可观测、可暂停、可恢复的。判断依据很简单——只要你的图结构足够清晰Checkpoint 就能在任意两步之间保存快照下次从快照位置继续跑。手写循环做不到这点因为在while里执行到哪一行并不构成一个可恢复的边界。提示节点函数里尽量不要直接修改传入的 state 对象而是返回一个部分更新的 dict。LangGraph 的底层合并逻辑更可靠也不容易串数据。2.2 PostgreSQL Checkpoint 接入不只是存个快照Checkpoint 这个词容易让人误解成“数据库备份”。实际上LangGraph 的 Checkpoint 是每个超步super-step的细粒度状态记录它保存的信息包括当前正在执行的节点、节点间传递的消息、状态数据的序列化结果、以及图执行的时间线。要接入 PostgreSQL先安装依赖pip install langgraph-checkpoint-postgres然后在代码里创建连接池和 Checkpoint 实例from langgraph.checkpoint.postgres import PostgresSaver from psycopg import Connection conn Connection.connect( postgresql://user:passwordlocalhost:5432/agent_runtime, autocommitTrue, ) checkpointer PostgresSaver(conn) # 首次使用需要初始化数据表 checkpointer.setup() app g.compile(checkpointercheckpointer)关键点在于compile(checkpointer...)这一步。一旦编译时传入了 CheckpointerLangGraph 就会自动在每个节点执行完之后把当前状态写入 PostgreSQL。你不需要在业务代码里手动埋点save 的时机是框架控制的。还有一个我必须提醒你的细节连接参数必须开启 autocommit。我第一次接入时没注意结果状态一直写不进数据库卡了整整一个下午排查最后发现是事务没有自动提交所有 Checkpoint 都回滚了。psycopg 的连接默认不是 autocommit 模式这在普通 SQL 操作里没事但 LangGraph 的 saver 会频繁写入快照如果不自动提交事务会越积越多最终要么锁死要么丢失。2.3 AG-UI 在 Runtime 中的定位把“内部状态”变成“可消费事件”AG-UI 你可能不熟它不是 LangGraph 的一部分而是一套定义Agent 与用户/前端之间事件流格式的协议规范。运行时内部的状态是图节点自己用的可前端界面却需要实时显示“哪个工具正在调用”“当前轮到谁说话”“是否在等待用户确认”。如果没有统一协议你就会陷入自己设计消息格式的泥潭前端一个字段一个字段跟你对烦不胜烦。AG-UI 的核心是提供了一套标准的事件类型比如工具调用开始、工具调用结束、Agent 消息增量、会话暂停等。我在 LangGraph 节点里会写一个回调函数把这些事件按 AG-UI 协议输出到前端from ag_ui.core import AGUISession def agent_node(state: AgentState) - dict: session AGUISession.from_state(state) result llm.invoke(session.messages) session.emit(agent_message, contentresult.content) return {messages: [result]}有了 AG-UI可恢复 Runtime 的价值才能真正显现不仅后端能恢复前端的展示状态也能同步恢复。你刷新页面之后可以从数据库中读取历史 Checkpoint 对应的事件流把界面恢复到崩溃前的样子。这一点在手写 Loop 里想都不敢想。2.4 超时、中断与边界条件生产环境的三个隐藏炸弹除了核心概念还有三个边界条件我希望你从一开始就想清楚不然上线后早晚出事。第一个是单步超时。OpenAI 这样的大模型接口偶尔会超时工具调用也可能卡死。我在每个节点上会包一层超时控制不让任何一步无限阻塞。LangGraph 本身没有原生的 timeout 参数但你可以用asyncio.timeout或者functools.partial结合concurrent.futures实现反正不能裸调 API。第二个是中断拦截。LangGraph 的interrupt机制是专门的“暂停”节点设计调用它会抛出中断让图执行暂停并等待外部输入。这个特性天然适合人工确认场景。同时你需要自己判断“循环何时该停”如果某个工具连续被调用超过 N 次就应该让条件边返回中断而不是傻乎乎地继续喂给模型。第三个是死循环防御。有时候模型会自己陷入循环即使没有工具调用它也会说一堆空话导致状态无限膨胀。我的做法是在状态里加一个step_count字段每经过一次 agent 节点就 1条件边检查如果超过最大步数直接跳到 END。这三个炸弹不解决你的 Runtime 即使能恢复也会在一个小时内被生产流量打垮。3. 实操过程与核心环节实现一步步搭出可恢复 Runtime3.1 定义状态结构与图结构我不建议一上来就写代码好的开始不是敲代码而是先梳理状态结构。我的做法是先画一张状态字段表明确哪些数据需要被持久化哪些只是在单次执行中临时用。字段类型是否持久化用途messageslist是所有对话和工具消息恢复时要完整回放tool_resultsdict是最近一次工具调用的结果用于继续推理step_countint是循环步数防止死循环pending_actionstr是当前等待的人类确认动作session_metadict否会话元信息如用户 ID不参与恢复表里“是否持久化”的判断标准很简单只要是恢复后续执行所需的数据就必须持久化只影响 UI 或日志的临时数据可以放在 State 之外。这里我踩过一个坑把大模型的原始 response 对象直接塞进了 messages但这个对象里有不可序列化的字段导致 PostgreSQL Checkpoint 写入时报错。所以我在消息入 State 之前都会转成统一的 dict 结构。3.2 LangGraph 图的完整代码实现可以直接抄下面这段代码基本是我生产环境下的最小版本去掉业务细节保留骨架。我会把关键步骤讲透。from typing import TypedDict, Literal from langgraph.graph import StateGraph, END from langgraph.checkpoint.postgres import PostgresSaver from psycopg import Connection class AgentState(TypedDict): messages: list tool_results: dict step_count: int pending_action: str MAX_STEPS 10 def agent_node(state: AgentState) - dict: # 1. 组织 prompt把历史消息交给 LLM # 2. 如果有工具调用就把调用信息附加到 messages # 3. step_count 1 return { messages: new_messages, step_count: state.get(step_count, 0) 1, } def tools_node(state: AgentState) - dict: # 批量执行 messages[-1].tool_calls results [] for call in state[messages][-1].tool_calls: results.append(execute_tool(call)) return {tool_results: {calls: results}} def route_after_agent(state: AgentState) - Literal[tools, end]: last_msg state[messages][-1] if state.get(step_count, 0) MAX_STEPS: return end if last_msg.get(tool_calls): return tools return end graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tools, tools_node) graph.add_edge(tools, agent) graph.add_conditional_edges(agent, route_after_agent) graph.set_entry_point(agent) graph.add_edge(end, END) # PostgreSQL 连接与 Checkpoint conn Connection.connect( postgresql://agent:secretlocalhost:5432/agent_runtime, autocommitTrue, ) saver PostgresSaver(conn) saver.setup() app graph.compile(checkpointersaver)执行时你需要给每次运行分配一个thread_id这个 ID 是关键中的关键。它标识的不只是会话更是一条独立的执行流。后续恢复就是拿着这个 thread_id 去查历史 Checkpoint。config {configurable: {thread_id: order-20250101-001}} result app.invoke( {messages: [{role: user, content: 帮我查一下订单 12345 的状态}]}, configconfig, )写入 PostgreSQL 的具体结构长什么样你可以去数据表里看一眼主要表有两张checkpoints存储每个超步的快照checkpoint_blobs存储序列化后的二进制内容。通过这两个表你能清楚看到整个图在什么时间执行到哪个节点这为问题定位提供了非常有力的证据。3.3 中断恢复的完整流程从崩溃到原地复活假设你的服务在工具调用节点执行过程中崩溃了重启后你想要恢复这个订单流程。代码逻辑非常简单依然拿着原来的 thread_id 调用invokeconfig {configurable: {thread_id: order-20250101-001}} app.invoke(None, configconfig)有没有发现我传给invoke的是None。这是 LangGraph 官方设计的行为模式——当传入 None 且存在 Checkpoint 时它会自动从上次停止的位置继续。如果 Checkpoint 中没有记录它会从 entry_point 重新开始。实际项目里重启后你会发现进程恢复的位置不一定从崩溃的节点中间而是从崩溃前最后一个完成写入的快照点继续。这是一个非常重要的工程认知Checkpoint 恢复不是“精确到字节”的还原而是“恢复到最近一个可靠边界”。那些“执行了一半但还没形成快照”的副作用比如已经调用了外部 API 但结果没写库可能在恢复后不会重新执行也可能会因外部 API 幂等性不足而重复执行。所以工具本身一定要设计成幂等的否则即使有 Checkpoint也无法保证端到端的一致性。3.4 人工确认节点怎么做interrupt 与 update_state 的正确姿势结合前面的两个核心机制我把“人工确认”这个动作做成了图中一个反复使用的模式。from langgraph.types import interrupt def human_review_node(state: AgentState) - dict: action state[pending_action] # 挂起图等待外部输入 user_decision interrupt({action: action}) return {pending_action: None, review_result: user_decision}当节点执行到interrupt时LangGraph 会抛出一个中断异常整个图的执行暂停状态已经持久化。你可以在另一个接口里拿到暂停信息然后给用户展示确认按钮。用户点“同意”后调用app.update_state( config, {review_result: approved}, as_nodehuman_review_node, )这一步会把新的状态注入到图中并解除中断。随后再次invoke即可继续。这个流程对手写 Loop 来说实现起来相当繁琐而用 LangGraph 的标准机制前后代码量不足 20 行。注意一个细节update_state的as_node参数很重要。如果你不指定它LangGraph 可能在你恢复后找不到要执行的节点位置。我遇到过一次把 as_node 写成空结果它直接从入口节点从头跑了一遍所有的历史消息全乱了。3.5 AG-UI 事件流如何和前端对接完整示例为了让前端能实时感知后端运行状态我会把 AG-UI 事件通过 WebSocket 推给前端。事件流的时序如下{type: AGENT_MESSAGE_START, message_id: m1} {type: TOOL_CALL_START, tool: order_query} {type: TOOL_CALL_END, tool: order_query, output: 订单状态: 已发货} {type: AGENT_MESSAGE_DELTA, content: 根据查询结果} {type: AGENT_MESSAGE_DELTA, content: 您的订单已发货}前端收到这些事件后可以做流式展示也可以在终端用户告知“刚刚卡住了”时直接根据后端返回的历史事件流恢复界面。这套协议的好处是标准、自解释。你不需要和下一位前端开发者吵架告诉对方“我给你发一个 modelOutput还有一个 toolOutput还有一个 pendingAction”而是让对方直接对照 AG-UI 的事件表取数据。在实际协作中这个协议帮我节省的沟通成本远超预期。4. 常见问题与排查技巧实录这些坑我替你踩过了4.1 问题速查表直接对照症状找方案症状可能原因排查方法解决方案checkpoint 一直写不进库psycopg 连接未开 autocommit查看 PostgreSQL 日志确认事务是否回滚连接参数增加 autocommitTrue恢复后从头执行而不是续跑thread_id 未正确传递或 update_state 缺 as_node检查 config 里 thread_id 是否与第一次执行一致统一线程 ID 生成规则绘制状态流图节点内部异常导致整个应用崩溃节点函数未捕获异常加 try-except记录节点上下文在节点外层包一层通用异常处理返回错误消息模型陷入死循环反复调用同一工具缺少最大步数限制查看 step_count 变化使用 Conditional Edge 检查 step_count超限跳 ENDAG-UI 事件流顺序混乱多线程写出事件顺序不一致查看 WebSocket 推送日志给事件加递增 seq 字段前端按 seq 排序4.2 排查实录一次“恢复后状态错乱”的完整分析有一次我遇到了一个很诡异的现象系统恢复后工具调用的返回结果里居然出现了上一次会话的数据。查了很久最后定位到根因是我的工具函数用了全局缓存缓存 key 只包含参数没包含会话 ID。第一次会话查询订单 111第二次恢复查询订单 222但工具读到了缓存里的 111 结果直接返回了。这个问题的启示是接入 Checkpoint 只是让“状态”可恢复但你的工具调用必须是纯粹的函数——相同的输入产生相同的输出或者至少完全上下文无关。凡是依赖隐式上下文的工具在恢复场景下一定会出问题。后来我把所有工具函数都改成显式传参会话 ID并把缓存策略改成按会话隔离问题就再没出现过。4.3 我的独家避坑经验提前说给你听第一不要滥用 checkpointer 的 setup 方法。它只应该在首次部署时执行一次如果在每次启动时都调用就会在高并发场景下产生 DDL 锁竞争。生产环境我通常用 ORM 迁移或手动 SQL 来建表。第二给 PostgreSQL 表建索引。LangGraph 官方建表语句里checkpoints 表和 checkpoint_blobs 表的查询条件都基于 thread_id所以一定要在 thread_id 字段上建立索引。我项目刚上线时没有加索引恢复一次要扫全表数据量一旦上去就明显感觉到查询变慢。第三不要把 Checkpoint 表和业务表混在一个连接池里。如果你让业务模块和 Checkpointer 共用同一个连接池一旦业务表有大事务Checkpoint 的写入可能被阻塞。我在生产环境给 LangGraph 单独开了一个连接池大小控制在 10 个连接以内效果很稳定。第四监控 Checkpoint 写入延迟。Checkpoint 是我们这套可恢复 Runtime 的“命根子”如果写入延迟突然飙高意味着系统在崩溃发生时可能丢掉更多状态。我给 Checkpoint 写入加了指标上报和告警阈值设置在 200ms超过就报警这会让你在状态真正丢失之前就介入处理。4.4 从 LangChain Agent 迁移到 LangGraph 时COM 层错误与 Runtime 报错速览社区里有不少朋友是从 langchain agent executor 迁移过来的常会碰到一个看起来很吓人的报错比如在 Windows 环境下偶发 runtime error 相关字样或者 webview2 runtime 相关的报错。这些报错其实和 LangGraph 本身没关系多半是系统环境问题比如缺少 Microsoft Edge WebView2 Runtime或者 Visual C 运行库。LangGraph 只是纯 Python 包它不依赖这些系统组件所以遇到这类报错时先检查你的部署环境别把锅甩给框架。有一个容易混淆的点langgraph 依赖 langchain-core但不强制依赖 langchain 全套。如果你在项目里只用 LangGraph 和模型接口哪怕完全没有 LangChain 的 Chain 也完全没问题。反过来如果你已经在用 LangChain 的各种工具集成LangGraph 也可以作为一个外层运行时把它们挂载进图里两者是互补关系。5. 可恢复 Runtime 的架构边界与扩展空间不吹不黑讲点实际价值5.1 它能解决什么问题不能解决什么问题做完这套改造最直接的价值有三个。第一进程崩溃后可以原地恢复不丢中间状态这是当初做这件事的初衷实际效果也是在一次真的宕机后验证的。当时某个订单处理任务执行到第三步服务重启后我拿着 thread_id 调了一次 invoke它从断点继续把订单状态更新完成了整个过程没有让用户重新发起请求。第二人工介入成为一等公民。审核、确认、多轮对话暂停都可以通过 interrupt 和 update_state 实现不再需要自己用临时表去设计中转状态。第三执行过程完全可观测。由于 Checkpoint 落库了每个超步的状态你可以像回放录像一样查看任何一个任务的执行轨迹。这在排查问题、审计合规场景下价值巨大。但也要说清楚局限性。首先它不会帮你自动解决外部系统的副作用一致性问题如果你调用了一个只能成功不能失败的下游系统Checkpoint 并不能让它支持事务回滚。其次如果你的节点函数不是纯函数有随机性或全局状态那么从 Checkpoint 恢复后行为可能和崩溃前不一致。最后LangGraph 本身不支持跨进程分布式图执行——多实例部署时同一线程 ID 最好只由一个实例处理否则两个实例同时从同一个 Checkpoint 恢复就不只是状态冲突而是直接逻辑错乱了。5.2 后续可以怎么扩展消息队列、子图拆分、流式输出目前这套 Runtime 已经能支撑单条执行流的完整生命周期。如果业务量起来我会考虑做三个扩展方向。方向一是接入消息队列做异步任务分发。LangGraph 原生是同步阻塞调用但如果你的任务本身适合放进任务队列就可以在图的入口节点把任务投递到队列由 worker 进程异步执行图结果通过回调回来。这样能把 Runtime 和 Web 服务生命周期解耦适合长时间运行的任务也能避开“同步 HTTP 接口等待 5 分钟”这种尴尬。方向二是子图拆分。如果单个图节点数超过十来个维护成本会急剧上升。LangGraph 支持把一部分节点封装成子图在主图的节点里调用它。子图也有自己的 State 和 Checkpoint这样每个功能域可以独立开发和测试调试体验会舒服很多。方向三是流式输出增强。目前 AG-UI 事件已经支持流式你可以进一步结合 SSE 或 WebSocket 把事件推送做得更平滑。官方提供的 stream 方法和 astream_events 接口就是为此设计的节点内部的 token 级输出都可以作为事件发出去配合前端打字机效果再合适不过。结尾最后一次迁移之后我再也不写裸 while 循环了从手写 Loop 到 LangGraph 可恢复 Runtime这次重构对我的项目来说不只是一个技术栈替换更像是一次思维方式的转变。现在每次写 Agent 流程我下意识的第一步永远是“这个流程的状态边界在哪”第二步是“如果这一步挂了从哪一步能爬起来”。带着这两个问题去设计任何一个 Agent 项目都不会太走样。如果你也打算做类似的改造我给的建议是千万别开头就追求完美。先用最简的图结构接入 PostgreSQL Checkpoint把thread_id统一管理起来再去加人工确认和 AG-UI 事件流。一步一步来每一步都能独立验证这样即使出问题你也能很快定位到是图结构的问题、存储的问题还是事件协议的问题。最后分享一个小技巧每次部署新版本前我会用一份历史 Checkpoint 做一次完整的恢复演练脚本会自动检查恢复后的状态数据和原状态数据是否一致。如果检查不过我宁愿延迟发布也不让一个不可恢复的新版本悄悄上线。这个习惯帮我避免了至少三次线上事故希望能对你有用。
