人工智能大模型AI AgentAgent 框架多智能体工具调用RAGAgent 工作流【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址https://gitcode.com/GitHub_Trending/ag/agno点击查看免费下载本篇技术指南围绕 agno 开源仓库中 cookbook/02_agents/10_human_in_the_loop 目录的示例集合及其 TEST_LOG.md 测试记录展开系统讲解 Human-in-the-LoopHITL在 Agent 中的三类核心交互工具调用前的用户确认confirmation、执行前需要用户补充的输入user input以及交由外部进程执行的工具external execution。读完本文你将掌握tool装饰器上requires_confirmation、requires_user_input、external_execution三个标志位的声明方式理解RunRequirement驱动的暂停—收集—续跑pause / continue循环并能照抄示例代码在自己的 Agent 中落地可复现的 HITL 流程。一、什么是 agno 的 Human-in-the-Loop在纯自动化的 Agent 链路中模型可以自由调用工具但面对**有副作用side-effecting**的操作——如发邮件、写数据库、对外发布内容、执行 Shell 命令——让模型自行决定往往不可接受。agno 提供了标准化的 HITL 机制当 Agent 决定调用一个声明了 HITL 属性的工具时运行会暂停paused把待办要求以RunRequirement的形式暴露给宿主程序宿主程序收集用户/外部系统的反馈后调用continue_run恢复运行。整个过程中 Agent 的工具逻辑可以完全保持原样只是执行边界被推到了 Agent 之外。从 cookbook/02_agents/10_human_in_the_loop/README.md 的定位看这一目录集中了确认流程、用户输入提示和外部工具处理三类示例并额外包含会话状态持久化、副作用工具审批等进阶验证用例。与之配套的 TEST_LOG.md测试于 2026-02-13环境为.venvs/demo/bin/python、pgvector 运行中记录了 8 个脚本的实测结果是理解各示例应当如何表现的第一手依据汇总如下示例脚本状态说明agentic_user_input.pyPASS (interactive)Agent 主动请求用户输入交互式文件非交互模式下等待输入confirmation_advanced.pyFAIL缺少依赖wikipediaModuleNotFoundError应归类为 SKIP 或补充依赖confirmation_required.pyPASS (interactive)需要用户确认非交互模式按预期抛EOFErrorconfirmation_required_mcp_toolkit.pyPASS (interactive)MCP toolkit 场景的确认流程同上confirmation_toolkit.pyPASS (interactive)Toolkit 场景的确认流程同上external_tool_execution.pyPASS外部执行工具11 秒内完成并产生预期输出mixed_external_and_regular_tools.pyPASS外部执行工具与常规工具混用常规工具自动执行、外部工具暂停等待user_input_required.pyPASS (interactive)需要用户输入非交互模式按预期抛EOFError测试记录揭示了两个重要事实其一交互式示例在非交互模式下抛EOFError属于预期行为input()读到文件末尾所致并非故障其二confirmation_advanced.py的失败原因是测试环境缺少wikipedia依赖属于环境配置问题而非代码逻辑错误。二、核心机制暂停pause与续跑continue所有 HITL 示例都遵循同一个运行模式agent.run(...)返回RunResponse若run_response.is_paused为 True说明有工具调用等待外部处理遍历run_response.active_requirements对每个RunRequirement判断其需求类型并处理调用agent.continue_run(run_id..., requirements...)恢复运行必要时重复 24 步while循环。RunRequirement是这一机制的载体定义在 libs/agno/agno/run/requirement.pyclass RunRequirement文档字符串即 Requirement to complete a paused run (used in HITL flows)。从源码可以看到它通过几个属性暴露需求类型needs_confirmation当工具设置了requires_confirmation且尚未被确认/拒绝时为 Truerequirement.py 中needs_confirmation属性实现needs_user_input当工具设置了requires_user_input或user_input_schema中存在value is None的字段时为 Trueneeds_user_feedback当存在未回答的结构化问题user_feedback_schema时为 Trueneeds_external_execution当工具设置了外部执行且尚未收到结果时为 True。对应的处理方法同样定义在RunRequirement上confirm()/reject(note)用于确认流程provide_user_input(values)用于填入用户输入provide_user_feedback(selections)用于提交结构化反馈set_external_execution_result(result)用于回填外部执行结果。其中reject支持携带说明性note如选错了工具而provide_user_input只有在所有字段都被赋值后才会把工具标记为已作答answered True确保不会带着残缺输入继续运行。三、工具声明三个互斥的 HITL 标志位HITL 行为在工具定义层面声明而非在 Agent 调用处。tool装饰器实现于 libs/agno/agno/tools/decorator.py提供三个核心参数参数作用requires_confirmationTrue工具执行前必须获得用户确认confirm/rejectrequires_user_inputTrue工具执行前必须由用户提供输入字段external_executionTrue工具不在 Agent 上下文内执行由外部宿主执行并回填结果注意一个容易被忽视的约束这三个标志在同一工具上最多只能设置一个为 True。decorator.py 中有一段显式的互斥校验——true_flags_count 1时直接抛出ValueError(Only one of requires_user_input, requires_confirmation, or external_execution can be set to True at the same time.)。这是因为三种模式对应的暂停语义不同、处理接口也不同混用会造成需求歧义。此外requires_user_input可以搭配user_input_fields使用传入字段名列表时只有这些字段需要用户提供其余参数仍由模型生成留空则意味着所有参数都由用户提供。external_execution还有配套的external_execution_silent用于抑制暂停时的冗长提示信息。Function类libs/agno/agno/tools/function.py中定义了UserInputFieldname、field_type、description、value工具的参数 schema 在运行时据此生成user_input_schema宿主程序逐字段收集后回填。四、模式一执行前确认Confirmation4.1 单工具确认confirmation_required.py最简单的确认流程是对单个工具加上requires_confirmationTrue标记。以 confirmation_required.py 为例tool(requires_confirmationTrue) def get_top_hackernews_stories(num_stories: int) - str: Fetch top stories from Hacker News. # ... 调用 Hacker News API 并返回 JSON ...Agent 侧的主循环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, ) pprint.pprint_run_response(run_response)流程要点agent.run返回后宿主遍历active_requirements对每个需要确认的要求用rich的Prompt.ask询问用户选择n则reject()工具将不会执行选择y则confirm()工具恢复执行随后continue_run携带所有 requirements 续跑。这正是 TEST_LOG.md 中该示例被标记为PASS (interactive)的原因——必须有真实终端输入才能完成全流程非交互模式下Prompt.ask读取不到输入而抛EOFError属于预期行为。4.2 Toolkit 级确认confirmation_toolkit.py对于现成 Toolkit无需改写其中每个工具只需在构造 Toolkit 时声明需要确认的工具名即可。见 confirmation_toolkit.pyagent Agent( modelOpenAIResponses(idgpt-5-mini), tools[WebSearchTools(requires_confirmation_tools[web_search])], markdownTrue, dbSqliteDb(db_filetmp/confirmation_required_toolkit.db), )当模型决定调用web_search时运行会暂停宿主走与 4.1 相同的确认循环。这里的判断条件使用了run_response.is_paused源码注释提示也可以用agent.run_response.is_paused语义更加明确。4.3 MCP 工具集确认confirmation_required_mcp_toolkit.py确认机制同样适用于通过 MCP 协议接入的外部工具。见 confirmation_required_mcp_toolkit.pymcp_tools MCPTools( transportstreamable-http, urlhttps://docs.agno.com/mcp, requires_confirmation_tools[SearchAgno], # Note: Tool names are case-sensitive ) agent Agent( modelOpenAIResponses(idgpt-5.2), tools[mcp_tools], markdownTrue, dbSqliteDb(db_filetmp/confirmation_required_toolkit.db), )两个值得注意的细节一是工具名大小写敏感源码注释明确提醒requires_confirmation_tools里的名称必须与 MCP 服务器暴露的工具名完全一致二是该示例使用了异步流式接口agent.arun(..., streamTrue)与agent.acontinue_run(..., streamTrue)暂停事件通过run_event.is_paused判断确认完成后在acontinue_run的异步迭代中继续输出内容。这展示了 HITL 在流式场景下的用法暂停前流式输出、暂停时收集确认、续跑后继续输出。4.4 多工具与拒绝原因confirmation_advanced.py当 Agent 同时挂载多个需要确认的工具本例中既有requires_confirmationTrue的自定义工具也有WikipediaTools(requires_confirmation_tools[search_wikipedia])且一次任务可能多次暂停时应使用while run_response.is_paused循环confirmation_advanced.pywhile run_response.is_paused: for requirement in run_response.active_requirements: if requirement.needs_confirmation: # ... 询问用户 y/n ... if message n: requirement.reject(This is not the right tool to use. Use the other tool!) else: requirement.confirm() run_response agent.continue_run( run_idrun_response.run_id, requirementsrun_response.requirements, ) pprint.pprint_run_response(run_response)这里reject()携带了说明文字拒绝后 Agent 可以读到该 noteconfirmation_note并据此调整策略例如本例要求换一个来源获取文章。这也解释了为什么 TEST_LOG.md 将其单独标记为 FAIL 而非逻辑问题——失败纯粹是环境缺失wikipedia模块所致属于依赖管理问题。五、模式二执行前用户输入User Input5.1 指定字段user_input_required.py如果工具缺少关键参数如收件人地址可以声明requires_user_inputTrue并指定user_input_fields让用户补充。见 user_input_required.py# 指定 user_input_fields 时仅这些字段由用户提供留空则所有字段都由用户提供 tool(requires_user_inputTrue, user_input_fields[to_address]) def send_email(subject: str, body: str, to_address: str) - str: Send an email. return fSent email to {to_address} with subject {subject} and body {body}主循环负责把user_input_schema中的每个字段展示给用户并收集值run_response agent.run(Send an email with the subject Hello and the body Hello, world!) for requirement in run_response.active_requirements: if requirement.needs_user_input: input_schema: List[UserInputField] requirement.user_input_schema for field in input_schema: field_type field.field_type field_description field.description 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: print(fValue: {field.value}) 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)注意模型已经通过subject、body生成了内容只有to_address被排除在模型参数之外等待用户输入用户填写后field.value被回写continue_run后工具以完整参数执行。由于依赖终端input()该示例在非交互模式下按预期抛EOFError对应 TEST_LOG.md 中的PASS (interactive)。5.2 Agent 主动请求输入agentic_user_input.py/user_input.py上述方式要求工具参数天然缺位而 Agentic 场景下Agent 自己判断信息不足并主动向用户要数据。示例agentic_user_input.py当前仓库树中对应内容为 user_input.py通过UserControlFlowTools实现——该工具集定义于 libs/agno/agno/tools/user_control_flow.py其中get_user_input工具本身不做任何执行函数体直接返回 User input received真正的暂停逻辑由 Agent 框架截获。class EmailTools(Toolkit): def __init__(self, *args, **kwargs): super().__init__( nameEmailTools, tools[self.send_email, self.get_emails], *args, **kwargs ) # send_email / get_emails 定义略 ... agent Agent( modelOpenAIResponses(idgpt-5-mini), tools[EmailTools(), UserControlFlowTools()], markdownTrue, dbSqliteDb(db_filetmp/agentic_user_input.db), )用户侧收集循环与 5.1 几乎一致区别在于这里必须使用while循环——因为 Agent 可能多次、分批地向用户索要字段模型看到字段缺失就再次调用get_user_input。UserControlFlowTools自带默认指令源码中DEFAULT_INSTRUCTIONS约束模型信息不足时必须用工具索取、不得编造、不得重复询问同一字段、布尔字段只认显式肯定回答等行为保证交互收敛。TEST_LOG 记录其状态为PASS (interactive)——setup succeeded, waiting for user input as expected即验证了Agent 能正确暂停等待输入这一核心行为。六、结构化反馈模式user_feedback.py与自由文本输入不同UserFeedbackTools允许 Agent 以预定义选项的问卷形式向用户提问适用于让用户在有限选择中表态的场景user_feedback.pyagent 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), )其底层数据结构定义在 libs/agno/agno/tools/user_feedback.pyAskUserQuestion含question、header、options、multi_select与AskUserOption含label、可选description。宿主侧的处理循环支持单选与多选逗号分隔编号while run_response.is_paused: for requirement in run_response.active_requirements: if requirement.needs_user_feedback: feedback_schema requirement.user_feedback_schema selections {} for question in feedback_schema: print(f\n{question.header or Question}: {question.question}) # 打印选项编号等待用户输入 if question.multi_select: raw input(Select options (comma-separated numbers): ) # ... 解析为 label 列表 ... else: raw input(Select an option (number): ) # ... 解析为单个 label ... selections[question.question] selected requirement.provide_user_feedback(selections) run_response agent.continue_run(...)该模式与用户输入的差异在于约束强度provide_user_feedback接收的是选项 label 的映射Agent 只能从预设答案中消费用户选择适合偏好收集、决策确认等结构化场景。七、模式三外部工具执行External Execution7.1 单一外部工具external_tool_execution.py有些工具不适合在 Agent 进程内执行如需要专用环境、权限边界或人工操作。声明external_executionTrue后工具被留白由宿主在 Agent 之外执行并回填结果。见 external_tool_execution.pytool(external_executionTrue) def execute_shell_command(command: str) - str: Execute a shell command. if command.startswith(ls): return subprocess.check_output(command, shellTrue).decode(utf-8) else: raise Exception(fUnsupported command: {command}) # ... 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) requirement.set_external_execution_result(result) run_response agent.continue_run( run_idrun_response.run_id, requirementsrun_response.requirements, )关键点宿主通过requirement.tool_execution.tool_name/tool_args拿到待执行的调用信息在外部执行本例直接调用函数entrypoint生产环境可以是独立服务、人工审批系统等然后必须用set_external_execution_result(result)把结果写回Agent 续跑后才能继续推理。该示例在测试记录中为纯PASS11 秒内完成无需交互输入因为它把外部执行模拟成了宿主进程内的直接调用。7.2 混用常规工具与外部工具mixed_external_and_regular_tools.py一个 Agent 可以同时拥有自动执行的常规工具和需要暂停的外部工具。行为语义在 mixed_external_and_regular_tools.py 的文档字符串中写得很清楚常规工具自动执行外部工具调用时暂停外部结果提供后续跑。# 常规工具 —— Agent 自动执行 def get_current_date() - str: Get the current date and time. return datetime.now().strftime(%A, %B %d, %Y at %I:%M %p) # 外部工具 —— Agent 暂停由宿主执行 tool(external_executionTrue) def get_user_location() - str: Get the users current location. return json.dumps({city: San Francisco, country: US}) agent Agent( modelOpenAIResponses(idgpt-5-mini), tools[get_user_location, get_current_date], markdownTrue, dbSqliteDb(session_tablemixed_tools_session, db_filetmp/mixed_tools.db), )宿主侧只需处理needs_external_execution的要求get_current_date由框架自动执行续跑后 Agent 将两路结果合并输出。TEST_LOG 记录其结果为 Regular tool executes automatically, external tool pauses for manual handling, then agent continues with combined results与源码语义完全吻合。八、进阶验证状态持久化与副作用审批的确定性8.1 暂停/续跑中的会话状态confirmation_with_session_state.py一个关键工程问题是工具在暂停前修改的session_state能否在续跑后保留confirmation_with_session_state.py 专门验证这一点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 Agent( modelOpenAIChat(idgpt-5.6-luna), tools[add_to_watchlist], session_state{watchlist: []}, instructionsYou MUST use the add_to_watchlist tool when the user asks to add a stock. The users watchlist is: {watchlist}, dbSqliteDb(db_filetmp/hitl_state.db), markdownTrue, )示例打印暂停后的 session state与确认续跑后的最终 state进行对比并显式检查run_response.status ! RunStatus.paused的情况提示模型可能未调用工具建议重跑。这为暂停不是进程重启提供了验证——状态通过db本例为 SQLite与 requirements 一起在暂停/续跑往返中存活。8.2 确定性审批测试side_effecting_tool_approval.py为了无网络、无凭据地验证审批边界side_effecting_tool_approval.py 用本地 mock 模型DeterministicModel精确控制模型输出第一次调用返回publish_report工具调用第二次返回最终回复。核心断言如下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: if approve: requirement.confirm() 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 # 拒绝时副作用不得发生批准时恰好执行一次 run_case(approveFalse) assert published_reports [], Rejected calls must not execute the tool. run_case(approveTrue) assert published_reports [Weekly status]该示例把确认边界上升到可自动验证的层面被拒绝的调用必须跳过副作用被批准的调用必须恰好执行一次exactly-once。其 docstring 明确指出同样的边界适用于邮件、支付、数据库写入等所有副作用工具——工具体publish_report中published_reports.append只有在 requirement 被确认后才会运行。这是把 HITL 从演示推向可测试的工程契约的范例。九、运行环境与测试结论解读根据 cookbook/02_agents/10_human_in_the_loop/README.md 的前置条件运行这些示例需要加载环境变量含OPENAI_API_KEY可通过direnv allow用 scripts/demo_setup.sh 创建 demo 环境随后以.venvs/demo/bin/python运行部分示例依赖本地可选服务如 pgvector或特定 provider 的 API Key。运行单个示例的命令为.venvs/demo/bin/python cookbook/02_agents/10_human_in_the_loop/file.py结合 TEST_LOG.md 的实测记录可归纳出三条实践结论交互式示例的EOFError不是 Bug凡是依赖input()/Prompt.ask的确认与输入示例在非交互模式下都会以EOFError终止测试记录将其归类为 Expected behavior for interactive file in non-interactive mode。若要接入 CI 自动化应改用程序化注入如provide_user_input传入预置字典或仿照 8.2 使用确定性模型 直接调用confirm()/reject()。依赖完整性决定测试分类confirmation_advanced.py的 FAIL 源于ModuleNotFoundError: No module named wikipedia测试记录建议将其重分类为 SKIP 或补充依赖到 demo 环境。生产使用前请核对示例的 import 清单如httpx、wikipedia、rich是否齐全。状态与结果的持久化是可靠性的基础continue_run依赖run_id与requirements恢复运行工具必须通过set_external_execution_result/field.value/provide_user_feedback显式回写数据配合dbSQLite 等存储与session_state暂停—续跑往返才能保证不丢状态、不重复执行。十、总结agno 的 Human-in-the-Loop 由一条主线贯穿工具声明 HITL 属性 → 运行暂停暴露RunRequirement→ 宿主收集反馈 →continue_run续跑。三条支线分别解决三类问题确认confirmation为副作用工具加审批闸门支持单工具、Toolkit、MCP 工具集可携带拒绝原因适合发布、支付、写操作等场景用户输入user input / user feedback向用户补齐缺失参数或收集结构化选择适合表单补齐、偏好澄清等场景外部执行external execution把工具调用移交到 Agent 之外的执行环境并回填结果适合受控 Shell、专用运行时与人工操作场景。从 requirement.py 的RunRequirement实现到 decorator.py 的互斥校验再到 side_effecting_tool_approval.py 的确定性断言这一整套机制既可用于快速原型也经得起自动化测试与生产级审批流程的检验。上手时建议从 confirmation_required.py 与 user_input_required.py 两个最小示例开始再逐步引入 Toolkit、MCP、流式接口与状态持久化。赞分享人工智能大模型AI AgentAgent 框架多智能体工具调用RAGAgent 工作流【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址https://gitcode.com/GitHub_Trending/ag/agno点击查看免费下载相关推荐Open Headunit媒体键路由完整指南MediaKeyRoutingPolicy如何仲裁蓝牙与Android Auto的按键冲突Open Headunit媒体键路由完整指南MediaKeyRoutingPolicy如何仲裁蓝牙与Android Auto的按键冲突 Open Headun音视频如何为 LlamaIndex Agent 的工具调用实现人工确认human in the loop如何为 LlamaIndex Agent 的工具调用实现人工确认human in the loop 当你用 LlamaIndex 的 Agent Agen人工智能RAG大模型Haystack 人机协同Human-in-the-LoopAPI 完全指南在 Agent 工具执行前插入人工确认、拒绝与参数修改Haystack 人机协同Human in the LoopAPI 完全指南在 Agent 工具执行前插入人工确认、拒绝与参数修改 在构建生产级 LLM人工智能大模型RAGAI AgentNLP上一篇FigmaToCode深度解析从设计思维到代码实现的智能桥梁下一篇MouseTester鼠标性能测试工具专业级精准评测完整指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考