本文详解LangGraph的动态中断机制，实现真正的人机交互功能。对比静态中断与动态中断区别，介绍四大核心组件：Checkpointer、thread_id、interrupt()和Command。通过"转账审批"实战案例展示人工审批工作流构建，强调"幂等性"和"节点重跑"两大黄金法则。掌握此技术可使Agent进化为可控可靠的智能助手。

---

引言

在此前的教程中，我们已经掌握了 LangGraph 的核心组件，包括（在早先版本中）用于调试的静态中断。许多读者在实战中发现，静态中断（interrupt_before / interrupt_after）虽然可以作为“断点”来观察状态，但它并不能实现我们真正需要的人机交互（Human-in-the-Loop, HITL）功能，例如：

• • 在 Agent 执行关键操作（如删除数据库、调用付费 API）前，人工“批准”或“拒绝”。
• • 在 Agent 生成初稿后，人工“审查”并“编辑” 其内容。
• • 在 Agent 需要额外信息时，人工“输入” 所需的数据。

这些功能是构建可靠、可控 Agent 的基石。值得注意的是，LangGraph 提供了另一套更强大的机制来解决这个问题，这就是动态中断 (Dynamic Interrupts)。

本期教程将作为一篇补充内容，专门深入解析 LangGraph 的人机交互机制，帮你彻底分清两种“中断”的区别，并掌握如何构建真正需要人工干预的“审批”流程。 在本期教程中，你将掌握：

• • 两种中断模式的对比：静态中断（调试）vs 动态中断（人机交互）。
• • 动态中断组件Checkpointer, thread_id, interrupt(), Command。
• • 综合实战：从零构建一个需要人工批准的“审批”工作流。
• • 黄金法则：使用动态中断时必须遵守的“幂等性”与“节点重跑”原则。

---

第一部分: 重新定义“中断” - 两种模式的对比

要掌握人机交互，我们必须首先厘清 LangGraph 中两种“中断”的根本区别。它们的设计目的、使用方法和恢复机制截然不同。

模式一：静态中断 (Static Interrupts) - “调试断点”

定义方式: 在 graph.compile() 或 graph.invoke() 时，作为参数传入。

# 在编译时设置graph = builder.compile(    interrupt_before=["node_name"])# 或在运行时设置graph.invoke(..., interrupt_after=["node_name"])

核心目的: 调试 (Debugging)。它允许开发者在某个节点执行前或执行后暂停图，以便检查（get_state）当时的 State 状态，类似于 IDE 中的“断点”。

恢复方式:

# 传入 None 来恢复，表示“继续执行”graph.invoke(None, config=config)

局限性: 这是一个“只出不进”的暂停。它无法在恢复时将“人工决策”（如 True 或 "已修改的内容") 传递回 节点内部。因此，它不能用于真正的人机交互。

模式二：动态中断 (Dynamic Interrupts) - “人机交互” (HITL)

定义方式: 在节点函数内部，直接调用 langgraph.types.interrupt 函数。

from langgraph.types import interruptdef approval_node(state: AgentState):# ...# 在代码逻辑中动态触发    decision = interrupt("请批准此操作")# ...

核心目的: 人机交互 (Human-in-the-Loop)。它允许图在运行时根据特定逻辑暂停，等待用户的输入（如批准、编辑、提供数据）。 恢复方式:

from langgraph.types import Command# 传入 Command(resume=...) 来恢复# resume 传入的值将成为 interrupt() 的返回值graph.invoke(Command(resume=True), config=config)

强大之处: 这是一个“有进有出”的暂停。它不仅暂停，还能在恢复时接收一个值（resume=... 传入的值），这个值会成为 interrupt() 函数的返回值（上例中的 decision 变量），从而驱动后续的业务逻辑。

---

第二部分: “动态中断”的四个核心组件

要使“动态中断”按预期工作，我们必须同时使用到四个核心组件，它们缺一不可。

Checkpointer (状态记录)

• • 职责：持久化。当 interrupt() 被调用时，Checkpointer 负责将当前 Graph 的完整状态保存到数据库（如 MemorySaver, SqliteSaver, RedisSaver）。
• • 关键：没有 Checkpointer，动态中断无法工作。Graph 必须有办法“存档”，才能在未来“读档”并恢复。

