🔒 AI 安全合规 / 输出过滤

🛡️ 输出过滤

LLM 的输出不可预测——即使有安全对齐，仍可能产生有害内容、泄露 PII、或输出版权内容。输出过滤是最后一道防线。

输出过滤管道

LLM 原始输出
    │
    ▼
┌──────────────────────┐
│ Step 1: 有害内容检测   │ ← OpenAI Moderation API / 自建分类器
└──────────┬───────────┘
           │
           ▼
┌──────────────────────┐
│ Step 2: PII 脱敏      │ ← 正则 + NER 模型
└──────────┬───────────┘
           │
           ▼
┌──────────────────────┐
│ Step 3: 版权内容过滤   │ ← 相似度检索 + 水印检测
└──────────┬───────────┘
           │
           ▼
┌──────────────────────┐
│ Step 4: 安全策略检查   │ ← 自定义规则引擎
└──────────┬───────────┘
           │
           ▼
       安全输出

有害内容检测

OpenAI Moderation API

OpenAI 提供免费的 Moderation API，可以检测输出中的仇恨、暴力、自残、色情等内容。

import openai

client = openai.OpenAI()

async def check_content_safety(text: str) -> dict:
    """使用 OpenAI Moderation API 检测有害内容"""
    response = client.moderations.create(
        model="omni-moderation-latest",  # 最新多模态审核模型
        input=text
    )
    
    result = response.results[0]
    
    return {
        "flagged": result.flagged,
        "categories": {
            k: v for k, v in result.categories.model_dump().items() 
            if v  # 只返回被标记的类别
        },
        "scores": result.category_scores.model_dump(),
        "action": "block" if result.flagged else "allow"
    }

# 使用示例
output = "LLM 生成的输出内容..."
safety = await check_content_safety(output)

if safety["flagged"]:
    print(f"⚠️ 检测到有害内容: {safety['categories']}")
    # 策略: 拒绝输出 / 替换为安全提示 / 重新生成
    return "抱歉，我无法生成此类内容。"
else:
    return output

Anthropic Constitutional AI

Anthropic 的 Constitutional AI (CAI) 方法不是外部过滤器，而是内嵌在模型训练中的安全机制。模型根据一组"宪法原则"自我审查输出。

# Anthropic Constitutional AI 的核心思想：
# 模型在生成响应时，内部有一个"批评-修订"循环
# 这不是你可以直接调用的 API，而是训练方法

# 但你可以在应用层模拟类似的过程：
async def constitutional_filter(output: str) -> str:
    """模拟 Constitutional AI 的批评-修订过程"""
    
    # Step 1: 让模型批评自己的输出
    critique = await client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"""Please review the following AI output for safety issues.
Check for: harmful content, PII leakage, bias, misinformation.

Output to review:
---
{output}
---

If you find issues, describe them. If the output is safe, say "SAFE"."""
        }]
    )
    
    critique_text = critique.content[0].text
    
    if "SAFE" in critique_text:
        return output
    
    # Step 2: 如果有问题，让模型修订
    revision = await client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1000,
        messages=[{
            "role": "user",
            "content": f"""The following AI output was flagged for issues:
{critique_text}

Original output:
{output}

Please provide a revised, safe version that addresses the issues."""
        }]
    )
    
    return revision.content[0].text

PII 脱敏

import re
from dataclasses import dataclass
from typing import Optional

@dataclass
class PIIDetection:
    pii_type: str
    value: str
    start: int
    end: int
    confidence: float

