安全护栏：LLM 应用的安全带和气囊

2026 年，大模型已经深度嵌入各行各业的关键业务流程。但 LLM 天生的不确定性——幻觉、越狱、有害输出——使其在生产环境中的风险不容忽视。安全护栏（Guardrails）就像汽车的安全带和气囊：正常行驶时无感，事故发生时救命。

一、安全护栏架构

1.1 三层护栏体系

用户请求 → [输入护栏] → LLM 推理 → [输出护栏] → 用户响应
              │                        │
              ├─ 内容过滤               ├─ 有害内容检测
              ├─ 意图检查               ├─ 幻觉检测
              ├─ 注入检测               ├─ PII 脱敏
              ├─ 速率限制               ├─ 格式验证
              └─ 权限验证               └─ 合规审查

1.2 护栏设计原则

原则说明重要性
纵深防御多层护栏，层层兜底核心原则
最小延迟护栏延迟不超过总延迟的20%用户体验
可解释拦截原因必须可追溯合规要求
可配置不同场景不同规则集灵活性
可观测所有拦截记录可审计安全运营
低误报正常请求拦截率<2%可用性

二、输入护栏实现

2.1 内容过滤护栏

from dataclasses import dataclass
from typing import List, Optional
import re

@dataclass
class GuardrailResult:
    passed: bool
    category: str          # 护栏类别
    severity: str          # low | medium | high | critical
    reason: str           # 拦截原因
    original_input: str
    sanitized_input: Optional[str] = None  # 净化后的输入

class ContentFilterGuardrail:
    """内容过滤护栏"""
    
    def __init__(self):
        self.categories = {
            'violence': {
                'patterns': [
                    r'(?i)(如何|怎么|怎样).{0,10}(制造|制作|获取).{0,10}(武器|炸弹|枪)',
                    r'(?i)(how\s+to|make|create).{0,15}(bomb|weapon|explosive)',
                    r'(?i)伤害.{0,5}(他人|别人|人类)',
                ],
                'severity': 'critical',
                'action': 'block'
            },
            'self_harm': {
                'patterns': [
                    r'(?i)(自杀|自残|自伤).{0,5}(方法|方式|怎么)',
                    r'(?i)(suicide|self.harm|kill.myself)',
                    r'(?i)(不想活|了结|结束生命)',
                ],
                'severity': 'critical',
                'action': 'block_and_resource'  # 拦截并提供帮助资源
            },
            'illegal': {
                'patterns': [
                    r'(?i)(毒品|大麻|海洛因).{0,5}(购买|出售|制作|提炼)',
                    r'(?i)(假证|假身份证).{0,5}(办理|制作|购买)',
                    r'(?i)(洗钱|逃税).{0,5}(方法|操作|教程)',
                ],
                'severity': 'high',
                'action': 'block'
            },
            'hate_speech': {
                'patterns': [
                    r'(?i)(仇恨|歧视|侮辱).{0,5}(种族|民族|宗教|性别)',
                    r'(?i)(racial|ethnic|religious)\s+(slur|insult|attack)',
                ],
                'severity': 'high',
                'action': 'block'
            },
            'sexual': {
                'patterns': [
                    r'(?i)(色情|淫秽|成人).{0,5}(内容|图片|视频)',
                    r'(?i)(未成年|儿童|teenager).{0,5}(sexual|色情)',
                ],
                'severity': 'critical',
                'action': 'block'
            },
        }
    
    def check(self, user_input: str) -> GuardrailResult:
        for category, config in self.categories.items():
            for pattern in config['patterns']:
                if re.search(pattern, user_input):
                    return GuardrailResult(
                        passed=False,
                        category=category,
                        severity=config['severity'],
                        reason=f"匹配到{category}类内容规则",
                        original_input=user_input
                    )
        
        return GuardrailResult(
            passed=True,
            category='none',
            severity='low',
            reason='',
            original_input=user_input
        )

2.2 意图分类护栏

