第11章：防御策略与安全对齐

┌─────────────────────────────────────────┐
│         应用层防御                         │
│  - 业务逻辑验证                           │
│  - 用户权限控制                           │
└─────────────────────────────────────────┘
┌─────────────────────────────────────────┐
│         输出层防御                         │
│  - 内容过滤器                             │
│  - 毒性检测                               │
│  - 安全分类器                             │
└─────────────────────────────────────────┘
┌─────────────────────────────────────────┐
│         模型层防御                         │
│  - 安全微调                               │
│  - 对抗训练                               │
│  - Constitutional AI                      │
└─────────────────────────────────────────┘
┌─────────────────────────────────────────┐
│         输入层防御                         │
│  - 提示过滤                               │
│  - 注入检测                               │
│  - 输入净化                               │
└─────────────────────────────────────────┘
┌─────────────────────────────────────────┐
│         训练层防御                         │
│  - RLHF                                  │
│  - 数据清洗                               │
│  - 差分隐私                               │
└─────────────────────────────────────────┘

原则示例：

1. 有用性："我应该提供准确、相关、有帮助的信息"
2. 无害性："我不应该生成可能造成身体或心理伤害的内容"
3. 诚实性："我应该承认不确定性，避免编造信息"
4. 公平性："我应该避免偏见和歧视"
5. 隐私性："我应该保护个人隐私信息"

输入提示 → 初始回复 → 批判 → 修订 → 最终回复

def automated_red_teaming(model, constitution):
    # 生成对抗性提示
    adversarial_prompts = generate_attacks(model)

    # 模型自我评估
    for prompt in adversarial_prompts:
        response = model.generate(prompt)
        critique = model.critique(response, constitution)

        if is_harmful(critique):
            # 添加到训练数据
            training_data.add(prompt, response, critique)
            # 生成改进版本
            revised = model.revise(response, critique)
            training_data.add(prompt, revised, "safe")

    # 重新训练模型
    model.train(training_data)

标注维度：
├── 安全等级：安全/边界/危险
├── 危害类型：偏见/暴力/隐私/错误信息
├── 响应策略：直接回答/委婉拒绝/教育引导
└── 上下文依赖：独立/需要上下文

def safety_aware_instruction_tuning(model, dataset):
    # 数据预处理
    safe_data = []
    for example in dataset:
        # 添加安全前缀
        safe_prompt = add_safety_prefix(example.prompt)

        # 生成多个回复候选
        responses = model.generate_multiple(safe_prompt, n=5)

        # 安全过滤
        safe_response = safety_filter(responses)

        # 构建训练样本
        safe_data.append({
            "input": safe_prompt,
            "output": safe_response,
            "safety_label": classify_safety(safe_response)
        })

    # 加权损失函数
    def weighted_loss(pred, target, safety_label):
        base_loss = cross_entropy(pred, target)
        # 对安全相关样本增加权重
        weight = 2.0 if safety_label == "safety_critical" else 1.0
        return weight * base_loss

    # 训练
    model.train(safe_data, loss_fn=weighted_loss)

原始权重：W_0 (frozen)
LoRA更新：ΔW = BA (trainable)
最终权重：W = W_0 + αΔW

安全风险：

1. 低秩更新可能绕过原始模型的安全机制
2. 不同LoRA模块组合可能产生意外行为
3. 恶意LoRA可以轻易改变模型行为

class BlacklistFilter:
    def __init__(self):
        self.patterns = [
            r'ignore.*previous.*instructions',
            r'you.*are.*now.*in.*developer.*mode',
            r'</?(script|img|iframe)',  # HTML注入
            r'(\x00|\x1f|\x7f)',  # 控制字符
        ]

    def check(self, text):
        for pattern in self.patterns:
            if re.search(pattern, text, re.IGNORECASE):
                return False, f"Blocked: {pattern}"
        return True, "Pass"

特征向量构建：
├── 词汇特征：危险词汇频率、罕见词比例
├── 句法特征：句子长度分布、依存关系模式
├── 语义特征：主题分布、情感极性
├── 统计特征：字符熵、n-gram分布
└── 行为特征：指令密度、角色切换标记

class EnsembleFilter:
    def __init__(self):
        self.models = [
            BERTClassifier(),    # 深度语义理解
            XGBoostClassifier(), # 统计特征
            RuleBasedFilter(),   # 确定性规则
        ]

    def predict(self, text):
        predictions = [m.predict(text) for m in self.models]
        # 加权投票
        weights = [0.5, 0.3, 0.2]
        score = sum(w * p for w, p in zip(weights, predictions))
        return score > threshold

原始输入：{user_input}
改写后：
"请以负责任和安全的方式回答以下问题，
避免有害或不当内容：{user_input}"

def detect_context_injection(text):
    # 检测常见的上下文分隔模式
    separators = [
        "### Assistant:",
        "System: ",
        "[INST]",
        "Human: ",
    ]

    for sep in separators:
        if sep in text:
            # 分析分隔符位置和内容
            parts = text.split(sep)
            if is_suspicious_context(parts):
                return True
    return False

class ToxicityDetector:
    def __init__(self):
        self.categories = {
            'hate': 0.0,      # 仇恨言论
            'threat': 0.0,    # 威胁恐吓
            'insult': 0.0,    # 侮辱谩骂
            'adult': 0.0,     # 成人内容
            'violence': 0.0,  # 暴力内容
            'self_harm': 0.0, # 自残内容
        }

    def detect(self, text):
        # 使用预训练的BERT模型
        embeddings = self.encoder(text)

        # 多头分类器
        for category, head in self.classification_heads.items():
            score = sigmoid(head(embeddings))
            self.categories[category] = score

        return self.categories

