agno 的 Human-in-the-Loop 实战指南：工具确认、用户输入与外部执行

agno 的 Human-in-the-Loop 实战指南：工具确认、用户输入与外部执行

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno

本文以 agno 仓库中 cookbook/02_agents/10_human_in_the_loop 的示例集为骨架，系统讲解如何在 Agent 运行流程中引入人工介入：包括在工具执行前要求用户确认（confirmation）、在工具调用时向用户索取缺失参数（user input）、以结构化选择题收集偏好（user feedback），以及把工具调用转移到 Agent 外部执行（external execution）。读完本文，你将掌握pause / continue_run的 HITL 循环机制，并能在自己的 agno Agent 中落地可中断、可恢复、可审计的人工审批与输入流程。

目录概览：本目录覆盖的四类 HITL 模式

README 将该目录定位为 "Examples for confirmation flows, user input prompts, and external tool handling"，共 11 个示例文件，可归为四条主线：

所有示例都遵循同一套运行契约：Agent 在需要人工介入时暂停（paused），应用代码处理active_requirements中的待办需求，然后调用continue_run(run_id=..., requirements=...)恢复执行。

运行前置条件

按 README 与仓库脚本，运行这些示例需要：

• 加载环境变量：执行direnv allow，其中必须包含OPENAI_API_KEY（示例默认使用OpenAIResponses/OpenAIChat模型，如gpt-5-mini）。
• 创建演示环境：运行 scripts/demo_setup.sh 创建.venvs/demo虚拟环境，之后统一用.venvs/demo/bin/python运行。
• 部分示例依赖可选服务：README 明确提到 "Some examples require optional local services (for example pgvector) or provider-specific API keys"；例如confirmation_advanced.py使用WikipediaTools，需要wikipedia包（TEST_LOG.md 中记录了该示例因缺少该依赖而 FAIL 的情况）。

运行单个示例的统一命令：

.venvs/demo/bin/python cookbook/02_agents/10_human_in_the_loop/<file>.py

需要说明的是：这些示例几乎全部是交互式脚本——它们在暂停点调用rich.prompt.Prompt.ask或input()等待人工操作，因此在非交互环境（CI、管道）中运行会以EOFError结束，这属于预期行为，详见 TEST_LOG.md。

模式一：工具执行前要求确认（Confirmation）

1.1 单个工具级确认：confirmation_required.py

该示例的核心是用装饰器参数把某个工具标记为"需要确认"：

@tool(requires_confirmation=True) def get_top_hackernews_stories(num_stories: int) -> str: """Fetch top stories from Hacker News. ... """

然后在Agent.run()返回后，遍历run_response.active_requirements，对needs_confirmation的需求逐个人工裁决：

run_response = agent.run("Fetch the top 2 hackernews stories.") for requirement in run_response.active_requirements: if requirement.needs_confirmation: console.print( f"Tool name [bold blue]{requirement.tool_execution.tool_name}({requirement.tool_execution.tool_args})[/] requires confirmation." ) message = Prompt.ask("Do you want to continue?", choices=["y", "n"], default="y").strip().lower() if message == "n": requirement.reject() else: requirement.confirm() run_response = agent.continue_run( run_id=run_response.run_id, requirements=run_response.requirements, )

这段代码展示了 HITL 的标准三段式：读取待办需求 → 人工裁决（confirm/reject）→ 携带裁决结果继续运行。模型侧需要注意：模型每次调用工具前会先看到"该工具需要确认"的约束，从而在第一次运行时不会真正执行工具体，而是产生一个待确认的ToolExecution，把流程停到应用层。

值得注意confirmation_required.py中 Agent 还配置了db=SqliteDb(session_table="test_session", db_file="tmp/example.db")，说明暂停/恢复依赖会话持久化，暂停中的需求与状态会写入数据库，continue_run通过run_id找回。

1.2 多工具场景与"驳回并给出理由"：confirmation_advanced.py

当 Agent 同时挂载多个工具、且只对其中一部分要求确认时，用requires_confirmation_tools参数做白名单。例如 confirmation_advanced.py：

