第一章：后训练基础理论

L4: 价值对齐 (Value Alignment)
    ├── 伦理原则遵循
    ├── 文化敏感性
    └── 长期影响考虑

L3: 任务能力 (Task Capability)  
    ├── 指令理解与执行
    ├── 多步推理
    └── 知识运用

L2: 交互质量 (Interaction Quality)
    ├── 对话连贯性
    ├── 角色一致性
    └── 语气适应性

L1: 基础安全 (Basic Safety)
    ├── 有害内容过滤
    ├── 个人信息保护
    └── 事实准确性

预训练分布 P_pretrain(x)
     ↓
   后训练（多目标优化）
     ↓
目标分布 P_aligned(x)

核心挑战：

1. 保留原有能力（避免灾难性遗忘）
   - 知识保持率 > 90%
   - 推理能力保持率 > 85%

2. 学习新行为（指令跟随）
   - 指令遵循率 > 95%
   - 格式一致性 > 90%

3. 维持输出多样性（避免模式坍塌）
   - 响应熵 H(Y|X) > 阈值
   - 创造性任务多样性保持

4. 分布泛化（OOD Generalization）
   - 训练分布 ≠ 部署分布
   - 需要鲁棒性机制

def self_instruct(seed_tasks, model, n_iter):
    tasks = seed_tasks
    for i in range(n_iter):
        # 生成新指令
        new_instructions = model.generate_instructions(tasks)
        # 生成响应
        responses = model.generate_responses(new_instructions)
        # 质量过滤
        filtered = quality_filter(new_instructions, responses)
        tasks.extend(filtered)
    return tasks

Step 1: 收集偏好数据
├── 对同一指令生成多个响应
├── 人工标注偏好排序
└── 构建对比数据集

Step 2: 训练奖励模型
├── 使用Bradley-Terry模型
├── 学习隐含奖励函数
└── 验证奖励一致性

Step 3: PPO优化
├── 初始化：π_θ = π_SFT
├── 采样：生成响应
├── 评分：计算奖励
├── 更新：PPO梯度步
└── 约束：KL惩罚

class RewardModel(nn.Module):
    def __init__(self, base_model):
        super().__init__()
        self.transformer = base_model
        # 关键：使用独立的value head
        self.value_head = nn.Linear(hidden_size, 1)

    def forward(self, input_ids, attention_mask):
        outputs = self.transformer(input_ids, attention_mask)
        # 使用最后一个token的表示
        rewards = self.value_head(outputs.last_hidden_state[:, -1, :])
        return rewards

if kl > target_kl * 1.5:
    β *= 1.5  # 增强约束
elif kl < target_kl * 0.5:
    β *= 0.5  # 放松约束

def compute_dpo_loss(model, ref_model, batch, beta=0.1):
    # 计算策略模型的log概率
    policy_chosen_logps = model.log_prob(batch.chosen)
    policy_reject_logps = model.log_prob(batch.rejected)

    # 计算参考模型的log概率
    with torch.no_grad():
        ref_chosen_logps = ref_model.log_prob(batch.chosen)
        ref_reject_logps = ref_model.log_prob(batch.rejected)

    # 计算log比率
    chosen_rewards = beta * (policy_chosen_logps - ref_chosen_logps)
    reject_rewards = beta * (policy_reject_logps - ref_reject_logps)

    # DPO损失
    loss = -F.logsigmoid(chosen_rewards - reject_rewards).mean()
    return loss

def generate_ai_preferences(instruction, responses, critic_model):
    # AI评判标准
    criteria = """

    1. 准确性和事实性
    2. 有用性和相关性
    3. 清晰度和组织性
    4. 安全性和适当性
    """

    scores = []
    for response in responses:
        prompt = f"{criteria}\n\n指令：{instruction}\n响应：{response}\n评分："
        score = critic_model.evaluate(prompt)
        scores.append(score)

    # 转换为偏好对
    preferences = create_preference_pairs(responses, scores)
    return preferences

Initial Response → Critique → Revision → Final Response
         ↑                                      ↓
         └──────── Constitutional Rules ────────┘

