06：中断恢复与人机交互

大家好，我是大志。

在上一篇文章中，我们详细介绍了什么是流式输出，以及在LangGraph中如何使用流式输出，本文将会介绍LangGraph的中断(interrupt)和恢复(resume)使用方法和使用场景。

文中所有示例代码：https://github.com/wzycoding/langchain-study

一、什么是中断和恢复

在LangGraph执行过程中，在一些场景下需要在图执行过程中停止执行，并在处理完成后，继续在程序停止的地方继续执行，这里的“停止执行”就是中断，这里的“继续执行”就是恢复，当程序触发中断，LangGraph会使用持久化技术（checkpointer）保存图状态，无限期的等待用户恢复执行。

举一个中断和恢复的例子：在一个LangGraph图应用运行过程中，执行到某个节点时，要执行如转账、扣款等高风险操作，需要人类进行判断是否继续执行下一步，这时就需要先将程序中断，等待人类输入后，恢复图执行，再根据人类的输入来决定是否继续执行，这样就通过中断和恢复实现了人机交互。

二、中断的使用

LangGraph的中断和恢复底层是通过线程和检查点、检查点管理器实现的，因此想要使用中断，必须在编译时指定checkpointer。

在前面讲解线程检查点时使用的checkpointer都是InMemorySaver，但InMemorySaver一般是用于在学习和测试时使用的检查点管理器，并且它是基于内存的，当程序一旦停止执行后，数据就会丢失。

因此，本文将使用一个新的检查点管理器PostgresSaver，它可以将检查点信息存储到数据库中，即使程序退出，重新启动程序，LangGraph依然能获取到这些检查点信息，PostgresSaver适合用于生产环境，LangSmith就是使用该技术对执行过程进行监控。

首先，需要安装PostgresSaver所需的依赖：

pip install -U langgraph-checkpoint-postgres

并安装Postgres数据库驱动，这里有一个坑点，要注意这里安装的是v3版本的数据库驱动，如果安装成v2版本会报错。

pip install -U "psycopg[binary,pool]"

2.1 人工审核

在图执行过程中，在某些关键步骤我们需要人工进行审核来决定是否执行，这里就可以使用中断恢复，假设，有一个业务流程：要给指定公司打款，在打款前需要人工进行审核，审核通过了就完成打款操作并生成账单，审核不通过则直接结束。

将业务逻辑抽象成图，可以把打款作为一个节点，生成账单作为一个节点，在打款节点内部进行中断，直到恢复执行，拿到人类输入的内容，根据内容来决定下一步的执行路径，并且需要给打款节点添加一个条件边和对应的路由函数，通过判断人工审核是否通过来决定下一步执行生成账单节点还是走到结束节点。

首先，定义State

# 1.定义State
class State(TypedDict):
    # 支付金额
    pay_amount: int
    # 收款公司
    receiving_company: str
    # 是否通过审核
    approved: bool

定义打款节点

def pay_node(state: State):
    print("======进入支付节点======")
    approved = interrupt(f"你是否允许给{state['receiving_company']}打款{state['pay_amount']}元？")

    if approved:
        print("======人工审核通过，支付成功======")
    else:
        print("======人工审核拒绝，支付失败======")
    return {"approved": approved}

定义生成账单节点

def generate_bill_node(state: State):
    print("======开始生成账单======")

定义条件边路由函数

def pay_route(state: State) -> Literal["generate_bill", END]:
    if state['approved']:
        return "generate_bill"
    return END

构建图、节点、边。

graph = StateGraph(State)

graph.add_node("pay", pay_node)
graph.add_node("generate_bill", generate_bill_node)

graph.add_edge(START, "pay")
graph.add_conditional_edges("pay", pay_route, ["generate_bill", END])

创建PostgresSaver检查点管理器

conn = connect("postgres://postgres:postgres@localhost:5432/langgraph", autocommit=True)
checkpointer = PostgresSaver(conn)
checkpointer.setup()

编译运行图

agent = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "1"}}
result = agent.invoke({"pay_amount": 6000, "receiving_company": "大米科技有限公司"}, config)

此时接收的result，就是当前图状态数据State，并且在其内部还包含一个特殊变量_interrupt_，在变量中包含interrupt方法中传递的内容。

