第 16 章 · 人机协同与 Guardrails
本章目标：给 Agent 装上"刹车"——用工具确认机制实现人机协同（HITL），用 pre/post hooks 与内置 Guardrail 拦截敏感输入和危险输出，并设计合理的失败降级策略。
16.1 为什么 Agent 需要"刹车"
自主 Agent 的风险来自两端：入口（用户可能注入恶意指令或提交隐私数据）和出口（模型可能调用删库、转账这类不可逆操作）。工程上的对策是三层防线：
- 工具确认（Human-in-the-Loop）：危险操作执行前暂停，等人批准；
- Guardrails（pre_hooks）：输入进模型前做校验/脱敏/拦截；
- 输出检查（post_hooks）：回复返回用户前做合规审查。
版本说明
Guardrails 与 hooks 体系自 Agno v2.1.0 引入；工具确认（requires_confirmation）同样是 2.x 的 HITL 标准做法。本教程基于 Agno 2.9。
16.2 工具确认：让危险操作等一个人
给自定义工具加 @tool(requires_confirmation=True)，Agent 执行到该工具时会暂停（is_paused=True），把待执行的工具名与参数交给你审批，批准后用 continue_run() 续跑：
# hitl_confirm.py
import os
from agno.agent import Agent
from agno.db.sqlite import SqliteDb # 续跑需要数据库保存运行状态
from agno.models.openai import OpenAIChat
from agno.tools import tool
@tool(requires_confirmation=True)
def refund_order(order_id: str, amount: float) -> str:
"""为指定订单执行退款，属于敏感资金操作。"""
return f"订单 {order_id} 已退款 {amount} 元"
agent = Agent(
model=OpenAIChat(
id="deepseek-chat",
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1",
),
tools=[refund_order],
db=SqliteDb(db_file="tmp/hitl.db"), # 必须持久化 run 状态才能续跑
)
run_response = agent.run("帮我把订单 A1001 退款 99 元")
# 审批环节：遍历待确认需求
for requirement in run_response.active_requirements:
if requirement.needs_confirmation:
print(f"待确认: {requirement.tool_execution.tool_name}"
f"{requirement.tool_execution.tool_args}")
if input("批准吗? (y/n): ").lower() == "y":
requirement.confirm()
else:
requirement.reject()
# 带着审批结果继续执行
response = agent.continue_run(
run_id=run_response.run_id,
requirements=run_response.requirements,
)
print(response.content)对 Toolkit 内置工具，可以用 requires_confirmation_tools=["工具名"] 只保护指定函数，其余照常自动执行。
16.3 内置 Guardrail：PII 与提示注入防护
Guardrail 以 pre_hook 形式挂载，在输入到达模型前执行。Agno 内置了 PII 检测、提示注入防御、OpenAI 内容审核三种：
from agno.agent import Agent
from agno.guardrails import PIIDetectionGuardrail, PromptInjectionGuardrail
support_agent = Agent(
model=model,
pre_hooks=[
PIIDetectionGuardrail(), # 拦截身份证号/手机号等个人隐私
PromptInjectionGuardrail(), # 识别"忽略之前所有指令"类注入攻击
],
markdown=True,
)
# 含手机号的输入会被直接拦截并抛出 InputCheckError16.4 自定义 Guardrail
继承 BaseGuardrail 并实现 check() / async_check()（分别服务同步 .run() 和异步 .arun()），发现违规时抛出 InputCheckError：
import re
from agno.exceptions import CheckTrigger, InputCheckError
from agno.guardrails import BaseGuardrail
from agno.run.agent import RunInput
class SensitiveWordGuardrail(BaseGuardrail):
"""拦截包含竞品名的对外咨询输入。"""
def check(self, run_input: RunInput) -> None:
if isinstance(run_input.input_content, str):
if re.search(r"某友|某蝶", run_input.input_content):
raise InputCheckError(
"输入包含不允许讨论的内容。",
check_trigger=CheckTrigger.INPUT_NOT_ALLOWED,
)
async def async_check(self, run_input: RunInput) -> None:
self.check(run_input) # 规则简单时异步版可直接复用同步逻辑
# 用法同内置 guardrail：pre_hooks=[SensitiveWordGuardrail()]输出侧审查则写 post_hook：它拿到生成结果但尚未返回用户，可做敏感词扫描、合规改写或打标。
from datetime import datetime
def audit_log_hook(run_input, run_output):
"""post_hook：把每次输入与输出落盘，供合规审计。"""
with open("tmp/audit.log", "a", encoding="utf-8") as f:
f.write(f"[{datetime.now():%F %T}] IN: {run_input.input_content}\n")
f.write(f"[{datetime.now():%F %T}] OUT: {run_output.content}\n")
compliance_agent = Agent(
model=model,
pre_hooks=[SensitiveWordGuardrail()], # 事前拦截
post_hooks=[audit_log_hook], # 事后审计（也可以做改写/打标）
)16.5 失败降级策略
拦截不是终点，生产系统要回答"拦下之后怎么办"：
- 可修复错误（格式问题）：在 post_hook 里自动改写重试一次；
- 需人工介入：捕获
InputCheckError后转人工队列，并把原因写入审计日志； - 拒绝但不崩溃：对外统一返回礼貌的兜底话术，内部保留完整错误码；
- 灰度上线：新 guardrail 先以"只记日志不拦截"模式观察误杀率，再逐步切换为强制拦截。
本章小结
- 三层防线：工具确认（事中）、pre_hooks 输入防护（事前）、post_hooks 输出审查（事后）；
@tool(requires_confirmation=True)+active_requirements+continue_run()是标准 HITL 流程，且必须配置db以持久化运行状态；- 内置 Guardrail 通过
pre_hooks=[...]挂载；自定义时继承BaseGuardrail实现check()/async_check()并抛出InputCheckError； - 拦截之后要有明确的降级路径：重试、转人工、兜底话术、灰度观察。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. 标记 requires_confirmation=True 的工具被触发时，Agent 会发生什么？
2. 使用工具确认机制时为什么必须给 Agent 配置 db？
3. 自定义 Guardrail 需要继承哪个类？检测到违规时应如何处理？
4. pre_hooks 在什么时机执行？
🛠️ 动手实践
- 写一个"删除文件"工具并加
requires_confirmation=True，完整走一遍暂停→拒绝→续跑流程，观察拒绝后 Agent 如何回应用户。 - 实现一个自定义 Guardrail 拦截超过 500 字的输入，并用一段长文本验证
InputCheckError被正确抛出。 - 为客服 Agent 组合
PIIDetectionGuardrail与一个退款确认工具，模拟"用户提供手机号 + 申请退款"的双重拦截场景。
下一章把视野扩展到外部世界：第 17 章：MCP 协议集成。