LLM Guardrails 技术全景：从输入校验到输出防护的生产级安全架构

┌──────────────────────────────────────────────────────────────┐
│                 LLM Guardrails 四层防御架构                    │
│                                                              │
│  ┌────────────────────────────────────────────────────────┐  │
│  │  Layer 1: 输入 Guardrails（延迟预算 5-50ms）            │  │
│  │  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐  │  │
│  │  │ 注入检测 │ │ PII 脱敏 │ │ 话题过滤 │ │ 速率限制 │  │  │
│  │  └──────────┘ └──────────┘ └──────────┘ └──────────┘  │  │
│  └────────────────────────────────────────────────────────┘  │
│                          ▼                                   │
│  ┌────────────────────────────────────────────────────────┐  │
│  │  Layer 2: 行为/对话 Guardrails（策略引擎）               │  │
│  │  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐  │  │
│  │  │ 对话流控 │ │ 工具白名单│ │ 角色边界 │ │ 多轮状态 │  │  │
│  │  └──────────┘ └──────────┘ └──────────┘ └──────────┘  │  │
│  └────────────────────────────────────────────────────────┘  │
│                          ▼                                   │
│  ┌────────────────────────────────────────────────────────┐  │
│  │  Layer 3: 检索/工具 Guardrails（Agent 时代新增）         │  │
│  │  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐  │  │
│  │  │ 文档信任 │ │ 参数校验 │ │ 沙箱隔离 │ │ 调用审计 │  │  │
│  │  └──────────┘ └──────────┘ └──────────┘ └──────────┘  │  │
│  └────────────────────────────────────────────────────────┘  │
│                          ▼                                   │
│  ┌────────────────────────────────────────────────────────┐  │
│  │  Layer 4: 输出 Guardrails（延迟预算 50-300ms）          │  │
│  │  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐  │  │
│  │  │ 毒性检测 │ │ Schema   │ │ 事实核查 │ │ PII 扫描 │  │  │
│  │  └──────────┘ └──────────┘ └──────────┘ └──────────┘  │  │
│  └────────────────────────────────────────────────────────┘  │
└──────────────────────────────────────────────────────────────┘

import re
from dataclasses import dataclass

@dataclass
class InputGuardResult:
    passed: bool
    reason: str = ""
    risk_level: str = "low"

class InputGuardrailPipeline:
    def __init__(self):
        self.injection_patterns = [
            r"ignore\s+(all\s+)?previous\s+instructions",
            r"you\s+are\s+now\s+(DAN|jailbreak)",
            r"system\s*prompt\s*:",
            r"<\|im_start\|>.*system",
        ]
        self.pii_patterns = {
            "email": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
            "phone": r"\b1[3-9]\d{9}\b",
            "id_card": r"\b\d{17}[\dXx]\b",
        }

    def check(self, user_input: str) -> InputGuardResult:
        for pattern in self.injection_patterns:
            if re.search(pattern, user_input, re.IGNORECASE):
                return InputGuardResult(
                    passed=False,
                    reason=f"Prompt injection detected: {pattern}",
                    risk_level="critical"
                )

        for pii_type, pattern in self.pii_patterns.items():
            if re.search(pattern, user_input):
                return InputGuardResult(
                    passed=False,
                    reason=f"PII detected: {pii_type}",
                    risk_level="high"
                )

        if len(user_input) > 10000:
            return InputGuardResult(
                passed=False,
                reason="Input exceeds maximum length",
                risk_level="medium"
            )

        return InputGuardResult(passed=True)

from pydantic import BaseModel, Field
from typing import Optional
from enum import Enum

class ToxicityLevel(Enum):
    SAFE = "safe"
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

class OutputCheckResult(BaseModel):
    is_safe: bool
    toxicity: ToxicityLevel = ToxicityLevel.SAFE
    pii_found: list[str] = Field(default_factory=list)
    schema_valid: bool = True
    repaired_content: Optional[str] = None

