15. Human-in-the-Loop:什么时候让 Agent 停下来问人

关于 Agent 有个很广泛的误解:越自主越好。实际生产经验正相反——知道什么时候停下来问人,是 Agent 最重要的能力。

原因朴素。Agent 的失败率永远不是零——最好的 Claude Agent SDK 配最好的模型,在多步任务上也有 10~20% 的失败率。低 stakes 场景(写草稿、整理笔记)跑错了重来就行;高 stakes 场景(发邮件、花钱、改数据库)一次错就是不可挽回的损失。

Human-in-the-Loop(HITL)的本质就是给 Agent 装上"在高 stakes 时强制暂停"的机制。这一篇讲几种常见的 HITL 模式、怎么设计好的审批体验、以及怎么做"可中断 Agent"——让人类在 Agent 跑到一半时接管。

几种 HITL 的常见时机

不是每个动作都要问人,那样 Agent 就回退成了人工助手。关键在区分什么需要审批、什么不需要。

几个经过验证的分类:

低影响、可逆——读文件、查数据、做草稿、运行分析代码。不需要审批,让 Agent 自由做。

中影响、可逆——发给自己的邮件草稿、生成可删除的文件、创建测试订单。可以做,但要展示结果,让用户事后看到。

高影响、难逆——发邮件给客户、提交订单、改生产数据库、转账、删文件、提 commit。必须问人,而且要让审批内容一目了然。

不可逆的破坏性动作——rm -rf、生产数据库的 TRUNCATE、发布新闻稿。必须明确二次确认,甚至要求输入特定文字("输入 'DELETE' 确认")才能执行。

Claude Code 里的做法就是这样:读文件不问,编辑文件可选是否问(权限模式),执行 rm 之类破坏性命令默认强问,网络请求视工具而定。不同工具对应不同"授权等级"。

审批怎么设计不烦人

一个差的 HITL 体验是每做一件事都问一次,用户很快烦到关掉审批让 Agent 完全自主,反而出事。好的 HITL 有几个特点:

批量展示,而非逐条审批。Agent 把 10 个要发的邮件一起拿给用户审,用户看一眼勾选哪些可发。比"发这封吗? 发这封吗? 发这封吗?"好 10 倍。

预览实际动作,而非描述。不要说"Agent 将要创建一个 Spreadsheet 存分析结果"。直接展示"将要创建的文件预览:标题、前 10 行数据"。用户能扫一眼就判断对错。

结构化审批内容。把 diff 做成清晰视图。删文件显示文件名和大小;发邮件显示收件人、主题、正文;改数据库显示 before / after 的 diff。

允许部分批准。10 个操作里有 2 个有问题,让用户勾选"其他 8 个通过"而不是"全部重做"。

记住用户的选择。用户说过"发给 @example.com 的邮件不用再问我",下次就不问。但要明确这个规则持续多久——本次会话?7 天?永久?

这些原则你能在 Cursor / Claude Code 的 permission 系统里看到。Cursor 的 "Ask for Each Edit" vs "Ask Once" vs "Always Accept" 就是经验上被证明合理的几档。

可中断 Agent:停和继续

更有意思的 HITL 场景是Agent 跑到一半,人类想介入改变方向。这要求 Agent 本身设计上是"可中断"的。

所谓可中断,意味着在 Agent 执行循环的任何时刻:

用户可以查看当前状态(走到第几步、目前的理解是什么)
用户可以发送消息改变 Agent 的方向
用户可以取消整个任务
Agent 必须优雅地合并新指令和已有工作,不丢失前面的成果

实现上的关键点:

把 Agent 的 messages 做成可持久化的状态。每一步结束把完整 messages 写进数据库。中断时从最后一个检查点恢复。

把 Agent 的"一步"做成幂等。Agent 在某一步发起了 tool_call,系统还没返回 tool_result 时被中断——恢复时要知道"这一步做完了吗",以免重复执行。

在每轮循环开头检查"是否有用户消息"。用户打断时,把新消息加到 messages 里继续循环。Agent 的下一轮 Thought 会综合历史和新指令。

最简骨架:

class InterruptibleAgent:
    def __init__(self, state_store):
        self.store = state_store  # 数据库/KV

    async def step(self, task_id: str):
        state = self.store.load(task_id)
        messages = state["messages"]

        # 检查是否有用户插入的新消息
        if new_msg := self.store.pop_user_message(task_id):
            messages.append({"role": "user", "content": new_msg})

        # 检查是否被取消
        if state.get("cancelled"):
            return "任务已取消"

        # 正常一步
        resp = await call_model(messages)
        messages.append(resp.message)
        self.store.save(task_id, {"messages": messages, "step": state["step"] + 1})

        if resp.tool_calls:
            for tc in resp.tool_calls:
                result = await run_tool(tc)
                messages.append({"role": "tool", ..., "content": result})
                self.store.save(task_id, {"messages": messages})

        return "continue" if resp.tool_calls else "done"

    async def run(self, task_id: str, max_steps=20):
        for _ in range(max_steps):
            status = await self.step(task_id)
            if status == "done" or status == "任务已取消":
                return status

