LangGraph 人工介入机制实战：interrupt 功能设计与应用模式解析

作者：佑瞻
在开发智能体流程时，我们常常会遇到这样的场景：当智能体需要做出关键决策或处理敏感信息时，必须引入人工审核环节。比如金融交易审批、医疗诊断复核等场景，单纯的自动化流程可能存在风险。这时候，如何在智能体流程中无缝嵌入人工介入机制，就成为了一个关键挑战。今天，我们就来聊聊 LangGraph 框架中强大的 interrupt 功能，看看它是如何优雅地实现 "人在回路" 的工作流设计。
interrupt 功能基础：流程控制的 "暂停键"

LangGraph 中的 interrupt 函数就像是流程控制的 "暂停键"，它能在指定节点暂停图的执行，将信息呈现给人工，然后根据人工输入继续执行。这种机制非常适合需要人工干预的场景，比如内容审核、数据校正等。
中断机制的核心流程：从定义到触发

human_node 的完整工作流程

先来看一个完整的中断示例，注意代码中三个关键阶段的衔接：
python

1. from langgraph.types import interrupt, Command
2. from langgraph.graph import GraphBuilder
3. from langgraph.checkpoint import Checkpointer
5. # 1. 定义包含interrupt的节点函数
6. def human_node(state: dict):
7.     """当图执行到该节点时会暂停，等待人工输入"""
8.     # 传递需要人工处理的文本内容
9.     revised_text = interrupt({
10.         "text_to_revise": state["original_text"],
11.         "instruction": "请将这段文本简化为50字以内"
12.     })
13.     # 返回人工处理后的结果，更新图状态
14.     return {"processed_text": revised_text}
16. # 2. 构建并编译图
17. graph_builder = GraphBuilder()
18. graph_builder.add_node("human_review", human_node)
19. graph_builder.add_start_node("start", next_node="human_review")
20. checkpointer = Checkpointer()  # 必须指定检查点保存状态
21. graph = graph_builder.compile(checkpointer=checkpointer)
23. # 3. 执行图并处理中断
24. if __name__ == "__main__":
25.     # 首次调用触发中断
26.     thread_config = {"configurable": {"thread_id": "review_001"}}
27.     first_result = graph.invoke(
28.         {"original_text": "这是一段需要简化的长文本，包含了许多细节信息"},
29.         config=thread_config
30.     )
31.    
32.     # 解析中断结果
33.     if "__interrupt__" in first_result:
34.         print("检测到中断，需要人工处理")
35.         interrupt_info = first_result["__interrupt__"][0]
36.         print(f"中断节点: {interrupt_info.ns}")
37.         print(f"待处理文本: {interrupt_info.value['text_to_revise']}")
38.         
39.         # 4. 模拟人工输入并恢复流程
40.         human_input = "简化后的文本：长文本包含细节信息"
41.         resumed_result = graph.invoke(
42.             Command(resume=human_input),
43.             config=thread_config
44.         )
45.         
46.         print("流程恢复后结果:")
47.         print(resumed_result)  # 输出: {'processed_text': '简化后的文本：长文本包含细节信息'}

复制代码

这段代码展示了中断机制的完整生命周期，关键要理解：human_node并不是直接调用的，而是通过图的执行流程触发的。当图运行到 "human_review" 节点时，interrupt 函数会被激活，导致流程暂停。
中断恢复的核心原理：重新执行节点的底层逻辑

恢复执行的关键特性

LangGraph 的中断恢复机制与 Python 的input()函数有本质区别：当使用Command恢复流程时，节点会从函数开头重新执行，而非从中断位置继续。我们可以通过一个计数器示例直观理解：
python

1. counter = 0
2. def node_with_interrupt(state: dict):
3.     global counter
4.     counter += 1
5.     print(f"节点第{counter}次执行")
6.     user_input = interrupt("请输入内容")
7.     print(f"当前counter值: {counter}")
8.     return {"input": user_input}
10. # 首次调用触发中断
11. graph.invoke({"initial": "data"}, config=thread_config)
12. # 输出: 节点第1次执行
14. # 恢复后再次执行
15. graph.invoke(Command(resume="用户输入"), config=thread_config)
16. # 输出: 节点第2次执行
17. # 输出: 当前counter值: 2