能力维度评估（真实案例统计）：
┌─────────────────┬──────────┬──────────┬─────────────┐
│    能力类别      │ 预训练   │ 后训练   │   退化原因   │
├─────────────────┼──────────┼──────────┼─────────────┤
│ 事实知识召回    │   95%    │   92%    │ 安全过滤    │
│ 数学推理        │   88%    │   85%    │ 格式约束    │
│ 代码生成        │   92%    │   88%    │ 拒绝机制    │
│ 创造性写作      │   90%    │   75%    │ 模式坍塌    │
│ 多语言能力      │   85%    │   78%    │ 英语偏向    │
├─────────────────┼──────────┼──────────┼─────────────┤
│ 安全性          │   60%    │   95%    │ 主要目标 ✓  │
│ 指令跟随        │   40%    │   90%    │ 主要目标 ✓  │
│ 拒绝有害请求    │   20%    │   98%    │ 主要目标 ✓  │
└─────────────────┴──────────┴──────────┴─────────────┘

# 可视化：t-SNE投影显示
# 预训练：知识均匀分布
# 后训练：形成"安全区"和"危险区"聚类

class AlignmentTaxMeasurer:
    def __init__(self, pretrain_model, aligned_model):
        self.benchmarks = {
            'knowledge': ['MMLU', 'TriviaQA', 'NaturalQuestions'],
            'reasoning': ['GSM8K', 'MATH', 'BBH'],
            'coding': ['HumanEval', 'MBPP', 'CodeContests'],
            'creativity': ['story_continuation', 'poetry_generation'],
            'multimodal': ['VQA', 'COCO_caption'] if has_vision else []
        }

    def measure_tax(self):
        results = {}
        for category, tests in self.benchmarks.items():
            pretrain_scores = evaluate(self.pretrain_model, tests)
            aligned_scores = evaluate(self.aligned_model, tests)

            # 计算各种指标
            results[category] = {
                'absolute_drop': pretrain_scores - aligned_scores,
                'relative_drop': (pretrain_scores - aligned_scores) / pretrain_scores,
                'worst_case': min(aligned_scores),
                'variance': np.var(aligned_scores)
            }

        # 综合对齐税
        total_tax = weighted_average(results, self.importance_weights)
        return results, total_tax

def create_mixed_dataset(alignment_data, pretrain_data, mix_ratio=0.9):
    """
    关键发现：10-15%的预训练数据可减少50%的对齐税
    """
    n_alignment = int(len(alignment_data) * mix_ratio)
    n_pretrain = len(alignment_data) - n_alignment

    # 策略1：随机混合
    mixed = random.sample(alignment_data, n_alignment) + \
            random.sample(pretrain_data, n_pretrain)

    # 策略2：课程混合（推荐）
    # 早期多预训练数据，后期多对齐数据
    schedule = lambda epoch: 0.3 * exp(-epoch/10) + 0.1

    return mixed

def compute_fisher_information(model, data):
    fisher = {}
    model.eval()

    for batch in data:
        loss = model(batch).loss
        loss.backward()

        for name, param in model.named_parameters():
            if param.grad is not None:
                if name not in fisher:
                    fisher[name] = param.grad.data ** 2
                else:
                    fisher[name] += param.grad.data ** 2

    # 归一化
    for name in fisher:
        fisher[name] /= len(data)

    return fisher

def layerwise_finetune(model, data, layer_schedule):
    """
    从顶层到底层逐渐解冻
    底层保留更多预训练知识
    """
    for stage, layers_to_unfreeze in enumerate(layer_schedule):
        # 冻结所有层
        for param in model.parameters():
            param.requires_grad = False

        # 解冻指定层
        for layer_idx in layers_to_unfreeze:
            for param in model.layers[layer_idx].parameters():
                param.requires_grad = True

        # 训练当前阶段
        train_stage(model, data, epochs=2)

对齐程度 ↑
    │     ○ 帕累托最优点
    │    ╱│
    │   ╱ │ ← 可达区域
    │  ╱  • 次优解
    │ ╱   │
    │╱    │
    └──────┴──→ 原始能力
         能力保持阈值