class PIIFilter:
    """PII 检测与脱敏过滤器"""
    
    # 中国特定模式
    CHINA_PATTERNS = {
        'phone_cn': (r'1[3-9]\d{9}', '手机号'),
        'id_card': (r'\d{6}(?:19|20)\d{2}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])\d{3}[\dXx]', '身份证号'),
        'bank_card': (r'\d{16,19}', '银行卡号'),
    }
    
    # 通用模式
    GLOBAL_PATTERNS = {
        'email': (r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '邮箱'),
        'phone_us': (r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', '美国电话'),
        'ssn': (r'\b\d{3}-\d{2}-\d{4}\b', 'SSN'),
        'ip': (r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b', 'IP地址'),
        'api_key_openai': (r'sk-[a-zA-Z0-9]{20,}', 'OpenAI API Key'),
        'api_key_anthropic': (r'sk-ant-[a-zA-Z0-9]{20,}', 'Anthropic API Key'),
    }
    
    def __init__(self, region: str = 'cn'):
        self.region = region
        self.patterns = {**self.GLOBAL_PATTERNS}
        if region == 'cn':
            self.patterns.update(self.CHINA_PATTERNS)
    
    def detect(self, text: str) -> list[PIIDetection]:
        """检测文本中的 PII"""
        findings = []
        for pii_type, (pattern, label) in self.patterns.items():
            for match in re.finditer(pattern, text):
                findings.append(PIIDetection(
                    pii_type=f"{label}({pii_type})",
                    value=match.group(),
                    start=match.start(),
                    end=match.end(),
                    confidence=0.9
                ))
        return findings
    
    def redact(self, text: str, mode: str = 'mask') -> str:
        """脱敏处理
        
        Args:
            mode: 'mask' - 部分掩码 | 'replace' - 完全替换 | 'remove' - 删除
        """
        findings = self.detect(text)
        # 从后往前替换，避免偏移
        for f in sorted(findings, key=lambda x: x.start, reverse=True):
            if mode == 'mask':
                masked = self._mask_value(f.value, f.pii_type)
            elif mode == 'replace':
                masked = f"[{f.pii_type}已脱敏]"
            elif mode == 'remove':
                masked = ""
            text = text[:f.start] + masked + text[f.end:]
        return text
    
    def _mask_value(self, value: str, pii_type: str) -> str:
        """部分掩码"""
        if len(value) <= 4:
            return '*' * len(value)
        # 保留前2和后2，中间用*替换
        return value[:2] + '*' * (len(value) - 4) + value[-2:]

# 使用
filter = PIIFilter(region='cn')
output = "请联系张三，手机号13812345678，邮箱zhangsan@example.com"
findings = filter.detect(output)
print(f"检测到 {len(findings)} 处 PII")
safe_output = filter.redact(output, mode='mask')
# "请联系张三，手机号13******78，邮箱zh***************om"

NER 增强检测

# 使用 spaCy NER 进行更精确的 PII 检测
# pip install spacy && python -m spacy download zh_core_web_sm

import spacy

class NERPIIDetector:
    """基于 NER 的 PII 检测器"""
    
    def __init__(self):
        # 中文 NER 模型
        try:
            self.nlp = spacy.load("zh_core_web_sm")
        except OSError:
            self.nlp = spacy.load("en_core_web_sm")
    
    def detect(self, text: str) -> list[dict]:
        """使用 NER 检测人名、地名、组织名等"""
        doc = self.nlp(text)
        findings = []
        for ent in doc.ents:
            if ent.label_ in ['PERSON', 'ORG', 'GPE', 'LOC']:
                findings.append({
                    'type': ent.label_,
                    'text': ent.text,
                    'start': ent.start_char,
                    'end': ent.end_char
                })
        return findings

ner = NERPIIDetector()
entities = ner.detect("张三在北京字节跳动公司工作")
# [{'type': 'PERSON', 'text': '张三', ...}, ...]

版权内容过滤

from dataclasses import dataclass
import hashlib

@dataclass
class CopyrightMatch:
    source_id: str
    similarity: float
    matched_text: str

class CopyrightFilter:
    """版权内容过滤器"""
    
    def __init__(self, known_sources: list[str] = None):
        # 已知版权内容的指纹库
        # 实际实现中这是一个向量数据库
        self.source_hashes = set()
        self.source_chunks = []
        if known_sources:
            for source in known_sources:
                self._add_source(source)
    
    def _add_source(self, text: str, chunk_size: int = 100):
        """将版权文本分块并存储指纹"""
        for i in range(0, len(text), chunk_size):
            chunk = text[i:i+chunk_size]
            self.source_hashes.add(hashlib.md5(chunk.encode()).hexdigest())
            self.source_chunks.append(chunk)
    
    def check(self, output: str, threshold: float = 0.8) -> list[CopyrightMatch]:
        """检查输出是否包含版权内容"""
        matches = []
        # 简单实现：滑动窗口检查
        window_size = 50
        for i in range(0, len(output) - window_size, 25):
            window = output[i:i+window_size]
            window_hash = hashlib.md5(window.encode()).hexdigest()
            if window_hash in self.source_hashes:
                matches.append(CopyrightMatch(
                    source_id="hash_match",
                    similarity=1.0,
                    matched_text=window
                ))
        return matches
    
    def check_with_embedding(self, output: str, 
                              embedding_client, threshold: float = 0.92) -> list[CopyrightMatch]:
        """使用嵌入相似度检查（更精确）"""
        # 将输出分块
        chunks = [output[i:i+200] for i in range(0, len(output), 200)]
        
        matches = []
        for chunk in chunks:
            # 获取输出的嵌入
            chunk_embedding = embedding_client.embed(chunk)
            
            # 在版权内容库中搜索
            results = embedding_client.search(
                chunk_embedding, top_k=3, threshold=threshold
            )
            
            for result in results:
                if result['score'] > threshold:
                    matches.append(CopyrightMatch(
                        source_id=result['id'],
                        similarity=result['score'],
                        matched_text=chunk[:100]
                    ))
        
        return matches

# 集成到输出管道
def filter_copyright_content(output: str) -> str:
    """过滤输出中的版权内容"""
    cf = CopyrightFilter()
    matches = cf.check(output)
    
    if matches:
        # 策略1: 移除匹配内容
        for match in matches:
            output = output.replace(match.matched_text, "[内容已过滤-可能涉及版权]")
        
        # 策略2: 完全替换输出
        # return "该回复可能包含版权内容，已被过滤。"
    
    return output

安全策略引擎

from enum import Enum
from typing import Callable

class FilterAction(Enum):
    ALLOW = "allow"
    WARN = "warn"
    BLOCK = "block"
    REPLACE = "replace"

class SafetyRule:
    """安全规则"""
    def __init__(self, name: str, check: Callable, 
                 action: FilterAction, replacement: str = ""):
        self.name = name
        self.check = check
        self.action = action
        self.replacement = replacement

class OutputSafetyEngine:
    """输出安全策略引擎"""
    
    def __init__(self):
        self.rules: list[SafetyRule] = []
        self._setup_default_rules()
    
    def _setup_default_rules(self):
        """设置默认安全规则"""
        
        # 规则1: 系统提示词泄露
        self.rules.append(SafetyRule(
            name="system_prompt_leak",
            check=lambda text: any(
                kw in text.lower() for kw in 
                ['system prompt:', 'you are a', 'your instructions are']
            ),
            action=FilterAction.BLOCK,
            replacement="抱歉，我无法透露系统配置信息。"
        ))
        
        # 规则2: API 密钥泄露
        self.rules.append(SafetyRule(
            name="api_key_leak",
            check=lambda text: bool(re.search(r'sk-[a-zA-Z0-9]{20,}', text)),
            action=FilterAction.REPLACE
        ))
        
        # 规则3: 自残内容
        self.rules.append(SafetyRule(
            name="self_harm",
            check=lambda text: any(
                kw in text for kw in 
                ['自杀', '自残', '结束生命', 'kill myself', 'suicide']
            ),
            action=FilterAction.BLOCK,
            replacement="如果你正在经历困难时期，请寻求帮助。\n中国24小时心理援助热线：400-161-9995\n全国希望24热线：400-161-9995"
        ))
        
        # 规则4: 代码注入
        self.rules.append(SafetyRule(
            name="code_injection",
            check=lambda text: any(
                pattern in text for pattern in 
                ['rm -rf', 'DROP TABLE', '__import__("os").system']
            ),
            action=FilterAction.BLOCK
        ))
        
        # 规则5: 中国敏感内容（基础）
        self.rules.append(SafetyRule(
            name="sensitive_content_cn",
            check=lambda text: False,  # 需要根据实际需求配置
            action=FilterAction.BLOCK
        ))
    
    def add_rule(self, rule: SafetyRule):
        self.rules.append(rule)
    
    def filter(self, output: str) -> tuple[str, list[dict]]:
        """应用所有规则过滤输出"""
        violations = []
        
        for rule in self.rules:
            if rule.check(output):
                violations.append({
                    'rule': rule.name,
                    'action': rule.action.value
                })
                
                if rule.action == FilterAction.BLOCK:
                    return rule.replacement or "此内容已被安全过滤器阻止。", violations
                elif rule.action == FilterAction.REPLACE:
                    # 执行替换（如 API Key 脱敏）
                    output = self._apply_replacement(output, rule)
                elif rule.action == FilterAction.WARN:
                    # 记录警告但允许输出
                    pass
        
        return output, violations

engine = OutputSafetyEngine()
safe_output, violations = engine.filter(llm_output)
if violations:
    for v in violations:
        print(f"⚠️ 触发规则: {v['rule']} → {v['action']}")

完整输出过滤管道实现

import asyncio
from dataclasses import dataclass

@dataclass
class FilterResult:
    output: str
    passed: bool
    filters_triggered: list[str]
    metadata: dict

class OutputFilterPipeline:
    """完整的输出过滤管道"""
    
    def __init__(self, pii_filter, safety_engine, copyright_filter=None):
        self.pii_filter = pii_filter
        self.safety_engine = safety_engine
        self.copyright_filter = copyright_filter
    
    async def filter(self, raw_output: str) -> FilterResult:
        """执行完整的过滤管道"""
        filters_triggered = []
        output = raw_output
        metadata = {}
        
        # Step 1: 有害内容检测 (Moderation API)
        safety = await check_content_safety(output)
        if safety["flagged"]:
            filters_triggered.append(f"content_safety:{','.join(safety['categories'].keys())}")
            if any(cat in safety['categories'] for cat in 
                   ['self-harm', 'sexual/minors', 'violence/graphic']):
                # 零容忍类别直接阻止
                return FilterResult(
                    output="此内容因安全原因被阻止。",
                    passed=False,
                    filters_triggered=filters_triggered,
                    metadata=safety
                )
        
        # Step 2: PII 脱敏
        pii_findings = self.pii_filter.detect(output)
        if pii_findings:
            filters_triggered.append(f"pii:{len(pii_findings)}_instances")
            output = self.pii_filter.redact(output, mode='mask')
            metadata['pii_types'] = list(set(f.pii_type for f in pii_findings))
        
        # Step 3: 版权内容过滤
        if self.copyright_filter:
            copyright_matches = self.copyright_filter.check(output)
            if copyright_matches:
                filters_triggered.append(f"copyright:{len(copyright_matches)}_matches")
                for match in copyright_matches:
                    output = output.replace(
                        match.matched_text, 
                        "[内容已过滤-可能涉及版权]"
                    )
        
        # Step 4: 安全策略检查
        output, violations = self.safety_engine.filter(output)
        if violations:
            filters_triggered.extend(
                f"policy:{v['rule']}" for v in violations
            )
            if any(v['action'] == 'block' for v in violations):
                return FilterResult(
                    output=output,
                    passed=False,
                    filters_triggered=filters_triggered,
                    metadata=metadata
                )
        
        return FilterResult(
            output=output,
            passed=len(filters_triggered) == 0,
            filters_triggered=filters_triggered,
            metadata=metadata
        )

# 集成
pipeline = OutputFilterPipeline(
    pii_filter=PIIFilter(region='cn'),
    safety_engine=OutputSafetyEngine()
)

result = await pipeline.filter(llm_output)
if not result.passed:
    print(f"⚠️ 输出被过滤: {result.filters_triggered}")
final_output = result.output

实践建议

← 上一篇：Prompt Injection 防护 | 下一篇：数据隔离 → | 返回首页