Guardrails 输出防护

2026-04-16 AI Agent 编程 4.6k 字 18 分钟阅读
📑 目录
  1. 概述
  2. 输出校验流水线与重试-降级策略的容错设计
  3. 正文
  4. 相关阅读
  5. 参考文档
  6. 完整实战示例：输出质量校验与自动修复流水线
  7. 输出 Guardrails 在内容生产流水线中的位置
  8. 常见问题与调试
  9. 与其他方案对比
  10. 模板方法模式与指数退避重试机制
    1. 模板方法模式构建输出校验流水线
    2. 指数退避重试策略
  11. 生产环境部署与性能优化
    1. 输出质量网关的实践要点
    2. 内容安全合规的关键指标
    3. 流式输出校验的架构考量
    4. 运维团队的协作建议

概述

输出校验流水线与重试-降级策略的容错设计

输出 Guardrails 的目标是确保 Agent 的响应符合质量、安全和格式要求。与输入防护不同，输出防护面临一个独特的挑战：修改成本。如果输入被拦截，用户只需重新输入；但如果输出被拦截，Agent 已经消耗了计算资源（LLM token、工具调用），且用户可能已经在等待响应。

因此，输出 Guardrails 的设计需要在"拦截"和"修复"之间找到平衡。常见的策略包括：

  1. 拦截并重试：如果输出不符合要求，将错误信息反馈给模型，请求重新生成。这适用于格式错误、缺少必要字段等可修复问题。
  2. 拦截并降级：如果重试仍然失败，返回一个预设的安全响应（如"抱歉，我无法回答这个问题"）。这适用于内容安全违规等不可修复问题。
  3. 拦截并人工审核：对于高价值但高风险的内容（如医疗建议、法律意见），将输出放入人工审核队列，审核通过后再展示给用户。
  4. 自动修复：对于简单的格式问题（如多余的引号、缺失的闭合标签），直接在应用层修复，不打扰模型。

从实现机制来看，输出 Guardrails 在 Agent 产生最终输出后触发，它可以访问完整的运行结果（包括消息历史和工具调用记录），这使得它能够做出比输入 Guardrails 更精准的上下文感知判断。

性能影响方面，输出 Guardrails 的延迟加在请求的尾部，直接影响用户感知的总响应时间。与输入 Guardrails 类似，建议采用分层策略：先用快速的规则层处理格式校验，只在必要时启用语义层分析。

Output Guardrails：如何在 Agent 返回结果前校验输出内容。

正文

相关阅读

参考文档

完整实战示例：输出质量校验与自动修复流水线

以下示例展示了如何在生产环境中构建一个支持格式校验、内容审核和自动修复的输出防护系统：

import asyncio
import json
import re
from dataclasses import dataclass
from typing import Any
from agents import Agent, Runner, OutputGuardrail, GuardrailFunctionOutput
from agents import RunContextWrapper


@dataclass
class OutputQualityConfig:
    min_length: int = 10
    max_length: int = 2000
    required_sections: list[str] = None
    banned_phrases: list[str] = None
    json_schema: dict = None