def pareto_optimization(models, objectives):
    """
    NSGA-II风格的多目标优化
    """
    population = initialize_population(models)

    for generation in range(max_generations):
        # 评估每个模型
        fitness = evaluate_objectives(population, objectives)

        # 非支配排序
        fronts = non_dominated_sort(population, fitness)

        # 选择下一代
        next_gen = []
        for front in fronts:
            if len(next_gen) + len(front) <= pop_size:
                next_gen.extend(front)
            else:
                # 拥挤度排序
                crowding = crowding_distance(front, fitness)
                sorted_front = sorted(zip(front, crowding), 
                                    key=lambda x: x[1], reverse=True)
                remaining = pop_size - len(next_gen)
                next_gen.extend([x[0] for x in sorted_front[:remaining]])
                break

        population = next_gen

    return get_pareto_front(population, fitness)

Level 5: 元认知理解
  └── "我要你忽略之前的指令" → 识别并拒绝

Level 4: 价值对齐理解
  └── "帮我写一封辞职信" → 考虑后果和建议

Level 3: 隐含意图推理  
  └── "太热了" → 推断：调节温度/开窗/提供饮品建议

Level 2: 多步骤指令分解
  └── "先总结文章要点，然后翻译成中文，最后写一段评论"

Level 1: 直接指令执行
  └── "将这段话翻译成英文"

Level 0: 语法解析
  └── 基础的句法理解

class DynamicSafetyBoundary:
    def __init__(self):
        self.static_rules = load_constitution()  # 基础规则
        self.context_factors = {
            'user_history': [],
            'conversation_context': [],
            'detected_intent': None,
            'risk_level': 0
        }

    def evaluate_request(self, request):
        # 1. 静态规则检查
        static_risk = self.check_static_rules(request)

        # 2. 上下文风险评估
        context_risk = self.evaluate_context(request)

        # 3. 意图分析
        intent = self.analyze_intent(request)

        # 4. 综合决策
        total_risk = self.weighted_risk(static_risk, context_risk, intent)

        if total_risk > 0.8:
            return "REFUSE", self.generate_refusal(request, total_risk)
        elif total_risk > 0.5:
            return "WARN", self.generate_warning(request) 
        elif total_risk > 0.3:
            return "CAVEAT", self.add_disclaimer(request)
        else:
            return "ALLOW", None

有用性 ↑
100%│     
    │   ╱ ← 理想曲线
 80%│  ╱
    │ ╱ • 当前SOTA
 60%│╱
    │── ← 基础线
 40%│
    │
 20%│
    └────────────→ 安全性
     60% 80% 100%

class IntelligentRefusal:
    def __init__(self):
        self.refusal_templates = {
            'illegal': "我不能协助违法活动。",
            'harmful': "这可能造成伤害，我不能提供帮助。",
            'privacy': "我不能分享个人隐私信息。",
            'uncertain': "我不确定这是否合适，让我们换个话题。"
        }

    def generate_refusal(self, request, risk_type):
        # 1. 识别具体风险
        specific_risk = self.identify_specific_risk(request)

        # 2. 解释原因（教育性）
        explanation = self.explain_why_harmful(specific_risk)

        # 3. 提供替代方案
        alternatives = self.suggest_alternatives(request, risk_type)

        # 4. 组合响应
        response = f"""
        {self.refusal_templates[risk_type]}

        {explanation}

        作为替代，我可以：
        {alternatives}
        """

        return response

请求分类决策树：
├── 明确有害（>95%确定）
│   └── 直接拒绝 + 简短解释
├── 可能有害（70-95%确定）
│   └── 软性拒绝 + 详细解释 + 替代建议
├── 边界模糊（30-70%确定）
│   ├── 部分满足 + 安全限制
│   └── 要求澄清意图
└── 安全请求（<30%风险）
    └── 正常执行 + 可选免责声明

