3.1 中断机制与人工干预

本节导读：通过本节学习，你将掌握LangGraph的中断机制，学会在智能体执行过程中插入人工干预点，实现人机协作的智能体系统。

学习目标

• 理解中断机制的核心概念和作用
• 掌握interrupt_before和interrupt_after的使用方法
• 学会在智能体流程中添加人工审核点
• 实现基于条件的中断策略
• 构建完整的人机协作工作流

核心概念

LangGraph的中断机制是其区别于其他AI框架的核心特性之一。它允许在智能体执行过程中暂停执行，等待人工干预或外部条件满足后再继续。

中断的类型

1. 中断前中断（interrupt_before）

在节点执行前暂停，允许审核输入或修改状态：

# 在节点执行前中断
graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["human_review"]  # 在human_review节点前中断
)
​

2. 中断后中断（interrupt_after）

在节点执行后暂停，允许审核输出或决定下一步：

# 在节点执行后中断
graph.compile(
    checkpointer=checkpointer,
    interrupt_after=["data_analysis"]  # 在data_analysis节点后中断
)
​

环境准备 / 前置知识

依赖安装

# 基础依赖
pip install -U langgraph langchain-openai

# 可选依赖：Redis用于生产环境检查点
pip install -U redis
​

前置要求

• 熟悉StateGraph基本概念
• 了解节点和边的定义
• 掌握状态管理基础

分步实战

步骤 1：基础中断机制实现

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated, Sequence

class AgentState(TypedDict):
    messages: Annotated[Sequence[dict], add_messages]
    current_step: str
    requires_approval: bool

def analysis_node(state: AgentState):
    """数据分析节点"""
    llm = ChatOpenAI(model="gpt-4-turbo")
    
    # 模拟数据分析过程
    user_query = state["messages"][-1]["content"]
    
    # 执行分析
    analysis_prompt = f"""
    你是数据分析专家，对以下问题进行分析：
    {user_query}
    
    请提供详细的分析报告。
    """
    
    response = llm.invoke([
        {"role": "system", "content": "你是数据分析专家"},
        {"role": "user", "content": analysis_prompt}
    ])
    
    return {
        "messages": [response],
        "current_step": "analysis_complete",
        "requires_approval": True  # 标记需要人工审核
    }

def review_node(state: AgentState):
    """人工审核节点"""
    print("\n=== 人工审核 ===")
    print("分析结果：")
    for msg in state["messages"]:
        if msg["role"] == "assistant":
            print(f"- {msg['content']}")
    
    # 这里在实际应用中应该等待用户输入
    # 为了演示，我们自动批准
    approval = input("是否批准分析结果? (y/n): ")
    
    return {
        "messages": [{"role": "assistant", "content": f"人工审核结果: {approval}"}],
        "current_step": "review_complete",
        "requires_approval": False
    }

# 构建图
graph = StateGraph(AgentState)
graph.add_node("analysis", analysis_node)
graph.add_node("review", review_node)
graph.add_edge(START, "analysis")
graph.add_edge("analysis", "review")
graph.add_edge("review", END)

# 编译时启用中断
checkpointer = MemorySaver()
compiled_graph = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["review"]  # 在审核节点前中断
)

# 执行示例
initial_state = {
    "messages": [{"role": "user", "content": "分析2024年Q1销售数据趋势"}],
    "current_step": "start",
    "requires_approval": False
}

print("第一次执行（将在审核点中断）")
result1 = compiled_graph.invoke(initial_state)
print(f"中断时的状态: {result1['current_step']}")

print("\n第二次执行（继续审核流程）")
final_result = compiled_graph.invoke(result1)
print(f"最终结果: {final_result['messages'][-1]['content']}")
​

步骤 2：条件中断策略

from langgraph.prebuilt import ToolInvocation
from langgraph.graph import END

class AdvancedAgentState(TypedDict):
    messages: Annotated[Sequence[dict], add_messages]
    current_step: str
    analysis_type: str
    confidence_score: float
    needs_review: bool

