LangGraph 中断与人工介入

这篇适合谁、读完能做到什么

这篇适合在 LangGraph 里写流程、需要在某一步停下来等人确认或补充信息的开发者。读完你能在节点里调用 interrupt() 暂停执行，知道状态是怎么保存、怎么恢复的，会用 Command 把人的回答交回去，并且知道节点重跑会带来哪些必须遵守的规则。

是什么：可以停在任意位置的中断

中断（interrupt）允许你在图执行到某一点时暂停，等待外部输入后再继续。这正是人机协作的基础：需要人批准一次 API 调用、一次数据库改动或一笔交易时，流程停在那里，等人的答复。^[1]

中断触发时，LangGraph 会用持久化层（persistence layer）保存图的状态，然后一直等待，直到你恢复执行。^[1] 只要没有人给出恢复值，图就停在暂停状态；之后随时用同一个 thread ID 就能把它接着跑下去。^[1]

中断通过在节点里调用 interrupt() 函数实现，它接受任何可以 JSON 序列化的值，并把这个值交给调用方。^[1] 这个值通常是一句问题，或者一份需要审核的内容摘要，调用方拿到后再决定怎么向人展示。它的形状没有固定要求，只要能被序列化即可。^[1]

中断和静态断点（breakpoint）不同。静态断点只能停在某个节点的前后；中断是动态的，可以放在代码的任意位置，也可以根据业务逻辑决定要不要停。^[1]

准备

• 持久化：中断必须配合 checkpointer 才能保存状态。生产环境要用持久化的 checkpointer，测试和原型阶段可以用内存版。^[1]
• thread ID：调用时在 config 里传入 thread_id，它告诉 checkpointer 该加载哪一份状态。换一个新的值就是一个全新的线程，状态从空开始。^[1]
• 数据：传给 interrupt() 的值必须能被 JSON 序列化。^[1]

上手：暂停、等待、恢复

下面是一个最小的审批节点。节点调用 interrupt 时，执行在这一点暂停，问题文字就是交给调用方的值。恢复时，Command(resume=...) 传入的值会成为这次 interrupt() 的返回值，节点再接着往下走。^[1]

from langgraph.types import interrupt

def approval_node(state: State):
    # Pause and ask for approval
    approved = interrupt("Do you approve this action?")

    # When you resume, Command(resume=...) returns that value here
    return {"approved": approved}

恢复执行时，有三点要记住。第一，必须使用触发中断时的那个 thread ID。第二，Command(resume=...) 的值就是 interrupt() 的返回值。第三，节点会从头重新开始，所以 interrupt() 之前的代码会再跑一遍。^[1]

在 Python 里，推荐的做法是用事件流（graph.stream_events(..., version="v3")）驱动图。中断的值会出现在 stream.interrupts 上，stream.interrupted 为 True 表示这次运行暂停在等待输入。如果用默认的 invoke()，中断值则在 result["__interrupt__"] 里。^[1]

审批例子：按人的回答决定走向

审批是中断最常见的用途。下面的节点把问题和细节交给人，然后根据回答用 Command(goto=...) 跳到 "proceed" 或 "cancel"：

from typing import Literal
from langgraph.types import interrupt, Command

def approval_node(state: State) -> Command[Literal["proceed", "cancel"]]:
    # Pause execution; payload shows up on stream.interrupts (with stream_events) or result["__interrupt__"] (with invoke)
    is_approved = interrupt({
        "question": "Do you want to proceed with this action?",
        "details": state["action_details"]
    })

    # Route based on the response
    if is_approved:
        return Command(goto="proceed")  # Runs after the resume payload is provided
    else:
        return Command(goto="cancel")

要批准，就用 Command(resume=True) 恢复；要拒绝，就用 Command(resume=False)，两种情况都要传入同一个 config：

# To approve
graph.stream_events(Command(resume=True), config=config, version="v3").output

# To reject
graph.stream_events(Command(resume=False), config=config, version="v3").output

这里要注意：Command(resume=...) 是唯一应该作为 invoke()、stream() 或 stream_events() 输入的 Command 写法。要继续多轮对话，应当传普通的输入字典，而不是 Command(update=...)。^[1]

多个中断与回答的结构

interrupt 的 response_schema 参数描述恢复时期望的输入。客户端可以拿它渲染带类型的输入表单。这个参数需要 langgraph 1.2.12 或更高版本。^[1] 你可以传入 Pydantic 模型、typing_extensions 里的 TypedDict 或 dataclass 来校验恢复的值。LangGraph 会把这个类型转成 JSON Schema，放进 Interrupt.response_schema，和 value 分开。^[1]

如果并行的分支同时中断，就要在一次调用里恢复多个中断，方法是把每个中断 ID 映射到它对应的恢复值。^[1]

常见场景包括：审批流程、处理多个中断、审阅并修改状态、中断工具调用，以及校验人工输入。^[1]

节点重跑带来的规则

中断的实现方式决定了几条规则。调用 interrupt() 时，LangGraph 通过抛出一个特殊的异常来挂起执行，运行时捕获这个异常后保存状态。^[1] 恢复时，运行时会从节点的开头重新执行，而不是从调用 interrupt() 的那一行继续。^[1] 由此得出的规则如下。

不要用 try/except 包住 interrupt。 包住后，这个特殊异常会被捕获，中断就不会传回图里。应当把 interrupt 调用和容易出错的代码分开。^[1]

不要在节点里调换 interrupt 的顺序。 恢复值的匹配严格按索引进行，所以同一个节点里 interrupt 调用的先后顺序很重要。^[1]

传给 interrupt 的值要简单。 有些 checkpointer 无法序列化函数之类的复杂对象，所以尽量只传 JSON 可以序列化的简单类型。^[1]

interrupt 之前的副作用要幂等。 节点恢复时会再跑一遍，所以在 interrupt 之前做的写库、调接口等操作，最好可以重复执行而结果不变；能放在 interrupt 之后的副作用，尽量放到后面。^[1]

子图和调试

子图被当作函数在节点里调用时，父图会从调用它的那个节点的开头恢复，子图则从调用 interrupt 的那个节点的开头恢复。^[1]

调试时可以用静态中断。静态中断只能停在节点的前后，需要根据业务数据做判断时，还是要在代码里调用 interrupt()。在编译图的时候设置 interrupt_before 和 interrupt_after，就能把它们当作断点，一次只走一个节点。^[1]

限制和常见问题

• 没有 checkpointer 或没有 thread ID，中断后的状态就无法恢复。^[1]
• 恢复时换了 thread ID，相当于开始了一段新的空线程，原来暂停的状态不会被继续使用。^[1]
• 节点里 interrupt 之前的代码会重复执行，所以要检查那些代码能否安全重跑。^[1]
• 传入不能序列化的值，在某些 checkpointer 下会出错。^[1]

下一步

想了解如何校验人工输入、如何审阅和修改状态，可以接着读 LangGraph 官方文档里的中断章节。子图和流式输出的内容也在同一套文档里，可以按需查阅。^[1]

本文改编自 LangChain 官方文档《Interrupts》（LangGraph 部分）^[1]，原文以 MIT 许可证发布。