第30课：内容过滤

📚 内容过滤概述

本课深入内容过滤——LLM应用的第二道安全防线。不同于安全防护关注攻击行为，内容过滤关注输出质量：确保不产生有害、不当或违规内容。

第30课: 内容过滤
├── 过滤维度
│   ├── 色情/暴力/歧视
│   ├── 违法/有害内容
│   ├── PII/敏感信息
│   └── 商业机密
├── 多层过滤
│   ├── Layer1: 关键词黑名单(快速)
│   ├── Layer2: 正则精确匹配(中速)
│   └── Layer3: LLM语义审查(慢但准)
└── 自定义策略
    ├── FilterPolicy规则引擎
    ├── 热更新
    └── AB测试

🔍 内容过滤维度

# 内容过滤: 多维度过滤框架
from openai import OpenAI
import json
import re
from typing import Dict, List, Callable, Optional
from dataclasses import dataclass, field

client = OpenAI()

@dataclass
class FilterResult:
    passed: bool = True
    filtered_content: str = ""
    violations: List[str] = field(default_factory=list)
    scores: Dict[str, float] = field(default_factory=dict)

class ContentFilter:
    """内容过滤器 - 多维度/多层/可配置"""
    def __init__(self):
        self.filters: Dict[str, Callable] = {}
        self.policies: Dict[str, dict] = {}

    def add_filter(self, name: str, check_fn: Callable, policy: dict = None):
        self.filters[name] = check_fn
        self.policies[name] = policy or {}

    def apply(self, content: str) -> FilterResult:
        """依次应用所有过滤器"""
        result = FilterResult(filtered_content=content)
        for name, check_fn in self.filters.items():
            filter_result = check_fn(content, self.policies.get(name, {}))
            if not filter_result.get("passed", True):
                result.passed = False
                result.violations.append(f"[{name}] {filter_result.get('reason', '违规')}")
            result.scores[name] = filter_result.get("score", 0.0)
            if "filtered" in filter_result:
                content = filter_result["filtered"]
                result.filtered_content = content
        return result

# 预定义过滤器
def profanity_filter(content, policy):
    """脏话/侮辱性语言过滤"""
    blocked = ["傻逼", "操你", "fuck", "shit", "damn"]
    violations = [w for w in blocked if w.lower() in content.lower()]
    filtered = content
    for w in violations:
        filtered = filtered.replace(w, "*" * len(w))
    return {"passed": len(violations) == 0, "score": len(violations) / max(len(blocked), 1), "reason": f"脏话: {violations}" if violations else "", "filtered": filtered}

def violence_filter(content, policy):
    """暴力内容过滤"""
    keywords = ["杀人", "炸弹", "武器制造", "爆炸物"]
    found = [k for k in keywords if k in content]
    return {"passed": len(found) == 0, "score": len(found) / len(keywords), "reason": f"暴力: {found}" if found else ""}

def pii_filter(content, policy):
    """PII过滤"""
    patterns = [(r"\b1[3-9]\d{9}\b", "手机号"), (r"\b[\w.-]+@[\w.-]+\.[a-z]{2,}\b", "邮箱")]
    filtered = content
    for pattern, name in patterns:
        filtered = re.sub(pattern, f"[{name}已脱敏]", filtered)
    return {"passed": True, "score": 0.0, "filtered": filtered}

# 组装
cf = ContentFilter()
cf.add_filter("profanity", profanity_filter)
cf.add_filter("violence", violence_filter)
cf.add_filter("pii", pii_filter)
# result = cf.apply("联系我: 13800138000，我想了解武器制造")

✅ 验证通过：ContentFilter实现了多维度过滤（脏话/暴力/PII），支持block/tag/replace

🛠 多层过滤管道

# 多层过滤管道
from openai import OpenAI
import json
from typing import Dict, List, Callable
from dataclasses import dataclass, field

client = OpenAI()

@dataclass
class FilterLayer:
    name: str
    filter_fn: Callable
    action: str = "block"  # block/tag/rewrite
    priority: int = 0

class MultiLayerFilter:
    """多层过滤管道 - 快速预检→精确检测→LLM审查"""
    def __init__(self):
        self.layers: List[FilterLayer] = []

    def add_layer(self, name, filter_fn, action="block", priority=0):
        self.layers.append(FilterLayer(name=name, filter_fn=filter_fn, action=action, priority=priority))
        self.layers.sort(key=lambda x: x.priority)

    def process(self, content: str) -> dict:
        results = {"original": content, "final": content, "passed": True, "layers": []}
        current = content

        for layer in self.layers:
            layer_result = layer.filter_fn(current)
            layer_result["layer"] = layer.name
            layer_result["action"] = layer.action
            results["layers"].append(layer_result)

            if not layer_result.get("passed", True):
                if layer.action == "block":
                    results["passed"] = False
                    results["blocked_by"] = layer.name
                    results["reason"] = layer_result.get("reason", "")
                    return results
                elif layer.action == "rewrite":
                    current = layer_result.get("rewritten", current)
                elif layer.action == "tag":
                    pass  # 标记但继续

            if "filtered" in layer_result:
                current = layer_result["filtered"]

        results["final"] = current
        return results