def analysis_with_confidence(state: AdvancedAgentState):
    """带置信度评分的分析节点"""
    llm = ChatOpenAI(model="gpt-4-turbo")
    
    user_query = state["messages"][-1]["content"]
    
    # 判断分析类型
    analysis_type = "detailed" if "深入" in user_query else "basic"
    
    # 执行分析
    prompt = f"""
    分析以下请求：
    {user_query}
    
    请提供：
    1. 分析类型：{analysis_type}
    2. 置信度评分（0-1）
    3. 分析结果
    """
    
    response = llm.invoke([
        {"role": "system", "content": "你是专业分析师"},
        {"role": "user", "content": prompt}
    ])
    
    # 模拟置信度评分
    confidence = 0.8 if analysis_type == "detailed" else 0.6
    
    return {
        "messages": [response],
        "current_step": "analysis_complete",
        "analysis_type": analysis_type,
        "confidence_score": confidence,
        "needs_review": confidence < 0.8  # 置信度低于0.8需要审核
    }

def conditional_interrupt(state: AdvancedAgentState):
    """条件中断判断"""
    if state.get("needs_review", False):
        print(f"检测到低置信度分析 ({state['confidence_score']:.2f})，需要人工审核")
        return True
    return False

# 构建带条件中断的图
graph = StateGraph(AdvancedAgentState)
graph.add_node("analysis", analysis_with_confidence)
graph.add_node("review", review_node)
graph.add_edge(START, "analysis")

# 添加条件边
graph.add_conditional_edges(
    "analysis",
    conditional_interrupt,
    {
        True: "review",
        False: END
    }
)

graph.add_edge("review", END)

# 编译
compiled_graph = graph.compile(checkpointer=checkpointer, interrupt_before=["review"])

# 测试高置信度请求
print("=== 测试高置信度请求 ===")
high_confidence_input = {
    "messages": [{"role": "user", "content": "简单分析用户增长趋势"}],
    "current_step": "start",
    "analysis_type": "",
    "confidence_score": 0.0,
    "needs_review": False
}

result_high = compiled_graph.invoke(high_confidence_input)
print(f"高置信度请求直接完成: {result_high['current_step']}")

# 测试低置信度请求
print("\n=== 测试低置信度请求 ===")
low_confidence_input = {
    "messages": [{"role": "user", "content": "深入分析产品销售策略"}],
    "current_step": "start",
    "analysis_type": "",
    "confidence_score": 0.0,
    "needs_review": False
}

result_low1 = compiled_graph.invoke(low_confidence_input)
print(f"低置信度请求中断: {result_low1['current_step']}")

print("\n继续执行审核流程")
final_low = compiled_graph.invoke(result_low1)
print(f"最终审核结果: {final_low['messages'][-1]['content']}")
​

常见问题 FAQ

Q1：中断机制与普通流程控制的区别是什么？

A：中断机制与普通流程控制有本质区别：

• 普通流程控制：基于代码逻辑的分支判断，如if-else、switch等
• 中断机制：基于外部条件或人工决策的暂停，需要在运行时动态决定

中断机制特别适合需要人工参与或不确定条件的场景，而普通流程控制适合确定性逻辑。

Q2：如何在生产环境中处理中断？

A：生产环境中的中断需要考虑：

1. 用户界面：提供友好的审核界面
2. 超时机制：设置人工审核超时时间
3. 通知系统：及时通知相关人员
4. 审计日志：记录所有审核决策
5. 回滚机制：允许撤销错误的审核决策

本节小结

通过本节学习，我们掌握了LangGraph的中断机制，包括interrupt_before和interrupt_after的使用、条件中断策略以及中断恢复机制。中断机制使得LangGraph能够在智能体执行过程中融入人工智能，实现真正的人机协作。

下一节我们将探讨LangGraph的流式控制技术，这是实现实时反馈的关键特性。

延伸阅读

• LangGraph官方文档-中断机制
• LangSmith调试工具

关键词：中断机制, 人工干预, interrupt_before, interrupt_after, 人机协作, 智能体控制
难度：进阶
预计阅读：25分钟