class ProductionOutputGuardrail:
    """生产级输出质量校验器。"""

    DEFAULT_BANNED = [
        "as an ai", "i cannot", "i'm sorry, i can't",
        "i don't have the ability", "my training data",
    ]

    def __init__(self, config: OutputQualityConfig = None):
        self.config = config or OutputQualityConfig()
        self.config.banned_phrases = self.config.banned_phrases or self.DEFAULT_BANNED
        self.repair_log: list[dict] = []

    async def check(self, ctx: RunContextWrapper[Any], agent: Agent, output_text: str) -> GuardrailFunctionOutput:
        violations = []

        # 层1：基础格式校验（<5ms）
        if len(output_text) < self.config.min_length:
            violations.append(f"Too short: {len(output_text)} < {self.config.min_length}")
        if len(output_text) > self.config.max_length:
            violations.append(f"Too long: {len(output_text)} > {self.config.max_length}")

        # 层2：内容审核
        lowered = output_text.lower()
        for phrase in self.config.banned_phrases:
            if phrase in lowered:
                violations.append(f"Banned phrase detected: '{phrase}'")

        # 层3：JSON Schema 校验（如果配置了）
        if self.config.json_schema:
            try:
                parsed = json.loads(output_text)
                # 简化校验：实际应使用 jsonschema 库
                for key in self.config.json_schema.get("required", []):
                    if key not in parsed:
                        violations.append(f"Missing required field: {key}")
            except json.JSONDecodeError:
                if self.config.json_schema.get("strict", False):
                    violations.append("Output is not valid JSON")

        # 层4：自动修复尝试
        repaired = output_text
        if violations and not any(v.startswith("Banned") for v in violations):
            repaired = self._attempt_repair(output_text, violations)
            if repaired != output_text:
                self.repair_log.append({"original": output_text, "repaired": repaired})
                violations = []  # 修复成功，清除违规

        is_allowed = len(violations) == 0
        return GuardrailFunctionOutput(
            output_info={"violations": violations, "repaired": repaired != output_text},
            tripwire_triggered=not is_allowed,
        )

输出 Guardrails 在内容生产流水线中的位置

下图展示了输出 Guardrails 在内容生产流水线中的环节和与其他组件的协作关系：
mermaid
flowchart TD
A[Agent 生成内容] --> B[输出 Guardrails]
B --> C{合规检查}
C -->|通过| D[后处理 脱敏/格式化]
C -->|敏感内容| E[内容重写]
E --> B
C -->|严重违规| F[阻断并告警]
D --> G[返回用户]
F --> H[安全团队通知]
style B fill:#e8d5b5,stroke:#5a4a3a,stroke-width:2px
style F fill:#f4b183,stroke:#5a4a3a


输出防护不仅关乎安全，也关乎品牌声誉。一个生成不当内容的 AI 助手可能导致法律风险、用户流失和公关危机。

## 输出内容的合规性检查矩阵

不同行业对 AI 生成内容有不同的合规要求。构建一个全面的合规检查矩阵是输出 Guardrails 设计的基础。

| 行业 | 禁止内容 | 必须包含 | 检查方式 |
|------|----------|----------|----------|
| 金融 | 投资建议、收益率承诺 | 风险提示 | 关键词+LLM 双重检查 |
| 医疗 | 诊断、处方、用药建议 | 请咨询医生 disclaimer | 医学知识库比对 |
| 教育 | 替考、作弊方法 | 学习引导 | 语义分类模型 |
| 法律 | 具体法律建议 | 请咨询律师 disclaimer | 法律实体识别 |

**内容重写机制**：

当输出 Guardrails 检测到敏感内容时，不是简单拒绝，而是可以触发内容重写流程。重写 Agent 接收原始输出和违规原因，生成符合规范的替代版本。

