跳到主要内容

中断与人工在环

中断机制​

中断允许暂停图的执行并等待外部输入。

基本用法​

from langgraph.types import interrupt

@entrypoint
def workflow_with_approval(data):
    """需要人工批准的工作流"""
    
    # 处理第一步
    processed = process_data(data)
    
    # 中断请求批准
    approval = interrupt({
        "message": "需要批准处理结果",
        "data": processed,
        "options": ["approve", "reject", "edit"]
    })
    
    # 根据批准结果继续
    if approval == "approve":
        return finalize(processed)
    elif approval == "edit":
        edited = edit_data(processed)
        return finalize(edited)
    else:
        return "已拒绝"

恢复中断​

# 第一次执行到中断
config = {"configurable": {"thread_id": "task_1"}}

try:
    result = app.invoke(initial_data, config=config)
except Exception:
    # 被中断
    pass

# 用户审查后，带着输入恢复
result = app.invoke(None, config=config, input_={"approval": "approve"})

多步骤批准工作流​

from langgraph.types import interrupt

def approval_workflow(request_data):
    # 第一步：初步处理
    processed = analyze_request(request_data)
    
    # 第一次中断：初步审核
    initial_check = interrupt({
        "stage": "initial_review",
        "data": processed
    })
    
    if not initial_check:
        return "初步审核失败"
    
    # 第二步：详细处理
    detailed = detailed_processing(processed)
    
    # 第二次中断：详细审核
    detailed_check = interrupt({
        "stage": "detailed_review",
        "data": detailed
    })
    
    if not detailed_check:
        return "详细审核失败"
    
    # 最终批准
    return finalize(detailed)

人工在环模式​

1. 审批模式​

决策者审查并批准/拒绝操作

2. 编辑模式​

用户可以修改建议

3. 验证模式​

确保操作不会产生不良结果

实现建议​

✅ 确保中断请求清晰说明为什么需要中断 ✅ 提供足够的上下文供决策 ✅ 有合理的超时时间 ✅ 记录所有批准/拒绝决定 ✅ 支持多种恢复选项

常见模式​

多步骤审批： 根据复杂性进行多级审核 条件批准： 基于阈值自动批准低风险操作 反馈循环： 允许用户迭代改进

下一步​