Human in Loop的几种实现方式

发表于 2025-08-24 更新于 2025-08-31 分类于 AI

human-in-loop是指在agent的循环执行中，让人能够介入，从而让自动化流程中可以使用人这个工具来实现输入密码、授予权限等操作。
实现human-in-loop主要取决于两个能力：

阻塞通知能力，原地阻塞，等待用户输入，用户输入将通过事件通知的方式来触发流程的继续执行，例如：LlamaIndex和ADK的实现
中断恢复能力，在关键执行节点保存执行上下文，让任务能恢复执行，例如：LangGraph的实现

LangGraph的实现模式

LangGraph 中的人机循环工作流建立在 checkpoint 系统之上，使用检查点保存每一步的图形状态，并在人工干预后恢复。

中断原语

中断时序图.png

Common Patterns（常见模式）

LangGraph中的人机循环实现提供了几种常见的模式，用于不同的应用场景：

1. 审批或拒绝模式（Approve or Reject）

Node中断执行流.png
用于需要人工审批的场景，比如内容审核、决策确认等。

def approval_node(state):
    # 准备需要审批的内容
    content_to_review = state["generated_content"]
    
    # 中断等待人工审批
    approval = interrupt({
        "content": content_to_review,
        "action": "approve_or_reject"
    })
    
    if approval == "approve":
        return {"status": "approved", "content": content_to_review}
    else:
        return {"status": "rejected", "content": None}

2. 审查和编辑状态模式（Review and Edit State）

允许人工审查当前状态并进行编辑修改。

def review_edit_node(state):
    current_state = {
        "text": state.get("text", ""),
        "metadata": state.get("metadata", {})
    }
    
    # 中断等待人工编辑
    edited_state = interrupt({
        "current_state": current_state,
        "instruction": "请审查并编辑当前状态"
    })
    
    return edited_state

3. 审查工具调用模式（Review Tool Calls）

Tool中断执行流.png
在工具执行前进行人工审查，确保工具调用的正确性。

def tool_review_node(state):
    tool_calls = state["proposed_tool_calls"]
    
    # 中断等待人工确认工具调用
    approved_calls = interrupt({
        "tool_calls": tool_calls,
        "action": "review_tool_calls"
    })
    
    return {"approved_tool_calls": approved_calls}

4. 为任何工具添加中断模式（Add Interrupts to Any Tool）

可以在任何工具执行前后添加人工干预点。

def tool_with_interrupt(state):
    # 工具执行前的确认
    pre_confirmation = interrupt({
        "message": "即将执行工具，请确认",
        "tool_name": "api_call"
    })
    
    # 执行工具
    result = execute_tool(state["parameters"])
    
    # 工具执行后的审查
    post_review = interrupt({
        "result": result,
        "action": "review_result"
    })
    
    return {"final_result": post_review}

5. 验证人工输入模式（Validate Human Input）

对人工输入进行验证，确保输入的正确性和完整性。

def validation_node(state):
    while True:
        # 获取人工输入
        user_input = interrupt({
            "message": "请输入数据",
            "validation_rules": ["required", "email_format"]
        })
        
        # 验证输入
        if validate_input(user_input):
            return {"validated_input": user_input}
        else:
            # 输入无效，继续循环
            continue