ag2-hitl

用途

该技能用于为 AG2 的 Agent 增加人机协同（HITL）能力，让智能体在运行过程中暂停并引入人类判断。典型场景包括：智能体在执行有风险操作前请求确认；在对话中途需要用户提供信息（如密码、API 密钥、缺失的上下文）；某个特定工具调用在运行前需要人工审批（不可逆、昂贵或敏感操作）；以及在最终定稿前展示草稿并获取人类编辑或批准。

两种机制

根据意图选择不同机制：

需要工具提出开放式问题并等待用户输入：在工具内部使用 context.input()，并在智能体上配置 hitl_hook。
需要在特定工具调用主体运行前批准或拒绝：使用 approval_required() 工具中间件。

模式一：context.input() 用于开放式提问

工具通过 Context.input(message, timeout=...) 请求输入。智能体必须配置一个知道如何收集该输入的 hitl_hook。

from ag2 import Agent, Context, tool
from ag2.events import HumanInputRequest, HumanMessage

@tool
async def execute_query(context: Context) -> str:
    answer = await context.input(
        "Are you sure you want to run this query? (yes/no)",
        timeout=60.0,
    )
    if answer.strip().lower() != "yes":
        return "Query cancelled."
    return "Query executed successfully."

def hitl_hook(event: HumanInputRequest) -> HumanMessage:
    print(f"Agent asks: {event.content}")
    return HumanMessage(content=input("Your answer: "))

agent = Agent("dba", tools=[execute_query], hitl_hook=hitl_hook)

该 hook 接收一个 HumanInputRequest（提示内容在 event.content 中），并返回 HumanMessage 或普通字符串（框架会通过 HumanMessage.ensure_message 包装字符串）。同时支持 def 和 async def 两种 hook。

也可以在构造之后注册 hook：

agent = Agent("dba", tools=[execute_query])

@agent.hitl_hook
async def async_hitl_hook(event: HumanInputRequest) -> HumanMessage:
    answer = await collect_from_ui(event.content)
    return HumanMessage(content=answer)

装饰器会覆盖构造函数中设置的任何 hook——但如果之前已经设置过（例如通过构造函数），应用 @agent.hitl_hook 会发出 RuntimeWarning（"You already set HITL hook, provided value overrides it"）。建议只在一个地方设置 hook 以避免该警告。

hook 参与依赖注入——Context、Inject、Variable、Depends 的工作方式与工具中相同。

如果调用了 context.input() 但没有注册 hook，框架会抛出 HumanInputNotProvidedError。

模式二：approval_required() 用于特定工具调用

使用内置审批中间件为单个工具设置门禁。在工具主体运行前提示用户，用户可以批准或拒绝。

from ag2 import Agent, tool
from ag2.config import OpenAIConfig
from ag2.middleware import approval_required

@tool(middleware=[approval_required()])
def delete_account(user_id: str) -> str:
    """Deletes a user account by ID permanently."""
    return f"Account {user_id} deleted."

agent = Agent(
    "support",
    config=OpenAIConfig(model="gpt-4o-mini"),
    tools=[delete_account],
    hitl_hook=lambda event: input(event.content),
)

当智能体调用 delete_account 时，用户会看到（默认 allow_always=True）：

Agent wants to call the tool:
`delete_account`, {"user_id": "abc-123"}
Please approve or deny this request.
Y/N/Always?

回答在匹配前会被转为小写。y、yes 或 1 批准本次调用；always 批准本次及同一上下文中该工具的后续所有调用（它会设置一个按上下文生效的绕过标志）。其他任何回答都会拒绝；智能体收到 denied_message（默认为 "User denied the tool call request"）并可据此调整。默认超时为 30 秒。设置 allow_always=False 可去掉 "Always" 选项（提示则只显示 Y/N?）。

approval_required() 底层调用 context.input()，因此同样需要 hitl_hook——没有它会出现运行时错误。

自定义提示

@tool(middleware=[approval_required(
    message="⚠️ Run `{tool_name}` with {tool_arguments}? (y/n)",
    denied_message="Operation blocked by user.",
)])
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to the given address."""
    ...

{tool_name} 和 {tool_arguments} 会被插值替换。

组合两种模式

对于既需要在运行中途收集输入又需要审批的工具：

@tool(middleware=[approval_required()])
async def schedule_report(name: str, context: Context) -> str:
    """Schedule a report — asks the user for the cadence, then runs after approval."""
    cadence = await context.input("How often? (daily / weekly / monthly)")
    return f"Scheduled '{name}' on {cadence} cadence."

审批中间件先运行（最外层）。一旦获批，工具主体执行，context.input() 触发第二次人类交互。

深入了解

源文档：website/docs/user-guide/context/human_in_the_loop.mdx（context.input、hitl_hook）、website/docs/user-guide/tools/approval_required.mdx（approval_required 中间件）。

工具中间件通用文档：website/docs/user-guide/tools/tool_middleware.mdx。另请参见 ag2-middleware，了解通过 BaseMiddleware.on_human_input() 实现的智能体级 HITL 拦截。

HITL hook 支持与工具完全相同的依赖注入——参见 ../ag2-add-custom-tool/references/dependency_injection.md。

常见陷阱

使用 approval_required() 但没有 hitl_hook——该中间件会调用 context.input()，因此智能体需要 hook，否则会出现 HumanInputNotProvidedError。
忘记处理拒绝路径——context.input() 返回 hook 返回的任何内容。如果只对 "yes" 分支处理，其他任何回答（包括静默/默认）都会让操作继续。务必进行验证。
在异步 UI 中使用同步 input()——input() 会阻塞事件循环。对于任何非 CLI 应用，使用异步 hook（async def）和异步输入收集器（web socket、消息队列）。
没有超时——context.input(prompt) 可能永远等待。对任何生产路径都应传入 timeout=60.0（秒）。
装饰器 hook 覆盖构造函数 hook——如果两者都设置，装饰器获胜并发出 RuntimeWarning。只选一个地方设置。
期望 HumanMessage 自动流入对话历史——对于发起请求的工具的返回值，确实如此，但通过 ctx.input() 收集的运行中输入并不是单独的用户轮次，它们存在于工具的作用域中。