```python
rewrite_agent = Agent(
    name="ContentRewriter",
    instructions="你是一个内容重写专家。请根据违规原因，将原始内容重写为合规版本，保留核心信息但去除敏感内容。"
)

async def safe_output(agent, input_text: str, max_retries: int = 2):
    for _ in range(max_retries):
        result = await Runner.run(agent, input_text)
        check = await output_guardrail.check(result.final_output)
        if check.passed:
            return result.final_output
        result = await Runner.run(rewrite_agent,
            f"违规原因: {check.reason}\\n原始内容: {result.final_output}")
    return "抱歉，无法生成符合规范的回答。"

重写机制的关键是收敛性保证：必须设置最大重写次数，防止无限循环。如果多次重写后仍无法通过检查，应该返回兜底消息而非继续尝试。

输出内容的后处理与格式化同样重要。即使输出 Guardrails 通过了校验，返回给用户的文本仍可能需要后处理。例如：敏感信息脱敏将手机号、身份证号等替换为掩码形式；Markdown 清理移除 LLM 可能生成的 HTML 标签，防止 XSS；长度截断确保响应不超过前端展示限制。后处理逻辑应与业务规则解耦，通过插件化架构实现。不同业务线可以注册不同的后处理器，而核心框架保持通用。

常见问题与调试

问题一：输出 Guardrails 导致响应为空

如果 Guardrails 过于严格，可能拦截掉所有输出，导致用户看到空白回复。排查方法：

  1. 在 Guardrails 的 output_info 中记录详细的拦截原因，便于定位问题。
  2. 实现分级响应：如果主输出被拦截，返回一个预置的友好提示而非空字符串。
  3. 在开发环境中关闭 Guardrails，确认 Agent 本身能否正常输出，排除 Agent 配置问题。

问题二：重试循环导致延迟激增

如果输出 Guardrails 每次都将错误反馈给模型重试，而模型又持续犯同样的错误，可能形成无限重试。解决方案：

  1. 设置最大重试次数（如 2 次），超过后直接降级到安全响应。
  2. 在反馈信息中提供具体的修正指导（如"JSON 缺少 closing brace，请添加 }"）。
  3. 收集失败案例，用于微调模型或优化提示词。

问题三：输出校验与模型能力不匹配

如果要求模型输出复杂的 JSON Schema，但使用的模型太小（如 nano），可能 consistently 无法生成符合要求的输出。建议：

  1. 为复杂格式校验选择更强的模型。
  2. 简化 Schema，将复杂嵌套结构拆分为多个简单调用。
  3. 在后置处理中使用应用层代码补充模型遗漏的字段，而非强制模型一次生成完美输出。

与其他方案对比

维度Agents SDK Output GuardrailsLlama GuardNeMo Guardrails
校验时机Agent 输出后模型输出后对话回合后
修复能力需自行实现无（仅分类）支持对话引导
集成成本低（SDK 原生）中（需部署模型）中（需配置流程）
覆盖范围自定义逻辑预定义安全类别预定义 + 自定义

Llama Guard 是 Meta 开源的内容安全分类模型，在识别有害内容方面表现优异，但它本身不提供修复或重试机制，需要与 Agents SDK 的输出 Guardrails 结合使用。NVIDIA 的 NeMo Guardrails 则是一个完整的对话管理系统，支持通过 Colang 脚本定义复杂的对话流程和防护策略，适合需要精细控制对话走向的企业场景。Agents SDK 的输出 Guardrails 是最轻量的方案，适合已经拥有成熟应用架构、只需要在 Agent 层增加校验点的团队。

模板方法模式与指数退避重试机制

输出 Guardrails 不仅需要校验内容，还需要在发现问题时尝试修复。这种"校验-修复-重试"的流程天然适合**模板方法模式（Template Method Pattern）**的抽象。模板方法定义了输出处理的骨架，将可变的校验和修复逻辑留给子类实现，既保证了流程的一致性，又保留了业务的灵活性。

模板方法模式构建输出校验流水线

一个完整的输出处理流程通常包含四个阶段：格式校验、自动修复、内容审核和业务规则校验。模板方法模式将这些阶段固定下来，形成不可变的骨架，同时允许各个子类根据业务特点自定义每个阶段的具体实现。例如代码审查 Agent 的格式校验关注代码块完整性，而客服 Agent 的格式校验关注回复结构是否包含问候语和结束语。

这种模式的最大价值在于流程标准化。在多人协作开发 Agent 系统时，不同的开发者可能倾向于不同的校验顺序和错误处理方式。模板方法通过强制统一的处理骨架，确保所有 Agent 的输出都经过相同的基本检查，不会因为某个开发者的疏忽而遗漏关键环节。同时，新增的 Agent 类型只需要实现几个抽象方法即可接入现有体系，学习成本极低。

在实现自动修复时，需要明确修复的边界。对于格式问题（如缺少闭合标签、多余的换行符），应用层代码可以直接修复。但对于内容安全问题（如泄露了敏感信息），修复不应该由程序自动完成，而应该拦截并降级到安全响应。模板方法模式通过将这两类处理放在不同的抽象方法中，天然实现了这种边界划分。

指数退避重试策略

当输出需要反馈给模型重试时，简单的循环重试可能导致资源浪费。如果模型连续多次生成相同的违规输出，无间隔的重试只是在快速消耗 token 配额。指数退避（Exponential Backoff）结合抖动（Jitter）是更健壮的策略：每次重试的等待时间呈指数增长，并加入随机抖动以避免多个并发请求同时重试造成的流量峰值。

指数退避的数学原理很简单：第 n 次重试的基准延迟为 base_delay 乘以 2 的 n 次方。但这个值需要设置上限，否则在极端情况下等待时间会增长到不可接受的程度。通常将最大延迟限制在 30 秒以内。抖动的作用是打散重试请求的时间分布，避免所谓的"惊群效应"。

在实际工程中，重试次数也需要谨慎设置。过多的重试会显著增加用户等待时间，过少的重试则可能导致本可以修复的问题被过早放弃。建议根据业务场景设置不同的重试上限：实时聊天场景最多重试 1 次，异步任务处理场景可以放宽到 3 次。同时，每次重试都应该在反馈信息中提供具体的修正指导，而不是简单地告诉模型"上次错了"，这样才能提高重试的成功率。

graph TD
    A[Agent输出] --> B[格式校验层]
    B -->|违规| C[自动修复层]
    C -->|修复成功| D[返回修复结果]
    C -->|修复失败| E[指数退避重试]
    E -->|重试成功| D
    E -->|超过上限| F[降级响应]
    B -->|通过| G[内容审核层]
    G -->|违规| F
    G -->|通过| H[正常返回]
    
    style A fill:#4a90d9,color:#fff
    style H fill:#5cb85c,color:#fff
    style D fill:#5cb85c,color:#fff
    style F fill:#f0ad4e,color:#fff

以下示例展示了模板方法模式和指数退避重试的实现：

from abc import ABC, abstractmethod
from typing import Any
import asyncio
import random

class OutputPipeline(ABC):
    """输出处理流水线模板。"""
    
    async def process(self, raw_output: str, context: dict) -> dict:
        violations = await self._validate_format(raw_output)
        if violations:
            repaired = await self._attempt_repair(raw_output, violations)
            if repaired != raw_output:
                return self._build_success(repaired, repaired=True)
        
        safety_issues = await self._validate_content(
            repaired if violations else raw_output
        )
        if safety_issues:
            return self._build_blocked(safety_issues)
        
        final_output = repaired if violations else raw_output
        biz_violations = await self._validate_business_rules(final_output, context)
        if biz_violations:
            return self._build_blocked(biz_violations)
        
        return self._build_success(final_output)
    
    @abstractmethod
    async def _validate_format(self, output: str) -> list[str]:
        pass
    
    @abstractmethod
    async def _attempt_repair(self, output: str, violations: list[str]) -> str:
        pass
    
    @abstractmethod
    async def _validate_content(self, output: str) -> list[str]:
        pass
    
    @abstractmethod
    async def _validate_business_rules(self, output: str, context: dict) -> list[str]:
        pass
    
    def _build_success(self, output: str, repaired: bool = False) -> dict:
        return {"status": "success", "output": output, "repaired": repaired}
    
    def _build_blocked(self, reasons: list[str]) -> dict:
        return {"status": "blocked", "reasons": reasons}

class CodeReviewOutputPipeline(OutputPipeline):
    """代码审查 Agent 的专用输出流水线。"""
    
    async def _validate_format(self, output: str) -> list[str]:
        issues = []
        if not output.startswith("```"):
            issues.append("Output must start with code block")
        if output.count("```") % 2 != 0:
            issues.append("Unclosed code block detected")
        return issues
    
    async def _attempt_repair(self, output: str, violations: list[str]) -> str:
        repaired = output
        for issue in violations:
            if "Unclosed code block" in issue and not output.rstrip().endswith("```"):
                repaired = output.rstrip() + "\n```"
        return repaired
    
    async def _validate_content(self, output: str) -> list[str]:
        import re
        if re.search(r'sk-[a-zA-Z0-9]{48}', output):
            return ["Potential API key leak detected"]
        return []
    
    async def _validate_business_rules(self, output: str, context: dict) -> list[str]:
        max_lines = context.get("max_lines", 100)
        if output.count("\n") > max_lines:
            return [f"Output exceeds {max_lines} lines limit"]
        return []