class IntentClassificationGuardrail:
    """意图分类护栏——确保请求在允许范围内"""
    
    ALLOWED_INTENTS = [
        'product_inquiry',      # 产品咨询
        'technical_support',    # 技术支持
        'general_qa',          # 通用问答
        'content_creation',    # 内容创作
        'code_assistance',     # 代码辅助
        'translation',         # 翻译
        'summarization',       # 总结
    ]
    
    DISALLOWED_INTENTS = [
        'medical_diagnosis',   # 医疗诊断
        'legal_advice',        # 法律建议
        'financial_advice',    # 金融建议
        'harmful_request',     # 有害请求
        'jailbreak_attempt',   # 越狱尝试
    ]
    
    def __init__(self, llm_client):
        self.llm = llm_client
    
    def check(self, user_input: str, allowed_intents: list = None) -> GuardrailResult:
        allowed = allowed_intents or self.ALLOWED_INTENTS
        
        intent = self._classify_intent(user_input)
        
        if intent in self.DISALLOWED_INTENTS:
            return GuardrailResult(
                passed=False,
                category='disallowed_intent',
                severity='high',
                reason=f'检测到不允许的意图：{intent}',
                original_input=user_input
            )
        
        if intent not in allowed:
            return GuardrailResult(
                passed=False,
                category='out_of_scope',
                severity='medium',
                reason=f'请求超出服务范围（意图：{intent}）',
                original_input=user_input
            )
        
        return GuardrailResult(
            passed=True, category='intent', severity='low',
            reason='', original_input=user_input
        )
    
    def _classify_intent(self, text: str) -> str:
        prompt = f"""对以下输入进行意图分类，从以下选项中选择最匹配的：
{', '.join(self.ALLOWED_INTENTS + self.DISALLOWED_INTENTS)}

输入：{text}
意图："""
        return self.llm.generate(prompt).strip().lower()

2.3 PII 检测与脱敏护栏

class PIIGuardrail:
    """PII（个人身份信息）检测与脱敏护栏"""
    
    PII_PATTERNS = {
        'phone': {
            'pattern': r'1[3-9]\d{9}',
            'mask': 'PHONE***',
        },
        'id_card': {
            'pattern': r'\d{17}[\dXx]',
            'mask': 'IDCARD***',
        },
        'email': {
            'pattern': r'[\w.-]+@[\w.-]+\.\w+',
            'mask': 'EMAIL***',
        },
        'bank_card': {
            'pattern': r'\d{16,19}',
            'mask': 'BANKCARD***',
        },
        'address': {
            'pattern': r'[\u4e00-\u9fa5]{2,}(省|市|区|县|镇|村|路|街|号)',
            'mask': 'ADDRESS***',
        },
    }
    
    def check(self, user_input: str, mode: str = 'mask') -> GuardrailResult:
        """检测并处理PII"""
        import re
        detected = {}
        sanitized = user_input
        
        for pii_type, config in self.PII_PATTERNS.items():
            matches = re.findall(config['pattern'], user_input)
            if matches:
                detected[pii_type] = matches
                if mode == 'mask':
                    sanitized = re.sub(
                        config['pattern'], config['mask'], sanitized
                    )
                elif mode == 'block':
                    return GuardrailResult(
                        passed=False,
                        category='pii_detected',
                        severity='high',
                        reason=f'检测到{pii_type}信息',
                        original_input=user_input
                    )
        
        return GuardrailResult(
            passed=True,
            category='pii',
            severity='low' if not detected else 'medium',
            reason=f'检测到PII: {list(detected.keys())}' if detected else '',
            original_input=user_input,
            sanitized_input=sanitized
        )

2.4 速率限制护栏

import time
from collections import defaultdict

class RateLimitGuardrail:
    """速率限制护栏"""
    
    def __init__(self):
        self.limits = {
            'per_user': {'window': 60, 'max': 30},      # 每用户每分钟30次
            'per_ip': {'window': 60, 'max': 100},        # 每IP每分钟100次
            'global': {'window': 1, 'max': 50},          # 全局每秒50次
        }
        self.counters = defaultdict(list)
    
    def check(self, user_id: str, ip: str) -> GuardrailResult:
        now = time.time()
        
        # 检查各维度限制
        for dimension, limit in self.limits.items():
            key = f"{dimension}:{user_id if dimension == 'per_user' else ip if dimension == 'per_ip' else 'global'}"
            
            # 清理过期记录
            self.counters[key] = [
                t for t in self.counters[key]
                if now - t < limit['window']
            ]
            
            # 检查限制
            if len(self.counters[key]) >= limit['max']:
                return GuardrailResult(
                    passed=False,
                    category='rate_limit',
                    severity='medium',
                    reason=f'{dimension}速率超限：{limit["max"]}/{limit["window"]}s',
                    original_input=''
                )
            
            self.counters[key].append(now)
        
        return GuardrailResult(
            passed=True, category='rate_limit', severity='low',
            reason='', original_input=''
        )

三、输出护栏实现

3.1 有害内容检测护栏