┌─────────────┐     通过      ┌─────────────┐     通过      ┌─────────────┐
│  快速过滤器  │ ──────────> │  精确分类器  │ ──────────> │  专家模型   │
│  (规则基础)  │              │  (轻量模型)  │              │  (大模型)   │
└─────────────┘              └─────────────┘              └─────────────┘
      ↓ 拦截                        ↓ 拦截                        ↓ 拦截
   [直接拒绝]                    [警告提示]                   [人工审核]

def cascade_detection(text):
    # 第一层：快速规则检查
    if rule_filter.is_obvious_harmful(text):
        return "BLOCKED", 1.0

    # 第二层：轻量级模型
    light_score = light_model.predict(text)
    if light_score > 0.9:
        return "BLOCKED", light_score
    elif light_score < 0.1:
        return "SAFE", light_score

    # 第三层：精确模型（仅对边界案例）
    precise_score = heavy_model.predict(text)
    return ("BLOCKED" if precise_score > 0.5 else "SAFE", precise_score)

class StreamingSafetyMonitor:
    def __init__(self):
        self.buffer = []
        self.safety_score = 0.0
        self.threshold = 0.7

    def check_token(self, token):
        self.buffer.append(token)

        # 滑动窗口检测
        if len(self.buffer) >= self.window_size:
            text = tokenizer.decode(self.buffer[-self.window_size:])
            score = self.detector.quick_check(text)

            # 指数移动平均更新
            self.safety_score = 0.9 * self.safety_score + 0.1 * score

            if self.safety_score > self.threshold:
                return "STOP", self.safety_score

        return "CONTINUE", self.safety_score

class EnsembleSafetyChecker:
    def __init__(self):
        self.models = [
            ('perspective_api', 0.3),
            ('custom_bert', 0.3),
            ('gpt_guard', 0.2),
            ('rule_based', 0.2),
        ]

    def evaluate(self, text):
        scores = []
        for model_name, weight in self.models:
            score = self.get_model_score(model_name, text)
            scores.append(score * weight)

        # 加权平均
        final_score = sum(scores)

        # 一票否决机制
        if any(s > 0.95 for s in scores):
            return 1.0

        return final_score

1. 初始化策略 π_0
2. For k = 0, 1, 2, ...:
   a. 策略评估：计算 V^{π_k} 和 V_{safe}^{π_k}
   b. 策略改进：
      π_{k+1}(a|s) ∝ exp(Q^{π_k}(s,a)/τ) · I[V_{safe}^{π_k}(s') ≥ θ]
   c. 如果收敛，停止

def analyze_attention_for_attacks(model, input_text):
    # 获取所有层的注意力权重
    attentions = model.get_attention_weights(input_text)

    # 检测异常注意力模式
    suspicious_patterns = []

    for layer_idx, attn in enumerate(attentions):
        # 1. 检测注意力劫持
        if has_attention_hijacking(attn):
            suspicious_patterns.append(f"Layer {layer_idx}: Attention hijacking")

        # 2. 检测位置编码攻击
        if has_position_encoding_anomaly(attn):
            suspicious_patterns.append(f"Layer {layer_idx}: Position anomaly")

        # 3. 检测语义漂移
        if has_semantic_drift(attn):
            suspicious_patterns.append(f"Layer {layer_idx}: Semantic drift")

    return suspicious_patterns

class SafetyNeuronAnalyzer:
    def __init__(self, model):
        self.model = model
        self.safety_neurons = {}

    def identify_safety_neurons(self, safe_samples, unsafe_samples):
        # 收集激活模式
        safe_activations = self.collect_activations(safe_samples)
        unsafe_activations = self.collect_activations(unsafe_samples)

        # 计算每个神经元的判别能力
        for layer_idx in range(self.model.num_layers):
            layer_neurons = []
            for neuron_idx in range(self.model.hidden_size):
                # 使用互信息衡量相关性
                mi_score = mutual_information(
                    safe_activations[layer_idx][:, neuron_idx],
                    unsafe_activations[layer_idx][:, neuron_idx]
                )

                if mi_score > threshold:
                    layer_neurons.append((neuron_idx, mi_score))

            self.safety_neurons[layer_idx] = layer_neurons

def find_safe_counterfactual(model, harmful_input):
    # 初始化为原始输入
    counterfactual = harmful_input.copy()

    for iteration in range(max_iterations):
        # 计算梯度
        grad = compute_gradient(model, counterfactual, target="safe")

        # 更新输入
        counterfactual = counterfactual - lr * grad

        # 投影到有效输入空间
        counterfactual = project_to_valid_input(counterfactual)

        # 检查是否安全
        if is_safe(model(counterfactual)):
            return counterfactual

    return None

class SelectivePruning:
    def __init__(self, model):
        self.model = model

    def prune_harmful_neurons(self, harmful_neurons):
        for layer_idx, neuron_indices in harmful_neurons.items():
            # 创建剪枝掩码
            mask = torch.ones(self.model.layers[layer_idx].weight.shape)
            mask[:, neuron_indices] = 0

            # 应用掩码
            self.model.layers[layer_idx].weight.data *= mask

            # 补偿剪枝造成的性能损失
            self.compensate_pruning(layer_idx, neuron_indices)