Config 中的 thread_id (会话 ID)

• • 职责：唯一标识。thread_id 就像是你的“游戏存档文件名”。
• • 关键：你必须使用一个固定的 thread_id (通过 config={"configurable": {"thread_id": "..."}} 传入) 来调用 Graph。当恢复时，LangGraph 才知道要加载哪一个被暂停的会话。

interrupt() (暂停函数)

• • 职责：执行暂停。在节点中调用它时，它会：

1. 1. 抛出一个特殊信号，通知 LangGraph 框架“暂停”。
2. 2. 框架命令 Checkpointer 保存当前 thread_id 的状态。
3. 3. invoke() 调用立即返回。interrupt() 中传递的参数（如 "请批准"）会包含在返回结果的 __interrupt__ 字段中，用于展示给用户。

Command(resume=…) (恢复指令)

• • 职责：恢复执行。当用户做出决策后，再次调用 graph.invoke()，但这次传入的不是输入数据，而是一个 Command(resume=...) 对象。
• • 关键：传入 resume 的值（如 True, False 或一个包含编辑后文本的字典）将在节点恢复执行时，被 interrupt()函数捕获并作为其返回值。

---

第三部分: 综合实战 - 构建“人工审批”流程

现在，我们从零开始构建一个“转账审批”流程，演示这四个组件如何协同工作。

3.1 基础设置与状态定义

我们定义一个简单的 State，包含操作详情和当前状态。

import timefrom typing import TypedDict, Literal, Optionalfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.checkpoint.memory import MemorySaver from langgraph.types import Command, interrupt       class AgentState(TypedDict):    action_details: str  status: Optional[Literal["pending", "approved", "rejected"]]

3.2 节点定义 (审批节点)

这是实战的核心。我们将 interrupt() 嵌入到节点逻辑中。

def approval_node(state: AgentState):    """    一个需要人工批准的节点。    注意：这个节点在“恢复”时会从头重新运行。    """    print("--- 节点 [approval_node] 开始执行 ---")    print(f"  > 待办操作: {state['action_details']}")    # 仅在 'pending' 状态时才触发中断    if state['status'] == 'pending':    # 调用 interrupt() 来暂停    # 'payload' 将会返回给调用者    payload = {        "question": "您是否批准此操作？",        "details": state["action_details"]    }    print("  > 暂停，等待人工批准...")    # 第一次运行：Graph 在此暂停。    # 恢复运行时：decision 将被赋予 Command(resume=...) 中的值。    decision = interrupt(payload)     print(f"  > 收到人工决策: {decision}")    # 根据决策更新状态    if decision:        return {"status": "approved"}    else:        return {"status": "rejected"}    # 如果状态不是 'pending' (例如在重跑时)，则跳过    print(f"  > 状态为 {state['status']}, 跳过中断。")    return {}def proceed_node(state: AgentState):    print("--- 节点 [proceed_node] 执行 ---")    print(f"正在执行操作: {state['action_details']}")    return {} def cancel_node(state: AgentState):    print("--- 节点 [cancel_node] 执行 ---")    print(f"取消操作: {state['action_details']}")    return {}

3.3 构建 Graph (带条件路由)

我们使用条件路由，根据 approval_node 之后的 status 状态决定下一步。

builder = StateGraph(AgentState)# 添加节点builder.add_node("approval", approval_node)builder.add_node("proceed", proceed_node)builder.add_node("cancel", cancel_node)# 设置入口builder.set_entry_point("approval")# 定义条件路由def route_decision(state: AgentState):    if state["status"] == "approved":        return"proceed"    else:        return"cancel"# 'approval' 节点完成后，根据 'status' 决定去向builder.add_conditional_edges(    "approval",    route_decision,    {        "proceed": "proceed",        "cancel": "cancel"    })# 最终节点builder.add_edge("proceed", END)builder.add_edge("cancel", END)

3.4 编译与执行（暂停与恢复）

这是演示“暂停-恢复”流程的关键。

