06.3 — 护栏与人类监督
定位: Agent 安全的"最后防线"——理解如何通过预设规则（护栏）和人类介入（HITL）来确保 Agent 行为安全可控 面试高频度: ⭐⭐⭐ 考查方式: 设计题（设计 Agent 护栏系统）、策略题（HITL vs Auto 的权衡）
一、这是什么？为什么需要它？
是什么
护栏（Guardrails） 是 Agent 行为的预设约束规则。人类监督（Human-in-the-Loop, HITL） 是在关键节点引入人类判断的机制。
为什么需要护栏和监督？
如果没有护栏：
Agent 可能无限循环调用 API
Agent 可能生成危险内容
Agent 可能做出不可逆的操作（删除数据、发送邮件）
出了问题时不知道谁该负责
有了护栏和监督：
关键操作需要人类确认
危险行为会被自动拦截
行为边界清晰可控
出了问题可以回滚核心洞察：护栏是"刹车"，HITL 是"方向盘"——刹车保证不出事，方向盘保证方向正确。
二、原理拆解
2.1 护栏的三层模型
┌──────────────────────────────────────────────────────────┐
│ 护栏三层架构 │
│ │
│ 输入层护栏 │
│ ┌──────────────────────────────────────────┐ │
│ │ 内容过滤 指令验证 上下文检查 │ │
│ │ 用户输入 → 是否包含危险内容/指令？ │ │
│ └──────────────────┬───────────────────────┘ │
│ │ │
│ 决策层护栏 │ │
│ ┌──────────────────▼───────────────────────┐ │
│ │ 操作验证 权限检查 风险评估 │ │
│ │ Agent 决策 → 是否在允许范围内？ │ │
│ └──────────────────┬───────────────────────┘ │
│ │ │
│ 输出层护栏 │ │
│ ┌──────────────────▼───────────────────────┐ │
│ │ 内容审核 格式验证 敏感信息检测 │ │
│ │ Agent 输出 → 是否包含不当内容？ │ │
│ └──────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────┘2.2 Guardrails 实现
python
class Guardrail:
"""单个护栏规则"""
def __init__(self, name: str, check_fn, action: str = "block"):
self.name = name
self.check_fn = check_fn # 检查函数，返回 (pass: bool, reason: str)
self.action = action # block / warn / log
def check(self, context: dict) -> tuple:
return self.check_fn(context)
class GuardrailSystem:
"""护栏系统"""
def __init__(self):
self.guardrails = []
def add_guardrail(self, guardrail: Guardrail):
self.guardrails.append(guardrail)
def check_all(self, context: dict) -> list:
"""执行所有护栏检查"""
violations = []
for guardrail in self.guardrails:
passed, reason = guardrail.check(context)
if not passed:
violations.append({
"guardrail": guardrail.name,
"reason": reason,
"action": guardrail.action
})
if guardrail.action == "block":
break # 阻断：停止后续检查
return violations
# 定义常用护栏
guardrails = GuardrailSystem()
# 1. 速率限制护栏
guardrails.add_guardrail(Guardrail(
"rate_limit",
lambda ctx: (ctx.get("call_count", 0) < 50, "调用频率过高"),
"block"
))
# 2. 敏感操作护栏
guardrails.add_guardrail(Guardrail(
"sensitive_operation",
lambda ctx: (
"delete" not in ctx.get("tool", "") or ctx.get("confirmed", False),
"敏感操作需要确认"
),
"block"
))
# 3. 内容安全护栏
guardrails.add_guardrail(Guardrail(
"content_safety",
lambda ctx: (
not any(word in str(ctx.get("content", ""))
for word in ["暴力", "色情", "歧视"]),
"内容包含敏感词"
),
"warn" # 仅警告
))2.3 三种人类监督模式
| 模式 | 做法 | 延迟 | 安全性 | 适用场景 |
|---|---|---|---|---|
| HITL（人在回路） | 每步都需人类确认 | 高 | 最高 | 高风险场景（金融交易） |
| HOTL（人在上面） | 人类监控，异常时介入 | 中 | 中 | 中等风险（代码生成） |
| HOTL（人在外面） | 完全自动，人类事后审查 | 低 | 中 | 低风险（信息查询） |
python
class HumanInTheLoop:
"""人类监督系统"""
def __init__(self, mode: str = "hotl"):
self.mode = mode # "hitl", "hotl", "outside"
self.pending_approvals = []
def request_approval(self, action: dict) -> dict:
"""请求人类批准"""
if self.mode == "hitl":
# 阻塞等待人类批准
return self._wait_for_human(action)
elif self.mode == "hotl":
# 先执行，同时通知人类
print(f"[HOTL] Agent 正在执行: {action['tool']}({action['args']})")
print("[HOTL] 人类可随时输入 'stop' 终止")
return {"status": "executing", "can_interrupt": True}
elif self.mode == "outside":
# 记录日志，事后审查
self.pending_approvals.append(action)
return {"status": "auto_executed"}
def _wait_for_human(self, action: dict) -> dict:
"""等待人类决策"""
print(f"\n需要人类确认: {action['tool']}({action['args']})")
print("描述:", action.get("description", ""))
# 现实中这里会发送通知等待用户确认
# 这里用一个简单的模拟
decision = input("允许执行？(y/n): ")
if decision.lower() == 'y':
return {"status": "approved"}
else:
return {"status": "rejected", "reason": "人类拒绝"}
# 使用示例
hitl = HumanInTheLoop(mode="hitl")
# Agent 想执行一个危险操作
action = {
"tool": "delete_file",
"args": {"path": "/data/temp.txt"},
"description": "删除临时文件"
}
result = hitl.request_approval(action)
print(f"操作结果: {result}")2.4 审计与可追溯性
python
class AuditLogger:
"""审计日志系统"""
def __init__(self):
self.entries = []
def log(self, event_type: str, agent: str,
action: str, result: str, approved_by: str = None):
self.entries.append({
"timestamp": now(),
"event_type": event_type, # tool_call / approval / rejection / error
"agent": agent,
"action": action,
"result": result,
"approved_by": approved_by,
})
def get_agent_report(self, agent_id: str) -> list:
"""获取某个 Agent 的所有操作记录"""
return [e for e in self.entries if e["agent"] == agent_id]
def export_for_review(self) -> str:
"""导出供人类审查"""
report = "Agent 行为审计报告\n" + "=" * 40 + "\n"
for entry in self.entries:
report += f"[{entry['timestamp']}] {entry['agent']}: {entry['action']}\n"
report += f" 结果: {entry['result'][:50]}...\n"
if entry["approved_by"]:
report += f" 批准人: {entry['approved_by']}\n"
return report三、图解全景
┌──────────────────────────────────────────────────────────────┐
│ 护栏与监督的完整工作流 │
│ │
│ Agent 执行 │
│ │ │
│ ▼ │
│ 护栏检查（自动） │
│ ┌──────────────────────────────────┐ │
│ │ 所有护栏规则逐项检查 │ │
│ │ 通过 → 继续 │ │
│ │ 阻断 → 返回错误信息 │ │
│ │ 警告 → 输出警告但允许继续 │ │
│ └──────────────────┬───────────────┘ │
│ │ │
│ ▼ │
│ HITL 决策（可选） │
│ ┌──────────────────────────────────┐ │
│ │ 如果是敏感操作： │ │
│ │ 等待人类确认 → 批准 → 继续 │ │
│ │ → 拒绝 → 返回拒绝信息 │ │
│ │ 如果不是敏感操作：自动继续 │ │
│ └──────────────────┬───────────────┘ │
│ │ │
│ ▼ │
│ 执行工具 │
│ │ │
│ ▼ │
│ 审计日志记录 │
│ ┌──────────────────────────────────┐ │
│ │ 记录：谁、什么时间、调用了什么工具、 │ │
│ │ 参数是什么、结果如何、谁批准的 │ │
│ └──────────────────────────────────┘ │
│ │
└──────────────────────────────────────────────────────────────┘四、实战验证
python
# 完整的安全执行链路
class SafeAgentExecutor:
"""安全执行器"""
def __init__(self):
self.guardrails = GuardrailSystem()
self.hitl = HumanInTheLoop(mode="hotl")
self.audit = AuditLogger()
# 配置护栏
self._setup_guardrails()
def _setup_guardrails(self):
self.guardrails.add_guardrail(Guardrail(
"max_steps",
lambda ctx: (ctx.get("step", 0) < 20, "超过最大执行步数"),
"block"
))
self.guardrails.add_guardrail(Guardrail(
"max_cost",
lambda ctx: (ctx.get("total_cost", 0) < 1.0, "超过预算"),
"block"
))
def execute(self, agent_id: str, action: dict) -> dict:
context = {
"tool": action.get("tool"),
"args": action.get("args", {}),
"step": action.get("step", 0),
"total_cost": action.get("total_cost", 0)
}
# 护栏检查
violations = self.guardrails.check_all(context)
if violations:
for v in violations:
self.audit.log("guardrail_blocked", agent_id,
str(action), v["reason"])
return {"blocked": True, "violations": violations}
# HITL 确认
if action.get("sensitive", False):
approval = self.hitl.request_approval(action)
if approval["status"] != "approved":
self.audit.log("rejected", agent_id, str(action), "人类拒绝")
return {"blocked": True, "reason": "人类未批准"}
# 执行
try:
result = execute_tool(action["tool"], action["args"])
self.audit.log("success", agent_id, str(action), str(result))
return {"success": True, "result": result}
except Exception as e:
self.audit.log("error", agent_id, str(action), str(e))
return {"error": str(e)}五、面试视角
| 追问 | 答案要点 |
|---|---|
| 护栏的三层架构？ | 输入层（检查输入内容）、决策层（检查 Agent 决策）、输出层（检查输出内容）。每层独立部署，某层失效不影响其他层 |
| HITL 的三种模式及选择？ | HITL（每步确认，高安全高延迟）、HOTL（监控可中断，中等）、Outside（全自动，事后审查）。选择看：操作风险等级、延迟容忍度、自动化目标 |
| 如何确定什么操作需要人类确认？ | 三原则：不可逆操作（删除、修改）、高影响操作（发送消息、转账）、新操作（从未执行过的操作）。可配置规则引擎 |
| 护栏的误报（阻断正常操作）怎么处理？ | 分层处理：阻断前给出详细原因、提供"我确认，继续执行"的申诉通道、记录误报用于护栏优化、护栏可配置严格程度 |
📚 相关链接
- **Prompt注入与防护** — 输入层安全
- **工具滥用防御** — 决策层安全
- **感知-行动循环** — Agent 循环中的安全节点
- ← 返回 **Agent安全索引**