Interrupt 与人工审批
Interrupt（中断）是 LangGraph 实现 Human-in-the-Loop（人机协同）的核心机制，允许图在执行过程中暂停，等待人工决策后再继续。
为什么需要 Interrupt
在 Agent 工作流中，有些操作需要人工介入：
- 敏感操作：发送邮件、付款、删除数据
- 不确定场景：LLM 输出有歧义
- 审批流程：内容审核、合规检查
- 信息补充：需要用户提供更多信息
基本用法
添加 Interrupt
python
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from typing import TypedDict
class State(TypedDict):
input: str
approved: bool
result: str
def review_node(state: State):
"""节点被中断后，人工审查可以在这里修改 state"""
return {"approved": True}
def generate_node(state: State):
return {"result": f"最终输出: {state['input']}"}
# 构建图
builder = StateGraph(State)
builder.add_node("review", review_node)
builder.add_node("generate", generate_node)
builder.add_edge(START, "review")
builder.add_edge("review", "generate")
builder.add_edge("generate", END)
# 编译时传入 checkpointer（interrupt 需要 checkpoint 支持）
checkpointer = InMemorySaver()
graph = builder.compile(
checkpointer=checkpointer,
interrupt_before=["review"] # 在执行 review 之前中断
)
# 执行（会在 review 前暂停）
config = {"configurable": {"thread_id": "1"}}
result = graph.invoke({"input": "hello", "approved": False}, config)
print(result) # 返回中断时的状态恢复执行
python
# 人工审查后，通过 update_state 继续
graph.update_state(
config,
{"approved": True}, # 人工修改状态
)
# 或者直接恢复（不修改状态）
result = graph.invoke(None, config) # 传入 None 表示继续Interrupt 的配置方式
interrupt_before：在节点执行前中断
python
graph.compile(
checkpointer=checkpointer,
interrupt_before=["sensitive_node", "approval_node"]
)interrupt_after：在节点执行后中断
python
graph.compile(
checkpointer=checkpointer,
interrupt_after=["llm_call_node"]
)组件内 Interrupt
LangGraph 也支持在节点函数内部触发中断，更精确地控制中断位置：
python
from langgraph.types import interrupt
def node_with_interrupt(state: State):
# 执行到一半
result = llm.invoke(state["messages"])
if result.get("needs_approval"):
# 在节点内部触发中断
user_response = interrupt(
# 中断时向用户显示的消息
{"question": "是否批准这个操作?", "data": result}
)
# 用户响应后继续执行
if user_response.get("approved"):
return {"result": result.content}
else:
return {"result": "已取消"}
return {"result": result.content}完整的人机协同流程
python
import uuid
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt
def email_agent(state: MessagesState):
"""发送邮件需要人工审批"""
response = llm.invoke(state["messages"])
if response.tool_calls:
for tc in response.tool_calls:
if tc["name"] == "send_email":
# 发送邮件需要人工确认
approval = interrupt({
"type": "approval",
"tool": "send_email",
"args": tc["args"],
"message": "是否发送这封邮件?"
})
if not approval["approved"]:
return {"messages": [
response,
{"role": "tool", "content": "用户取消了邮件发送"}
]}
return {"messages": [response]}
return {"messages": [response]}
# 构建
builder = StateGraph(MessagesState)
builder.add_node("agent", email_agent)
builder.add_edge(START, "agent")
builder.add_edge("agent", END)
graph = builder.compile(checkpointer=InMemorySaver())对应的前端（或交互层）如何处理：
python
# 1. 执行，遇到 interrupt 会暂停
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
try:
result = graph.invoke(
{"messages": [{"role": "user", "content": "给张三发邮件"}]},
config
)
except:
pass # interrupt 会抛出异常或暂停
# 2. 获取待处理的中断
state = graph.get_state(config)
print(state.tasks) # 显示被中断的任务
# 3. 接收人工响应后恢复
graph.update_state(
config,
{"messages": [{"role": "tool", "content": "已发送"}]}
)
result = graph.invoke(None, config)StateSnapshot 中的 Interrupt 信息
python
state = graph.get_state(config)
for task in state.tasks:
if task.interrupts:
for interrupt in task.interrupts:
print(f"中断节点: {task.name}")
print(f"中断消息: {interrupt.value}")生产实践
- Checkpointer 必须配置：interrupt 依赖 persistence 机制
- 超时处理：interrupt 可能永远等待，建议设置超时
- 幂等性：节点应能安全处理多次 resume
- 用户体验：interrupt 的消息要清晰，让用户知道需要做什么