print(result)

之后，我们再恢复图的执行，在resume中传入False，这里传入的内容就是interrupt中断方法返回的内容，这样就可以根据返回内容，来决定下一步如何执行。

agent.invoke(Command(resume=False), config=config)

执行结果如下：

2.2 内容修改

除了，可以使用中断做人工审核，还可以对图状态数据进行更改，假设有一个业务场景，需要先让LLM生成一封邮件，之后通过人工检查，对生成的邮件进行最终确认，并给出最终发出的邮件内容。

具体实现这里不再罗列重复的代码，基础代码和前面的程序基本一致，这里只给出关键代码，完整代码可以参考代码仓库。

定义图状态数据，其中包括消息列表和最终发送邮件的内容。

class State(TypedDict):
    # 图运行的消息列表
    messages: Annotated[list[AnyMessage], operator.add]
    # 发送邮件内容
    email_content: str

定义图节点，在llm_node中，将调用LLM返回的结果保存到图状态数据中的email_content中，在check_node中，通过中断函数interrupt，让人类对最终邮件内容进行确认。

def llm_node(state: State):
    """调用LLM节点"""
    ai_message = llm.invoke(state["messages"])

    return {"messages": [ai_message], "email_content": ai_message.content}

def check_node(state: State):
    print("======进入人工审核节点======")
    new_email_content = interrupt(f"请核对以下邮件内容,给出最终发送内容：{state['email_content']}")
    return {"email_content": new_email_content}

调用图并输出结果

config = {"configurable": {"thread_id": "101"}}
result = agent.invoke(
    {"messages": [HumanMessage(content="写一封邮件给Bob，通知已经给Bob打款10000元货款，邮件格式保持正式信件格式")]},
    config)

# 9.输出中断后返回内容
print(result['__interrupt__'])

# 10.恢复中断执行
state = agent.invoke(Command(resume="你好，Bob，货款10000元已转账，请知悉"), config=config)
print(state['email_content'])

执行结果如下：

除此之外，还可以使用中断功能完成工具中断、输入校验等功能。

三、使用中断注意点

1.不能使用try/except包裹interrupt方法

在LangGraph节点内调用interrupt时，会通过抛出一个异常来完成暂停执行的操作，异常会不断向上层传递，最终通知图将当前状态完整保存，等待外部输入恢复执行。

因此，不能使用try...except包裹interrupt()方法，如果使用try...except包裹将会捕获这个异常，这样中断就不会生效了。

2.要保证节点中中断的顺序是确定的

在同一个节点中可能会有多个中断，在一个节点中有多个中断，要确保这些中断的执行顺序，不能使用条件语句（如 if）动态跳过某些中断，否则，恢复时会出现错误。

3.不要在中断中返回复杂信息

由于使用的检查点管理器不同，有些检查点管理器可能无法序列化复杂的对象，会导致序列化失败，比如函数、对象等等。

4.保证节点中断前的代码是幂等的

当在节点中进行中断，后续恢复图执行，重新执行时，会从头执行这个节点的代码，因此从节点开始到断点处中间的代码会被重复执行，因此，要保证多次重复执行该段代码，对业务没有影响。

四、总结

本文围绕 LangGraph 的中断（Interrupt）与恢复（Resume）机制 进行了系统讲解，重点说明了它们在实际业务中的作用以及具体使用方式。

首先，介绍了什么是中断与恢复：在图执行过程中，允许程序在某个关键节点暂停执行，并将当前图状态通过检查点持久化保存起来，等待外部输入后，再从中断位置继续执行。这一机制是实现 Human-in-the-Loop （人在环路）场景的基础。

接下来，通过两个示例，演示了中断与恢复在真实业务中的应用：

• 人工审核场景：在高风险操作（如打款）前触发中断，等待人工决策，再根据恢复时传入的结果决定后续执行路径；
• 内容确认与修改场景：先由 LLM 生成内容，再通过中断将内容交给人类校对与修改，最终以人工确认后的结果继续图执行。

最后，总结了使用中断时需要注意的几个关键点。

通过本文，相信你已经对 LangGraph 的中断与恢复机制 有了一个比较完整的了解。在后续文章中，我们还会继续深入 LangGraph 的其他核心能力，结合更多实战场景进行讲解。