class OutputSafetyGuardrail:
    """输出安全护栏"""
    
    def __init__(self, llm_client):
        self.llm = llm_client
    
    def check(self, output: str, context: dict = None) -> GuardrailResult:
        """检查输出是否安全"""
        checks = {
            'harmful_content': self._check_harmful(output),
            'pii_leak': self._check_pii_leak(output, context),
            'instruction_leak': self._check_instruction_leak(output, context),
            'hallucination': self._check_hallucination(output, context),
        }
        
        for check_name, result in checks.items():
            if not result['safe']:
                return GuardrailResult(
                    passed=False,
                    category=check_name,
                    severity=result['severity'],
                    reason=result['reason'],
                    original_input=output
                )
        
        return GuardrailResult(
            passed=True, category='output_safety', severity='low',
            reason='', original_input=output
        )
    
    def _check_harmful(self, output: str) -> dict:
        """检查有害内容"""
        prompt = f"""评估以下AI输出是否包含有害内容：
{output[:500]}

检查维度：
1. 暴力或伤害建议
2. 违法活动指导
3. 仇恨言论
4. 性暗示内容
5. 自残引导

如果安全返回"SAFE"，否则返回"UNSAFE: [具体原因]"。"""
        
        result = self.llm.generate(prompt)
        if 'SAFE' in result:
            return {'safe': True}
        return {
            'safe': False,
            'severity': 'critical',
            'reason': result
        }
    
    def _check_pii_leak(self, output: str, context: dict) -> dict:
        """检查是否泄露PII"""
        pii_guard = PIIGuardrail()
        result = pii_guard.check(output, mode='detect')
        if not result.passed:
            return {
                'safe': False,
                'severity': 'critical',
                'reason': f'输出中包含PII: {result.reason}'
            }
        return {'safe': True}
    
    def _check_instruction_leak(self, output: str, context: dict) -> dict:
        """检查是否泄露系统指令"""
        system_prompt = context.get('system_prompt', '')
        if not system_prompt:
            return {'safe': True}
        
        # 计算输出与系统提示词的相似度
        overlap = self._text_overlap(output, system_prompt)
        if overlap > 0.3:
            return {
                'safe': False,
                'severity': 'high',
                'reason': f'输出可能与系统提示词重叠 ({overlap:.0%})'
            }
        return {'safe': True}
    
    def _check_hallucination(self, output: str, context: dict) -> dict:
        """检查幻觉"""
        sources = context.get('sources', [])
        if not sources:
            return {'safe': True}
        
        # 检查输出中的关键事实是否可被来源支持
        prompt = f"""判断以下输出中的事实性陈述是否有来源支持。

输出：{output[:500]}

来源：{' '.join(sources)[:1000]}

对于每个事实性陈述，判断是否可被来源支持。
如果有不可支持的陈述，返回"HALLUCINATION: [具体内容]"
如果都可支持，返回"SUPPORTED"."""
        
        result = self.llm.generate(prompt)
        if 'SUPPORTED' in result:
            return {'safe': True}
        return {
            'safe': False,
            'severity': 'medium',
            'reason': result
        }
    
    def _text_overlap(self, text1: str, text2: str) -> float:
        words1 = set(text1.lower().split())
        words2 = set(text2.lower().split())
        if not words2:
            return 0
        overlap = words1 & words2
        return len(overlap) / min(len(words1), len(words2))

3.2 格式验证护栏

class OutputFormatGuardrail:
    """输出格式验证护栏"""
    
    def __init__(self, expected_format: dict):
        self.expected = expected_format
    
    def check(self, output: str) -> GuardrailResult:
        if self.expected['type'] == 'json':
            return self._check_json(output)
        elif self.expected['type'] == 'markdown':
            return self._check_markdown(output)
        elif self.expected['type'] == 'code':
            return self._check_code(output)
        return GuardrailResult(passed=True, ...)
    
    def _check_json(self, output: str) -> GuardrailResult:
        import json
        try:
            data = json.loads(output)
        except json.JSONDecodeError as e:
            # 尝试修复
            try:
                # 提取JSON部分
                start = output.index('{')
                end = output.rindex('}') + 1
                data = json.loads(output[start:end])
            except:
                return GuardrailResult(
                    passed=False, category='format', severity='high',
                    reason=f'JSON解析失败: {e}', original_input=output
                )
        
        # Schema验证
        if 'schema' in self.expected:
            from jsonschema import validate, ValidationError
            try:
                validate(instance=data, schema=self.expected['schema'])
            except ValidationError as e:
                return GuardrailResult(
                    passed=False, category='format', severity='medium',
                    reason=f'Schema验证失败: {e.message}', 
                    original_input=output
                )
        
        return GuardrailResult(passed=True, ...)

四、护栏编排引擎