核心在:每一步的开头 load state,末尾 save state。这样 Agent 跑到一半死了、或者用户中断了,下次从持久化的最后一步恢复,不丢工作。用户发新消息通过 store.pop_user_message 进入 messages,Agent 下一步的 Thought 会看到并处理。

这个模式是 LangGraph 的 Checkpointer 和 Claude Agent SDK 的 Session 的核心。框架层面把持久化和状态管理包起来,你只写业务逻辑。

审批的 UI 和 API

生产 Agent 的审批有两种常见形态:

同步审批——Agent 跑到某一步停住,阻塞等待用户决定。适合桌面应用(Cursor、Claude Code)这种"用户正在看屏幕"的场景。

异步审批——Agent 把动作挂起,放到审批队列,继续做不依赖这个动作的其他事。用户在任何时候看队列做决定,Agent 再把审批结果合并回来。适合后台 Agent、工作流集成。

异步审批的实现:

async def submit_for_approval(task_id: str, action: dict) -> str:
    approval_id = await approval_queue.push({
        "task_id": task_id,
        "action": action,
        "created_at": now(),
        "status": "pending",
    })
    # 不等待,返回让 Agent 继续
    return approval_id

async def wait_for_approval(approval_id: str, timeout=3600):
    # Agent 的另一步,或周期性检查
    result = await approval_queue.get(approval_id)
    if result["status"] == "pending":
        return None  # 还没人批
    return result

async def agent_with_approval(task: str):
    # Agent 做完了需要审批的动作
    approval = await submit_for_approval(task_id, {"type": "send_email", ...})

    # 先做不依赖审批的其他事
    await do_independent_work()

    # 周期检查审批结果
    while not (result := await wait_for_approval(approval)):
        await asyncio.sleep(10)

    if result["status"] == "approved":
        await actually_send_email(...)
    else:
        await handle_rejected(...)

这种异步模式对长时运行的 Agent(例如每小时自动做一堆事的后台 Agent)特别重要——它不会被"等审批"卡死,能继续推进独立任务。

让 Agent 学会"自己申请审批"

另一个值得做的设计:把审批做成一个工具,让 Agent 在觉得 stakes 高的时候主动触发:

def request_approval(action_description: str, risk_level: str) -> str:
    """在执行高影响动作前请求人类审批。
    risk_level: 'low' | 'medium' | 'high' | 'critical'
    """
    approval_id = approval_queue.push({...})
    # ... 等待或异步

Agent 在 system prompt 里被告知:

以下动作必须先调用 request_approval 工具:

发送对外邮件

提交订单或付款

删除文件或数据

任何涉及金额超过 ¥1000 的操作

这样 Agent 自己判断什么需要审批,而不是工程师硬编码每个工具的审批规则。模型能处理 prompt 里没提到但语义相似的情况——比如"转账"虽然没显式列在列表里,但模型会识别成"涉及金额"而主动请求审批。

要注意的是,这种"让模型自我判断"不能是唯一防线。硬编码的白名单/黑名单仍然要有——涉及真实金钱、生产数据的动作,必须有不依赖模型判断的硬规则。模型判断是第一道网,硬规则是最后一道。

失败恢复:从错误中继续

HITL 的另一面是失败时怎么让人接管。Agent 跑到第 8 步失败了,用户看到错误,应该能:

看到完整 trace——前 7 步做了什么、第 8 步为什么失败
编辑状态——发现 Agent 理解错了某个事实,直接改 messages 里的那条
从编辑后的状态继续——不重做 1~7 步,只让 Agent 用新的理解重试第 8 步

这种"人在 trace 里直接改"的体验在 LangGraph Studio 和 Langfuse 里有。用户看到一个可视化的状态图,哪里错了改哪里,继续跑。

小结

Human-in-the-Loop 不是"把 Agent 阉割成半自动工具"。好设计的 HITL 是让 Agent 在低风险区完全自主,高风险区必然停下,中间的灰色区间有工程化的审批流。这样的 Agent 在生产环境的综合价值远高于"全自主但偶尔闯祸"的 Agent。

记住这个原则:自主性不是评判 Agent 好坏的唯一维度,对 stakes 的敏感性同样重要。知道什么时候该问的 Agent,比看起来更聪明的 Agent,更值得生产投入。

下一篇讲这一切的前提——可观测性与评测。Agent 这种非线性系统不能靠直觉调,需要 trace、replay、eval 构成一整套工程化的开发闭环。