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 个示例文件可归为四条主线模式示例文件核心 API工具执行前确认confirmation_required.py、confirmation_advanced.py、confirmation_toolkit.py、confirmation_required_mcp_toolkit.py、confirmation_with_session_state.py、side_effecting_tool_approval.pytool(requires_confirmationTrue)、requirement.confirm() / reject()请求用户输入user_input.pyagentic、user_input_required.pytool(requires_user_inputTrue, user_input_fields...)、UserControlFlowTools结构化问题收集user_feedback.pyUserFeedbackTools外部工具执行external_tool_execution.py、mixed_external_and_regular_tools.pytool(external_executionTrue)、requirement.set_external_execution_result()所有示例都遵循同一套运行契约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。模式一工具执行前要求确认Confirmation1.1 单个工具级确认confirmation_required.py该示例的核心是用装饰器参数把某个工具标记为需要确认tool(requires_confirmationTrue) 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( fTool 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], defaulty).strip().lower() if message n: requirement.reject() else: requirement.confirm() run_response agent.continue_run( run_idrun_response.run_id, requirementsrun_response.requirements, )这段代码展示了 HITL 的标准三段式读取待办需求 → 人工裁决confirm/reject→ 携带裁决结果继续运行。模型侧需要注意模型每次调用工具前会先看到该工具需要确认的约束从而在第一次运行时不会真正执行工具体而是产生一个待确认的ToolExecution把流程停到应用层。值得注意confirmation_required.py中 Agent 还配置了dbSqliteDb(session_tabletest_session, db_filetmp/example.db)说明暂停/恢复依赖会话持久化暂停中的需求与状态会写入数据库continue_run通过run_id找回。1.2 多工具场景与驳回并给出理由confirmation_advanced.py当 Agent 同时挂载多个工具、且只对其中一部分要求确认时用requires_confirmation_tools参数做白名单。例如 confirmation_advanced.pyagent Agent( modelOpenAIResponses(idgpt-5-mini), tools[ get_top_hackernews_stories, # 装饰器级 requires_confirmationTrue WikipediaTools(requires_confirmation_tools[search_wikipedia]), # 工具包级白名单 ], markdownTrue, dbSqliteDb(db_filetmp/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.pyconfirmation_toolkit.py 展示了把确认策略声明在工具包上的通用做法与按函数装饰器声明等价agent Agent( modelOpenAIResponses(idgpt-5-mini), tools[WebSearchTools(requires_confirmation_tools[web_search])], markdownTrue, dbSqliteDb(db_filetmp/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 Serverstreamable-http 传输并声明需要确认的工具名mcp_tools MCPTools( transportstreamable-http, urlhttps://docs.agno.com/mcp, requires_confirmation_tools[SearchAgno], # 注意工具名大小写敏感 ) agent Agent( modelOpenAIResponses(idgpt-5.2), tools[mcp_tools], markdownTrue, dbSqliteDb(db_filetmp/confirmation_required_toolkit.db), )该示例同时演示了异步 流式的 HITL 循环用agent.arun(..., streamTrue)迭代事件事件对象run_event同样具备is_paused与active_requirements确认完成后调用agent.acontinue_run(run_id..., requirements..., streamTrue)恢复流式输出。这为在 FastAPI / GUI / 聊天界面中集成 HITL 提供了直接范本。1.5 确认与 session_state 的组合confirmation_with_session_state.pyconfirmation_with_session_state.py 验证了一个关键语义工具在暂停前对session_state的修改在确认后的continue_run中必须保留。示例工具在函数签名中注入run_context直接读写会话状态tool(requires_confirmationTrue) def add_to_watchlist(run_context: RunContext, symbol: str) - str: Add a stock symbol to the users 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 fAdded {symbol} to watchlist. Current watchlist: {watchlist}Agent 通过session_state{watchlist: []}初始化并在 instructions 中引用状态占位符{watchlist}暂停后通过agent.get_session_state()检查中间状态恢复后再取最终状态打印。这印证了暂停/恢复是一条有状态的完整往返状态修改、待确认需求、run 上下文都会被持久化而不是在暂停时丢失。1.6 副作用工具的确定性验证side_effecting_tool_approval.pyside_effecting_tool_approval.py 是一个无需任何模型凭证即可运行的确定性审批测试对理解确认语义最有帮助。它用一个本地DeterministicModel继承Model第一次返回工具调用、第二次返回最终回复精确控制模型行为然后断言驳回reject的调用不执行工具体published_reports []批准confirm的调用恰好执行一次published_reports [Weekly status]。def run_case(approve: bool) - None: agent Agent(modelDeterministicModel(), tools[publish_report], dbInMemoryDb()) 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_idresponse.run_id, requirementsresponse.requirements) assert not response.is_paused如文件注释所述邮件发送、支付、数据库写入等一切副作用工具都应遵循同一边界工具体不得在需求确认前执行。这也解释了为何要额外引入external_executionTrue模式见模式四——把真正有副作用的调用从 Agent 进程内彻底移走。1.7 确认机制的源码依据暂停需求的数据模型位于 libs/agno/agno/run/requirement.py 的RunRequirement类needs_confirmation属性当confirmation与tool_execution.confirmed均为空、且tool_execution.requires_confirmation为真时返回 Trueconfirm()/reject(noteNone)写入裁决并回写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 Input2.1 工具级输入需求user_input_required.py当一个工具的部分参数必须由用户提供时用requires_user_inputTrue加user_input_fields白名单声明。模型可为其余字段自动填值被列入白名单的字段会从模型可见 schema 中剔除转而等待人工输入对应 function.py 第 651 行的excluded_params逻辑# 只要求用户提供 to_addresssubject/body 可由模型推断 tool(requires_user_inputTrue, user_input_fields[to_address]) def send_email(subject: str, body: str, to_address: str) - str: Send an email. ... 处理循环读取requirement.user_input_schemaList[UserInputField]逐字段打印名称、描述、类型再调用input()取值并回填field.valuefor field in input_schema: print(f\nField: {field.name}) print(fDescription: {field_description}) print(fType: {field_type}) if field.value is None: user_value input(fPlease enter a value for {field.name}: ) else: user_value field.value field.value user_value run_response agent.continue_run(run_idrun_response.run_id, requirementsrun_response.requirements)注释还提示了一个等价调用agent.continue_run(run_responserun_response)以及调试用的agent.print_response(...)简化流。2.2 Agent 主动发起询问user_input.pyagentic 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 将其与自定义EmailToolssend_email/get_emails组合Agent 配置为agent Agent( modelOpenAIResponses(idgpt-5-mini), tools[EmailTools(), UserControlFlowTools()], markdownTrue, dbSqliteDb(db_filetmp/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.pyAskUserQuestionquestion、header 最多 12 字符、options 2–4 个、multi_select与AskUserOptionlabel、可选 description工具名为ask_user。user_feedback.py 构建了一个旅行助手instructions 指示模型在规划行程时用ask_user澄清偏好agent Agent( modelOpenAIResponses(idgpt-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., ], markdownTrue, dbSqliteDb(db_filetmp/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 Execution3.1 把工具调用移出 Agent 进程external_tool_execution.py某些工具如执行本地 Shell 命令、调用企业内网服务不应由模型直接触发而应由宿主程序代为执行。用tool(external_executionTrue)声明后Agent 遇到该工具调用时只暂停并移交不会自己运行工具体tool(external_executionTrue) 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(fExecuting {requirement.tool_execution.tool_name} with args {requirement.tool_execution.tool_args} externally) result execute_shell_command.entrypoint(**requirement.tool_execution.tool_args) # 必须把结果设置回 tool_executionAgent 才能继续 requirement.set_external_execution_result(result) run_response agent.continue_run(run_idrun_response.run_id, requirementsrun_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 的文件注释中写得很明确常规工具如get_current_date由 Agent 自动执行遇到外部工具如get_user_location时暂停等待宿主人为处理提供外部结果后恢复Agent 合并两类结果继续作答。agent Agent( modelOpenAIResponses(idgpt-5-mini), tools[get_user_location, get_current_date], # 一个 external_executionTrue一个普通函数 markdownTrue, dbSqliteDb(session_tablemixed_tools_session, db_filetmp/mixed_tools.db), )该示例没有外层循环——用if run_response.is_paused:单次判断即可因为外部执行只有一轮。这一点与确认/输入场景可能多轮暂停形成对照可根据实际暂停次数选择if或while。模式四HITL 循环的完整心智模型综合上述四类模式与 libs/agno/agno/run/requirement.py 的实现可以把 agno 的 HITL 抽象为一张通用时序暂停Agent 运行中模型产生一个带requires_confirmation/requires_user_input/user_feedback_schema/external_execution_required标记的ToolExecution运行暂停返回RunOutput其中is_pausedTrue、active_requirements列出所有待办RunRequirement人工介入宿主程序按requirement.needs_*属性分发处理——确认类走confirm()/reject(note)输入类走field.value...或provide_user_input反馈类走provide_user_feedback(selections)外部执行类走set_external_execution_result(result)恢复调用agent.continue_run(run_id..., requirements...)异步流式场景为acontinue_runAgent 依据裁决继续执行批准则执行工具体驳回则把confirmation_note反馈给模型重试外部结果则直接注入上下文收敛当所有需求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.pyPASSinteractive——交互式脚本在非交互模式下的EOFError属预期行为external_tool_execution.py、mixed_external_and_regular_tools.pyPASS可无人工参与跑通约 11 秒confirmation_advanced.pyFAIL——缺少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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考