class ExponentialBackoffRetry:
    """指数退避重试器。"""
    
    def __init__(self, max_retries: int = 3, base_delay: float = 1.0, max_delay: float = 30.0):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
    
    def _calculate_delay(self, attempt: int) -> float:
        exponential = self.base_delay * (2 ** attempt)
        capped = min(exponential, self.max_delay)
        jitter = random.uniform(0, capped * 0.3)
        return capped + jitter
    
    async def execute(self, fn, *args, **kwargs) -> Any:
        last_exception = None
        for attempt in range(self.max_retries + 1):
            try:
                return await fn(*args, **kwargs)
            except GuardrailViolationError as e:
                last_exception = e
                if attempt < self.max_retries:
                    delay = self._calculate_delay(attempt)
                    print(f"Retry {attempt + 1}/{self.max_retries} after {delay:.1f}s")
                    await asyncio.sleep(delay)
        raise last_exception

生产环境部署与性能优化

输出质量网关的实践要点

将本章节的技术应用到生产环境时，首要考虑的是稳定性与可观测性。建议采用渐进式 rollout 策略：先在开发环境验证核心逻辑，再迁移到预发布环境进行压力测试，最后才全量上线。部署过程中应配置完善的日志收集和指标监控，确保任何问题都能被快速发现和定位。