复制代码

核心结论：每次恢复都会导致节点从头执行，这是理解中断机制的关键前提。
恢复执行的流程示意图

三大设计模式：人工介入的场景化实现

通常，在有人参与的工作流程中，可以执行三种不同的操作：
1.批准或拒绝：在关键步骤（如API调用）之前暂停图表，以审查并批准该操作。如果该操作被拒绝，你可以阻止图表执行该步骤，并可能采取替代行动。这种模式通常涉及根据人工输入对图表进行路由。
2.编辑图状态：暂停图表以查看和编辑图状态。这对于纠正错误或使用其他信息更新状态很有用。此模式通常涉及根据人类输入更新状态。
3.获取输入：在图形的特定步骤中明确请求人工输入。这对于收集更多信息或上下文以辅助智能体的决策过程很有用。
[color=rgba(0, 0, 0, 0.87)]Approve or reject¶ 批准或拒绝

在涉及敏感操作时，我们需要在执行前让人工确认。比如调用外部 API 发送营销邮件前，先让运营人员审核内容：

1. from langgraph.types import Command
2. from typing import Literal  # 若未导入需补充
4. def human_approval(state: State) -> Command[Literal["some_node", "another_node"]]:
5.     # 通过interrupt函数暂停图执行，向人工展示待审核内容并获取审批结果
6.     is_approved = interrupt(
7.         {
8.             "question": "这是否正确？",              # 审核问题
9.             # 展示需要人工审核和批准的输出（如LLM生成结果）
10.             "llm_output": state["llm_output"]       # 待审核的LLM输出
11.         }
12.     )
14.     # 根据人工审批结果返回不同的流程控制指令（Command对象）
15.     if is_approved:
16.         return Command(goto="some_node")  # 批准则跳转到"some_node"节点
17.     else:
18.         return Command(goto="another_node")  # 拒绝则跳转到"another_node"节点
20. # 在合适的位置将节点添加到图中，并连接相关节点
21. graph_builder.add_node("human_approval", human_approval)
22. graph = graph_builder.compile(checkpointer=checkpointer)
24. # 运行图并触发interrupt后，图会暂停执行
25. # 使用批准（resume=True）或拒绝（resume=False）操作来恢复执行
26. thread_config = {"configurable": {"thread_id": "some_id"}}
27. graph.invoke(Command(resume=True), config=thread_config)  # 示例：模拟批准操作

复制代码

这种模式的核心是根据人工输入进行流程路由，确保敏感操作的安全性。在金融、医疗等对合规性要求高的场景中尤为重要。
[color=rgba(0, 0, 0, 0.87)]Review & edit state¶ 查看与编辑状态

当智能体生成的内容需要人工修正时，我们可以使用编辑状态模式。比如 LLM 生成的新闻摘要需要编辑优化：

1. from langgraph.types import interrupt
3. def human_editing(state: State):
4.     ...
5.     result = interrupt(
6.         # 要呈现给客户端的中断信息，可为任意 JSON 可序列化的值
7.         {
8.             "task": "审核 LLM 输出内容并进行必要编辑",
9.             "llm_generated_summary": state["llm_generated_summary"]
10.         }
11.     )
13.     # 使用编辑后的文本更新状态
14.     return {
15.         "llm_generated_summary": result["edited_text"]
16.     }
18. # 在合适位置将节点添加到图中，并连接相关节点
19. graph_builder.add_node("human_editing", human_editing)
20. graph = graph_builder.compile(checkpointer=checkpointer)
22. ...
24. # 运行图直至触发中断，此时图将暂停
25. # 使用编辑后的文本恢复图执行
26. thread_config = {"configurable": {"thread_id": "some_id"}}
27. graph.invoke(
28.     Command(resume={"edited_text": "The edited text"}),
29.     config=thread_config
30. )