agent = Agent( model=OpenAIResponses(id="gpt-5-mini"), tools=[ get_top_hackernews_stories, # 装饰器级 requires_confirmation=True WikipediaTools(requires_confirmation_tools=["search_wikipedia"]), # 工具包级白名单 ], markdown=True, db=SqliteDb(db_file="tmp/confirmation_required_multiple_tools.db"), )

这里同时展示了两种声明方式：函数装饰器 +Toolkit构造参数。更重要的进阶点在于reject()支持携带说明文字，可以引导模型改用别的工具：

if message == "n": requirement.reject( "This is not the right tool to use. Use the other tool!" )

Agent 收到驳回及理由后，会重新规划并尝试其他工具（该示例提示词要求 "only use one source"，若 Hacker News 被驳回，模型应转向 Wikipedia）。另外该示例用while run_response.is_paused:循环包裹整个处理逻辑，因为多工具场景可能连续暂停多次，直到所有需求被处理完毕。

1.3 Toolkit 级确认：confirmation_toolkit.py

confirmation_toolkit.py 展示了把确认策略声明在工具包上的通用做法，与"按函数装饰器声明"等价：

agent = Agent( model=OpenAIResponses(id="gpt-5-mini"), tools=[WebSearchTools(requires_confirmation_tools=["web_search"])], markdown=True, db=SqliteDb(db_file="tmp/confirmation_required_toolkit.db"), )

注意工具名web_search需要与 Toolkit 内部注册的工具名一致（大小写敏感）。在确认分支上，该示例还给出了一个便利写法：run_response.is_paused与agent.run_response.is_paused等价，可在外层先判断是否暂停再进入需求循环。

1.4 MCP 工具的确认：confirmation_required_mcp_toolkit.py

通过 MCP 接入的外部工具同样可以纳入确认流程，confirmation_required_mcp_toolkit.py 使用MCPTools连接远程 MCP Server（streamable-http 传输），并声明需要确认的工具名：

mcp_tools = MCPTools( transport="streamable-http", url="https://docs.agno.com/mcp", requires_confirmation_tools=["SearchAgno"], # 注意：工具名大小写敏感 ) agent = Agent( model=OpenAIResponses(id="gpt-5.2"), tools=[mcp_tools], markdown=True, db=SqliteDb(db_file="tmp/confirmation_required_toolkit.db"), )

该示例同时演示了异步 + 流式的 HITL 循环：用agent.arun(..., stream=True)迭代事件，事件对象run_event同样具备is_paused与active_requirements；确认完成后调用agent.acontinue_run(run_id=..., requirements=..., stream=True)恢复流式输出。这为在 FastAPI / GUI / 聊天界面中集成 HITL 提供了直接范本。

1.5 确认与 session_state 的组合：confirmation_with_session_state.py

confirmation_with_session_state.py 验证了一个关键语义：工具在暂停前对session_state的修改，在确认后的continue_run中必须保留。示例工具在函数签名中注入run_context，直接读写会话状态：

@tool(requires_confirmation=True) def add_to_watchlist(run_context: RunContext, symbol: str) -> str: """Add a stock symbol to the user's watchlist. Requires confirmation.""" if run_context.session_state is None: run_context.session_state = {} watchlist = run_context.session_state.get("watchlist", []) symbol = symbol.upper() if symbol not in watchlist: watchlist.append(symbol) run_context.session_state["watchlist"] = watchlist return f"Added {symbol} to watchlist. Current watchlist: {watchlist}"

Agent 通过session_state={"watchlist": []}初始化，并在 instructions 中引用状态占位符{watchlist}；暂停后通过agent.get_session_state()检查中间状态，恢复后再取最终状态打印。这印证了暂停/恢复是一条"有状态"的完整往返：状态修改、待确认需求、run 上下文都会被持久化，而不是在暂停时丢失。

1.6 副作用工具的确定性验证：side_effecting_tool_approval.py

side_effecting_tool_approval.py 是一个无需任何模型凭证即可运行的"确定性审批测试"，对理解确认语义最有帮助。它用一个本地DeterministicModel（继承Model，第一次返回工具调用、第二次返回最终回复）精确控制模型行为，然后断言：

• 驳回（reject）的调用不执行工具体：published_reports == []；
• 批准（confirm）的调用恰好执行一次：published_reports == ["Weekly status"]。

def run_case(approve: bool) -> None: agent = Agent(model=DeterministicModel(), tools=[publish_report], db=InMemoryDb()) response = agent.run("Publish the weekly status report.") assert response.is_paused, "The side-effecting tool should require approval." for requirement in response.active_requirements: if requirement.needs_confirmation: requirement.confirm() if approve else requirement.reject("The report is not ready to publish.") response = agent.continue_run(run_id=response.run_id, requirements=response.requirements) assert not response.is_paused

如文件注释所述，邮件发送、支付、数据库写入等一切副作用工具都应遵循同一边界：工具体不得在需求确认前执行。这也解释了为何要额外引入external_execution=True模式（见模式四）——把真正有副作用的调用从 Agent 进程内彻底移走。

1.7 确认机制的源码依据

暂停需求的数据模型位于 libs/agno/agno/run/requirement.py 的RunRequirement类：

• needs_confirmation属性：当confirmation与tool_execution.confirmed均为空、且tool_execution.requires_confirmation为真时返回 True；
• confirm()/reject(note=None)：写入裁决并回写tool_execution.confirmed，驳回还可附带confirmation_note供模型读取（源码第 109–123 行）；
• is_resolved()：只有 confirmation / user input / user feedback / external execution 四条需求线全部满足才为 True（源码第 180–187 行）；
• pause_type属性按优先级给出暂停类型：feedback > external > input > confirmation（源码第 190–201 行）。

@tool装饰器支持requires_confirmation、external_execution、requires_user_input、user_input_fields等参数，声明逻辑见 libs/agno/agno/tools/function.py（参数定义集中在第 1263–1276 行附近）。此外ToolExecution中requires_user_input、user_input_schema、user_feedback_schema、external_execution_required等字段共同驱动RunRequirement各属性判断。

模式二：向用户索取输入（User Input）

2.1 工具级输入需求：user_input_required.py

当一个工具的部分参数必须由用户提供时，用requires_user_input=True加user_input_fields白名单声明。模型可为其余字段自动填值，被列入白名单的字段会从模型可见 schema 中剔除，转而等待人工输入（对应 function.py 第 651 行的excluded_params逻辑）：

# 只要求用户提供 to_address，subject/body 可由模型推断 @tool(requires_user_input=True, user_input_fields=["to_address"]) def send_email(subject: str, body: str, to_address: str) -> str: """Send an email. ... """

处理循环读取requirement.user_input_schema（List[UserInputField]），逐字段打印名称、描述、类型，再调用input()取值并回填field.value：

for field in input_schema: print(f"\nField: {field.name}") print(f"Description: {field_description}") print(f"Type: {field_type}") if field.value is None: user_value = input(f"Please enter a value for {field.name}: ") else: user_value = field.value field.value = user_value run_response = agent.continue_run(run_id=run_response.run_id, requirements=run_response.requirements)

注释还提示了一个等价调用：agent.continue_run(run_response=run_response)，以及调试用的agent.print_response(...)简化流。

2.2 Agent 主动发起询问：user_input.py（agentic user input）

如果想让模型自主决定何时缺参数、缺哪些参数，则挂载UserControlFlowTools（见 libs/agno/agno/tools/user_control_flow.py）。该 Toolkit 暴露get_user_input(user_input_fields: list[dict])工具，内置 instructions 明确要求模型：信息不足时不得"说做不了"，而是调用该工具把字段以表单形式交给用户；布尔字段只把显式肯定回答（true/yes/y/1/on/t）视为 True，其余一律视为 False。

user_input.py 将其与自定义EmailTools（send_email/get_emails）组合，Agent 配置为：

agent = Agent( model=OpenAIResponses(id="gpt-5-mini"), tools=[EmailTools(), UserControlFlowTools()], markdown=True, db=SqliteDb(db_file="tmp/agentic_user_input.db"), )

运行期用while run_response.is_paused:循环处理需求：对每个needs_user_input的需求，同样逐字段提示用户输入并回填。示例先后演示两个场景：发邮件（模型缺邮件正文时主动询问 "What is the weather in Tokyo?"）与查邮件（缺起止日期时询问），展示同一套循环可复用。

2.3 结构化选择题收集：user_feedback.py

当需要的是"从选项中选择"而非"自由输入"时，使用UserFeedbackTools。其底层模型定义在 libs/agno/agno/tools/user_feedback.py：AskUserQuestion（question、header 最多 12 字符、options 2–4 个、multi_select）与AskUserOption（label、可选 description），工具名为ask_user。

user_feedback.py 构建了一个旅行助手，instructions 指示模型在规划行程时用ask_user澄清偏好：

agent = Agent( model=OpenAIResponses(id="gpt-5.2"), tools=[UserFeedbackTools()], instructions=[ "You are a helpful travel assistant.", "When the user asks you to plan a trip, use the ask_user tool to clarify their preferences.", ], markdown=True, db=SqliteDb(db_file="tmp/user_feedback.db"), )

处理侧通过requirement.needs_user_feedback与requirement.user_feedback_schema获取问题列表，打印编号选项，支持单选与多选（逗号分隔编号），最后调用requirement.provide_user_feedback(selections)提交（见 requirement.py 第 146–171 行的实现，它会把选择写回selected_options并标记answered）。

模式三：外部工具执行（External Tool Execution）

3.1 把工具调用移出 Agent 进程：external_tool_execution.py

某些工具（如执行本地 Shell 命令、调用企业内网服务）不应由模型直接触发，而应由宿主程序代为执行。用@tool(external_execution=True)声明后，Agent 遇到该工具调用时只"暂停并移交"，不会自己运行工具体：

@tool(external_execution=True) def execute_shell_command(command: str) -> str: """Execute a shell command. ... """

处理循环通过requirement.needs_external_execution识别这类需求，然后由宿主代码自行调用tool.entrypoint(**tool_args)执行，并把结果写回：

if run_response.is_paused: for requirement in run_response.active_requirements: if requirement.needs_external_execution: if requirement.tool_execution.tool_name == execute_shell_command.name: print(f"Executing {requirement.tool_execution.tool_name} with args {requirement.tool_execution.tool_args} externally") result = execute_shell_command.entrypoint(**requirement.tool_execution.tool_args) # 必须把结果设置回 tool_execution，Agent 才能继续 requirement.set_external_execution_result(result) run_response = agent.continue_run(run_id=run_response.run_id, requirements=run_response.requirements)

set_external_execution_result()在 requirement.py 第 173–178 行实现，会同时写入external_execution_result与tool_execution.result，随后needs_external_execution自动变为 False。这等于在"模型提方案、宿主管执行"之间划了一条清晰的安全边界。

3.2 混合外部工具与常规工具：mixed_external_and_regular_tools.py

当一个 Agent 同时挂载外部工具与常规工具时，运行规则在 mixed_external_and_regular_tools.py 的文件注释中写得很明确：

1. 常规工具（如get_current_date）由 Agent 自动执行；
2. 遇到外部工具（如get_user_location）时暂停，等待宿主人为处理；
3. 提供外部结果后恢复，Agent 合并两类结果继续作答。

agent = Agent( model=OpenAIResponses(id="gpt-5-mini"), tools=[get_user_location, get_current_date], # 一个 external_execution=True，一个普通函数 markdown=True, db=SqliteDb(session_table="mixed_tools_session", db_file="tmp/mixed_tools.db"), )

该示例没有外层循环——用if run_response.is_paused:单次判断即可，因为外部执行只有一轮。这一点与确认/输入场景（可能多轮暂停）形成对照，可根据实际暂停次数选择if或while。

模式四：HITL 循环的完整心智模型

综合上述四类模式与 libs/agno/agno/run/requirement.py 的实现，可以把 agno 的 HITL 抽象为一张通用时序：

1. 暂停：Agent 运行中，模型产生一个带requires_confirmation/requires_user_input/user_feedback_schema/external_execution_required标记的ToolExecution，运行暂停，返回RunOutput，其中is_paused=True、active_requirements列出所有待办RunRequirement；
2. 人工介入：宿主程序按requirement.needs_*属性分发处理——确认类走confirm()/reject(note)，输入类走field.value=...（或provide_user_input），反馈类走provide_user_feedback(selections)，外部执行类走set_external_execution_result(result)；
3. 恢复：调用agent.continue_run(run_id=..., requirements=...)（异步流式场景为acontinue_run），Agent 依据裁决继续执行：批准则执行工具体，驳回则把confirmation_note反馈给模型重试，外部结果则直接注入上下文；
4. 收敛：当所有需求is_resolved()后运行不再暂停，pprint.pprint_run_response(run_response)输出最终结果。

其中会话持久化（SqliteDb/InMemoryDb）保证了暂停与恢复之间的状态完整性——包括session_state的修改、需求列表、工具调用记录；RunRequirement.to_dict()/from_dict()（见 requirement.py 第 203–360 行）负责序列化往返。这也解释了为什么所有示例都显式配置了db参数。

验证与已知限制

TEST_LOG.md 记录了这些示例在.venvs/demo/bin/python环境下的实测结果，可作为预期行为参考：

• agentic_user_input.py、confirmation_required.py、confirmation_required_mcp_toolkit.py、confirmation_toolkit.py、user_input_required.py：PASS（interactive）——交互式脚本，在非交互模式下的EOFError属预期行为；
• external_tool_execution.py、mixed_external_and_regular_tools.py：PASS，可无人工参与跑通（约 11 秒）；
• confirmation_advanced.py：FAIL——缺少wikipedia依赖（ModuleNotFoundError），需先安装对应依赖再运行。

由这些记录可以总结出两条使用注意事项：一是交互式 HITL 示例不适合直接放进无输入流的自动化环境；二是Toolkit型依赖（如WikipediaTools）会引入额外的第三方包，运行前需确认 demo 环境已包含。

结语

本目录 11 个示例覆盖了 HITL 的全部核心形态：工具确认（函数级、Toolkit 级、MCP 级）、用户输入（自由输入与结构化选择题）、外部工具执行（纯外部与混合模式），并附带了会话状态保留与确定性审批测试两个进阶样本。配合 requirement.py 的RunRequirement数据模型与 function.py 的@tool参数体系，你可以在自己的 Agent 上按同样的pause → 人工介入 → continue_run循环接入审批、补参、人工执行等能力，把不可控的模型自主调用收敛为"模型提案、人来拍板"的可信流程。

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno