AI 干活你怕不怕有了 LangGraph 的中断功能你可以随时喊“停”审一审、改一改再让 AI 继续干。这篇文章用大白话把这个“刹车”功能从头到尾讲透彻。做 AI 应用的人最头疼的问题不是 AI 不会干活而是 AI 太“莽撞”。它可能一句话就发出邮件、完成转账、删除数据根本不问你能不能做。在一个需要按规章制度办事的真实业务场景里谁敢直接把这样的 AI 放到线上答案是没有。所以我们真正需要的是在 AI 的执行路上随时可以喊“等等让我看一眼”的能力。通俗地说当一个 AI 准备进行高风险操作时它应该乖乖停下来等你检查完、点头同意之后再继续执行。这种设计模式在 AI 行业里有一个专门的名字叫作“Human-in-the-Loop”也就是“人在回路”。LangGraph 提供了一套非常完整的“中断 恢复”机制正好就把这件事做了。本文会先把中断机制的基本原理讲清楚然后用四个可以放入项目中的实际例子展示其用法最后总结出几条关键的注意事项和最佳实践。一、先搞懂中断到底是怎么工作的要理解中断需要从最基础的概念入手。我们把一个业务流程图想象成一条流水线上面放着一个个任务。LangGraph 的作用就是把这些任务组织成一张“状态图”其中每一个“节点”代表一个执行步骤而整个图的状态也就是到目前为止拿到的所有数据会在每一个节点执行之后被保存下来。如果走到某一步需要让人来决定怎么办就需要用中断。中断的核心是 interrupt() 函数和 Command(resume...) 配合使用。interrupt() 的作用相当于在代码里面插了一面“暂停旗”。当一个节点执行到这个函数的时候图的运行会立刻停止然后把一段信息打包抛给调用方。调用方可以在这段信息里看到“图在等什么”比如“需要你审批一笔转账”。与此同时LangGraph 的持久化模块也就是 Checkpointer会把当前的整个图状态完整地存起来。这意味着就算电脑重启了只要用同一个 thread_id 重新连接就能精准地找回中断前的所有状态就像游戏读档一样。等到外部的人把决定做完了比如点了“批准”按钮调用方就再次调用图并传入一个 Command(resume...)resume 里塞着的就是人的决定。图收到 Command 之后会从头执行那个被中断的节点而 interrupt() 这次就不再抛异常停止而是直接把人传回的值当作返回值交还给节点。节点拿到这个值之后就可以根据结果做后续的处理。官方文档用一句话概括了整个过程暂停Pause→ 持久化Persist→ 恢复Resume。这三个环节缺了任何一个人机协作就做不完整。1.1 核心术语对照为了读懂后面的代码和官方文档有必要先记住几个必须的组件State状态就是图在执行过程中需要记下来的所有数据。通常用一个 Python 字典继承自 TypedDict来定义每个节点都可以读并且更新这个字典。Node节点图中一个执行单元就是一段普通的 Python 函数接收当前状态返回更新后的状态或者一种叫做 Command 的特殊指令。Checkpointer检查点负责把图的每一步状态都存下来。没有它中断了也没办法恢复。开发的时候可以用 MemorySaver上线之后必须换成能持久化的版本比如 PostgresSaver。thread_id线程标识是 config 里面的一个字段用来区分不同的执行流。恢复的时候必须用和中断时完全相同的 thread_idLangGraph 才知道该取哪一份存档。Command命令一个特殊的数据结构用来恢复执行。其中 resume 字段的值会成为 interrupt() 的返回值。另外 Command 还有一个 goto 字段可以在同一个节点里面动态决定下一步去哪个节点。1.2 中断时底层发生了什么当代码执行到 interrupt(value) 的时候LangGraph 会做下面几件事检查在本次 invoke或 stream的上下文里有没有预留给这个 interrupt() 的恢复值。如果没有恢复值它就抛出一个叫 GraphInterrupt 的异常。这个异常里面包裹着你在 interrupt() 里传入的那个 value。运行时捕获到 GraphInterrupt 后立即通知相关模块把当前的图状态完整保存到检查点中使用的正是你在 compile 时传入的那个 checkpointer。图的执行到此为止控制权交还给调用方调用方可以从返回结果的 __interrupt__ 字段中拿到 interrupt() 里面传出的那个值。如果调用方在后续的恢复正常流程里传入了 Command(resumesomething)当图再次执行到同一个 interrupt() 时LangGraph 就不再抛异常而是直接把这个 something 作为返回值交给节点继续往下走。特别重要的一点是恢复执行的时候节点从头开始执行而不是从 interrupt() 调用的下一行继续。这意味着如果某个操作写在 interrupt() 之前它有可能会被执行两次。这个特性叫“可重入性”下文会有专项说明。1.3 两条关键的使用前提官方文档明确强调使用 interrupt() 必须满足两个必要条件编译图的时候必须配置 checkpointer。不管是 MemorySaver仅适合开发和测试还是生产环境里用的 PostgresSaver一个都不能省。调用 invoke 时必须传入 config 参数并且在 configurable 里面设置一个唯一的 thread_id。同一个图的不同用户或不同对话要用不同的 thread_id 来区分。只要没有 checkpointer 或者没有 thread_id中断机制就不会生效。1.4 静态断点与动态中断的区别在 LangGraph 里面还有另一种“中断”方式叫作静态断点。静态断点是用 compile() 的时候通过 interrupt_before 和 interrupt_after 参数配置的比如写成 graph.compile(checkpointercheckpointer, interrupt_before[node_name]) 的形式会让图在进入某个节点之前强制暂停。但静态断点有两个局限性第一断点位置是在编译时就固定了的不能在程序运行时根据实际情况动态决定是否暂停第二interrupt_before 只是让图停下来并不能像 interrupt() 那样在同一个节点内部自由地嵌入暂停逻辑也无法用 resume 的方式把从外部收到的数据直接返回给正在执行的函数。所以在真实的生产业务里绝大多数人用 interrupt() 而不是静态断点。静态断点更多地被用在调试开发阶段用来逐步检查图的执行情况。1.5 特性速查对比为了方便快速查阅把静态断点和动态中断的几个关键区别整理如下表格特性静态断点interrupt_before / after动态中断interrupt配置时机编译图时固定运行时由业务逻辑决定暂停位置节点之前或之后节点内部的任意行阻塞粒度整个节点级别代码行级别携带载荷无法携带只暂停可携带任意 JSON 载荷获取返回值只能通过状态快照间接读取interrupt()直接返回适用场景调试、固定拦截点审批流程、人工输入、工具审批等有了这些基础知识的铺垫就可以看看在实际项目里怎么用了。二、快速入门一个能跑起来的最小例子在深入业务场景之前先把最核心的流程用一个最简单的例子串起来。这个例子不涉及任何复杂逻辑只是为了让你亲眼看到“停一下然后恢复数据传回来了”这个完整过程。代码中的注释会讲清楚每一行在做什么。# ---------- 1. 导入必要的模块 ---------- from typing import TypedDict, Optional from langgraph.types import interrupt, Command from langgraph.graph import StateGraph, START from langgraph.checkpoint.memory import MemorySaver # ---------- 2. 定义状态 ---------- # 继承自 TypedDict定义图里需要记录哪些数据 class MyState(TypedDict): info: str # 存放一些初始信息 user_response: list[str] # 存放用户通过恢复流程传回的数据 # ---------- 3. 定义节点 ---------- def ask_user_node(state: MyState): 这个节点只会做一件事停下来问用户一个问题然后把用户的回答存进状态。 注意如果这个节点是在中断恢复后被重新执行interrupt() 不会抛异常而是直接返回恢复值。 print(系统我需要你输入一些内容否则没法继续。) # 中断并且把问题的文本作为载荷传给调用方 # 调用方在返回结果里的 __interrupt__ 字段中能看到这个载荷 answer interrupt({question: 请随便说点什么}) print(f系统收到回复啦你说的是{answer}) # 把用户的回答追加到 user_response 列表里这样状态中就多了一条记录 return {user_response: [answer]} # ---------- 4. 构建状态图 ---------- builder StateGraph(MyState) builder.add_node(ask, ask_user_node) # 加入节点 builder.add_edge(START, ask) # 从 START 连到 ask图一启动就执行 ask 节点 # ---------- 5. 配置检查点 ---------- # 重要checkpointer 是中断能够持久化的关键。在开发阶段用 MemorySaver适合测试运行。 checkpointer MemorySaver() # 编译图把 checkpointer 传进去 graph builder.compile(checkpointercheckpointer) # 配置 configthread_id 用于标识当前这个对话流程 config {configurable: {thread_id: demo-001}} # ---------- 6. 第一次执行触发中断 ---------- print( 第一次执行会触发中断 ) result graph.invoke( {info: 初始化数据, user_response: []}, configconfig ) # 中断被触发后invoke 的返回结果里会包含一个特殊的 __interrupt__ 字段 print(中断载荷:, result[__interrupt__][0].value) # ---------- 7. 第二次执行恢复 ---------- print(\n 第二次执行恢复继续带上用户回复 ) final graph.invoke( Command(resume用户说这篇文章写得不错。), configconfig ) print(最终状态:, final)这个例子跑起来后第一步先中断返回结果中的 __interrupt__ 会显示出你传入 interrupt() 的那个字典。第二步把恢复值塞进 Command(resume...)图重新执行 ask_user_node这次 interrupt() 不再停止而是直接返回用户给的字符串。把这个例子跑通之后后面的四个场景看起来就完全不陌生了因为它们只是在同一个核心机制的基础上增加了真正的业务逻辑。三、场景一高风险操作必须经过人工审批3.1 场景描述金融转账、数据库删除操作、发出批量邮件等场景随便哪个都不应该由 AI 自己做主。最直接的做法是AI 在执行这些操作之前主动停下来并要求一个具备权限的管理员做出“批准”或“拒绝”的决定。然后根据决定的结果要么执行要么取消。3.2 技术要点在执行真正有风险的操作前插入一个“审批节点”。审批节点里面调用 interrupt()并把操作详情作为载荷抛出去。审批人做出的决定True 或 False通过 Command(resume...) 传回。节点根据恢复值决定路由批准就走向执行节点拒绝就直接走向取消节点。3.3 完整代码from typing import Literal, Optional, TypedDict from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph, START, END from langgraph.types import Command, interrupt class ApprovalState(TypedDict): action_details: str # 描述需要审批的操作 status: Optional[Literal[pending, approved, rejected]] def approval_node(state: ApprovalState): 审批节点向外部抛出需要审批的操作详情然后根据人的决定决定下一步去哪个节点。 print(f待审批的操作{state[action_details]}) print(等待人工决策...) # 中断把操作详情打包成一个字典抛出去交给审批人看 decision interrupt({ title: 请审批此操作, question: 您是否批准执行上述高风险操作, details: state[action_details], }) # 根据恢复值决定路由 if decision: print(决策通过路由至执行节点。) return Command(gotoexecute) else: print(决策拒绝路由至取消节点。) return Command(gotocancel) def execute_node(state: ApprovalState): 模拟执行真正的业务操作。 print(f正在执行操作{state[action_details]}) return {status: approved} def cancel_node(state: ApprovalState): 模拟操作被取消后的处理。 print(f已拒绝执行操作{state[action_details]}) return {status: rejected} # 构建状态图 builder StateGraph(ApprovalState) builder.add_node(approval, approval_node) builder.add_node(execute, execute_node) builder.add_node(cancel, cancel_node) builder.add_edge(START, approval) builder.add_edge(execute, END) builder.add_edge(cancel, END) checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: approval-123}} # 第一次调用触发中断 initial graph.invoke( {action_details: 删除所有用户测试数据, status: pending}, configconfig ) print(中断载荷:, initial[__interrupt__][0].value) # 模拟审批人批准也可以改成 False 模拟拒绝 final graph.invoke(Command(resumeTrue), configconfig) print(最终状态:, final[status])在实际生产环境里中断载荷中的 details: state[action_details] 发送给前端的审批中心管理员在界面上点击批准之后后端就会调用 graph.invoke(Command(resumeTrue)) 把流程走下去。这个过程中图的状态一直完好地保存在检查点里面即使审批人是三个小时后才点击的批准也能原样恢复。四、场景二AI 生成的文本允许人工审阅和编辑4.1 场景描述AI 自动生成了一段回复邮件、产品描述或新闻稿但因为 AI 偶尔会犯错所以需要在真正使用之前让一个有经验的人来审阅一遍甚至可以当场修改其中的语句。修改完成之后把改好的文本作为最终版本存回状态。4.2 技术要点在需要人工审阅的节点里用 interrupt() 抛出原文。Command(resume...) 传回的是人工修改后的版本。节点把改过的文本存回状态。4.3 完整代码from typing import TypedDict from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph, START, END from langgraph.types import Command, interrupt class ReviewState(TypedDict): generated_text: str def review_node(state: ReviewState): 审阅节点把 AI 生成的内容抛出去让人修改人修完之后把新内容存回状态。 print(AI 生成的原始草稿) print(state[generated_text]) # 中断把原文交给审阅人。载荷不仅包含原文还可以包含指导语。 edited interrupt({ instruction: 请审阅并修改以下内容, original: state[generated_text], }) print(已收到修改后的版本。) return {generated_text: edited} # 构建图 builder StateGraph(ReviewState) builder.add_node(review, review_node) builder.add_edge(START, review) builder.add_edge(review, END) checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: review-001}} # AI 先自动生成一份草稿 ai_draft 尊敬的用户我们愉快速通知你您的订单已经提交成功我们将会尽快给您发货。 initial graph.invoke({generated_text: ai_draft}, configconfig) print(中断载荷:, initial[__interrupt__][0].value) # 模拟人工修改修正错别字并调整语气 human_corrected 尊敬的用户我们愉快地通知您您的订单已提交成功我们将尽快安排发货。 final graph.invoke(Command(resumehuman_corrected), configconfig) print(最终版本:, final[generated_text])这个模式适用于几乎所有“AI 初稿 人工定稿”的场景比如社交媒体运营草稿、客服答复内容、产品公告等。干扰和人工输入的边界清晰而且编辑过程天然就是异步的。五、场景三AI 调用工具之前让人拦截并且可以修改参数5.1 场景描述一个自主智能体Agent在决策过程中可能会调用一些外部工具例如发邮件、调用第三方 API、写数据库等。这些工具调用往往带有参数。理想的做法是当智能体准备执行某个敏感工具时先中断让人看一眼调用参数可以批准并执行也可以当场修改某个参数以后再执行甚至可以完全拒绝这次调用。5.2 技术要点直接在工具函数内部调用 interrupt()。中断的载荷中包含工具的名称、参数以及其他上下文信息。Command(resume...) 传回的字典不仅包含批准与否的信号还可以包含修改后的参数。工具内部根据恢复值决定实际执行还是放弃。5.3 完整代码from typing import TypedDict from langchain_core.tools import tool from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph, START, END from langgraph.types import Command, interrupt class AgentState(TypedDict): messages: list[dict] tool def send_email(to: str, subject: str, body: str): 模拟发送邮件的工具。 在实际发送之前中断并等待人工审批。 人类可以通过恢复值来修改收件人、主题或正文。 print(f准备发送邮件收件人{to}主题{subject}) # 中断并把完整的邮件信息抛出去 decision interrupt({ tool: send_email, to: to, subject: subject, body: body, message: 请审批此邮件是否允许发送如果需要可修改任何字段。, }) if decision.get(approve): # 允许发送使用决策中可能修改过的参数如果没有修改就用原始值 final_to decision.get(to, to) final_subject decision.get(subject, subject) final_body decision.get(body, body) print(f邮件已发送至 {final_to}主题{final_subject}) return f成功发送邮件至 {final_to} else: print(邮件发送已被拒绝。) return 邮件发送已取消 def agent_node(state: AgentState): 模拟一个简单的 Agent 节点第一次调入时模拟 AI 决定调用 send_email 工具。 if len(state[messages]) 1: # 模拟从 LLM 解析出的工具调用参数 tool_args { to: adminexample.com, subject: 今晚部署计划, body: 预计今晚 22:00 开始部署服务可能短暂不可用。 } result send_email.invoke(tool_args) return { messages: state[messages] [ {role: assistant, content: 请求调用 send_email 工具}, {role: tool, name: send_email, content: result} ] } return {messages: state[messages]} # 构建图 builder StateGraph(AgentState) builder.add_node(agent, agent_node) builder.add_edge(START, agent) builder.add_edge(agent, END) checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: tool-789}} # 第一次调用触发中断 initial graph.invoke( {messages: [{role: user, content: 给管理员发一封部署通知邮件}]}, configconfig ) print(中断载荷:, initial[__interrupt__][0].value) # 模拟人工操作批准邮件并且修改了主题使其更正式 human_decision { approve: True, subject: 【正式通知】关于今晚部署计划的详细说明 } final graph.invoke(Command(resumehuman_decision), configconfig) print(最终消息:, final[messages][-1][content])生产场景中可以把 interrupt() 的载荷内容发送到企业微信或钉钉审批卡片上管理人员直接在卡片上点同意或拒绝甚至可以不切换界面就修改工具参数。整个流程既能保证 AI 的自主性又在最关键的地方保留了人工的绝对控制权。六、场景四表单反复校验直到用户输入正确6.1 场景描述有些场景需要用户手动输入信息例如姓名、年龄、邮箱等。用户可能第一次输入时格式就不对比如年龄填了“三十”而不是数字。此时系统不应该直接报错退出而是应该在同一个节点里面反复追问直到拿到符合格式要求的数据才继续往后走。这种方法在交互式工作流里很常见。6.2 技术要点在一个节点内部配合 while True 循环多次调用 interrupt()。第一次进来时 interrupt() 抛异常中断恢复后再进入节点interrupt() 会直接返回传入的 resume 值。检查返回值格式是否正确如果正确就退出循环返回更新后的状态如果不正确就继续用 interrupt() 再次抛出新的提示信息。每个节点负责校验一个字段姓名、年龄等完成一个之后图的状态更新再进下一个节点。6.3 完整代码from typing import TypedDict from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph, START, END from langgraph.types import Command, interrupt class FormState(TypedDict): name: str age: int def ask_name(state: FormState): 收集姓名反复验证直到用户输入了一个非空的纯文字字符串。 # 第一次中断抛出提示信息 user_input interrupt(请输入您的姓名) while True: if isinstance(user_input, str) and len(user_input.strip()) 0: # 输入有效存入状态并退出节点 return {name: user_input.strip()} # 输入无效提示用户重新输入并再次中断 user_input interrupt(姓名不能为空请重新输入) def ask_age(state: FormState): 收集年龄反复验证直到用户输入的是一个大于0的整数。 print(f您好{state[name]}) user_input interrupt(请输入您的年龄整数) while True: try: age int(user_input) if age 0: return {age: age} except (ValueError, TypeError): pass # 不是有效的整数继续追问 user_input interrupt(f{user_input} 不是有效的年龄请输入一个正整数) # 构建图 builder StateGraph(FormState) builder.add_node(ask_name, ask_name) builder.add_node(ask_age, ask_age) builder.add_edge(START, ask_name) builder.add_edge(ask_name, ask_age) builder.add_edge(ask_age, END) checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: form-001}} # 开始填表 state graph.invoke({name: , age: 0}, configconfig) print(中断信息:, state[__interrupt__]) # 第一次输入姓名有效 state graph.invoke(Command(resume张三), configconfig) print(当前状态:, state) # 故意输错年龄字符串 state graph.invoke(Command(resume十八), configconfig) print(再次中断:, state[__interrupt__]) # 最后输入正确的年龄 final graph.invoke(Command(resume30), configconfig) print(最终结果:, final)这种“节点内部循环 每隔一次 interrupt() 追问一次”的设计模式是 LangGraph 中实现多轮人工输入交互最简洁有效的写法它避免了你用额外的方式在外部再来一遍复杂的重试管理因为重试逻辑本来就包含在节点代码里。七、四条铁律别在这些地方翻车LangGraph 官方文档针对 interrupt() 的使用总结了四条非常重要的注意事项。下面从实用角度一一说明如果忽略这些中断时会出现各种难以排查的问题。7.1 不要用try/except包住interrupt()interrupt() 在第一次被调用的时候会抛出一个名叫 GraphInterrupt 的异常LangGraph 运行时恰恰是靠捕获这个异常才知道要暂停的。如果你用 try 把它包起来又不在 except 里面把这个异常重新抛出去就相当于把 LangGraph 的一个关键信号给吃掉了后果就是图不会正常暂停程序会按照你的错误处理代码胡乱走下去。正确的做法是try 只用于捕获自己的业务异常而让 interrupt() 留在 try 块之外或者如果确实需要放在 try 里就在 except GraphInterrupt 中把它重新抛出。7.2 节点内不要用分支改变interrupt()的顺序当在同一个节点里调用多个 interrupt() 时LangGraph 是严格按照它们在代码中出现的顺序来匹配恢复值的。换句话说第一个 interrupt() 对应第一次恢复时传给 Command(resume...) 的值第二个 interrupt() 对应第二次恢复的值。如果你用 if ... else ... 有条件地调用 interrupt()可能会导致两种不同的执行路径下这些 interrupt() 出现的顺序不一样。一旦顺序错位恢复值就会被送到错误的 interrupt() 调用上去造成数据混乱。错误顺序正确做法用分支控制哪些interrupt()会执行将每个interrupt()放进独立的节点在同一个节点内动态决定interrupt()的位置每个节点保持固定的interrupt()结构依赖条件决定是否跳过某个interrupt()用Command(goto...)跳过不需要的节点解决这个问题最好的办法就是把可能需要分支处理的逻辑拆成不同的节点每个节点里面最多只有一个 interrupt()节点之间通过 Command(goto...) 来控制路由。7.3 写在interrupt()之前的代码可能会执行两次必须保证幂等性恢复执行的时候节点不是从 interrupt() 的下一行继续运行的而是从头开始重新执行整个节点。这意味着 interrupt() 之前的所有代码每次恢复的时候都会再次执行一遍。举例来说如果在 interrupt() 之前就执行了扣费操作第一次执行时扣了一次恢复后节点从头执行又扣了一次就造成重复扣费。解决的方法有两个方向。第一种方法是把有副作用写数据库、发送消息、扣费等的操作挪到 interrupt() 之后去执行并且把它放在一个不会被重复执行的独立节点中。第二种方法是给外部操作带上幂等键idempotency key即使重复调用也能被外部系统识别出来只产生一次有效效果。这是实现可重入性的核心要求。7.4 必须配置checkpointer和thread_id这个规则虽然多次提到但依然值得单独强调没有 checkpointer中断状态就存不住程序一重启就没法恢复没有 thread_id图不知道从哪个会话存档里恢复数据恢复时也找不到对应的中断点。开发测试阶段可以用 MemorySaver但一旦部署到生产环境就必须换成能持久化到数据库的检查点实现例如 PostgresSaver以确保中断数据不会因为应用重启而丢失。八、生产环境进阶把检查点从内存搬到数据库MemorySaver 只在开发调试阶段适用因为它只把检查点存在内存里。一旦程序挂了或者重启了之前的所有中断状态就全部丢失了。在企业级生产环境里必须用一个可以持久化的数据库作为 checkpointer目前最简单也最常用的方案是 PostgreSQL。LangGraph 提供了 PostgresSaver 类它把每一步的图状态写进 PostgreSQL 数据库中的特定表里。用起来和 MemorySaver 的写法几乎一样只不过在创建 PostgresSaver 的时候需要指定数据库的连接池。首次使用的时候还需要调用 .setup() 方法来创建必要的表结构。一个简单的用法示例from psycopg_pool import ConnectionPool from langgraph.checkpoint.postgres import PostgresSaver # 创建数据库连接池 pool ConnectionPool(conninfopostgresql://user:passwordlocalhost:5432/langgraph_db) checkpointer PostgresSaver(sync_connectionpool) # 初次使用时需要创建表 checkpointer.setup() # 后面的用法和 MemorySaver 完全一样 graph builder.compile(checkpointercheckpointer)官方推荐在生产环境中使用连接池模式来管理 PostgresSaver这样可以支持更高的并发量也能更好地保证中断机制的可靠性。切换到数据库后最大的不同是就算整个应用重启中断状态也不会消失用户可以随时再恢复。因此 thread_id 也就具备了真正的跨会话、跨流程的意义。九、总结LangGraph 的中断机制是解决人机协作问题的关键方案。它不是简单地在代码中插入一个 input()而是一套完整的“暂停—持久化—恢复”体系。通过本文的讲解和四个代码示例应该能看到这套机制在实际项目中的价值高风险操作审批场景给了管理员一个在 AI 真正动刀之前放行的机会。AI 内容人工审阅和编辑场景让模型生成的草稿在发布之前能被人类润色把关。工具调用拦截审查场景提供了在智能体自主执行工具之前拦住并修改参数的灵活能力。表单循环验证场景确保用户填入的数据质量让交互工作流的健壮性大大提升。与此同时也必须时刻记住四个关键规则避免捕获 GraphInterrupt 异常、保证节点内 interrupt() 顺序固定、确保 interrupt() 前的操作具备幂等性、持久化检查点绝不能用 MemorySaver 对付生产环境。如果把上述内容全部吃透再加上合适的数据库持久化方案就可以放心地让自主智能体跑在核心业务路径上同时让人在最重要、最敏感的地方保持最终控制权。现在就可以动手在项目里尝试一下 interrupt() 了哪怕是先从最简单的“AI 发重要消息必须经过二次确认”做起也会让整个系统变得更可控、更令人信赖。