具体来说，需要在基础设施层面做好以下准备：容器资源限制（CPU/内存）、网络策略配置（防火墙规则、服务网格）、持久化存储选型（SSD vs 标准盘）以及备份恢复方案。对于高可用要求严格的场景，建议部署多实例并配置负载均衡，避免单点故障导致服务中断。

内容安全合规的关键指标

监控是生产系统的生命线。针对本章节涉及的功能，建议重点跟踪以下指标：请求延迟（P50/P95/P99）、错误率（4xx/5xx/超时）、吞吐量（QPS/TPS）以及资源利用率（CPU/内存/磁盘/网络）。这些指标应接入统一的监控大盘，并设置合理的告警阈值。

除了基础指标，还应关注业务层面的指标。例如功能成功率、用户满意度、成本消耗趋势等。通过将技术指标与业务指标关联分析，可以更准确地评估系统改进的实际价值，避免陷入"为了优化而优化"的陷阱。

流式输出校验的架构考量

随着业务规模增长，单实例部署很快会成为瓶颈。扩展性设计应在项目初期就纳入考量，而非事后补救。水平扩展通常比垂直扩展更具成本效益，但也引入了分布式系统的复杂性（数据一致性、服务发现、负载均衡等）。

在扩展过程中，建议遵循"无状态优先"原则：将状态外置到独立的存储层（如 Redis、PostgreSQL），使计算层可以随时水平扩容。对于无法避免的状态（如会话、缓存），采用分布式一致性协议或最终一致性模型来管理。定期进行容量规划和压力测试，确保系统在流量峰值时仍能稳定运行。

运维团队的协作建议

技术方案的落地离不开高效的团队协作。建议建立清晰的运维手册（Runbook），涵盖常见故障的诊断步骤、应急处理流程和升级路径。同时，通过定期的复盘会议，将线上事故转化为团队的学习素材，持续完善系统的健壮性。

在工具链方面，推荐将本章节的配置和脚本纳入版本控制（Git），并使用 Infrastructure as Code（IaC）工具（如 Terraform、Ansible）管理基础设施变更。这不仅能提高部署效率，还能确保环境一致性，减少"在我机器上能跑"的问题。

输出 Guardrails 的延迟敏感性与业务场景相关。对于实时聊天应用，200ms 的额外校验可能尚可接受；但对于流式语音交互，任何超过 50ms 的延迟都会影响用户体验。