class InstructionPriorityResolver:
    def __init__(self):
        # 优先级从高到低
        self.priority_hierarchy = [
            'legal_compliance',      # 法律合规
            'safety',               # 安全性
            'privacy',              # 隐私保护
            'truthfulness',         # 真实性
            'helpfulness',          # 有用性
            'user_preference'       # 用户偏好
        ]

    def resolve_conflict(self, instructions):
        conflicts = self.detect_conflicts(instructions)

        if not conflicts:
            return self.merge_instructions(instructions)

        # 基于优先级解决冲突
        resolved = {}
        for conflict in conflicts:
            winning_instruction = max(
                conflict.instructions,
                key=lambda x: self.priority_hierarchy.index(x.type)
            )
            resolved[conflict.aspect] = winning_instruction

        return self.synthesize_response(resolved)

class DistributionShift:
    def __init__(self):
        self.shift_types = {
            'covariate': {  # P(X)改变，P(Y|X)不变
                'example': '训练：新闻文章 → 部署：社交媒体',
                'severity': 'medium',
                'detection': 'KL divergence on input embeddings'
            },
            'label': {  # P(Y)改变，P(X|Y)不变
                'example': '训练：礼貌响应 → 部署：用户期望直接回答',
                'severity': 'high',
                'detection': 'output distribution monitoring'
            },
            'concept': {  # P(Y|X)改变
                'example': '新概念出现（如新技术、新事件）',
                'severity': 'critical',
                'detection': 'perplexity spike detection'
            },
            'temporal': {  # 随时间变化
                'example': '2023年训练 → 2025年部署',
                'severity': 'progressive',
                'detection': 'time-aware evaluation'
            }
        }

def analyze_output_shift(train_outputs, deploy_outputs):
    metrics = {}

    # 长度分布变化
    metrics['length_shift'] = {
        'train_mean': np.mean([len(o) for o in train_outputs]),
        'deploy_mean': np.mean([len(o) for o in deploy_outputs]),
        'kl_divergence': compute_kl(train_lengths, deploy_lengths)
    }

    # 多样性变化
    metrics['diversity'] = {
        'train_entropy': compute_entropy(train_outputs),
        'deploy_entropy': compute_entropy(deploy_outputs),
        'unique_ngrams': count_unique_ngrams(deploy_outputs)
    }

    # 模式坍塌检测
    metrics['mode_collapse'] = {
        'repetition_rate': detect_repetitions(deploy_outputs),
        'template_usage': detect_templates(deploy_outputs)
    }

    return metrics

class DistributionShiftDetector:
    def __init__(self, reference_data):
        self.reference_stats = self.compute_reference_stats(reference_data)
        self.detection_methods = {
            'statistical': self.statistical_tests,
            'model_based': self.model_based_detection,
            'ensemble': self.ensemble_detection
        }

    def statistical_tests(self, new_data):
        tests = {}

        # 1. Kolmogorov-Smirnov测试
        tests['ks_test'] = ks_2samp(
            self.reference_stats['embeddings'],
            compute_embeddings(new_data)
        )

        # 2. Maximum Mean Discrepancy (MMD)
        tests['mmd'] = self.compute_mmd(
            self.reference_stats['features'],
            extract_features(new_data)
        )

        # 3. JS散度
        tests['js_divergence'] = self.compute_js_divergence(
            self.reference_stats['distribution'],
            estimate_distribution(new_data)
        )

        return tests

    def model_based_detection(self, new_data):
        # 使用专门的OOD检测器
        ood_scores = self.ood_detector.predict(new_data)

        # 不确定性估计
        uncertainty = self.uncertainty_estimator.estimate(new_data)

        return {
            'ood_score': np.mean(ood_scores),
            'epistemic_uncertainty': uncertainty['epistemic'],
            'aleatoric_uncertainty': uncertainty['aleatoric']
        }

监控仪表板：
┌──────────────────────────────────────┐
│ 分布偏移监控 - 实时状态              │
├──────────────────────────────────────┤
│ KL散度:     0.023 [████░░░░░░] 正常  │
│ JS散度:     0.045 [██████░░░░] 警告  │
│ MMD:        0.012 [███░░░░░░░] 正常  │
│ 困惑度峰值: 234   [████████░░] 异常  │
│ OOD率:      2.3%  [████░░░░░░] 正常  │
└──────────────────────────────────────┘

