玩转LlamaIndex：HITL让 Agent 在关键操作前“问一下人“

hyunbar

346人浏览 · 2026-07-30 09:04:33

hyunbar · 2026-07-30 09:04:33 发布

一、什么是 HITL?

HITL = Human-in-the-Loop。

在 Agent 自动化执行任务中,插入人类决策点,让 AI 在做关键操作前先询问。

在这里插入图片描述

如果让 Agent 完全自动跑,一旦出错代价太大。

二、HITL 的核心:两个事件

LlamaIndex 实现 HITL 用的是 Workflow 事件机制:

InputRequiredEvent:Agent 抛出这个事件,表示"我需要人的输入才能继续"
HumanResponseEvent:程序接收人输入后,把这个事件发回给 Agent,而后流程继续

只需要在工具函数里用 ctx.wait_for_event() 等待 HumanResponseEvent,其她的事件收发框架会处理。

三、代码:转账确认 Agent

3.1 关键代码

3.2 运行效果

输入 yes：

在这里插入图片描述

输入 no:

在这里插入图片描述

四、代码解析

4.1 工具函数为什么要返回 async 函数?

def transfer_money(to_account: str, amount: float) -> str:
    def _transfer(ctx: Context) -> str:  # async 函数
        ...
    return _transfer  # 返回函数,不执行

这是个反直觉但关键的设计。

原因:

FunctionAgent 调用工具时,会自动注入 ctx 参数
但 ctx 是异步上下文,只能在 async 函数里用
所以得用闭包把 to_account 和 amount 捕获,返回一个 async 函数
Agent 拿到这个 async 函数后会 await 它,并自动把 ctx 传进去

如果直接写成 async def transfer_money(...): 而不返回函数,ctx 参数就不会被自动注入,这就是为啥要走"返回函数"

4.2 wait_for_event 的工作机制

response = await ctx.wait_for_event(
    HumanResponseEvent,
    waiter_id = f"transfer_{to_account}_{amount}",
    waiter_event = InputRequiredEvent(
       prefix = prompt,
       user_name = "User"
    ),
    requirements = {"user_name": "User"},
)

waiter_event=InputRequiredEvent(...):框架会把这个事件抛到事件流里,所以外层 stream_events() 能收到
HumanResponseEvent:工具在这里暂停,等这个事件被发回
waiter_id:唯一标识,防止多个 wait 互相干扰
requirements:过滤条件,只接收 user_name=="User" 的事件

关键:整个机制是"声明式"的——不用手动管理事件队列,框架会做事件路由。

4.3 外层事件订阅

handler = workflow.run(user_msg = "请帮我向账户 'zhangsan' 转账 100 元")

async for event in handler.stream_events():
    if isinstance(event, InputRequiredEvent):
       # 框架告诉我们:需要人输入
       print("\n" + "=" * 50)
       print(event.prefix)  # 打印提示
       response = input("请输入: ")
       print("=" * 50 + "\n")
       # 把用户输入包装成 HumanResponseEvent 发回
       handler.ctx.send_event(
          HumanResponseEvent(
             response = response,
             user_name = event.user_name,
          )
       )
# 获取最终结果
final = await handler

handler:Agent 运行返回的句柄
stream_events():异步迭代所有事件
看到 InputRequiredEvent → 弹个输入框 → 收到输入后 send_event
注意 send_event 用的是 handler.ctx,不是工具内部的 ctx。但它们指向同一个 Context 对象。

在这里插入图片描述

五、HITL 内部工作流

在这里插入图片描述

六、踩坑指南

坑 1:升级版本!升级版本!升级版本!

LlamaIndex 0.14.0 ~ 0.14.4 的 wait_for_event 有个 bug,会把等待异常吞掉,导致用户输入后 Agent 直接结束。

必须升级到 0.14.10+:

pip install --upgrade llama-index-core

坑 2:工具函数忘记返回 async 函数

# ❌ 错误:直接 async def,ctx 注入不到
async def transfer_money(ctx: Context, to_account: str, amount: float):
    ...

# ✅ 正确:外层 def + 内层 async def + return
def transfer_money(to_account: str, amount: float):
    async def _transfer(ctx: Context):
        ...
    return _transfer

坑 3:InputRequiredEvent 抛出后没人接

如果你用 agent.run() 而不是 agent.run() 返回的 handler 来订阅事件流,事件会被框架"吃掉",人永远收不到提示。

必须用 handler 的 stream_events():

handler = agent.run(...)
async for event in handler.stream_events():  # ✅ 这才能收到
    ...

坑4：HITL 没触发

原因时候： DeepSeek 模型返回了自然语言而非 tool_calls。

system_prompt 中"等待用户确认"被模型误解为"用文本询问"。

修复方法是重写 system_prompt,明确告诉模型"必须调用工具,工具内部已实现确认逻辑",禁止用自然语言确认

修改前：system_prompt 说"涉及金额操作前必须等待用户确认"——这句话有歧义。

system_prompt=(
    "你是一个银行助手,可以使用 transfer_money 工具帮用户转账。"
    "涉及金额操作前必须等待用户确认。"
),

修改后：

system_prompt = (  # 角色设定
    "你是一个银行助手。\n"
    "【强制规则】当用户提出转账请求时,你必须调用 transfer_money 工具,"
    "严禁用自然语言询问或确认。\n"
    "工具内部已实现用户确认流程,工具返回什么你就回复什么。"
),