复制代码

这种模式常用于内容生成场景，人工可以修正智能体的输出，确保结果符合要求。
获取输入模式：补充缺失的关键信息

当智能体需要额外信息才能继续时，可以主动请求人工输入。比如客服系统中获取用户的具体需求：
python

1. def get_user_requirement(state: State):
2.     # 循环获取有效的用户需求
3.     while True:
4.         requirement = interrupt("请详细描述您的需求")
5.         # 简单验证输入
6.         if len(requirement) < 10:
7.             interrupt("需求描述过短，请补充更多细节")
8.         else:
9.             break
10.     return {"user_requirement": requirement}
12. # 在流程中使用
13. graph_builder.add_node("gather_requirement", get_user_requirement)
14. graph_builder.connect("start", "gather_requirement")

复制代码

这种模式通过多次 interrupt 调用，确保获取到足够的信息，常用于客服、工单系统等需要人机交互的场景。
高级技巧与常见陷阱

输入验证：打造鲁棒的人机交互

有时我们需要确保人工输入的有效性，比如年龄必须是正整数：
python

1. def validate_age(state: State):
2.     question = "您的年龄是多少？"
3.     while True:
4.         answer = interrupt(question)
5.         try:
6.             age = int(answer)
7.             if age <= 0:
8.                 raise ValueError
9.             break
10.         except:
11.             question = f"'{answer}'不是有效年龄，请重新输入"
12.             continue
13.     return {"age": age}
15. # 调用时会循环直到输入有效年龄
16. graph.invoke({"initial": "data"}, config=config)
17. # 输入"25"后正常返回

复制代码

这种循环验证机制可以确保流程获取到合法输入，避免因错误输入导致后续处理出错。
副作用处理：避免重复执行的坑

由于 interrupt 恢复时会重新执行节点，具有副作用的操作（如 API 调用）必须放在 interrupt 之后：
python

1. # 错误做法：API调用在interrupt之前
2. def bad_design(state: State):
3.     # 每次恢复都会重新调用API
4.     result = api_call("sensitive_operation")
5.     answer = interrupt("是否继续？")
6.     return {"result": result}
8. # 正确做法：API调用在interrupt之后
9. def good_design(state: State):
10.     answer = interrupt("是否允许调用API？")
11.     if answer:
12.         result = api_call("sensitive_operation")
13.     return {"result": result if answer else None}
15. # 更好的做法：将副作用操作放在单独节点
16. def separate_node(state: State):
17.     return {"should_call": interrupt("是否调用API？")}
19. graph_builder.add_node("decision", separate_node)
20. graph_builder.add_node("api_call", api_node)
21. graph_builder.connect("decision", "api_call", condition=lambda x: x["should_call"])

复制代码

将副作用操作放在单独节点或 interrupt 之后，可以避免重复执行带来的性能问题或数据不一致。
多中断处理：保持顺序一致性

当一个节点包含多个 interrupt 时，必须注意调用顺序，因为恢复时是按索引匹配的：
python

1. def multiple_interrupts(state: State):
2.     # 两个interrupt按顺序调用
3.     name = interrupt("您的姓名？")
4.     age = interrupt("您的年龄？")
5.     return {"name": name, "age": age}
7. # 恢复时按顺序传入两个值
8. graph.invoke(
9.     Command(resume=["张三", 25]),
10.     config=config
11. )

复制代码

如果在节点中动态改变 interrupt 的顺序或数量，会导致索引不匹配，因此应避免在执行期间修改节点结构。
结语：让智能体流程更灵活可控

通过 LangGraph 的 interrupt 机制，我们可以轻松构建 "人在回路" 的智能体流程，既发挥了 AI 的自动化优势，又保留了人工干预的灵活性。从审批流程到内容修正，从信息收集到决策支持，interrupt 功能为我们提供了强大的工具。
如果本文对你有帮助，别忘了点赞收藏，关注我，一起探索更高效的开发方式～

原文地址：https://blog.csdn.net/The_Thieves/article/details/148828318