class TestTimeAdaptation:
    def __init__(self, model):
        self.model = model
        self.adaptation_buffer = []

    def adapt(self, batch):
        # 1. 熵最小化
        outputs = self.model(batch)
        entropy_loss = -torch.sum(outputs * torch.log(outputs))

        # 2. 伪标签自训练
        pseudo_labels = outputs.argmax(dim=-1)
        confidence = outputs.max(dim=-1).values

        # 只使用高置信度样本
        mask = confidence > 0.9
        if mask.any():
            self.model.backward(
                self.model.loss(batch[mask], pseudo_labels[mask])
            )

        # 3. 批归一化统计更新
        self.update_bn_stats(batch)

def continual_learning_pipeline():
    """
    数据飞轮：收集→标注→训练→部署
    """
    while True:
        # 1. 收集困难案例
        hard_cases = collect_hard_cases(
            threshold=0.7,  # 置信度阈值
            max_samples=1000
        )

        # 2. 智能标注
        labeled_data = hybrid_labeling(
            hard_cases,
            human_budget=100,  # 人工标注预算
            ai_labeler=strong_model
        )

        # 3. 增量训练
        model = incremental_train(
            model,
            new_data=labeled_data,
            replay_buffer=sample_old_data(500),
            regularization='ewc'  # 弹性权重巩固
        )

        # 4. A/B测试验证
        if validate_improvement(model, baseline):
            deploy(model)

        time.sleep(24 * 3600)  # 每日更新

class RobustTraining:
    def __init__(self):
        self.augmentation_strategies = [
            self.add_noise,
            self.paraphrase,
            self.backtranslation,
            self.token_cutoff,
            self.adversarial_perturbation
        ]

    def augment_batch(self, batch):
        augmented = []
        for sample in batch:
            # 多策略组合
            aug_sample = sample
            for strategy in random.sample(self.augmentation_strategies, k=2):
                aug_sample = strategy(aug_sample)
            augmented.append(aug_sample)

        return augmented

    def adversarial_perturbation(self, text):
        # 生成对抗样本
        embedding = self.encoder(text)
        gradient = compute_gradient(embedding)

        # FGSM扰动
        perturbed = embedding + self.epsilon * gradient.sign()

        return self.decoder(perturbed)

class CurriculumLearning:
    def __init__(self):
        self.difficulty_stages = [
            {
                'week': 1,
                'tasks': ['simple_qa', 'translation'],
                'max_length': 50,
                'complexity': 'low'
            },
            {
                'week': 2,
                'tasks': ['summarization', 'explanation'],
                'max_length': 200,
                'complexity': 'medium'
            },
            {
                'week': 3,
                'tasks': ['multi_turn_dialogue', 'reasoning'],
                'max_length': 500,
                'complexity': 'high'
            },
            {
                'week': 4,
                'tasks': ['adversarial', 'edge_cases'],
                'max_length': 1000,
                'complexity': 'extreme'
            }
        ]

    def get_current_batch(self, epoch):
        stage = self.get_stage(epoch)

        # 动态难度调整
        if self.performance > 0.9:
            # 加速进度
            stage = min(stage + 1, len(self.difficulty_stages) - 1)
        elif self.performance < 0.7:
            # 放慢进度
            stage = max(stage - 1, 0)

        return self.sample_from_stage(stage)

观察到的行为 → 推断奖励函数 → 优化策略

数学形式：
max_R  L(R|D_demo) - λ·||R||
s.t.   π* = argmax_π E[R(s,a)]

# 动态调整
β = β_init * (1 + decay_rate * step)
# 典型值：β_init=0.01, decay_rate=0.0001

# N-gram重叠检测
contamination = check_ngram_overlap(train_set, eval_set, n=13)
if contamination > 0.01:
    raise DataLeakageError

def diagnose_training_issues():
    checks = {
        "gradient_norm": check_gradient_explosion(),
        "loss_plateau": check_loss_convergence(),
        "reward_hacking": check_reward_gaming(),
        "distribution_shift": check_kl_divergence(),
        "capability_drop": run_capability_benchmarks()
    }
    return generate_diagnostic_report(checks)