# 1. 核心组件：实例化 Checkpointercheckpointer = MemorySaver()# 编译 Graph，必须传入 checkpointergraph = builder.compile(checkpointer=checkpointer)# 2. 核心组件：定义一个唯一的 thread_idconfig = {"configurable": {"thread_id": "tx-12345"}}initial_input = {"action_details": "向用户 'A' 转账 500RMB", "status": "pending"}# 第一次调用，触发暂停print("--- 第一次运行 (将触发暂停) ---")# Graph 会运行到 approval_node，调用 interrupt()，然后暂停result = graph.invoke(initial_input, config=config)print("\n--- Graph 已暂停 ---")print("  > Graph 的当前状态 (已保存):")print(f"  > {graph.get_state(config).values}")print("\n  > 'interrupt' 返回的数据 (用于展示给用户):")# 注意这个特殊的 `__interrupt__` 字段print(f"  > {result['__interrupt__']}") # 此时应用程序（如 Web UI）会向用户展示 'result['__interrupt__']' 的内容# 用户审查后，决定 "批准" (True)time.sleep(1)human_decision = Trueprint(f"\n--- 用户已做出决策: {human_decision} ---")# 第二次调用，使用 Command 恢复print("--- 恢复 Graph 运行 ---")# 使用 Command(resume=...) 和 *相同的 config* 来恢复# 传入的 `resume=True` 将成为 `interrupt()` 的返回值resume_result=graph.invoke(Command(resume=human_decision), config=config)print("\n--- Graph 运行完毕 ---")print("  > Graph 的最终状态:")print(f"  > {graph.get_state(config).values}")print("\n  > 最后一步的输出:")print(f"  > {resume_result}")

---

第四部分: 黄金法则 - 动态中断的“天坑”

动态中断非常强大，但也引入了一个最容易出错的“天坑”：节点重跑。 当调用 Command(resume=...) 恢复时，LangGraph 不会从 interrupt() 函数的那一行代码继续执行，而是会从头开始重新执行包含 interrupt() 的整个节点函数（即 approval_node）。 这带来了两个必须遵守的“黄金法则”：

幂等性 (Idempotency)

规则：绝对不能在 interrupt() 调用之前放置任何“有副作用”且“非幂等”的操作（如写入数据库、发送 API 请求）。

举个反例 (错误):

def bad_node(state: AgentState):    # 错误！这个操作会执行两次！    # 第一次是暂停前，第二次是恢复后    db.append_to_log("Approval process started...")     decision = interrupt("Approve this action?")    if decision:        db.execute_transfer(...)    return ...

在上面的例子中，db.append_to_log 会在暂停前运行一次，在恢复后（节点重跑时）再次运行，导致数据库中出现重复日志。

正确做法:

• • 做法 A（推荐）：将所有“副作用”操作放在 interrupt() 之后，并由其返回值控制。
• • 做法 B：将“副作用”拆分到单独的节点中（如 proceed_node），利用路由来确保它只在批准后执行一次。

状态驱动 (State-Driven Logic)

规则：由于节点会重跑，你必须使用 State 来防止 interrupt() 被重复触发。 在我们的实战代码中，我们正是这么做的：

def approval_node(state: AgentState):    print("--- 节点 [approval_node] 开始执行 ---")    # 正确：使用状态来控制中断    if state['status'] == 'pending':        # 第一次运行：'pending'，触发中断        decision = interrupt(...)         if decision:            return {"status": "approved"}        else:            return {"status": "rejected"}    # 第二次运行 (恢复后)：    print(f"  > 状态为 {state['status']}, 跳过中断。")    return {}

这种模式确保了 interrupt() 在整个会话中只被触发一次。

---

本期总结

在本期补充教程中，我们彻底理清了 LangGraph 的两种中断机制：

• • 静态中断 (interrupt_before): 仅用于调试的“断点”，无法实现人机交互。
• • 动态中断 (interrupt()): 专为人机交互 (HITL) 设计，是构建审批、编辑、验证流程的核心。

我们通过一个“审批”实战，掌握了实现动态中断的四个组件Checkpointer, thread_id, interrupt, Command，并深刻理解了“节点重跑”和“幂等性”这两条黄金法则。 掌握了动态中断，你的 Agent 才真正从一个“自动化脚本”进化为了一个可控、可靠的“智能助手”。

后续