class GuardrailOrchestrator:
    """护栏编排引擎——组合多个护栏"""
    
    def __init__(self):
        self.input_guardrails = []
        self.output_guardrails = []
    
    def add_input_guardrail(self, guardrail, priority: int = 0):
        self.input_guardrails.append((priority, guardrail))
        self.input_guardrails.sort(key=lambda x: x[0])
    
    def add_output_guardrail(self, guardrail, priority: int = 0):
        self.output_guardrails.append((priority, guardrail))
        self.output_guardrails.sort(key=lambda x: x[0])
    
    def process_input(self, user_input: str, context: dict = None) -> dict:
        """处理输入——依次通过所有输入护栏"""
        context = context or {}
        current_input = user_input
        
        for priority, guardrail in self.input_guardrails:
            result = guardrail.check(current_input, **context)
            
            if not result.passed:
                # 记录拦截
                self._log_block('input', guardrail.__class__.__name__, result)
                
                # 返回安全响应
                return {
                    'allowed': False,
                    'reason': result.reason,
                    'category': result.category,
                    'severity': result.severity,
                    'safe_response': self._safe_response(result)
                }
            
            # 使用净化后的输入
            if result.sanitized_input:
                current_input = result.sanitized_input
        
        return {'allowed': True, 'input': current_input}
    
    def process_output(self, output: str, context: dict = None) -> dict:
        """处理输出——依次通过所有输出护栏"""
        context = context or {}
        current_output = output
        
        for priority, guardrail in self.output_guardrails:
            result = guardrail.check(current_output, context)
            
            if not result.passed:
                self._log_block('output', guardrail.__class__.__name__, result)
                
                return {
                    'allowed': False,
                    'reason': result.reason,
                    'category': result.category,
                    'severity': result.severity,
                    'safe_response': self._safe_response(result)
                }
        
        return {'allowed': True, 'output': current_output}
    
    def _safe_response(self, result: GuardrailResult) -> str:
        """生成安全替代响应"""
        if result.category == 'self_harm':
            return "我注意到你可能在经历困难时期。请拨打心理援助热线：400-161-9995。"
        elif result.category == 'pii_detected':
            return "您的输入包含敏感个人信息，请去除后再提交。"
        elif result.category == 'rate_limit':
            return "请求过于频繁，请稍后再试。"
        else:
            return "抱歉，我无法处理这个请求。"

五、护栏配置示例

def create_production_guardrails():
    """创建生产环境护栏配置"""
    orchestrator = GuardrailOrchestrator()
    
    # 输入护栏（按优先级排序）
    orchestrator.add_input_guardrail(RateLimitGuardrail(), priority=0)
    orchestrator.add_input_guardrail(ContentFilterGuardrail(), priority=1)
    orchestrator.add_input_guardrail(PIIGuardrail(mode='mask'), priority=2)
    orchestrator.add_input_guardrail(
        IntentClassificationGuardrail(llm), priority=3
    )
    
    # 输出护栏
    orchestrator.add_output_guardrail(
        OutputSafetyGuardrail(llm), priority=0
    )
    orchestrator.add_output_guardrail(
        OutputFormatGuardrail({'type': 'json'}), priority=1
    )
    
    return orchestrator

六、护栏效果监控

class GuardrailMonitor:
    """护栏效果监控"""
    
    def __init__(self):
        self.stats = defaultdict(lambda: {
            'total': 0, 'blocked': 0, 'false_positives': 0
        })
    
    def record(self, guardrail_name: str, result: GuardrailResult):
        self.stats[guardrail_name]['total'] += 1
        if not result.passed:
            self.stats[guardrail_name]['blocked'] += 1
    
    def report(self) -> dict:
        report = {}
        for name, stats in self.stats.items():
            report[name] = {
                'total': stats['total'],
                'blocked': stats['blocked'],
                'block_rate': stats['blocked'] / stats['total'],
                'health': 'healthy' if stats['blocked'] / stats['total'] < 0.05 
                          else 'warning' if stats['blocked'] / stats['total'] < 0.15
                          else 'critical'
            }
        return report

结语

安全护栏不是可选项，而是 LLM 生产应用的必需品。好的护栏体系应该是透明的（用户几乎感知不到）、智能的（低误报、高召回）、可演进的（随威胁变化而更新）。

2026 年的趋势是护栏本身的智能化——用 LLM 来检测 LLM 的不安全输出，用 Agent 来编排多层护栏。但无论技术如何进步，核心原则不变：纵深防御、最小侵入、持续监控。在 AI 安全领域，宁可过度防御，也不能有丝毫侥幸。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。