企业官网建设流程全解析

HITL 是 Human-in-the-Loop 的缩写，中文常译为 “人机回环” 或 “人在环中”。

简单来说，它指的是在AI（人工智能）系统自动化的决策或执行流程中，有意识地引入人工干预环节。

Agno 的 Human-in-the-Loop (HITL) 并非一个附加功能，而是作为一等设计原语（first-class design primitive） 被内置在框架中。它的核心价值在于，让开发者能在不牺牲 Agent 自主性的前提下，为关键操作建立清晰的人工监督边界

为什么需要 HITL？

当 AI 智能体从回答问题转向执行具体操作时，HITL 主要解决了三个问题：

1. 不可逆操作：如发送邮件、删除记录，需要人工确认来防止错误。

2. 缺失信息：Agent 知道该做什么，但缺少关键细节（如目标环境、预算），需要暂停并向人类提问。

3. 审计追踪：为敏感操作建立问责机制，保存完整的审批记录

三大核心机制：三层架构与三种交互模式

Agno 的 HITL 设计主要围绕两个维度展开：在哪介入与如何介入。

在哪介入：三层架构 (3 Layers)

你可以根据操作的风险级别，在框架的三个不同层面设置人工检查点。

如何介入：三种交互模式 (3 Modes)

在 Agent 暂停时，Agno 支持以下三种获取人工输入的方式。

• 用户确认 (User Confirmation)：“批准/拒绝”模型。Agent 提出操作方案，由人做出最终决定。

  • 触发方式：使用@tool(requires_confirmation=True)装饰器。

• 用户输入 (User Input)：“填空”模型。Agent 因缺少执行任务所需的关键信息而暂停，并向用户提问。

  • 触发方式：使用@tool(requires_user_input=True)装饰器或UserFeedbackTools。

• 外部工具执行 (External Tool Execution)：“交接”模型。Agent 将需要人类在外部系统完成的步骤委派出去，然后等待结果返回。

  • 触发方式：使用@tool(external_execution=True)装饰器

代码实现

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_confirmation=True) def refund(user_id: str): """ Refund a user's payment """ print(f"Refunding payment for user {user_id}") agent = Agent( name="RefundBot", model=DeepSeek(id="deepseek-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_response=run_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_response=run_response, run_id=run_response.run_id, requirements=run_response.requirements) print("最终回复:", final_response.content) # 3. 如果 Agent 直接完成（没有暂停），直接打印回复 else: print("回复:", run_response.content)

“填空”模型。Agent 因缺少执行任务所需的关键信息而暂停，并向用户提问。使用@tool(requires_user_input=True)装饰器

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_input=True, 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 f"Sending email to {to_address} with subject {subject} and body {body}" agent = Agent( name="RefundBot", model=DeepSeek(id="deepseek-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_response=run_response, run_id=run_response.run_id, requirements=run_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(id="deepseek-chat") db = SqliteDb(db_file="session_store1.db") search_agent = Agent( name="Researcher", model=deepseek, tools=[BaiduSearchTools()], description="负责研究主题，收集相关信息", instructions=[ "使用搜索工具查找最新信息", "总结3-5个关键要点", "提供信息来源" ], markdown=True ) writer_agent = Agent( name="Writer", model=deepseek, description="负责撰写文章", instructions=[ "基于研究结果撰写500字文章", "使用清晰的结构（引言、正文、结论）", "使用Markdown格式" ], markdown=True ) simple_workflow = Workflow( name="Simple Article Generator", description="简单的文章生成工作流", db=db, steps=[ Step(agent=search_agent, name="search"), Step( agent=writer_agent, name="write", human_review=HumanReview( requires_confirmation=True, confirmation_message="即将生成文章，是否继续？", on_reject=OnReject.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_response=result) 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_execution=True) def execute_shell_command(command: str) -> str: """ 执行Shell命令（外部执行） 注意：此函数不会被Agent直接调用！ Agent会暂停并等待你在外部执行。 """ # 这段代码只有在外部执行时才会运行 return subprocess.check_output(command, shell=True).decode("utf-8") agent = Agent( model=DeepSeek(id="deepseek-chat"), tools=[execute_shell_command], description="你是一个系统管理助手，可以执行Shell命令。", markdown=True, ) 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_response=run_response, run_id=run_response.run_id, requirements=run_response.requirements ) print(f"\n🤖 Agent响应：\n{response.content}")