HITL 是 Human-in-the-Loop 的缩写中文常译为 “人机回环” 或 “人在环中”。简单来说它指的是在AI人工智能系统自动化的决策或执行流程中有意识地引入人工干预环节。Agno 的 Human-in-the-Loop (HITL) 并非一个附加功能而是作为一等设计原语first-class design primitive 被内置在框架中。它的核心价值在于让开发者能在不牺牲 Agent 自主性的前提下为关键操作建立清晰的人工监督边界为什么需要 HITL当 AI 智能体从回答问题转向执行具体操作时HITL 主要解决了三个问题不可逆操作如发送邮件、删除记录需要人工确认来防止错误。缺失信息Agent 知道该做什么但缺少关键细节如目标环境、预算需要暂停并向人类提问。审计追踪为敏感操作建立问责机制保存完整的审批记录三大核心机制三层架构与三种交互模式Agno 的 HITL 设计主要围绕两个维度展开在哪介入与如何介入。在哪介入三层架构 (3 Layers)你可以根据操作的风险级别在框架的三个不同层面设置人工检查点。层级描述适用场景1. 工具级 (Tool-level)最细粒度的控制。在执行特定函数前暂停让人工审核、批准或拒绝该操作。高风险的单次操作如发送邮件、执行交易、删除数据。2. 工作流级 (Workflow-level)更广泛的流程控制。在步骤Step 前后暂停评估整体进展。多阶段的复杂流程需要人工在关键节点如审核草稿做决策。3. 审批级 (Approval-level)最正式的治理机制。引入一个独立的正式签批层通常用于合规审查。需要管理员或合规部门审批的高风险操作如大额付款如何介入三种交互模式 (3 Modes)在 Agent 暂停时Agno 支持以下三种获取人工输入的方式。用户确认 (User Confirmation)“批准/拒绝”模型。Agent 提出操作方案由人做出最终决定。触发方式使用tool(requires_confirmationTrue)装饰器。用户输入 (User Input)“填空”模型。Agent 因缺少执行任务所需的关键信息而暂停并向用户提问。触发方式使用tool(requires_user_inputTrue)装饰器或UserFeedbackTools。外部工具执行 (External Tool Execution)“交接”模型。Agent 将需要人类在外部系统完成的步骤委派出去然后等待结果返回。触发方式使用tool(external_executionTrue)装饰器代码实现1. 工具级 (Tool-level) HITL最直接的方式是使用tool装饰器的requires_confirmation参数from dotenv import load_dotenv from agno.models.deepseek import DeepSeek load_dotenv() from agno.agent import Agent from agno.approval.decorator import approval from agno.tools import tool tool(requires_confirmationTrue) def refund(user_id: str): Refund a users payment print(fRefunding payment for user {user_id}) agent Agent( nameRefundBot, modelDeepSeek(iddeepseek-chat), tools[refund], instructions你是一个智能助手可以调用工具来执行操作。, ) while True: text input(用户:) if text exit: break run_response agent.run(text) if run_response.is_paused: # for tool_call in run_response.tools_requiring_confirmation: # user_input input(f确认执行工具 {tool_call.tool_name} 吗(y/n): ).strip().lower() # if user_input y: # tool_call.confirmed True # else: # tool_call.confirmed False # tool_call.confirmation_note 用户取消了此操作 # # *** 关键步骤恢复执行 *** # final_response agent.continue_run(run_responserun_response) # print(最终回复:, final_response.content) for requirement in run_response.active_requirements: if requirement.needs_confirmation: # 显示工具调用信息 tool_exec requirement.tool_execution print(f\n 工具调用详情) print(f 工具名称: {tool_exec.tool_name}) print(f 参数: {tool_exec.tool_args}) # 请求用户确认 user_input input(\n是否批准此操作(y/n): ).strip().lower() if user_input y: requirement.confirm() else: requirement.reject() requirement.confirmation_note 删除操作过于危险用户取消了此操作 # *** 关键步骤恢复执行 *** final_response agent.continue_run(run_responserun_response, run_idrun_response.run_id, requirementsrun_response.requirements) print(最终回复:, final_response.content) # 3. 如果 Agent 直接完成没有暂停直接打印回复 else: print(回复:, run_response.content)“填空”模型。Agent 因缺少执行任务所需的关键信息而暂停并向用户提问。使用tool(requires_user_inputTrue)装饰器from typing import List from dotenv import load_dotenv from agno.models.deepseek import DeepSeek load_dotenv() from agno.agent import Agent from agno.approval.decorator import approval from agno.tools import tool from agno.tools.function import UserInputField tool(requires_user_inputTrue, user_input_fields[to_address]) def send_mail(subject: str, body: str, to_address: str): Send an email Args: subject (str): The subject of the email body (str): The body of the email to_address (str): The email address to send the email to Returns: str: A message indicating that the email has been sent return fSending email to {to_address} with subject {subject} and body {body} agent Agent( nameRefundBot, modelDeepSeek(iddeepseek-chat), tools[send_mail], instructions你是一个智能助手可以调用工具来执行操作。, ) run_response agent.run(发送一封邮件主题是会议通知内容是明天下午3点开会) # 处理用户输入需求 for requirement in run_response.active_requirements: if requirement.needs_user_input: input_schema: List[UserInputField] requirement.user_input_schema print( 需要您提供以下信息\n) for field in input_schema: print(f字段: {field.name}) print(f类型: {field.field_type}) print(f描述: {field.description}) if field.value is None: # 需要用户输入 user_value input(f请输入 {field.name}: ).strip() field.value user_value else: # Agent已经填充 print(f值: {field.value} (已由Agent填充)) # 继续执行 response agent.continue_run( run_responserun_response, run_idrun_response.run_id, requirementsrun_response.requirements ) print(f✅ {response.content})2. 工作流级 (Workflow-level) HITL对于工作流中的一个步骤可以通过HumanReview配置类来实现from agno.agent import Agent from agno.models.deepseek import DeepSeek from agno.workflow import HumanReview, OnReject, Step, Workflow from dotenv import load_dotenv from agno.tools.baidusearch import BaiduSearchTools from agno.db.sqlite import SqliteDb load_dotenv() deepseek DeepSeek(iddeepseek-chat) db SqliteDb(db_filesession_store1.db) search_agent Agent( nameResearcher, modeldeepseek, tools[BaiduSearchTools()], description负责研究主题收集相关信息, instructions[ 使用搜索工具查找最新信息, 总结3-5个关键要点, 提供信息来源 ], markdownTrue ) writer_agent Agent( nameWriter, modeldeepseek, description负责撰写文章, instructions[ 基于研究结果撰写500字文章, 使用清晰的结构引言、正文、结论, 使用Markdown格式 ], markdownTrue ) simple_workflow Workflow( nameSimple Article Generator, description简单的文章生成工作流, dbdb, steps[ Step(agentsearch_agent, namesearch), Step( agentwriter_agent, namewrite, human_reviewHumanReview( requires_confirmationTrue, confirmation_message即将生成文章是否继续, on_rejectOnReject.skip, ) ) ] ) if __name__ __main__: topic agno框架在agent上的运用 result simple_workflow.run(topic) while result.is_paused: print(⏸️ 工作流暂停需要人工确认...) for requirement in result.active_step_requirements: if requirement.needs_confirmation: confirmation_msg getattr(requirement, confirmation_message, 未提供确认信息) print(f 确认信息: {confirmation_msg}) user_input input(是否批准(y/n): ).strip().lower() if user_input y: requirement.confirm() print(✅ 已批准继续执行...) else: requirement.reject() print(❌ 已拒绝跳过该步骤...) # 只需传入 run_response 即可恢复 result simple_workflow.continue_run(run_responseresult) print(\n✅ 工作流完成) print(result.content)3.外部工具执行 (External Tool Execution)“交接”模型。Agent 将需要人类在外部系统完成的步骤委派出去然后等待结果返回。import subprocess from agno.models.deepseek import DeepSeek from agno.agent import Agent from agno.models.dashscope import DashScope from agno.tools import tool from dotenv import load_dotenv load_dotenv() tool(external_executionTrue) def execute_shell_command(command: str) - str: 执行Shell命令外部执行 注意此函数不会被Agent直接调用 Agent会暂停并等待你在外部执行。 # 这段代码只有在外部执行时才会运行 return subprocess.check_output(command, shellTrue).decode(utf-8) agent Agent( modelDeepSeek(iddeepseek-chat), tools[execute_shell_command], description你是一个系统管理助手可以执行Shell命令。, markdownTrue, ) run_response agent.run(列出当前目录的文件) # 处理外部执行需求 for requirement in run_response.active_requirements: if requirement.tool_execution.tool_name execute_shell_command.name: tool_exec requirement.tool_execution print(f Agent请求执行工具) print(f 工具: {tool_exec.tool_name}) print(f 参数: {tool_exec.tool_args}) print() # 安全检查 command tool_exec.tool_args[command] print(f⚠️ 准备执行命令: {command}) # 白名单检查 allowed_commands [ls, pwd, date, whoami] if not any(command.startswith(cmd) for cmd in allowed_commands): print(❌ 命令不在白名单中拒绝执行) requirement.external_execution_result 错误命令不被允许 else: # 在外部执行 print(✅ 命令安全开始执行...) try: result execute_shell_command.entrypoint(**tool_exec.tool_args) requirement.set_external_execution_result(result) print(f✅ 执行成功) except Exception as e: requirement.set_external_execution_result(f执行失败: {str(e)}) print(f❌ 执行失败: {e}) # 继续执行 response agent.continue_run( run_responserun_response, run_idrun_response.run_id, requirementsrun_response.requirements ) print(f\n Agent响应\n{response.content})