1. 为什么我要从手写 Loop 切换到 LangGraph Runtime最早做 AI Agent 编排的时候我和很多人一样第一反应就是写一个while True循环调模型、解析工具调用、执行工具、把结果塞回消息列表、再调模型直到模型不再请求工具为止。这个方案在 demo 阶段非常爽几十行代码就能跑通一个能查天气、能算数、能读文件的助手。但只要业务稍微复杂一点问题就全冒出来了。最典型的就是中断恢复。用户跟 Agent 聊到一半Agent 正在等一个耗时工具比如查数据库、调外部接口返回结果这时候服务重启了、进程被 kill 了、或者用户直接关掉页面第二天再回来手写 Loop 里的所有状态——消息历史、当前执行到哪一步、待执行的工具调用——全部丢失。你只能让用户从头再说一遍体验直接崩掉。第二个问题是Human-in-the-loop。有些操作比如转账、发邮件、删数据必须等人工确认才能继续手写 Loop 里你得自己设计一套暂停/恢复机制写着写着就变成了一坨状态机。LangGraph 解决的正是这类问题。它把 Agent 的执行过程建模成一张图Graph节点是执行单元边是流转逻辑而整个图的执行状态可以通过Checkpoint持久化到数据库。配合PostgreSQL Checkpoint每一次状态变更都能落库进程重启后从最近的 checkpoint 恢复中断恢复就变成了一个读档操作。再往上AG-UI负责把 Agent 的执行过程实时推给前端让用户看到Agent 正在思考正在调用工具等待确认这些状态而不是干等一个转圈。这篇文章我会完整讲清楚三件事LangGraph 的图模型和 Runtime 到底怎么运转、PostgreSQL Checkpoint 怎么配置才能真正做到可恢复、AG-UI 怎么把中断和恢复的过程可视化。适合已经写过手写 Loop、想往生产级 Agent 迁移的开发者也适合刚接触 LangGraph 想搞懂它和 LangChain 区别的人。下面所有代码和配置都是我在实际项目里跑通过的参数选择我会解释清楚为什么这么定。2. LangGraph 与 LangChain 的区别以及 Runtime 到底指什么2.1 一句话说清 LangChain 和 LangGraph 的定位差异很多人搜langchain和langgraph的区别其实核心就一句话LangChain 是组件库LangGraph 是编排运行时。LangChain 给你的是 LLM 封装、Prompt 模板、Retriever、Tool 这些积木你拿这些积木自己搭流程LangGraph 给你的是一个带状态、可持久化、可中断恢复的执行引擎你把流程描述成图它负责跑。打个比方LangChain 像是一箱乐高零件LangGraph 像是一台带存档功能的游戏机。你用 LangChain 的零件拼出一个机器人但机器人怎么跑、跑到一半断电了怎么办LangChain 不管LangGraph 则是把怎么跑这件事接管了还顺便帮你把存档做了。面试里经常被问的一个点是LangGraph 能不能不用 LangChain答案是能。LangGraph 的节点本质上就是普通函数输入输出是 state你完全可以在节点里手写 HTTP 请求调模型不依赖 LangChain 的任何封装。但实际项目里大家还是会用 LangChain 的ChatModel和Tool因为省事。所以两者是互补关系不是替代关系。2.2 Runtime 在 LangGraph 里具体指什么Runtime这个词在热词里出现频率很高但含义很杂。在 LangGraph 语境下Runtime 指的是驱动图执行的那套机制包括State 管理每个节点读写共享的 stateLangGraph 负责合并和传递。调度决定下一个执行哪个节点处理条件边、并行分支。Checkpoint 写入每个 super-step 结束后把 state 快照持久化。中断与恢复遇到interrupt或异常时暂停恢复时从 checkpoint 重建现场。这套 Runtime 是 LangGraph 区别于自己写个 for 循环的根本。手写 Loop 里state 就是一个 Python 变量进程没了就没了LangGraph 里state 是 checkpoint 里的一条记录进程没了还能读回来。2.3 图模型的核心概念速览在动手之前先把几个概念对齐不然后面配置会懵概念含义类比State图执行过程中共享的数据结构游戏存档里的角色数据Node一个执行单元接收 state 返回更新游戏里的一个关卡Edge节点之间的流转关卡之间的传送门Conditional Edge根据 state 决定走哪条边根据选择走不同剧情Checkpoint某个时刻 state 的完整快照存档点Thread一条独立的执行会话一个存档槽位interrupt主动暂停等待外部输入游戏弹出对话框等你选理解这张表后面所有配置你都能对上号。特别是Thread这个概念它是恢复的关键——同一个 thread_id 下的所有 checkpoint 构成一条完整的时间线恢复时你指定 thread_idRuntime 就能找到最近的存档。3. 用 PostgreSQL Checkpoint 实现真正可恢复的 Runtime3.1 为什么选 PostgreSQL 而不是内存或 SQLiteLangGraph 官方提供了多种 CheckpointerMemorySaver内存、SqliteSaver、PostgresSaver。选型逻辑很直接MemorySaver进程重启即丢只能用于本地调试生产绝对不能用。SqliteSaver单文件适合单机小项目但并发写入会锁库多实例部署直接歇菜。PostgresSaver支持并发、支持多实例共享、有成熟备份机制生产首选。我踩过的坑是早期用 SqliteSaver 做测试单进程跑得好好的一上多 worker比如 gunicorn 起 4 个进程就开始报database is locked。因为每个 worker 都想写同一个 sqlite 文件写锁互斥。换成 PostgreSQL 后这个问题彻底消失因为 PG 的行级锁和 MVCC 能扛住并发写。3.2 PostgreSQL Checkpoint 的建表与初始化PostgresSaver 需要几张表来存 checkpoint、写入记录和迁移版本。官方提供了setup()方法自动建表但生产环境我建议手动执行 SQL方便审计和权限控制。核心表结构大致是-- checkpoint 主表存每个存档点的元信息和 state CREATE TABLE checkpoints ( thread_id TEXT NOT NULL, checkpoint_ns TEXT NOT NULL DEFAULT , checkpoint_id TEXT NOT NULL, parent_checkpoint_id TEXT, type TEXT, checkpoint JSONB NOT NULL, metadata JSONB NOT NULL DEFAULT {}, created_at TIMESTAMPTZ DEFAULT NOW(), PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id) ); -- 待写入记录表存节点执行过程中产生的中间写入 CREATE TABLE checkpoint_writes ( thread_id TEXT NOT NULL, checkpoint_ns TEXT NOT NULL DEFAULT , checkpoint_id TEXT NOT NULL, task_id TEXT NOT NULL, idx INTEGER NOT NULL, channel TEXT NOT NULL, type TEXT, value JSONB, PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) );初始化代码from langgraph.checkpoint.postgres import PostgresSaver DB_URI postgresql://user:passwordlocalhost:5432/langgraph_db with PostgresSaver.from_conn_string(DB_URI) as checkpointer: checkpointer.setup() # 自动建表首次运行执行一次注意setup()只需要在首次部署时跑一次重复跑不会报错但也没必要。生产环境建议把建表 SQL 抽出来走数据库迁移流程别让应用启动时自动建表权限收不回来。3.3 把 Checkpointer 挂到图上有了 checkpointer编译图的时候传进去就行from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.postgres import PostgresSaver builder StateGraph(AgentState) builder.add_node(agent, call_model) builder.add_node(tools, tool_node) builder.add_edge(START, agent) builder.add_conditional_edges(agent, should_continue, {tools: tools, END: END}) builder.add_edge(tools, agent) with PostgresSaver.from_conn_string(DB_URI) as checkpointer: graph builder.compile(checkpointercheckpointer)关键点checkpointer 的生命周期要覆盖整个图执行过程。用with上下文管理连接别在函数里临时创建又关闭否则 checkpoint 写不进去。我见过有人把 checkpointer 建在节点函数内部结果每次执行都新建连接性能差还容易连接泄漏。3.4 thread_id 的设计与恢复逻辑恢复的核心是thread_id。调用图的时候通过 config 传入config {configurable: {thread_id: user-123-session-456}} result graph.invoke({messages: [HumanMessage(content帮我查下订单)]}, config)同一个 thread_id 再次 invokeLangGraph 会自动从最近的 checkpoint 恢复 state而不是从头开始。这就是中断恢复的底层机制——你不需要手动读存档Runtime 帮你做了。thread_id 的设计有几个实践建议不要用随机 UUID那样每次都是新会话恢复不了。要用能标识同一段对话的稳定 ID比如user_id session_id。区分会话边界用户点新对话时换一个 thread_id否则历史会一直累积。注意长度thread_id 会进数据库主键别塞太长的字符串控制在 128 字符内比较稳妥。3.5 中断恢复的完整验证流程光配置不够得验证真的能恢复。我的验证方法是启动图传入 thread_id让它执行到某个耗时节点。在节点里time.sleep(30)模拟长任务执行到一半直接kill -9进程。重启进程用同一个 thread_id 再次 invoke。观察是否从断点继续而不是重头再来。实测下来只要 checkpointer 配置正确重启后 LangGraph 会从最后一个完成的 super-step 恢复。注意是最后一个完成的 super-step如果进程是在某个节点执行中途被 kill 的那个节点会重新执行——所以节点逻辑要设计成幂等的这是很多人忽略的点。4. 用 interrupt 实现 Human-in-the-loop 与 AG-UI 可视化4.1 interrupt 的工作机制LangGraph 的interrupt是 Human-in-the-loop 的核心。在节点里调用from langgraph.types import interrupt def approval_node(state: AgentState): decision interrupt({ question: 是否确认执行转账, amount: state[amount], to: state[payee] }) if decision approve: return {approved: True} return {approved: False}执行到这里图会暂停state 被 checkpoint 保存interrupt的返回值会作为 invoke 的结果抛给调用方。调用方拿到这个中断信号后展示给用户用户确认后再用同一个 thread_id 恢复from langgraph.types import Command # 恢复并传入用户决策 graph.invoke(Command(resumeapprove), config)这里的关键是中断期间进程可以完全退出。因为 state 已经在 PostgreSQL 里了用户可能隔了一天才点确认这期间服务重启多少次都无所谓。这正是手写 Loop 做不到的。4.2 AG-UI 在中断恢复里的角色AG-UI 是一套面向 Agent 前端的协议它定义了 Agent 执行过程中的事件流怎么推给 UI。在中断恢复场景里AG-UI 负责把这几类事件传出去RUN_STARTED一次执行开始TEXT_MESSAGE_CONTENT模型流式输出TOOL_CALL_START/TOOL_CALL_END工具调用开始和结束STATE_SNAPSHOTstate 快照前端可以据此渲染RUN_FINISHED/RUN_ERROR执行结束或出错当图遇到interrupt暂停时AG-UI 会推一个特殊事件告诉前端Agent 在等你确认。前端渲染出确认按钮用户点击后通过后端把Command(resume...)发回去图继续执行。整个过程用户看到的是连续的而不是卡住了。4.3 后端把 LangGraph 事件桥接到 AG-UILangGraph 执行时可以流式产出事件用astream_events拿到再转成 AG-UI 格式推给前端。核心桥接逻辑async def stream_to_agui(graph, input_data, config): async for event in graph.astream_events(input_data, config, versionv2): kind event[event] if kind on_chat_model_stream: chunk event[data][chunk] if chunk.content: yield {type: TEXT_MESSAGE_CONTENT, delta: chunk.content} elif kind on_tool_start: yield {type: TOOL_CALL_START, name: event[name]} elif kind on_tool_end: yield {type: TOOL_CALL_END, name: event[name]} elif kind on_chain_end and event[name] LangGraph: yield {type: RUN_FINISHED}前端用 SSE 或 WebSocket 接收这些事件逐条渲染。中断事件需要单独处理因为interrupt不是标准的 astream_events 事件通常是在 invoke 返回结果里检查__interrupt__字段result await graph.ainvoke(input_data, config) if __interrupt__ in result: yield {type: INTERRUPT, payload: result[__interrupt__]}4.4 前端恢复交互的最小实现前端拿到 INTERRUPT 事件后渲染确认 UI用户点击后发一个恢复请求async function resumeRun(threadId, decision) { const res await fetch(/api/resume, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ thread_id: threadId, resume: decision }) }); // 继续消费 SSE 流 consumeStream(res.body); }后端/api/resume收到后用同一个 thread_id 和Command(resumedecision)重新 invoke 图继续把事件流推给前端。这样用户视角就是我点了确认Agent 接着干活。提示恢复时一定要用原来的 thread_id否则 LangGraph 会当成新会话从 START 重新跑中断就白做了。这是新手最容易犯的错。5. 实操中踩过的坑与排查速查表5.1 Checkpoint 写不进去的几种原因我遇到过 checkpoint 表一直是空的排查下来无非这几种checkpointer 没传进 compilebuilder.compile()忘了传checkpointer图照跑但没存档。连接提前关闭用with包了 checkpointer但图执行在 with 外面连接已关。thread_id 没传config 里没有configurable.thread_idLangGraph 不知道往哪个 thread 写。事务没提交自定义连接时忘了 commitPG 里看不到数据。5.2 恢复后状态不对的排查有时候恢复了但 state 是旧的或者缺字段。常见原因节点不幂等被 kill 的节点重新执行如果它有副作用比如发了邮件会重复执行。state 合并冲突多个节点并行写同一个字段LangGraph 的 reducer 没定义好后写的覆盖先写的。checkpoint 版本不匹配升级 LangGraph 后 state schema 变了旧 checkpoint 反序列化失败。5.3 常见问题速查表现象可能原因解决方向进程重启后从头开始thread_id 变了或没传检查 config 里的 thread_id 是否稳定checkpoint 表为空checkpointer 未挂载或连接关闭确认 compile 传了 checkpointer 且连接存活中断后恢复报错resume 值类型不匹配确认 Command(resume...) 的值和 interrupt 期望一致多实例下状态错乱用了 MemorySaver 或 SqliteSaver换 PostgresSaver节点重复执行有副作用节点不幂等加幂等键或把副作用移到确认后AG-UI 事件断流SSE 连接超时加心跳或改用 WebSocket5.4 幂等设计的实操心得被 kill 的节点会重跑这是 checkpoint 机制的固有行为不是 bug。我的做法是给每个有副作用的操作加一个幂等键比如用thread_id node_name step作为唯一标识执行前先查这个键有没有记录有就跳过。这样即使节点重跑副作用也只发生一次。对于转账这类操作幂等键直接存到业务表里配合数据库唯一约束双保险。6. 从手写 Loop 迁移到 LangGraph 的取舍建议6.1 什么场景值得迁移不是所有项目都值得上 LangGraph。我的判断标准是需要中断恢复用户会话可能跨进程、跨天必须持久化 state。需要 Human-in-the-loop有需要人工确认的敏感操作。流程复杂多分支、并行、循环手写 Loop 已经难以维护。多实例部署需要多个 worker 共享会话状态。如果只是单轮问答、无状态、单实例手写 Loop 反而更轻。别为了用框架而用框架。6.2 迁移时的渐进式路径我的建议是分三步走别一次性重写先把手写 Loop 的 state 抽出来定义成 LangGraph 的 State schema这一步不改逻辑只是把散落的变量收拢。把 Loop 里的每个阶段拆成节点用边连起来先跑通无 checkpoint 的版本验证逻辑等价。挂上 PostgresSaver加 thread_id验证恢复最后再接 interrupt 和 AG-UI。这样每一步都可回退出问题能快速定位是哪一层引入的。6.3 性能与成本的权衡LangGraph 每个 super-step 都写一次 checkpoint写库是有开销的。高频短任务场景下checkpoint 写入可能成为瓶颈。我的优化手段合并小节点把几个轻量操作合成一个节点减少 checkpoint 次数。异步写入PostgresSaver 支持异步用AsyncPostgresSaver配合 async 图。定期清理旧 checkpoint按 thread_id 保留最近 N 个老的归档或删除避免表无限膨胀。清理 SQL 大致是DELETE FROM checkpoints WHERE thread_id %s AND checkpoint_id NOT IN ( SELECT checkpoint_id FROM checkpoints WHERE thread_id %s ORDER BY created_at DESC LIMIT 20 );保留 20 个存档对绝大多数会话足够了既不影响恢复又能控制表大小。6.4 我个人的几点体会用下来最大的感受是LangGraph 把状态管理这件事从业务代码里彻底剥离了。以前写 Agent一半代码在处理现在到哪一步了上次的结果存哪了现在这些交给 Runtime业务代码只关心这一步该干什么。中断恢复从需要专门设计的功能变成了默认就有的能力这个心智负担的降低是实打实的。另一个体会是 AG-UI 的价值被低估了。很多人只关注后端能不能恢复但用户感知不到恢复过程体验还是断的。把 interrupt、恢复、工具调用这些事件实时推给前端用户才知道 Agent 在干什么、为什么停下来、点了确认之后又干了什么。这套可视化做起来不难但对体验的提升非常明显。最后提醒一句checkpoint 里存的是完整 state如果 state 里有敏感数据用户隐私、密钥数据库的访问权限和加密一定要做好。我见过有人把 API key 塞进 state结果 checkpoint 表里明文躺着这是很危险的做法。敏感信息要么不进 state要么进之前先脱敏。