1. 为什么手写 Loop 终究会撞上恢复这堵墙我最早做对话式 Agent 的时候和大多数人一样写了一个while True循环读用户输入、拼 prompt、调模型、解析工具调用、执行工具、把结果塞回消息列表、再调模型直到模型不再请求工具为止。这套东西跑 demo 特别爽几十行代码就能让模型查天气、算数学、读文件。但只要业务稍微真实一点问题就全冒出来了。最典型的一个场景用户让 Agent 帮忙做一份包含十几个步骤的数据处理任务跑到第七步的时候用户突然关掉了浏览器或者网络抖了一下或者后端服务重启了。这时候你那个while True循环里的所有状态——已经执行到哪一步、中间产出了什么、下一步该干嘛——全部随着进程一起蒸发了。用户回来一看得从头再来。这种体验在真实产品里是致命的。手写 Loop 的第二个死穴是人机协同Human-in-the-loop。很多任务需要中途让人确认一下比如我要给这个客户发邮件了你确认下内容或者这个数据库删除操作要不要执行。在裸循环里你只能靠input()阻塞或者自己造一套状态机把等待确认这个状态存到某个地方。一旦要支持多个用户并发、要支持确认后从断点继续你就会发现自己其实是在重新发明一个工作流引擎。第三个问题是可观测性。手写循环里每一步的输入输出都散落在各种变量和日志里你很难回答这个任务到底卡在哪一步了上一步的模型输出是什么工具调用参数对不对这类问题。排查线上问题时你只能靠翻日志猜。所以当我第一次接触 LangGraph 的时候最打动我的不是它能画流程图而是它把Checkpoint检查点做成了框架的一等公民。它本质上是在告诉你Agent 的执行状态不应该活在内存里而应该被持久化下来随时可以暂停、恢复、回放、甚至分叉。这篇文章我就想把这套东西讲透——从 LangGraph 的图结构到 PostgreSQL 做 Checkpoint 存储再到用 AG-UI 把中断恢复的能力暴露到前端跑通一条完整的链路。适合读这篇的人已经用 LangChain 或裸 API 写过 Agent、现在被状态管理折磨、想上 LangGraph 但被 checkpoint 概念绕晕的开发者。如果你还没写过 Agent建议先补一下基础不然有些坑你体会不到。2. LangGraph 的图模型到底和 LangChain 差在哪2.1 从链到图不是换个名字那么简单很多人问 LangChain 和 LangGraph 的区别网上答案一大堆但大多停留在一个是链一个是图这种废话层面。我用一句话概括我的理解LangChain 解决的是怎么把组件串起来LangGraph 解决的是怎么让一个有状态的流程可控地跑起来。LangChain 的 Chain 本质是线性的、无状态的或者说状态是隐式传递的。你写一个LLMChain它的输入输出是明确的跑完就结束了。就算你用AgentExecutor它内部那个循环也是黑盒的你没法在中途插手也没法把中间状态存下来。LangGraph 的核心抽象是StateGraph。你定义一个 State通常是个 TypedDict 或者 Pydantic 模型然后定义若干个 Node节点本质是函数接收 State 返回 State 的更新再用 Edge边把它们连起来。边可以是普通的也可以是条件边conditional edge根据当前 State 决定下一步走哪个节点。这个模型的关键在于State 是显式声明的每个 Node 对 State 的修改是可追踪的。这就为 Checkpoint 提供了基础——因为框架知道每一步之后 State 长什么样它就能把 State 存下来。2.2 State 与 Reducer状态合并的隐藏规则这里有个特别容易踩的坑我必须单独拎出来讲。LangGraph 里每个 Node 返回的不是完整的 State而是对 State 的部分更新。框架会把这些更新合并进当前 State。默认的合并策略是覆盖——你返回{messages: [新消息]}它就直接把messages字段覆盖掉。但对话场景里我们想要的是追加不是覆盖。这时候就要用到Reducer归约函数。LangGraph 提供了Annotated配合add_messages来声明某个字段用追加的方式合并from typing import Annotated from typing_extensions import TypedDict from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] current_step: str task_result: dictadd_messages这个 reducer 不只是简单 append它还会处理消息 ID 去重、处理RemoveMessage这类特殊指令。我见过太多人因为没加Annotated[list, add_messages]导致每轮对话历史都被覆盖模型永远失忆然后花半天时间怀疑是不是模型的问题。提示自定义 reducer 时函数签名是(current_value, new_update) - merged_value。如果你有复杂的合并逻辑比如字典深合并完全可以自己写一个 reducer 传进去。2.3 条件边与循环Agent 的思考-行动骨架一个 ReAct 风格的 Agent在 LangGraph 里通常长这样一个agent节点负责调模型一个tools节点负责执行工具然后用条件边判断——如果模型输出了工具调用就去tools如果模型直接给了最终答案就END。from langgraph.graph import StateGraph, END def should_continue(state: State): last_message state[messages][-1] if last_message.tool_calls: return tools return END builder StateGraph(State) builder.add_node(agent, call_model) builder.add_node(tools, tool_node) builder.set_entry_point(agent) builder.add_conditional_edges(agent, should_continue, {tools: tools, END: END}) builder.add_edge(tools, agent) graph builder.compile()注意tools - agent这条边它构成了一个循环。这就是为什么 LangGraph 能表达 Agent 的迭代行为而普通 DAG 不行。这个循环也是 Checkpoint 最有价值的地方——因为循环意味着执行到一半是常态你必须能随时把循环的当前状态存下来。3. Checkpoint 机制中断恢复的地基3.1 Checkpoint 到底存了什么很多人以为 Checkpoint 就是存了个消息列表其实远不止。一个完整的 Checkpoint 包含channel_values当前所有 State 字段的值也就是你的messages、current_step这些。channel_versions每个字段的版本号用于并发控制和增量更新。versions_seen每个节点看到的字段版本用来判断哪些节点需要重新执行。pending_sends待处理的消息发送用于 map-reduce 之类的并行场景。除此之外还有一个CheckpointTuple的概念它把 checkpoint 本身、它的 ID、父 checkpoint ID、以及 metadata比如step、source、writes打包在一起。这个父子链结构非常关键它让整个执行历史变成了一棵树你可以从任意一个历史节点分叉出去重新跑。3.2 thread_id会话隔离的钥匙Checkpoint 是按thread_id隔离的。你可以把它理解成会话 ID或者任务 ID。同一个 thread_id 下的所有 checkpoint 构成一条时间线graph.get_state(config)拿到的永远是这条时间线上最新的状态。config {configurable: {thread_id: user-123-task-456}} result graph.invoke({messages: [(user, 帮我分析这份数据)]}, config)这里有个经验thread_id 的生成策略要提前想清楚。如果你用用户 ID 做 thread_id那同一个用户的所有对话会共享一条时间线历史会无限增长如果你用用户ID 会话ID那每个会话独立但跨会话的上下文就断了。我的做法是会话级 thread_id 用 UUID同时在 State 里存一个 user_id 字段做业务关联这样既隔离了会话又能按用户维度查询。3.3 内存 Saver 与持久化 Saver 的分水岭LangGraph 内置了MemorySaver把 checkpoint 存在内存里。开发调试够用但进程一重启就没了而且多实例部署时各存各的完全不能用。生产环境必须换成持久化 Saver。官方支持的有SqliteSaver、PostgresSaver社区还有 Redis、MongoDB 的实现。选哪个我的判断标准很简单场景推荐 Saver理由本地开发、单机 demoMemorySaver零配置重启即清空单机小规模、边缘部署SqliteSaver单文件无需额外服务生产、多实例、高并发PostgresSaver事务保证、并发安全、可水平扩展已有 Redis 基础设施社区 Redis Saver复用现有组件读写快我选 PostgreSQL 的原因很实在Agent 的状态数据天然需要事务和一致性。你不想看到工具执行了但状态没存上这种脏数据吧Postgres 的 ACID 能兜住这个底。而且 checkpoint 表会随着对话增长变得很大Postgres 的分区、索引、归档能力都是现成的。4. 用 PostgreSQL 落地 Checkpoint 存储4.1 环境准备与依赖安装先把依赖装齐。注意版本LangGraph 迭代很快不同版本 API 有差异pip install langgraph langgraph-checkpoint-postgres psycopg[binary] psycopg-poolpsycopg是 psycopg3PostgresSaver 用的是它不是老的psycopg2。这点很多人会搞混装错了会报连接相关的奇怪错误。数据库这边你需要一个可用的 PostgreSQL 实例本地 Docker 起一个最省事docker run -d --name langgraph-pg \ -e POSTGRES_PASSWORDyourpassword \ -e POSTGRES_DBlanggraph \ -p 5432:5432 postgres:164.2 初始化 checkpoint 表结构PostgresSaver 需要几张表来存 checkpoint、writes、migrations。官方提供了setup()方法自动建表from langgraph.checkpoint.postgres import PostgresSaver from psycopg_pool import ConnectionPool DB_URI postgresql://postgres:yourpasswordlocalhost:5432/langgraph?sslmodedisable pool ConnectionPool(conninfoDB_URI, max_size20, kwargs{autocommit: True}) checkpointer PostgresSaver(pool) checkpointer.setup() # 建表只需执行一次注意setup()会创建checkpoints、checkpoint_writes、checkpoint_migrations等表。生产环境建议用独立的 migration 流程管理而不是每次启动都调setup()避免权限和并发问题。建完之后你可以连上去看看表结构理解一下数据是怎么落的SELECT thread_id, checkpoint_id, parent_checkpoint_id, metadata-step AS step, type FROM checkpoints WHERE thread_id user-123-task-456 ORDER BY checkpoint_id;你会看到每个 checkpoint 都有parent_checkpoint_id这就是那条时间线的链。metadata里的step告诉你这是第几步source告诉你是loop节点执行还是input用户输入触发的。4.3 把 checkpointer 挂到图上编译图的时候把 checkpointer 传进去整个图就具备了持久化能力graph builder.compile(checkpointercheckpointer)之后每次invoke或stream只要 config 里带了thread_id状态就会自动落库。这里有个细节checkpointer 是编译期绑定的不是调用期。也就是说你不能一个图实例用两个不同的 checkpointer需要不同存储就编译两个图。4.4 连接池与并发别让数据库成为瓶颈生产环境一定要用连接池。我上面用了ConnectionPoolmax_size根据你的并发量调。有个坑PostgresSaver 的写入是同步的如果你的图节点里有大量并发checkpoint 写入可能成为瓶颈。我的优化经验是连接池max_size至少设为预期并发数的 1.5 倍。给checkpoints表的thread_id建索引setup()默认会建但你要确认。定期归档老 checkpoint。对话历史无限增长的话单表几个 G 是迟早的事。可以按created_at分区或者写个定时任务把超过 N 天的 thread 数据迁到冷存储。CREATE INDEX IF NOT EXISTS idx_checkpoints_thread_id ON checkpoints (thread_id, checkpoint_id DESC);5. 中断恢复的完整实现链路5.1 用 interrupt 实现暂停等人确认LangGraph 提供了interrupt()函数在节点里调用它就能暂停执行把控制权交还给调用方。这是实现 Human-in-the-loop 最优雅的方式from langgraph.types import interrupt, Command def send_email_node(state: State): draft state[email_draft] # 暂停把草稿抛给前端确认 decision interrupt({ action: confirm_email, draft: draft }) if decision[approved]: # 真正发送 send_email(draft) return {status: sent} else: return {status: rejected, feedback: decision.get(feedback)}当图执行到这个节点时interrupt()会抛出一个特殊信号LangGraph 把当前状态存成 checkpoint然后invoke返回。此时你可以通过graph.get_state(config)看到state.next指向send_email_node说明它卡在这里等人。5.2 恢复执行Command 的正确用法用户确认之后怎么让它继续跑用Command(resume...)from langgraph.types import Command # 用户点了批准 resumed graph.invoke( Command(resume{approved: True}), config{configurable: {thread_id: user-123-task-456}} )resume的值会作为interrupt()的返回值注入到节点里节点从暂停的那一行继续往下执行。这里有个极其容易踩的坑恢复时必须用同一个 thread_id而且图实例必须是同一个编译产物节点定义一致。如果你改了节点代码再恢复LangGraph 会尝试从 checkpoint 恢复但节点逻辑变了行为可能不符合预期。提示interrupt()之前的代码在恢复时会重新执行一遍。所以interrupt()之前不要放有副作用的操作比如已经发了邮件、已经扣了款。把副作用放在interrupt()之后或者用幂等设计兜底。5.3 从任意历史点分叉时间旅行Checkpoint 的父子链结构让你可以做时间旅行——拿到某个历史 checkpoint 的 ID从那里重新跑history list(graph.get_state_history(config)) # history[0] 是最新history[-1] 是最早 target history[3] # 回到第 4 个状态 forked graph.invoke( None, # 不传新输入从该点继续 config{configurable: {thread_id: user-123-task-456, checkpoint_id: target.config[configurable][checkpoint_id]}} )这个能力在调试时简直是神器。线上出问题你可以把那个 thread 的历史拉出来从出错前的状态重放看看到底哪一步的模型输出跑偏了。也可以做如果当时用户选了另一个选项会怎样的假设分析。5.4 状态查询与前端对接的接口设计前端要展示当前卡在哪一步、等什么确认需要后端暴露查询接口。核心就是get_statestate graph.get_state(config) # state.values: 当前 State # state.next: 下一步要执行的节点元组 # state.tasks: 当前待处理的任务含 interrupt 信息state.tasks里会包含 interrupt 抛出的 payload前端拿到就能渲染确认弹窗。恢复时前端把用户选择 POST 回来后端调Command(resume...)。这套接口设计的关键是把 thread_id 和 checkpoint_id 都透传给前端前端每次请求带上后端就能精确定位到该恢复哪个状态。6. AG-UI 把中断恢复搬到前端6.1 AG-UI 解决的是什么问题前面讲的都是后端能力。但用户是在浏览器里操作的后端暂停了等人确认这件事怎么实时、结构化地传到前端传统做法是前端轮询或者自己定一套 WebSocket 协议每个项目重复造轮子。AG-UI 是一套面向 Agent 前端的协议它定义了 Agent 和后端之间的事件流格式——文本增量、工具调用开始/结束、状态更新、中断请求等等。它和 LangGraph 的配合点在于LangGraph 的 stream 事件可以被适配成 AG-UI 事件前端用统一的协议消费。6.2 事件流从 graph.stream 到前端LangGraph 的stream支持多种模式updates模式会在每个节点执行后吐出更新messages模式会流式吐 tokencustom模式可以吐你自定义的事件。做 AG-UI 对接时通常组合使用async for event in graph.astream(input, config, stream_mode[updates, messages, custom]): mode, payload event if mode messages: # 适配成 AG-UI 的 TEXT_MESSAGE_CONTENT yield to_agui_text(payload) elif mode updates: # 节点级更新适配成 STATE_DELTA 或 STEP 事件 yield to_agui_step(payload) elif mode custom: # 自定义事件比如工具调用进度 yield to_agui_custom(payload)当图执行到interrupt()时stream 会结束当前轮次前端收到一个中断信号弹出确认 UI。用户操作后前端发一个新的请求后端用Command(resume...)继续 stream。6.3 前端如何渲染等待确认状态前端拿到 interrupt payload 后需要根据action类型渲染不同的 UI。我的做法是约定一套 action 到组件的映射action 类型前端组件用户操作confirm_email邮件预览 批准/拒绝按钮返回{approved: bool}confirm_delete危险操作警告弹窗返回{confirmed: bool}select_option选项列表返回{selected: string}input_text文本输入框返回{text: string}这套映射让后端可以自由地插入人工确认点前端只需要按 action 渲染。恢复时把用户的选择作为resume值传回节点里interrupt()的返回值就是它。6.4 断线重连前端刷新后怎么找回状态这是实际产品里绕不开的问题。用户确认弹窗弹出来结果他刷新了页面弹窗没了任务卡在那。解决办法是前端启动时用 thread_id 调一次状态查询接口后端返回state.tasks如果里面有 pending 的 interrupt前端就重新渲染确认 UI。app.get(/threads/{thread_id}/state) async def get_thread_state(thread_id: str): config {configurable: {thread_id: thread_id}} state graph.get_state(config) return { values: state.values, next: state.next, interrupts: [ task.interrupts for task in state.tasks if task.interrupts ] }这样无论用户怎么刷新、换设备只要 thread_id 还在通常存在 URL 或 localStorage状态就能恢复。这也是 Checkpoint 持久化最直接的价值体现。7. 实测中那些文档不会告诉你的坑7.1 checkpoint 表膨胀与清理策略跑了一段时间后你会发现checkpoints表涨得飞快。每执行一个节点就写一条一个 20 步的任务就是 20 条记录加上checkpoint_writes表数据量是翻倍的。我的处理方式对不再活跃的 thread保留最近 N 个 checkpoint老的删掉。但注意不能随便删中间的因为父子链断了会导致时间旅行失效。要删就删整条 thread 的早期记录或者按时间归档。用checkpointer.delete_thread(thread_id)删整个 thread这是官方提供的干净方式。定期跑归档任务把超过 30 天的 thread 数据导出到对象存储然后从主表删除。7.2 恢复时节点重复执行的副作用前面提过interrupt()之前的代码会重跑。我踩过一个具体的坑在interrupt()之前调了一次外部 API 记录待确认日志结果恢复时又记了一次日志里出现两条。解决办法是把这类副作用移到interrupt()之后或者用幂等键去重。还有一个更隐蔽的如果你的节点里有time.sleep()或者耗时计算恢复时会重新执行导致响应变慢。所以节点设计要尽量轻重活拆到独立节点让 checkpoint 的粒度更细。7.3 thread_id 设计不当导致的串话我见过一个事故开发时图省事用固定的default做 thread_id结果所有用户的对话全串在一起A 用户看到了 B 用户的历史。这在生产环境是灾难级的。正确做法是 thread_id 必须全局唯一且和用户绑定。我的生成规则是f{user_id}:{session_uuid}既保证唯一又方便按用户查询。同时在后端做一层校验请求里的 user_id 必须和 thread_id 里的 user_id 匹配防止越权访问别人的会话。7.4 版本升级导致的 checkpoint 不兼容LangGraph 的 checkpoint 格式在不同版本间可能变化。我遇到过一次升级后老 checkpoint 反序列化失败的情况。经验是升级 LangGraph 前先在测试环境用生产数据的副本验证老 checkpoint 能否正常读取。大版本升级时考虑把活跃 thread 跑完再升级或者写迁移脚本。生产环境锁定版本别用latest。7.5 并发写同一 thread 的竞态如果同一个 thread_id 同时有两个请求进来比如用户手快点了两次两个请求都会尝试写 checkpoint可能产生冲突。PostgresSaver 内部有版本控制但业务层面最好加锁——用 Redis 分布式锁按 thread_id 加锁保证同一时刻只有一个执行流在跑。lock_key flanggraph:lock:{thread_id} if not acquire_lock(lock_key, timeout30): raise HTTPException(409, 该任务正在处理中请稍候)8. 从裸 Loop 迁移到 LangGraph 的实操建议如果你手上已经有一个跑得还行的手写 Loop想迁移到 LangGraph我的建议是别一次性重写。分三步走第一步先把 Loop 里的状态抽出来定义成显式的 State 结构。这一步不改逻辑只是把散落的变量收拢。你会发现很多隐式状态之前根本没意识到。第二步把 Loop 的每一轮拆成节点。调模型是一个节点执行工具是一个节点判断是否继续是一个条件边。这时候先不接 checkpointer用 MemorySaver 跑通确认行为和原来一致。第三步接上 PostgresSaver加上 thread_id测试中断恢复。这一步最容易暴露问题——你会发现有些节点有副作用、有些状态没被正确 reducer 合并。逐个修。迁移完之后你会获得几个原来没有的能力任务可以暂停恢复、可以时间旅行调试、可以多人协同确认、可以水平扩展因为状态在数据库不在内存。这些能力在裸 Loop 里要么做不了要么得自己造一套成本远高于直接用框架。最后分享一个我自己的判断标准当一个 Agent 任务的执行时间超过 30 秒或者需要人工介入或者需要跨请求保持上下文就该上 LangGraph Checkpoint 了。低于这个复杂度裸 Loop 反而更直接。工具是拿来解决问题的不是拿来炫技的。 LangGraph Checkpoint 实战：PostgreSQL 持久化与中断恢复
1. 为什么手写 Loop 终究会撞上恢复这堵墙我最早做对话式 Agent 的时候和大多数人一样写了一个while True循环读用户输入、拼 prompt、调模型、解析工具调用、执行工具、把结果塞回消息列表、再调模型直到模型不再请求工具为止。这套东西跑 demo 特别爽几十行代码就能让模型查天气、算数学、读文件。但只要业务稍微真实一点问题就全冒出来了。最典型的一个场景用户让 Agent 帮忙做一份包含十几个步骤的数据处理任务跑到第七步的时候用户突然关掉了浏览器或者网络抖了一下或者后端服务重启了。这时候你那个while True循环里的所有状态——已经执行到哪一步、中间产出了什么、下一步该干嘛——全部随着进程一起蒸发了。用户回来一看得从头再来。这种体验在真实产品里是致命的。手写 Loop 的第二个死穴是人机协同Human-in-the-loop。很多任务需要中途让人确认一下比如我要给这个客户发邮件了你确认下内容或者这个数据库删除操作要不要执行。在裸循环里你只能靠input()阻塞或者自己造一套状态机把等待确认这个状态存到某个地方。一旦要支持多个用户并发、要支持确认后从断点继续你就会发现自己其实是在重新发明一个工作流引擎。第三个问题是可观测性。手写循环里每一步的输入输出都散落在各种变量和日志里你很难回答这个任务到底卡在哪一步了上一步的模型输出是什么工具调用参数对不对这类问题。排查线上问题时你只能靠翻日志猜。所以当我第一次接触 LangGraph 的时候最打动我的不是它能画流程图而是它把Checkpoint检查点做成了框架的一等公民。它本质上是在告诉你Agent 的执行状态不应该活在内存里而应该被持久化下来随时可以暂停、恢复、回放、甚至分叉。这篇文章我就想把这套东西讲透——从 LangGraph 的图结构到 PostgreSQL 做 Checkpoint 存储再到用 AG-UI 把中断恢复的能力暴露到前端跑通一条完整的链路。适合读这篇的人已经用 LangChain 或裸 API 写过 Agent、现在被状态管理折磨、想上 LangGraph 但被 checkpoint 概念绕晕的开发者。如果你还没写过 Agent建议先补一下基础不然有些坑你体会不到。2. LangGraph 的图模型到底和 LangChain 差在哪2.1 从链到图不是换个名字那么简单很多人问 LangChain 和 LangGraph 的区别网上答案一大堆但大多停留在一个是链一个是图这种废话层面。我用一句话概括我的理解LangChain 解决的是怎么把组件串起来LangGraph 解决的是怎么让一个有状态的流程可控地跑起来。LangChain 的 Chain 本质是线性的、无状态的或者说状态是隐式传递的。你写一个LLMChain它的输入输出是明确的跑完就结束了。就算你用AgentExecutor它内部那个循环也是黑盒的你没法在中途插手也没法把中间状态存下来。LangGraph 的核心抽象是StateGraph。你定义一个 State通常是个 TypedDict 或者 Pydantic 模型然后定义若干个 Node节点本质是函数接收 State 返回 State 的更新再用 Edge边把它们连起来。边可以是普通的也可以是条件边conditional edge根据当前 State 决定下一步走哪个节点。这个模型的关键在于State 是显式声明的每个 Node 对 State 的修改是可追踪的。这就为 Checkpoint 提供了基础——因为框架知道每一步之后 State 长什么样它就能把 State 存下来。2.2 State 与 Reducer状态合并的隐藏规则这里有个特别容易踩的坑我必须单独拎出来讲。LangGraph 里每个 Node 返回的不是完整的 State而是对 State 的部分更新。框架会把这些更新合并进当前 State。默认的合并策略是覆盖——你返回{messages: [新消息]}它就直接把messages字段覆盖掉。但对话场景里我们想要的是追加不是覆盖。这时候就要用到Reducer归约函数。LangGraph 提供了Annotated配合add_messages来声明某个字段用追加的方式合并from typing import Annotated from typing_extensions import TypedDict from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] current_step: str task_result: dictadd_messages这个 reducer 不只是简单 append它还会处理消息 ID 去重、处理RemoveMessage这类特殊指令。我见过太多人因为没加Annotated[list, add_messages]导致每轮对话历史都被覆盖模型永远失忆然后花半天时间怀疑是不是模型的问题。提示自定义 reducer 时函数签名是(current_value, new_update) - merged_value。如果你有复杂的合并逻辑比如字典深合并完全可以自己写一个 reducer 传进去。2.3 条件边与循环Agent 的思考-行动骨架一个 ReAct 风格的 Agent在 LangGraph 里通常长这样一个agent节点负责调模型一个tools节点负责执行工具然后用条件边判断——如果模型输出了工具调用就去tools如果模型直接给了最终答案就END。from langgraph.graph import StateGraph, END def should_continue(state: State): last_message state[messages][-1] if last_message.tool_calls: return tools return END builder StateGraph(State) builder.add_node(agent, call_model) builder.add_node(tools, tool_node) builder.set_entry_point(agent) builder.add_conditional_edges(agent, should_continue, {tools: tools, END: END}) builder.add_edge(tools, agent) graph builder.compile()注意tools - agent这条边它构成了一个循环。这就是为什么 LangGraph 能表达 Agent 的迭代行为而普通 DAG 不行。这个循环也是 Checkpoint 最有价值的地方——因为循环意味着执行到一半是常态你必须能随时把循环的当前状态存下来。3. Checkpoint 机制中断恢复的地基3.1 Checkpoint 到底存了什么很多人以为 Checkpoint 就是存了个消息列表其实远不止。一个完整的 Checkpoint 包含channel_values当前所有 State 字段的值也就是你的messages、current_step这些。channel_versions每个字段的版本号用于并发控制和增量更新。versions_seen每个节点看到的字段版本用来判断哪些节点需要重新执行。pending_sends待处理的消息发送用于 map-reduce 之类的并行场景。除此之外还有一个CheckpointTuple的概念它把 checkpoint 本身、它的 ID、父 checkpoint ID、以及 metadata比如step、source、writes打包在一起。这个父子链结构非常关键它让整个执行历史变成了一棵树你可以从任意一个历史节点分叉出去重新跑。3.2 thread_id会话隔离的钥匙Checkpoint 是按thread_id隔离的。你可以把它理解成会话 ID或者任务 ID。同一个 thread_id 下的所有 checkpoint 构成一条时间线graph.get_state(config)拿到的永远是这条时间线上最新的状态。config {configurable: {thread_id: user-123-task-456}} result graph.invoke({messages: [(user, 帮我分析这份数据)]}, config)这里有个经验thread_id 的生成策略要提前想清楚。如果你用用户 ID 做 thread_id那同一个用户的所有对话会共享一条时间线历史会无限增长如果你用用户ID 会话ID那每个会话独立但跨会话的上下文就断了。我的做法是会话级 thread_id 用 UUID同时在 State 里存一个 user_id 字段做业务关联这样既隔离了会话又能按用户维度查询。3.3 内存 Saver 与持久化 Saver 的分水岭LangGraph 内置了MemorySaver把 checkpoint 存在内存里。开发调试够用但进程一重启就没了而且多实例部署时各存各的完全不能用。生产环境必须换成持久化 Saver。官方支持的有SqliteSaver、PostgresSaver社区还有 Redis、MongoDB 的实现。选哪个我的判断标准很简单场景推荐 Saver理由本地开发、单机 demoMemorySaver零配置重启即清空单机小规模、边缘部署SqliteSaver单文件无需额外服务生产、多实例、高并发PostgresSaver事务保证、并发安全、可水平扩展已有 Redis 基础设施社区 Redis Saver复用现有组件读写快我选 PostgreSQL 的原因很实在Agent 的状态数据天然需要事务和一致性。你不想看到工具执行了但状态没存上这种脏数据吧Postgres 的 ACID 能兜住这个底。而且 checkpoint 表会随着对话增长变得很大Postgres 的分区、索引、归档能力都是现成的。4. 用 PostgreSQL 落地 Checkpoint 存储4.1 环境准备与依赖安装先把依赖装齐。注意版本LangGraph 迭代很快不同版本 API 有差异pip install langgraph langgraph-checkpoint-postgres psycopg[binary] psycopg-poolpsycopg是 psycopg3PostgresSaver 用的是它不是老的psycopg2。这点很多人会搞混装错了会报连接相关的奇怪错误。数据库这边你需要一个可用的 PostgreSQL 实例本地 Docker 起一个最省事docker run -d --name langgraph-pg \ -e POSTGRES_PASSWORDyourpassword \ -e POSTGRES_DBlanggraph \ -p 5432:5432 postgres:164.2 初始化 checkpoint 表结构PostgresSaver 需要几张表来存 checkpoint、writes、migrations。官方提供了setup()方法自动建表from langgraph.checkpoint.postgres import PostgresSaver from psycopg_pool import ConnectionPool DB_URI postgresql://postgres:yourpasswordlocalhost:5432/langgraph?sslmodedisable pool ConnectionPool(conninfoDB_URI, max_size20, kwargs{autocommit: True}) checkpointer PostgresSaver(pool) checkpointer.setup() # 建表只需执行一次注意setup()会创建checkpoints、checkpoint_writes、checkpoint_migrations等表。生产环境建议用独立的 migration 流程管理而不是每次启动都调setup()避免权限和并发问题。建完之后你可以连上去看看表结构理解一下数据是怎么落的SELECT thread_id, checkpoint_id, parent_checkpoint_id, metadata-step AS step, type FROM checkpoints WHERE thread_id user-123-task-456 ORDER BY checkpoint_id;你会看到每个 checkpoint 都有parent_checkpoint_id这就是那条时间线的链。metadata里的step告诉你这是第几步source告诉你是loop节点执行还是input用户输入触发的。4.3 把 checkpointer 挂到图上编译图的时候把 checkpointer 传进去整个图就具备了持久化能力graph builder.compile(checkpointercheckpointer)之后每次invoke或stream只要 config 里带了thread_id状态就会自动落库。这里有个细节checkpointer 是编译期绑定的不是调用期。也就是说你不能一个图实例用两个不同的 checkpointer需要不同存储就编译两个图。4.4 连接池与并发别让数据库成为瓶颈生产环境一定要用连接池。我上面用了ConnectionPoolmax_size根据你的并发量调。有个坑PostgresSaver 的写入是同步的如果你的图节点里有大量并发checkpoint 写入可能成为瓶颈。我的优化经验是连接池max_size至少设为预期并发数的 1.5 倍。给checkpoints表的thread_id建索引setup()默认会建但你要确认。定期归档老 checkpoint。对话历史无限增长的话单表几个 G 是迟早的事。可以按created_at分区或者写个定时任务把超过 N 天的 thread 数据迁到冷存储。CREATE INDEX IF NOT EXISTS idx_checkpoints_thread_id ON checkpoints (thread_id, checkpoint_id DESC);5. 中断恢复的完整实现链路5.1 用 interrupt 实现暂停等人确认LangGraph 提供了interrupt()函数在节点里调用它就能暂停执行把控制权交还给调用方。这是实现 Human-in-the-loop 最优雅的方式from langgraph.types import interrupt, Command def send_email_node(state: State): draft state[email_draft] # 暂停把草稿抛给前端确认 decision interrupt({ action: confirm_email, draft: draft }) if decision[approved]: # 真正发送 send_email(draft) return {status: sent} else: return {status: rejected, feedback: decision.get(feedback)}当图执行到这个节点时interrupt()会抛出一个特殊信号LangGraph 把当前状态存成 checkpoint然后invoke返回。此时你可以通过graph.get_state(config)看到state.next指向send_email_node说明它卡在这里等人。5.2 恢复执行Command 的正确用法用户确认之后怎么让它继续跑用Command(resume...)from langgraph.types import Command # 用户点了批准 resumed graph.invoke( Command(resume{approved: True}), config{configurable: {thread_id: user-123-task-456}} )resume的值会作为interrupt()的返回值注入到节点里节点从暂停的那一行继续往下执行。这里有个极其容易踩的坑恢复时必须用同一个 thread_id而且图实例必须是同一个编译产物节点定义一致。如果你改了节点代码再恢复LangGraph 会尝试从 checkpoint 恢复但节点逻辑变了行为可能不符合预期。提示interrupt()之前的代码在恢复时会重新执行一遍。所以interrupt()之前不要放有副作用的操作比如已经发了邮件、已经扣了款。把副作用放在interrupt()之后或者用幂等设计兜底。5.3 从任意历史点分叉时间旅行Checkpoint 的父子链结构让你可以做时间旅行——拿到某个历史 checkpoint 的 ID从那里重新跑history list(graph.get_state_history(config)) # history[0] 是最新history[-1] 是最早 target history[3] # 回到第 4 个状态 forked graph.invoke( None, # 不传新输入从该点继续 config{configurable: {thread_id: user-123-task-456, checkpoint_id: target.config[configurable][checkpoint_id]}} )这个能力在调试时简直是神器。线上出问题你可以把那个 thread 的历史拉出来从出错前的状态重放看看到底哪一步的模型输出跑偏了。也可以做如果当时用户选了另一个选项会怎样的假设分析。5.4 状态查询与前端对接的接口设计前端要展示当前卡在哪一步、等什么确认需要后端暴露查询接口。核心就是get_statestate graph.get_state(config) # state.values: 当前 State # state.next: 下一步要执行的节点元组 # state.tasks: 当前待处理的任务含 interrupt 信息state.tasks里会包含 interrupt 抛出的 payload前端拿到就能渲染确认弹窗。恢复时前端把用户选择 POST 回来后端调Command(resume...)。这套接口设计的关键是把 thread_id 和 checkpoint_id 都透传给前端前端每次请求带上后端就能精确定位到该恢复哪个状态。6. AG-UI 把中断恢复搬到前端6.1 AG-UI 解决的是什么问题前面讲的都是后端能力。但用户是在浏览器里操作的后端暂停了等人确认这件事怎么实时、结构化地传到前端传统做法是前端轮询或者自己定一套 WebSocket 协议每个项目重复造轮子。AG-UI 是一套面向 Agent 前端的协议它定义了 Agent 和后端之间的事件流格式——文本增量、工具调用开始/结束、状态更新、中断请求等等。它和 LangGraph 的配合点在于LangGraph 的 stream 事件可以被适配成 AG-UI 事件前端用统一的协议消费。6.2 事件流从 graph.stream 到前端LangGraph 的stream支持多种模式updates模式会在每个节点执行后吐出更新messages模式会流式吐 tokencustom模式可以吐你自定义的事件。做 AG-UI 对接时通常组合使用async for event in graph.astream(input, config, stream_mode[updates, messages, custom]): mode, payload event if mode messages: # 适配成 AG-UI 的 TEXT_MESSAGE_CONTENT yield to_agui_text(payload) elif mode updates: # 节点级更新适配成 STATE_DELTA 或 STEP 事件 yield to_agui_step(payload) elif mode custom: # 自定义事件比如工具调用进度 yield to_agui_custom(payload)当图执行到interrupt()时stream 会结束当前轮次前端收到一个中断信号弹出确认 UI。用户操作后前端发一个新的请求后端用Command(resume...)继续 stream。6.3 前端如何渲染等待确认状态前端拿到 interrupt payload 后需要根据action类型渲染不同的 UI。我的做法是约定一套 action 到组件的映射action 类型前端组件用户操作confirm_email邮件预览 批准/拒绝按钮返回{approved: bool}confirm_delete危险操作警告弹窗返回{confirmed: bool}select_option选项列表返回{selected: string}input_text文本输入框返回{text: string}这套映射让后端可以自由地插入人工确认点前端只需要按 action 渲染。恢复时把用户的选择作为resume值传回节点里interrupt()的返回值就是它。6.4 断线重连前端刷新后怎么找回状态这是实际产品里绕不开的问题。用户确认弹窗弹出来结果他刷新了页面弹窗没了任务卡在那。解决办法是前端启动时用 thread_id 调一次状态查询接口后端返回state.tasks如果里面有 pending 的 interrupt前端就重新渲染确认 UI。app.get(/threads/{thread_id}/state) async def get_thread_state(thread_id: str): config {configurable: {thread_id: thread_id}} state graph.get_state(config) return { values: state.values, next: state.next, interrupts: [ task.interrupts for task in state.tasks if task.interrupts ] }这样无论用户怎么刷新、换设备只要 thread_id 还在通常存在 URL 或 localStorage状态就能恢复。这也是 Checkpoint 持久化最直接的价值体现。7. 实测中那些文档不会告诉你的坑7.1 checkpoint 表膨胀与清理策略跑了一段时间后你会发现checkpoints表涨得飞快。每执行一个节点就写一条一个 20 步的任务就是 20 条记录加上checkpoint_writes表数据量是翻倍的。我的处理方式对不再活跃的 thread保留最近 N 个 checkpoint老的删掉。但注意不能随便删中间的因为父子链断了会导致时间旅行失效。要删就删整条 thread 的早期记录或者按时间归档。用checkpointer.delete_thread(thread_id)删整个 thread这是官方提供的干净方式。定期跑归档任务把超过 30 天的 thread 数据导出到对象存储然后从主表删除。7.2 恢复时节点重复执行的副作用前面提过interrupt()之前的代码会重跑。我踩过一个具体的坑在interrupt()之前调了一次外部 API 记录待确认日志结果恢复时又记了一次日志里出现两条。解决办法是把这类副作用移到interrupt()之后或者用幂等键去重。还有一个更隐蔽的如果你的节点里有time.sleep()或者耗时计算恢复时会重新执行导致响应变慢。所以节点设计要尽量轻重活拆到独立节点让 checkpoint 的粒度更细。7.3 thread_id 设计不当导致的串话我见过一个事故开发时图省事用固定的default做 thread_id结果所有用户的对话全串在一起A 用户看到了 B 用户的历史。这在生产环境是灾难级的。正确做法是 thread_id 必须全局唯一且和用户绑定。我的生成规则是f{user_id}:{session_uuid}既保证唯一又方便按用户查询。同时在后端做一层校验请求里的 user_id 必须和 thread_id 里的 user_id 匹配防止越权访问别人的会话。7.4 版本升级导致的 checkpoint 不兼容LangGraph 的 checkpoint 格式在不同版本间可能变化。我遇到过一次升级后老 checkpoint 反序列化失败的情况。经验是升级 LangGraph 前先在测试环境用生产数据的副本验证老 checkpoint 能否正常读取。大版本升级时考虑把活跃 thread 跑完再升级或者写迁移脚本。生产环境锁定版本别用latest。7.5 并发写同一 thread 的竞态如果同一个 thread_id 同时有两个请求进来比如用户手快点了两次两个请求都会尝试写 checkpoint可能产生冲突。PostgresSaver 内部有版本控制但业务层面最好加锁——用 Redis 分布式锁按 thread_id 加锁保证同一时刻只有一个执行流在跑。lock_key flanggraph:lock:{thread_id} if not acquire_lock(lock_key, timeout30): raise HTTPException(409, 该任务正在处理中请稍候)8. 从裸 Loop 迁移到 LangGraph 的实操建议如果你手上已经有一个跑得还行的手写 Loop想迁移到 LangGraph我的建议是别一次性重写。分三步走第一步先把 Loop 里的状态抽出来定义成显式的 State 结构。这一步不改逻辑只是把散落的变量收拢。你会发现很多隐式状态之前根本没意识到。第二步把 Loop 的每一轮拆成节点。调模型是一个节点执行工具是一个节点判断是否继续是一个条件边。这时候先不接 checkpointer用 MemorySaver 跑通确认行为和原来一致。第三步接上 PostgresSaver加上 thread_id测试中断恢复。这一步最容易暴露问题——你会发现有些节点有副作用、有些状态没被正确 reducer 合并。逐个修。迁移完之后你会获得几个原来没有的能力任务可以暂停恢复、可以时间旅行调试、可以多人协同确认、可以水平扩展因为状态在数据库不在内存。这些能力在裸 Loop 里要么做不了要么得自己造一套成本远高于直接用框架。最后分享一个我自己的判断标准当一个 Agent 任务的执行时间超过 30 秒或者需要人工介入或者需要跨请求保持上下文就该上 LangGraph Checkpoint 了。低于这个复杂度裸 Loop 反而更直接。工具是拿来解决问题的不是拿来炫技的。