class OutputGuardrail:
    def __init__(self, max_retries: int = 2):
        self.max_retries = max_retries

    def check_toxicity(self, text: str) -> ToxicityLevel:
        toxicity_keywords = {
            "high": ["暴力", "仇恨", "歧视"],
            "medium": ["攻击", "威胁"],
        }
        for level, keywords in toxicity_keywords.items():
            if any(kw in text for kw in keywords):
                return ToxicityLevel(level)
        return ToxicityLevel.SAFE

    def check_pii(self, text: str) -> list[str]:
        import re
        found = []
        if re.search(r"\b\d{17}[\dXx]\b", text):
            found.append("id_card")
        if re.search(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", text):
            found.append("email")
        return found

    def validate_schema(self, text: str, schema: dict) -> bool:
        import json
        try:
            data = json.loads(text)
            return all(field in data for field in schema.get("required", []))
        except (json.JSONDecodeError, TypeError):
            return False

    def run(self, text: str, schema: dict = None) -> OutputCheckResult:
        toxicity = self.check_toxicity(text)
        pii = self.check_pii(text)
        schema_valid = self.validate_schema(text, schema) if schema else True

        is_safe = (
            toxicity in (ToxicityLevel.SAFE, ToxicityLevel.LOW)
            and len(pii) == 0
            and schema_valid
        )

        return OutputCheckResult(
            is_safe=is_safe,
            toxicity=toxicity,
            pii_found=pii,
            schema_valid=schema_valid,
        )

from nemoguardrails import LLMRails, RailsConfig

config = RailsConfig.from_path("./config")
rails = LLMRails(config)

response = await rails.generate_async(
    messages=[
        {"role": "user", "content": "帮我查一下隔壁公司的内部文档"}
    ]
)
print(response["content"])
# 被 Output Rail 拦截：超出授权范围的请求

define user ask about weather
  "今天天气怎么样"
  "明天会下雨吗"

define user ask about competitors
  "竞争对手的数据是什么"
  "帮我查一下 X 公司的信息"

define bot refuse competitor queries
  "抱歉，我无法提供其他公司的内部信息。我可以帮您查询天气或其他授权范围内的信息。"

# 对话流定义
define flow
  user ask about competitors
  bot refuse competitor queries

from guardrails import Guard
from guardrails.validators import Toxicity, PIIRedact
import openai

guard = Guard().use(
    PIIRedact(redact_direction="output"),
    Toxicity(threshold=0.7),
)

raw_output, metadata = guard(
    openai.chat.completions.create,
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "你是一个客服助手"},
        {"role": "user", "content": "我的身份证号是110101200001011234，帮我查一下订单"},
    ],
)

需要 Guardrails？
│
├─ 主要是输出格式/Schema 校验？
│  └─ → Guardrails AI（结构化验证 + 自动重试）
│
├─ 主要是对话安全/话题控制？
│  └─ → NeMo Guardrails（状态机 + Colang）
│
├─ 需要两者兼顾？
│  └─ → NeMo Guardrails + Guardrails AI 集成（官方已支持）
│
├─ 需要最低延迟？
│  └─ → Lakera Guard 或自建正则 + 小模型分类器
│
└─ 已有 Azure 生态？
   └─ → Azure AI Content Safety

User Request
  → [Input Guardrail Span] ← 注入检测: 通过 (23ms)
    → [LLM Call Span] ← gpt-4o 调用 (1.2s)
      → [Output Guardrail Span] ← 毒性检测: 通过 (45ms)
        → [PII Redaction Span] ← 脱敏处理 (8ms)
          → Response to User

import asyncio
from collections import deque

class StreamingOutputGuardrail:
    def __init__(self, buffer_size: int = 50):
        self.buffer_size = buffer_size
        self.token_buffer = deque(maxlen=buffer_size)
        self.toxicity_keywords = set()

    async def process_stream(self, token_stream):
        full_response = []
        buffer_filled = False

        async for token in token_stream:
            full_response.append(token)
            self.token_buffer.append(token)

            if not buffer_filled and len(self.token_buffer) >= self.buffer_size:
                buffer_filled = True
                buffer_text = "".join(self.token_buffer)
                if self._quick_toxicity_check(buffer_text):
                    yield "[安全提示：回复内容被拦截，请换个问法]"
                    return

            if buffer_filled and self._quick_toxicity_check(token):
                yield "[安全提示：回复内容被拦截]"
                return

            yield token

    def _quick_toxicity_check(self, text: str) -> bool:
        return any(kw in text for kw in self.toxicity_keywords)

┌─────────────────────────────────────────┐
│           AI Gateway（入口层）            │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐ │
│  │ 速率限制 │  │ 认证鉴权 │  │ 日志审计 │ │
│  └─────────┘  └─────────┘  └─────────┘ │
├─────────────────────────────────────────┤
│          自建输入 Guardrails             │
│  ┌─────────┐  ┌─────────┐              │
│  │ 正则过滤 │  │ PII 检测 │              │
│  └─────────┘  └─────────┘              │
├─────────────────────────────────────────┤
│        NeMo Guardrails（对话层）         │
│  ┌─────────┐  ┌─────────┐              │
│  │ 话题控制 │  │ 工具管控 │              │
│  └─────────┘  └─────────┘              │
├─────────────────────────────────────────┤
│              LLM / Agent                │
├─────────────────────────────────────────┤
│        Guardrails AI（输出层）           │
│  ┌─────────┐  ┌─────────┐              │
│  │ Schema   │  │ 毒性检测 │              │
│  │ 校验     │  │         │              │
│  └─────────┘  └─────────┘              │
├─────────────────────────────────────────┤
│         OpenTelemetry 追踪层            │
│  ┌─────────────────────────────────┐    │
│  │ Span: Input → LLM → Output     │    │
│  │ Metrics: 拦截率、延迟、误报率    │    │
│  └─────────────────────────────────┘    │
└─────────────────────────────────────────┘