人在回路（HITL）：Agent 流程中的确认与编辑

人在回路（HITL）：Agent 流程中的确认与编辑

学习目标

本章要解决什么问题：当 Agent 需要人类确认、编辑或拒绝其操作时，如何在不中断整个 Agent 流程的情况下插入人工审查环节？

读者将学到：

• LangChain HITL 中间件的核心概念（ActionRequest、Decision、ReviewConfig）
• 四种决策类型（approve/edit/reject/respond）的使用场景
• 如何使用 LangGraph 的 interrupt() 实现人工等待

前置知识

本章是 LangChain 特有的交互模式实现，不涉及通用知识原理。

项目实践

HITL 的核心数据结构

# 伪代码
class ActionRequest(TypedDict):
    name: str          # 动作名称
    args: dict         # 动作参数
    description: str   # 动作描述（供人类审查）

class ReviewConfig(TypedDict):
    action_name: str                # 关联的动作名称
    allowed_decisions: list[Literal["approve", "edit", "reject", "respond"]]  # 允许的决策
    args_schema: dict               # 参数 JSON Schema（用于编辑时的表单校验）

四种决策类型：

决策	含义	适用场景
approve	批准执行	操作看起来安全
edit	修改参数后执行	操作方向正确但参数需要调整
reject	拒绝执行	操作不安全或不相关
respond	直接回复	用户想提供额外信息而非审查操作

基于 LangGraph interrupt() 的实现

HITL 中间件利用 LangGraph 的 interrupt() 函数实现人工等待：

# 伪代码
def hitl_middleware(state, runtime):
    # 1. 从模型消息中提取工具调用请求
    action_requests = extract_action_requests(state["messages"])

    # 2. 构建 HITL 请求
    hitl_request = HITLRequest(
        action_requests=action_requests,
        review_configs=[...],  # 每个动作的审查配置
    )

    # 3. 中断 Agent 执行，等待人类输入
    human_response = interrupt(value=hitl_request)

    # 4. 根据人类决策处理
    for decision in human_response["decisions"]:
        if decision["type"] == "approve":
            # 继续执行原工具调用
            pass
        elif decision["type"] == "edit":
            # 使用修改后的参数重新构建工具调用
            modify_tool_call_args(decision["args"])
        elif decision["type"] == "reject":
            # 注入拒绝消息作为工具结果
            inject_tool_message(f"Action rejected: {decision.get('reason')}")
        elif decision["type"] == "respond":
            # 将用户响应注入对话
            inject_human_message(decision["content"])

关键机制：interrupt() 会暂停图的执行并等待外部输入。外部系统（如 Web UI）通过 LangGraph 的 API 提供人类决策，然后图从暂停处恢复。

ReviewConfig 的 per-action 配置

不同动作可以有不同的审查策略：

# 伪代码
review_configs = [
    {
        "action_name": "send_email",
        "allowed_decisions": ["approve", "edit", "reject"],
        "args_schema": {  # 编辑时校验的 Schema
            "to": {"type": "string"},
            "subject": {"type": "string"},
            "body": {"type": "string"}
        },
    },
    {
        "action_name": "search_db",
        "allowed_decisions": ["approve", "reject"],  # 不允许编辑
    },
]

问题与规避

interrupt() 的持久化依赖

问题：interrupt() 需要配合 checkpointer 使用，否则中断状态无法恢复。

对策：使用 HITL 中间件时，create_agent() 必须传入 checkpointer 参数（如 SqliteSaver 或 MemorySaver）。

超时处理

问题：人类可能长时间不响应中断，导致 Agent 挂起。

对策：在上层应用中设置中断超时，超时后自动注入 reject 决策或终止 Agent 执行。

设计取舍

interrupt() vs 工具模拟

方案	优势	代价
interrupt()	图级别暂停、状态持久化、恢复简单	需要 checkpointer、整个图暂停
工具模拟	细粒度控制、可以只暂停特定工具	实现复杂、需要模拟工具处理中断

LangChain 选择 interrupt() 方案，因为它利用了 LangGraph 内置的状态持久化机制，确保 Agent 在人类审查期间的状态不会丢失。

参考来源

• LangGraph Human-in-the-loop 文档