LangGraph 生产级笔记：Checkpointer + interrupt()，把人审做成可恢复状态机

千笔-AIWritePaper · https://www.aiwritepaper.com把「人审」塞进 Agent最常见翻车是进程一重启就丢现场或审批接口返回后图从错误节点重跑。LangGraph 的答案不是另写一套工作流引擎而是两块拼在一起Checkpointer 把线程状态落成可恢复快照interrupt()在节点任意位置动态暂停恢复时用同一个thread_id把Command(resume…)的值交回interrupt()调用点。本文只复述官方 Persistence、Memory 与 Interrupts 文档里能对照的行为并落到可核对的最小代码与检查清单。图编译挂上持久化 checkpointer → 带 thread_id 运行 → interrupt 暂停并表面 payload → Command(resume) 恢复底部为官方四条硬规则摘要。目标说明读完你应能独立完成五件事分清Checkpointer线程内短期记忆 / 图状态快照与Store跨线程长期键值人审恢复主要靠前者。在生产编译图时挂上PostgresSaver / SqliteSaver等持久化实现而不是把InMemorySaver直接上线。在节点里调用interrupt(payload)用stream_events(..., versionv3)读取stream.interrupted/stream.interrupts或用invoke看__interrupt__。用相同thread_id与Command(resume…)续跑并理解「恢复时节点会从开头重跑」。对照官方规则勿裸try/except吞中断、勿打乱 interrupt 顺序、payload 可序列化、中断前副作用幂等。规格钉死来自官方文档Checkpointer 持久化单线程图状态用途含对话续写、HITL、时间旅行、故障恢复。Store 持久化应用自定义数据跨线程适合用户偏好与共享知识。thread_id是游标复用则续状态换新值等于新线程空状态。interrupt()暂停执行、写入 checkpoint、无限等待外部输入恢复值成为该次interrupt()的返回值。静态interrupt_before/interrupt_after适合调试不推荐作为生产 HITL 主路径。Agent Server 场景下服务端可托管持久化应用侧仍需理解 thread 与 resume 语义。适用场景与边界适合做成可恢复人审发邮件、改库、转账、调用付费 API 前必须人点同意或改参数后再执行。审改 LLM 草稿、工具参数后再进下游节点。并行分支各自interrupt一次用 interrupt id 映射批量 resume。进程会重启、要隔夜审批状态必须在数据库或文件型 checkpointer 里。需要「时间旅行」按 checkpoint 查看历史状态再决定是否从某点续跑。不该指望 Checkpointer interrupt 单独搞定跨用户、跨会话的偏好与事实库用Store不是把一切塞进线程 checkpoint。无 checkpointer 的「纯函数图」interrupt无法持久等待。把人审当成同步 HTTP 里死等应把暂停态暴露给 UI审批回调再 resume。用裸MemorySaver/InMemorySaver冒充生产重启即丢。开放式、无法 JSON 化的「把整个 UI 组件塞进 interrupt」序列化会失败。风险提示Postgres 下thread_id过长会撞列宽官方建议控制在 255 字符内或用 UUID / 哈希。长对话 checkpoint 会膨胀延迟与存储成本上升需保留策略或定期清理。子图有独立 checkpoint 命名空间时父图未必立刻看见子图写入跨图共享数据优先考虑 Store。不可信来源的图定义同样可能诱导危险工具调用人审节点不能省略权限模型。机制Checkpointer 与 interrupt 如何咬合Checkpointer线程级快照编译时传入 checkpointerfromlanggraph.checkpoint.memoryimportInMemorySaver# 仅开发# 生产见官方PostgresSaver / SqliteSaver / Redis / Mongo 等checkpointerInMemorySaver()graphbuilder.compile(checkpointercheckpointer)config{configurable:{thread_id:approval-123}}每次步进后状态可落盘。人审暂停时运行时保存当前图状态之后即使进程退出只要 checkpointer 还在且thread_id不变就能续。官方 Persistence 文档把 Checkpointer 与 Store 对照成「短线程记忆 vs 长跨线程记忆」HITL 首先钉前者。生产示例骨架Postgres需安装对应包并setup()fromlanggraph.checkpoint.postgresimportPostgresSaverwithPostgresSaver.from_conn_string(DB_URI)ascheckpointer:# checkpointer.setup() # 首次graphbuilder.compile(checkpointercheckpointer)interrupt动态断点fromlanggraph.typesimportinterrupt,Commanddefapproval_node(state):decisioninterrupt({question:Approve this action?,details:state[action_details],})return{approved:bool(decision)}发生了什么执行挂起 → checkpoint 写入 → payload 出现在stream.interrupts或invoke的__interrupt__→ 等待 →Command(resume…)把值送回interrupt()。推荐驱动方式官方强调 event streamingstreamgraph.stream_events(inputs,configconfig,versionv3)_stream.outputifstream.interrupted:# 把 stream.interrupts 展示给 UIresumedgraph.stream_events(Command(resumeTrue),configconfig,versionv3)finalresumed.output交互式 HITL 循环可以一边消费stream.messages做 token 展示一边在 paused 时收集人输入再 resume直到stream.interrupted为假。为何「节点会整段重跑」恢复不是从interrupt下一行字节码续跑而是重新进入该节点。因此interrupt之前的代码会再执行一遍。这直接导出官方 Rules of interrupts副作用必须幂等或后移多个 interrupt 的顺序必须稳定不能用裸 except 吞掉暂停用的特殊异常。审批、审改、工具内暂停常见模式批准/拒绝interrupt返回布尔或枚举节点用Command(gotoproceed|cancel)路由。审改状态interrupt返回编辑后的文本写回 state。工具内 interrupt在tool里暂停审批时可覆盖to/subject/body再真正发送。校验人输入每节点只调用一次interrupt非法则更新pending_question条件边绕回禁止单节点while True多段 interrupt。与静态断点的区别类型触发用途interrupt()代码内、可条件生产 HITL、按业务暂停interrupt_before/after编译或调用时按节点名调试单步步骤最小可恢复审批图选持久化 checkpointer首次调用setup()Postgres 等需迁移。State只放可序列化字段审批细节用 dict不要塞函数对象或处理器实例。审批节点里先interrupt再根据返回值路由或写状态。副作用发邮件、插审计日志尽量放在 interrupt之后或做成幂等 upsert。驱动循环消费stream_events未 interrupted 则结束否则收集人输入再Command(resume…)。并行多 interruptresume 时传{interrupt_id: value}映射避免对错问题。观测用graph.get_state(config)/get_state_history核对暂停时的values与next便于排障。示意结构对齐官方 Full example省略业务细节defapproval_node(state):decisioninterrupt({question:Approve this action?,details:state[action_details],})returnCommand(gotoproceedifdecisionelsecancel)可验证检查清单检查项通过标准持久化杀掉进程后同thread_id仍能 resume 到待审状态表面stream.interrupted is True且 payload 与传入一致续跑resume 后interrupt()返回值等于Command(resume…)幂等故意多次走 interrupt 前路径外部副作用不重复脏写规则无裸 except 包 interrupt同节点 interrupt 顺序稳定边界thread_id长度合规开发/生产 checkpointer 类型可区分历史get_state_history能看到暂停前后的快照链本地可用 Sqlite 文件型 checkpointer 做冒烟跑到 interrupt → 退出解释器 → 新进程加载同 DB 与 thread → resume。把通过/失败记成表格比口头「我觉得可以恢复」更有用。踩坑踩坑后果改法生产用 InMemory重启丢审换 Postgres/Sqlite/Redis 等try/except Exception包 interrupt暂停异常被吞图「假继续」只捕业务异常类型interrupt 前create非幂等记录每次 resume 复制脏数据upsert 或副作用后移条件跳过某个 interruptresume 索引错位顺序固定或拆节点换新 thread_id 当「重试」另起空线程重试必须复用原 id把 HITL 做成静态 interrupt_before难按业务条件暂停改用interrupt()checkpoint 永不清理存储与延迟恶化保留策略 / 定期删除旧线程resume 时传Command(update…)当输入语义不符官方约定人审续跑用Command(resume…)生产编排补充把暂停态交给外部系统只在笔记本里invoke两次还不算生产。常见编排是API 收到用户请求创建或复用thread_id启动stream_events。若interrupted把interruptspayload 写入任务表状态标为waiting_humanHTTP 立刻返回「待审」。审核员在后台点同意/拒绝/编辑回调服务用同一thread_id调Command(resume…)。续跑完成后再通知用户失败则保留 checkpoint禁止「换个 id 重试」装作同一单。这样人审等待可以是小时或天级而不用占着工作进程。Checkpointer 此时就是工单系统的状态后端之一。注意UI 展示的文案应来自你传入interrupt的 JSON而不是事后拼出来的模糊提示否则审核员看不到模型当时真正要做什么。并发方面同一thread_id上应串行化 resume避免两个审核员同时提交导致状态竞争。并行节点上的多个 interrupt则按官方所述用 id→值映射一次性恢复减少半恢复态。和「自己写个审批表」比图状态机赢在哪自己做审批表也能存「待同意」但很难自动对齐「图跑到哪一节点、通道版本、待执行任务」这些运行时细节。LangGraph 把这些放进 checkpoint 元数据恢复时运行时知道next是谁。对 Agent 这种分支多、工具多的系统用图状态机比把审批逻辑散落在各处 if 更不容易漏边。代价是你必须遵守 interrupt 规则并把序列化边界划清。若业务只需一次性人工确认且无复杂分支轻量工单可能更简单一旦出现「改参数再跑工具」「多专家会签」「失败后从中段续」Checkpointer interrupt 的收益会明显高于散装 if。小抄上线前十问生产 checkpointer 类型是什么连接串是否在密钥管理里setup()/ 迁移是否纳入发布流程每个业务单的thread_id如何生成是否可能超过长度限制暂停态如何暴露给前端超时如何提醒审核员resume 接口是否鉴权到「只能审自己的单」interrupt payload 是否避免泄露敏感全文到日志节点内 interrupt 前是否还有非幂等写并行 interrupt 是否测试过 id 映射 resumecheckpoint 保留多久谁有权delete_thread故障演练杀进程后能否从待审态恢复十问都有书面答案再宣布「我们支持人审」不迟。总结生产级人审不是「多弹一个确认框」而是可恢复状态机Checkpointer 记住线程快照interrupt()在业务点暂停同thread_idCommand(resume…)把人的决定注回图。先把持久化与四条 interrupt 规则做对再谈 UI 与并行审批。文档入口LangGraph Persistence、Human-in-the-loopInterrupts、Memory。把检查清单跑通一次比堆更多节点名更接近「生产级」。把人审做成可恢复状态机之后产品经理与安全同学才有共同语言暂停态可查询恢复动作可审计线程标识可追踪。这比在演示视频里「人工点一下继续」更接近可运营系统。选工具时也可以把本文当验收单候选框架若声称支持 HITL就问三句。第一暂停态是否写入可独立于进程的存储。第二恢复是否保证业务 id 不变。第三节点重入时如何避免重复副作用。三句都答得清再谈可视化与模板。答不清多半只是演示级确认框。