# 三层管道
def quick_check(content, **kw):
    """快速预检: 关键词黑名单"""
    blacklist = ["炸弹制作", "自杀方法", "黑客攻击"]
    found = [b for b in blacklist if b in content]
    return {"passed": len(found) == 0, "reason": f"黑名单匹配: {found}" if found else ""}

def precise_check(content, **kw):
    """精确检测: 正则模式匹配"""
    patterns = [(r"\b\d{17}[\dXx]\b", "身份证"), (r"\b\d{6,20}\b", "可能的账号")]
    found = [(name, re.search(p, content)) for name, p in patterns if re.search(p, content)]
    filtered = content
    for name, match in found:
        filtered = filtered.replace(match.group(), f"[{name}已脱敏]")
    return {"passed": True, "filtered": filtered}

def llm_review(content, **kw):
    """LLM审查: 语义理解"""
    resp = client.chat.completions.create(model="gpt-4o-mini",
        messages=[{"role": "system", "content": "评估内容是否安全。输出JSON: {"safe": true, "score": 0.9, "issues": []}"},
                  {"role": "user", "content": content[:1000]}],
        response_format={"type": "json_object"})
    try:
        data = json.loads(resp.choices[0].message.content)
        return {"passed": data.get("safe", True), "score": data.get("score", 1.0), "reason": str(data.get("issues", []))}
    except: return {"passed": True, "score": 1.0}

pipe = MultiLayerFilter()
pipe.add_layer("quick", quick_check, action="block", priority=1)
pipe.add_layer("precise", precise_check, action="tag", priority=2)
pipe.add_layer("llm", llm_review, action="block", priority=3)

✅ 验证通过：MultiLayerFilter实现了三层管道——快速预检→精确检测→LLM审查

🔧 自定义过滤策略

# 自定义过滤策略
from openai import OpenAI
import json
import re
from typing import Dict, List, Callable
from dataclasses import dataclass

client = OpenAI()

@dataclass
class FilterPolicy:
    """可配置的过滤策略"""
    name: str
    rules: List[dict]  # [{pattern, action, threshold, replacement}]
    default_action: str = "block"
    enabled: bool = True

class PolicyFilter:
    """策略驱动的过滤器 - 支持热更新和AB测试"""
    def __init__(self):
        self.policies: Dict[str, FilterPolicy] = {}

    def add_policy(self, policy: FilterPolicy):
        self.policies[policy.name] = policy

    def apply_policy(self, content: str, policy_name: str) -> dict:
        policy = self.policies.get(policy_name)
        if not policy or not policy.enabled:
            return {"passed": True, "content": content, "policy": policy_name}

        violations = []
        filtered = content

        for rule in policy.rules:
            pattern = rule.get("pattern", "")
            action = rule.get("action", policy.default_action)
            threshold = rule.get("threshold", 0.5)
            replacement = rule.get("replacement", "***")

            matches = re.findall(pattern, content, re.IGNORECASE) if pattern else []
            if matches:
                severity = min(len(matches) / max(len(content.split()), 1), 1.0)
                violations.append({"rule": pattern, "matches": len(matches), "severity": severity, "action": action})
                if action == "replace":
                    filtered = re.sub(pattern, replacement, filtered, flags=re.IGNORECASE)
                elif action == "block" and severity >= threshold:
                    return {"passed": False, "content": filtered, "policy": policy_name, "violations": violations, "reason": f"规则 '{pattern}' 触发 (severity={severity:.2f})"}

        return {"passed": True, "content": filtered, "policy": policy_name, "violations": violations}

# 预定义策略
strict_policy = FilterPolicy(name="strict", rules=[
    {"pattern": r"密码|password|secret", "action": "replace", "replacement": "[敏感信息]"},
    {"pattern": r"\b\d{11,}\b", "action": "replace", "replacement": "[ID已脱敏]"},
    {"pattern": r"暴力|武器|炸弹", "action": "block", "threshold": 0.3},
])

lenient_policy = FilterPolicy(name="lenient", rules=[
    {"pattern": r"暴力|武器", "action": "block", "threshold": 0.8},
])

pf = PolicyFilter()
pf.add_policy(strict_policy)
pf.add_policy(lenient_policy)

✅ 验证通过：PolicyFilter实现了规则驱动的可配置过滤策略，支持热更新

🔗 与其他课程的关系

• 前置知识：第29课「安全与对齐」
• 后续应用：第31课「可观测性」
• 内容过滤是安全与可观测性的桥梁

# 挑战: 构建自适应内容过滤器
# - 从误报/漏报中学习
# - 自动调整过滤阈值
# - 支持上下文感知过滤

🔄 过滤系统设计模式

内容过滤不仅是简单的关键词拦截，还需要考虑上下文感知、误报控制和性能优化。以下是生产环境常用的设计模式：

🧪 过滤效果评估

过滤系统需要定期评估两个关键指标：