至此，我们已经掌握了构建一个复杂 Agent T 团队的全部核心技术：从单 Agent 的流式处理与中断，到多 Agent 的团队协作与路由。我们的 team_app 已经是一个功能强大、可观测的图。 然而，它目前还只是一个在 Python 脚本中运行的对象。要将其转变为一个真正可用的、能被前端（如网页、APP、微信小程序）调用的生产级服务，我们还差最后一步：服务化部署。 在下一期中，我们将进入实战部署阶段，学习如何使用 FastAPI 和 LangServe：

• • 将我们构建的 LangGraph app 封装为健壮的、可异步并发的 API 接口。
• • 利用 LangServe 轻松实现 astream 和 astream_log 的流式 WebSocket 接口。
• • 最终实现一个端到端、可观测、可部署的完整 RAG 服务。

如何学习AI大模型 ？

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。【保证100%免费】🆓

CSDN粉丝独家福利

这份完整版的 AI 大模型学习资料已经上传CSDN，朋友们如果需要可以扫描下方二维码&点击下方CSDN官方认证链接免费领取 【保证100%免费】

读者福利： 👉👉CSDN大礼包：《最新AI大模型学习资源包》免费分享 👈👈

（👆👆👆安全链接，放心点击）

对于0基础小白入门：

如果你是零基础小白，想快速入门大模型是可以考虑的。

一方面是学习时间相对较短，学习内容更全面更集中。
二方面是可以根据这些资料规划好学习计划和方向。

👉1.大模型入门学习思维导图👈

要学习一门新的技术，作为新手一定要先学习成长路线图，方向不对，努力白费。

对于从来没有接触过AI大模型的同学，我们帮你准备了详细的学习成长路线图&学习规划。可以说是最科学最系统的学习路线，大家跟着这个大的方向学习准没问题。（全套教程文末领取哈）

👉2.AGI大模型配套视频👈

很多朋友都不喜欢晦涩的文字，我也为大家准备了视频教程，每个章节都是当前板块的精华浓缩。

👉3.大模型实际应用报告合集👈

这套包含640份报告的合集，涵盖了AI大模型的理论研究、技术实现、行业应用等多个方面。无论您是科研人员、工程师，还是对AI大模型感兴趣的爱好者，这套报告合集都将为您提供宝贵的信息和启示。（全套教程文末领取哈）

👉4.大模型实战项目&项目源码👈

光学理论是没用的，要学会跟着一起做，要动手实操，才能将自己的所学运用到实际当中去，这时候可以搞点实战项目来学习。（全套教程文末领取哈）

👉5.大模型经典学习电子书👈

随着人工智能技术的飞速发展，AI大模型已经成为了当今科技领域的一大热点。这些大型预训练模型，如GPT-3、BERT、XLNet等，以其强大的语言理解和生成能力，正在改变我们对人工智能的认识。 那以下这些PDF籍就是非常不错的学习资源。（全套教程文末领取哈）

👉6.大模型面试题&答案👈

截至目前大模型已经超过200个，在大模型纵横的时代，不仅大模型技术越来越卷，就连大模型相关的岗位和面试也开始越来越卷了。为了让大家更容易上车大模型算法赛道，我总结了大模型常考的面试题。（全套教程文末领取哈）

为什么分享这些资料?

只要你是真心想学AI大模型，我这份资料就可以无偿分享给你学习，我国在这方面的相关人才比较紧缺，大模型行业确实也需要更多的有志之士加入进来，我也真心希望帮助大家学好这门技术，如果日后有什么学习上的问题，欢迎找我交流，有技术上面的问题，我是很愿意去帮助大家的！

这些资料真的有用吗?

这份资料由我和鲁为民博士共同整理，鲁为民博士先后获得了北京清华大学学士和美国加州理工学院博士学位，在包括IEEE Transactions等学术期刊和诸多国际会议上发表了超过50篇学术论文、取得了多项美国和中国发明专利，同时还斩获了吴文俊人工智能科学技术奖。目前我正在和鲁博士共同进行人工智能的研究。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

CSDN粉丝独家福利

这份完整版的 AI 大模型学习资料已经上传CSDN，朋友们如果需要可以扫描下方二维码&点击下方CSDN官方认证链接免费领取 【保证100%免费】

读者福利： 👉👉CSDN大礼包：《最新AI大模型学习资源包》免费分享 👈👈

（👆👆👆安全链接，放心点击）