agno 人工审批HITL Approvals全面解析11_approvals 示例集、测试日志与 approval 源码原理【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本篇文章以 agno 仓库中 cookbook/02_agents/11_approvals 目录为核心结合其 TEST_LOG.md、README.md 与 12 个可运行示例系统讲解 agno 的Human-in-the-LoopHITL审批机制两种审批类型阻塞式required与审计式audit、审批记录在数据库中的完整生命周期、以及不同 HITL 场景确认、用户输入、外部执行、团队协作的组合用法。读完本文你将能理解 12 个示例各自验证了什么行为、哪些用例在测试日志中被记录为失败及其断言含义并掌握用approval装饰器为自己的 Agent 工具添加审批门控的完整实战能力。一、关联文档与示例集概况TEST_LOG.md是11_approvals示例目录的测试记录文档记录于2026-02-13。它标明的测试环境为解释器.venvs/demo/bin/python运行环境pgvector: running测试日志记录 Postgres 向量库处于运行状态需要说明的是目录内示例在审批持久化上使用的是SqliteDbDB 文件位于各示例运行目录下的tmp/中该日志逐条列出了目录中 12 个.py示例的测试状态PASS/FAIL、Tier 层级均为 untagged、功能描述与执行结果。而README.md则从概念层面说明了这些示例背后的审批模型。两者结合恰好构成设计意图 → 实现代码 → 运行验证的完整证据链。目录内的 12 个示例文件按功能可划分为两组分组文件阻塞式审批approval默认typerequiredapproval_basic.py、approval_async.py、approval_team.py、approval_list_and_resolve.py、approval_user_input.py、approval_external_execution.py、approval_post_hook.py审计式审批approval(typeaudit)audit_approval_confirmation.py、audit_approval_user_input.py、audit_approval_external.py、audit_approval_async.py、audit_approval_overview.py二、审批机制核心概念approval装饰器2.1 从源码看装饰器语义所有示例都通过from agno.approval import approval导入装饰器。其实现位于 libs/agno/agno/approval/decorator.py支持三种调用形式approval # 无参等价于 typerequired approval() # 显式空参 approval(typeaudit) # 指定类型从源码看其处理逻辑有两个关键分支当approval位于tool之上装饰器收到的是一个Function对象直接把approval_type写入该工具当approval位于tool之下装饰器收到的是原始可调用对象会给原始函数打上一个_agno_approval_type哨兵属性decorator.py 第 10 行定义随后由tool在处理时读取。这意味着两种装饰器书写顺序均被支持。装饰器还会联动 HITL 标志decorator.py 第 53-62 行typerequired默认若tool()上没有设置任何 HITL 标志requires_confirmation/requires_user_input/external_execution会自动置requires_confirmationTruetypeaudit若tool()上没有任何 HITL 标志会抛出ValueError提示审计模式必须至少携带一个 HITL 标志。若传入非required/audit的类型装饰器直接抛出ValueError。2.2 两种审批类型的语义差异结合 README.md 的说明approval阻塞式approval_typerequired在工具执行之前创建一条持久化的审批记录Agent 运行暂停RunStatus.paused数据库中写入approval_typerequired、statuspending的记录。外部系统可以对该审批进行列举、查看与解析resolve只有解析完成后运行才继续。适合先批准、后执行的强管控场景。approval(typeaudit)审计式approval_typeaudit在 HITL 交互解析之后才创建审批记录记录立即处于终态approved或rejected写入approval_typeaudit。它不依赖外部审批系统做阻塞等待主要用于审计追踪——记录谁在什么时间批准/拒绝了什么。关键字段小结审批记录存放在agno_approvals表可通过approvals_table参数改名status取值包括pending、approved、rejected、expired、cancelledapproval_type字段区分required与audit。三、测试结果总览TEST_LOG.md 记录的 12 个用例结果汇总如下示例文件主题状态耗时结果要点approval_basic.py阻塞式审批同步PASS11s工具暂停 → DB 建记录 → 确认 → 续跑 → 解析approval_async.py阻塞式审批异步PASS11sarun/acontinue_run异步链路完整approval_external_execution.py阻塞审批 外部执行PASS4s注入外部执行结果后续跑approval_list_and_resolve.py审批全生命周期PASS12s暂停/列举/过滤/解析/删除含竞态防护验证approval_post_hook.py审批后置钩子PASS—post-hook 从run_output.metadata[approval]读到解析记录approval_team.py团队级审批FAIL—AssertionError: Expected paused, got RunStatus.completedapproval_user_input.py阻塞审批 用户输入PASS6sprovide_user_input补全参数后确认续跑audit_approval_async.py审计审批异步PASS4s异步 确认型审计记录audit_approval_confirmation.py审计审批确认PASS11s同时覆盖 approved 与 rejected 两条路径audit_approval_external.py审计审批 外部执行FAIL—AssertionError: Expected paused, got RunStatus.completedaudit_approval_overview.py两种审批混合概览FAIL—AssertionError: Expected paused, got RunStatus.completedaudit_approval_user_input.py审计审批 用户输入PASS5s用户输入 审计落库合计9 个 PASS3 个 FAIL。三个 FAIL 用例记录的错误信息完全一致均为AssertionError: Expected paused, got RunStatus.completed日志中标注为 code bug下文第五节专门分析。四、PASS 用例逐个拆解审批工作流实战4.1 阻塞式审批基础链路approval_basic.py这是理解整个审批机制的最佳入口。示例定义了一个被approvaltool(requires_confirmationTrue)双层装饰的 Hacker News 抓取函数approval tool(requires_confirmationTrue) def get_top_hackernews_stories(num_stories: int) - str: Fetch top stories from Hacker News. ... ...随后用SqliteDb显式指定审批表并创建 Agentdb SqliteDb( db_fileDB_FILE, session_tableagent_sessions, approvals_tableapprovals ) agent Agent( modelOpenAIResponses(idgpt-5-mini), tools[get_top_hackernews_stories], markdownTrue, dbdb, )运行脚本时的执行链路对应 approval_basic.py 中注释的五个 StepStep 1 — 运行并暂停agent.run(...)返回后断言run_response.is_paused为真。因为工具要求审批Agent 在执行前暂停Step 2 — 校验 DB 记录调用db.get_approvals(statuspending)应至少查到 1 条pending记录记录中携带id、run_id、status、source_type、context等字段Step 3 — 确认并续跑遍历run_response.active_requirements对needs_confirmation为真的 requirement 调用requirement.confirm()再以agent.continue_run(run_id..., requirements...)恢复运行Step 4 — 解析审批记录调用db.update_approval(id, expected_statuspending, statusapproved, resolved_bytest_user, resolved_atint(time.time()))返回非空即成功Step 5 — 收敛校验db.get_pending_approval_count()应回到 0。4.2 异步版本approval_async.pyapproval_async.py 与基础版逻辑完全一致仅将同步 API 换成异步await agent.arun(...)、await agent.acontinue_run(...)整个流程放进asyncio.run(main())。若你的 Agent 运行在 FastAPI/异步任务框架中这套链路可以直接照搬。4.3 需要用户输入的审批approval_user_input.py有些工具除了批不批准之外还缺少关键参数需要人在循环中补全。approval_user_input.py 用了一个转账工具做示范approval tool(requires_user_inputTrue, user_input_fields[recipient]) def send_money(amount: float, recipient: str, note: str) - str: Send money to a recipient. ...Agent 因缺recipient转账对象而暂停。续跑前脚本对 requirement 依次处理if requirement.needs_user_input: requirement.provide_user_input({recipient: Alice}) if requirement.needs_confirmation: requirement.confirm()注意user_input_fields[recipient]声明了哪些参数需要人工提供审批记录查询时用了db.get_approvals(statuspending, approval_typerequired)的过滤写法。这展示了approval与tool(requires_user_inputTrue)组合既能把工具参数留给人类补全又能让整次调用处于审批门控之下。4.4 外部执行的审批approval_external_execution.py真实世界里很多操作如 CI/CD 流水线、工单系统不由 Agent 进程本身完成而是交由外部系统执行。approval_external_execution.py 用生产部署工具做示范approval tool(external_executionTrue) def deploy_to_production(service_name: str, version: str) - str: ...Agent 暂停后外部系统真正执行了部署随后把结果回填给 requirementif requirement.needs_external_execution: requirement.set_external_execution_result(Deployed auth-service v2.1.0)continue_run之后 Agent 拿到外部执行结果继续生成后续回答。这是一个Agent 编排、外部系统执行、审批全程留痕的标准生产模式。4.5 审批完整生命周期与竞态防护approval_list_and_resolve.py这是单测覆盖最完整的用例展示了模拟后端管理 API 审批管理员审批的完整生命周期触发两个暂停分别执行删除用户数据delete_user_data与群发邮件send_bulk_email两个工具都挂了approvalrequires_confirmationTrue得到两个run_id列举与计数db.get_approvals(statuspending)得到 2 条db.get_pending_approval_count()返回 2按 run 过滤与单查db.get_approvals(run_idrun1.run_id)精确过滤db.get_approval(id)单条查询批准第一条update_approval(..., expected_statuspending, statusapproved, resolved_byadminexample.com, resolved_at...)验证竞态防护再次用expected_statuspending对同一条记录执行statusrejected的更新返回None—— 说明update_approval的expected_status参数充当原子性守卫防止并发环境下被二次解析double-resolve 被正确拦截拒绝第二条对剩余记录置rejected分别续跑两个 run已批准的第一个 run 走req.confirm()后正常完成被拒绝的第二个 run 走req.reject(Rejected by admin: too many recipients)Agent 会感知到拒绝删除记录db.delete_approval(id)逐条清理最终db.get_approvals()总数为 0。4.6 团队级审批approval_team.pyFAILapproval_team.py 展示了审批在Team多 Agent 团队层面的传递带approval工具的成员 Agent 被包进Team审批记录挂在team_sessions会话表上注意它与单 Agent 示例的agent_sessions不同db SqliteDb(db_fileDB_FILE, session_tableteam_sessions, approvals_tableapprovals) deploy_agent Agent(nameDeploy Agent, roleHandles deployments to production, ...) team Team(nameDevOps Team, members[deploy_agent], model..., dbdb)该用例还打印了审批记录中的source_type与source_name字段用于追溯审批来源来自哪个 Agent。团队场景的续跑调用方式是team.continue_run(response)——直接传入响应对象而非手拼run_id/requirements。测试结果该用例在 TEST_LOG 中被记录为FAIL断言response.is_paused失败实际得到RunStatus.completed即本轮运行未暂停。详见第五节分析。4.7 审批结果的后置钩子approval_post_hook.pyapproval_post_hook.py 面向审计/可观测性场景审批由外部管理员通过 DB/API 路径解析db.update_approval随后用不带 requirements 的continue_run(run_id...)恢复运行触发内部的check_and_apply_approval_resolution逻辑从 DB 读取已解析记录并挂到run_output.metadata[approval]上。示例里注册了 post-hookdef audit_resolved_approval(run_output: RunOutput) - None: approval_record run_output.metadata.get(approval) if approval_record is None: return print(f approval_id: {approval_record[id]}) print(f status: {approval_record[status]}) print(f resolved_by: {approval_record.get(resolved_by)}) print(f resolved_at: {approval_record.get(resolved_at)}) agent Agent(..., post_hooks[audit_resolved_approval], dbdb)脚本最后断言run.metadata[approval][resolved_by] adminexample.com验证元数据确实带上了解析人resolved_by与解析时间resolved_at。这正是审计钩子最需要的由谁在何时批准信息而不仅仅是工具是否被放行。TEST_LOG 对该用例的记录也确认Audit hook fires with approval_id, status, resolved_by, resolved_at populated from the DB record。4.8 审计式审批确认 / 用户输入 / 外部执行audit_*与阻塞式先记录后执行相反审计式审批在HITL 交互解析完成之后落库一条终态记录。以最完整的 audit_approval_confirmation.py 为例approval(typeaudit) tool(requires_confirmationTrue) def delete_user_data(user_id: str) - str: ...批准路径运行 → 暂停 →confirm()→ 续跑完成后db.get_approvals(approval_typeaudit)能查到statusapproved、approval_typeaudit的记录拒绝路径再次运行 →reject(Rejected by admin: not authorized)→ 续跑后能查到statusrejected的审计记录最终校验审计表中共 2 条记录approve 与 reject 各一approval_type均为audit。其余审计变体复用同一模式audit_approval_user_input.pyrequires_user_input、audit_approval_external.pyexternal_execution注意它在解析前db.get_approvals()总数应为 0证明审计记录确实事后才产生、audit_approval_async.py异步链路。4.9 两类审批混用总览audit_approval_overview.pyFAILaudit_approval_overview.py 在同一个 Agent中同时注册了一个approval工具critical_action和一个approval(typeaudit)工具sensitive_action用于演示两类记录可以在数据库中按approval_type清晰分池db.get_approvals(approval_typerequired)与db.get_approvals(approval_typeaudit)各返回且仅返回 1 条。从设计意图看它是对前文所有概念的整合验证但该用例在测试日志中同样被记录为FAIL断言暂停失败。五、三个 FAIL 用例解读何时该重新跑一遍TEST_LOG.md 中三个失败用例approval_team.py、audit_approval_external.py、audit_approval_overview.py失败信息完全一致AssertionError: Expected paused, got RunStatus.completed日志将其定性为Code bug。从断言本身可以还原的语义是脚本期望run()返回后 Agent 处于暂停状态is_pausedTrue因为带审批的工具应在执行前拦截而实际返回的RunStatus.completed表示该轮运行直接完成了——即审批拦截点没有按预期触发暂停。值得注意的观察点以下为基于代码与日志的合理推断而非定论三个失败用例恰好横跨团队审批、审计 外部执行、审计 阻塞混合三类相对复杂的组合而它们各自的基础版approval_basic、audit_approval_confirmation、approval_external_execution等均为 PASS说明单 Agent 单 HITL 标志的路径是稳定的失败可能源于示例对复杂组合的断言过强也可能与模型在该轮次未选择触发带审批工具模型没有调用工具运行自然直接完成有关还可能与框架在特定组合下的拦截时序有关。日志并未给出堆栈与根因因此不能据此断定框架存在缺陷。给读者的建议由于日志日期为 2026-02-13而代码库仍在演进最稳妥的做法是在本地按原环境复跑这三个用例确认失败是否仍然复现再决定是修正示例断言还是排查组合场景的拦截逻辑。复跑命令与下节一致。六、运行方式与验证环境单 Agent 示例的推荐运行方式来自 README.md 的 Running 一节.venvs/demo/bin/python cookbook/02_agents/11_approvals/approval_basic.py其余文件把末尾文件名替换即可例如.venvs/demo/bin/python cookbook/02_agents/11_approvals/audit_approval_confirmation.py .venvs/demo/bin/python cookbook/02_agents/11_approvals/approval_team.py运行前提从代码 import 可确认安装 agno 库提供agno.agent.Agent、agno.approval.approval、agno.db.sqlite.SqliteDb、agno.tools.tool、agno.team.team.Team等配置OPENAI_API_KEY环境变量因为示例统一使用OpenAIResponses(idgpt-5-mini)示例需要联网访问外部 API如 Hacker News 接口每个脚本启动时都会删除上次的tmp/*.db文件并重建SqliteDb因此可以反复执行日志记录的测试环境为.venvs/demo/bin/pythonpgvector 处于运行状态读者可自行选择等价的 Python 虚拟环境。由于示例内部遍布断言PASS 用例还会打印--- All checks passed! ---运行后可通过退出状态与终端输出快速判断当前版本下行为是否符合预期。七、可复用的关键 API 速查汇总以上示例中出现、且经源码验证存在的 API便于你直接嵌入自己的 Agent/管理后台关注点API / 属性作用装饰器approval/approval(typeaudit)标记工具需要审批详见 decorator.py工具层tool(requires_confirmationTrue / requires_user_inputTrue / external_executionTrue)声明 HITL 交互类型可多选组合运行结果run_response.is_paused、run_response.active_requirements、run_response.requirements、run_response.metadata判断暂停、取活跃待办项、续跑回传、读审批元数据待办操作requirement.confirm()/requirement.reject(msg)/requirement.provide_user_input({...})/requirement.set_external_execution_result(str)批准 / 拒绝 / 补参 / 回填外部结果续跑agent.continue_run(run_id, requirements)异步acontinue_run团队场景team.continue_run(response)携带人工处理结果恢复执行数据库SqliteDb(db_file, session_table, approvals_table)指定会话表与审批表默认agno_approvals语义由approvals_table控制审批 CRUDget_approvals(status, approval_type, run_id)、get_approval(id)、update_approval(id, expected_status, status, resolved_by, resolved_at)、delete_approval(id)、get_pending_approval_count()列举/过滤/单查/原子解析/删除/计数expected_status提供竞态保护审计钩子Agent(post_hooks[...])内读取run_output.metadata[approval]拿到approval_id/status/resolved_by/resolved_at做事后审计结语cookbook/02_agents/11_approvals是理解 agno HITL 审批能力最完整的入口从approval装饰器的源码语义decorator.py到阻塞式与审计式两类记录在数据库中的完整生命周期再到确认、用户输入、外部执行、团队与钩子等真实场景的组合。配套的 TEST_LOG.md 则如实标注了哪些链路稳定通过9/12哪些组合在当前版本下需要复跑验证3 个 FAIL 用例均报Expected paused, got RunStatus.completed。如果你正打算为删除数据、转账、生产部署、群发邮件这类高风险工具加上一层可审计的人工闸门这份示例集与日志将是你从概念到落地的最短路径。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考