安全护栏：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 生产应用的必需品。好的护栏体系应该是透明的（用户几乎感知不到）、智能的（低误报、高召回）、可演进的（随威胁变化而更新）。
...
