RLHF 实践指南 #

┌─────────────────────────────────────────────────────────────┐
│                    RLHF 项目规划清单                         │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  1. 明确目标                                                 │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  □ 模型的主要用途是什么？                            │   │
│  │  □ 目标用户群体是谁？                                │   │
│  │  □ 需要达到什么效果？                                │   │
│  │  □ 有哪些安全要求？                                  │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  2. 资源评估                                                 │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  □ 可用的计算资源（GPU 数量和类型）                  │   │
│  │  □ 数据收集预算                                      │   │
│  │  □ 团队规模和技能                                    │   │
│  │  □ 项目时间线                                        │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  3. 技术选择                                                 │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  □ 基础模型选择                                      │   │
│  │  □ 训练方法（RLHF/DPO/其他）                         │   │
│  │  □ 训练框架选择                                      │   │
│  │  □ 评估方案                                          │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  4. 风险评估                                                 │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  □ 潜在的技术风险                                    │   │
│  │  □ 数据质量风险                                      │   │
│  │  □ 安全和伦理风险                                    │   │
│  │  □ 缓解措施                                          │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

选择 RLHF 的场景：
────────────────────────
├── 追求最佳对齐效果
├── 有充足的计算资源
├── 需要精细控制优化过程
├── 可以接受复杂的训练流程
└── 有经验的团队

选择 DPO 的场景：
────────────────────────
├── 计算资源有限
├── 快速原型开发
├── 中小规模模型
├── 团队经验有限
└── 时间紧迫

选择其他方法的场景：
────────────────────────
├── KTO：偏好数据不完整
├── ORPO：不想维护参考模型
├── RLAIF：人工标注成本高
└── Constitutional AI：有明确的规则体系

高质量 SFT 数据标准：
────────────────────────
内容质量：
├── 回复准确、有帮助
├── 语言流畅、自然
├── 格式规范、统一
└── 无有害内容

多样性：
├── 覆盖多种任务类型
├── 不同难度级别
├── 不同主题领域
└── 不同回复风格

数据量：
├── 最小：1,000 - 5,000 条
├── 推荐：10,000 - 50,000 条
└── 理想：100,000+ 条

class PreferenceDataBestPractices:
    @staticmethod
    def design_annotation_interface():
        return """
        标注界面设计建议：
        ────────────────────────
        1. 清晰的对比展示
           - 并排显示多个回复
           - 高亮差异部分
           - 支持放大/缩小
        
        2. 明确的标注指南
           - 提供评估维度说明
           - 提供正反面示例
           - 边界情况处理规则
        
        3. 质量控制机制
           - 标注者资格测试
           - 黄金标准题
           - 多人标注一致性检查
        
        4. 效率优化
           - 快捷键支持
           - 自动保存
           - 批量操作
        """
    
    @staticmethod
    def calculate_sample_size():
        return """
        样本量估算：
        ────────────────────────
        经验法则：
        - 最小可行：10,000 对偏好数据
        - 推荐数量：50,000 - 100,000 对
        - 高质量需求：200,000+ 对
        
        影响因素：
        - 模型规模（越大需要越多）
        - 任务复杂度
        - 数据质量
        - 预算限制
        """

class DataCleaner:
    def __init__(self, min_length=10, max_length=2048):
        self.min_length = min_length
        self.max_length = max_length
    
    def clean_preference_data(self, data):
        cleaned = []
        stats = {
            "total": len(data),
            "filtered_length": 0,
            "filtered_similarity": 0,
            "filtered_quality": 0,
        }
        
        for item in data:
            if not self._check_length(item):
                stats["filtered_length"] += 1
                continue
            
            if self._too_similar(item["chosen"], item["rejected"]):
                stats["filtered_similarity"] += 1
                continue
            
            if not self._check_quality(item):
                stats["filtered_quality"] += 1
                continue
            
            cleaned.append(item)
        
        stats["cleaned"] = len(cleaned)
        return cleaned, stats
    
    def _check_length(self, item):
        if len(item["chosen"]) < self.min_length:
            return False
        if len(item["rejected"]) < self.min_length:
            return False
        if len(item["prompt"]) + len(item["chosen"]) > self.max_length:
            return False
        return True
    
    def _too_similar(self, text1, text2, threshold=0.9):
        from difflib import SequenceMatcher
        similarity = SequenceMatcher(None, text1, text2).ratio()
        return similarity > threshold
    
    def _check_quality(self, item):
        if item["chosen"] == item["rejected"]:
            return False
        
        if len(item["chosen"]) > 5 * len(item["rejected"]):
            return False
        if len(item["rejected"]) > 5 * len(item["chosen"]):
            return False
        
        return True

class RecommendedHyperparameters:
    SFT = {
        "learning_rate": 2e-5,
        "batch_size": 128,
        "num_epochs": 3,
        "warmup_ratio": 0.03,
        "weight_decay": 0.01,
    }
    
    REWARD_MODEL = {
        "learning_rate": 1e-5,
        "batch_size": 64,
        "num_epochs": 1,
        "warmup_ratio": 0.03,
    }
    
    PPO = {
        "learning_rate": 1e-6,
        "batch_size": 64,
        "mini_batch_size": 16,
        "ppo_epochs": 4,
        "clip_range": 0.2,
        "kl_coef": 0.05,
        "value_coef": 0.5,
        "entropy_coef": 0.01,
        "gamma": 1.0,
        "gae_lambda": 0.95,
    }
    
    DPO = {
        "learning_rate": 1e-6,
        "batch_size": 32,
        "beta": 0.1,
        "num_epochs": 3,
    }

class TrainingMonitor:
    def __init__(self):
        self.metrics_history = []
        self.alerts = []
    
    def check_health(self, metrics, step):
        alerts = []
        
        if "kl_divergence" in metrics:
            if metrics["kl_divergence"] > 10:
                alerts.append({
                    "level": "ERROR",
                    "message": f"KL divergence too high: {metrics['kl_divergence']:.2f}",
                    "suggestion": "Increase KL penalty coefficient or decrease learning rate"
                })
        
        if "reward" in metrics:
            if len(self.metrics_history) > 10:
                recent_rewards = [m.get("reward", 0) for m in self.metrics_history[-10:]]
                if all(r > recent_rewards[0] * 1.5 for r in recent_rewards[1:]):
                    alerts.append({
                        "level": "WARNING",
                        "message": "Reward increasing too fast, possible reward hacking",
                        "suggestion": "Check reward model quality, increase KL penalty"
                    })
        
        if "policy_loss" in metrics:
            if metrics["policy_loss"] < -10:
                alerts.append({
                    "level": "WARNING",
                    "message": "Policy loss very negative, check for instability",
                    "suggestion": "Monitor closely, consider reducing learning rate"
                })
        
        self.metrics_history.append(metrics)
        self.alerts.extend(alerts)
        
        return alerts
    
    def should_stop(self, metrics):
        if "kl_divergence" in metrics and metrics["kl_divergence"] > 20:
            return True, "KL divergence too high"
        
        if "reward" in metrics and metrics["reward"] < -5:
            return True, "Reward collapsed"
        
        return False, None

class CheckpointManager:
    def __init__(self, output_dir, max_checkpoints=5):
        self.output_dir = output_dir
        self.max_checkpoints = max_checkpoints
        self.checkpoints = []
    
    def save_checkpoint(self, model, tokenizer, step, metrics):
        checkpoint_dir = f"{self.output_dir}/checkpoint-{step}"
        model.save_pretrained(checkpoint_dir)
        tokenizer.save_pretrained(checkpoint_dir)
        
        self.checkpoints.append({
            "step": step,
            "path": checkpoint_dir,
            "metrics": metrics,
        })
        
        self._cleanup_old_checkpoints()
        
        return checkpoint_dir
    
    def _cleanup_old_checkpoints(self):
        if len(self.checkpoints) > self.max_checkpoints:
            oldest = self.checkpoints.pop(0)
            import shutil
            shutil.rmtree(oldest["path"])
    
    def load_best_checkpoint(self, model, metric_name="reward"):
        if not self.checkpoints:
            return model
        
        best = max(self.checkpoints, key=lambda x: x["metrics"].get(metric_name, 0))
        
        from transformers import AutoModelForCausalLM
        return AutoModelForCausalLM.from_pretrained(best["path"])

问题诊断流程：
────────────────────────
1. 检查损失曲线
   ├── 损失剧烈波动 → 降低学习率
   ├── 损失持续上升 → 检查数据质量
   └── 损失不下降 → 检查模型初始化

2. 检查 KL 散度
   ├── KL 突然增大 → 增大 KL 系数
   ├── KL 持续增大 → 使用自适应 KL
   └── KL 为负 → 检查参考模型

3. 检查奖励
   ├── 奖励下降 → 检查奖励模型
   ├── 奖励过高 → 警惕奖励黑客
   └── 奖励不变 → 检查优化流程

4. 检查输出质量
   ├── 输出重复 → 增加熵奖励
   ├── 输出不连贯 → 检查 KL 约束
   └── 输出退化 → 减少训练轮数

识别奖励黑客：
────────────────────────
症状：
├── 奖励很高但输出质量差
├── 输出模式化、重复
├── 特定模式总是得高分
└── 人工评估与奖励不一致

诊断方法：
────────────────────────
1. 人工检查高奖励输出
2. 分析奖励分布
3. 检查奖励模型准确率
4. 对比不同提示的奖励

解决方案：
────────────────────────
1. 增强 KL 约束
2. 更新奖励模型
3. 增加训练数据多样性
4. 使用多个奖励模型集成
5. 添加额外约束

识别能力退化：
────────────────────────
症状：
├── 语言流畅性下降
├── 知识遗忘
├── 指令遵循能力下降
└── 特定任务性能下降

原因分析：
────────────────────────
├── 过度优化偏好
├── KL 约束过松
├── 训练数据分布偏移
└── 训练轮数过多

解决方案：
────────────────────────
1. 增大 KL 约束系数
2. 减少训练轮数
3. 混合预训练数据
4. 使用早停策略
5. 多阶段渐进训练

class MemoryOptimizer:
    @staticmethod
    def enable_gradient_checkpointing(model):
        model.gradient_checkpointing_enable()
        return model
    
    @staticmethod
    def use_mixed_precision(model, use_bf16=True):
        if use_bf16:
            model = model.to(torch.bfloat16)
        else:
            model = model.to(torch.float16)
        return model
    
    @staticmethod
    def optimize_for_training(model, ref_model=None):
        model = MemoryOptimizer.enable_gradient_checkpointing(model)
        model = MemoryOptimizer.use_mixed_precision(model)
        
        if ref_model is not None:
            ref_model.eval()
            for param in ref_model.parameters():
                param.requires_grad = False
        
        return model, ref_model

class TrainingAccelerator:
    @staticmethod
    def setup_distributed():
        import deepspeed
        return deepspeed
    
    @staticmethod
    def optimize_dataloader(dataloader, num_workers=4, pin_memory=True):
        dataloader.num_workers = num_workers
        dataloader.pin_memory = pin_memory
        return dataloader
    
    @staticmethod
    def use_flash_attention(model):
        from flash_attn import replace_model_attention
        return replace_model_attention(model)

class EvaluationPipeline:
    def __init__(self, model, tokenizer, test_data):
        self.model = model
        self.tokenizer = tokenizer
        self.test_data = test_data
    
    def run_full_evaluation(self):
        results = {}
        
        results["perplexity"] = self.evaluate_perplexity()
        results["safety"] = self.evaluate_safety()
        results["instruction_following"] = self.evaluate_instruction_following()
        results["quality"] = self.evaluate_quality()
        
        return results
    
    def evaluate_perplexity(self):
        total_loss = 0
        total_tokens = 0
        
        for item in self.test_data:
            text = item["prompt"] + item["response"]
            inputs = self.tokenizer(text, return_tensors="pt", truncation=True)
            
            with torch.no_grad():
                outputs = self.model(**inputs, labels=inputs["input_ids"])
            
            total_loss += outputs.loss.item() * inputs["input_ids"].shape[1]
            total_tokens += inputs["input_ids"].shape[1]
        
        return {"perplexity": math.exp(total_loss / total_tokens)}
    
    def evaluate_safety(self):
        safety_keywords = ["harmful", "illegal", "dangerous"]
        issues = 0
        
        for item in self.test_data:
            response = self.generate_response(item["prompt"])
            for keyword in safety_keywords:
                if keyword in response.lower():
                    issues += 1
                    break
        
        return {"safety_rate": 1 - issues / len(self.test_data)}

人工评估最佳实践：
────────────────────────
评估维度：
├── 有用性（Helpfulness）
│   └── 回复是否解决了用户问题
│
├── 真实性（Truthfulness）
│   └── 信息是否准确、无幻觉
│
├── 安全性（Safety）
│   └── 是否包含有害内容
│
├── 连贯性（Coherence）
│   └── 语言是否流畅、逻辑清晰
│
└── 指令遵循（Instruction Following）
    └── 是否正确执行指令

评估流程：
────────────────────────
1. 准备评估数据集（100-500 条）
2. 设计评估界面和指南
3. 培训评估人员
4. 进行评估（多人独立评估）
5. 汇总和分析结果
6. 与基准模型对比

class ModelExporter:
    @staticmethod
    def export_for_inference(model, tokenizer, output_dir):
        model.save_pretrained(output_dir)
        tokenizer.save_pretrained(output_dir)
        
        with open(f"{output_dir}/config.json", "w") as f:
            json.dump({
                "model_type": "rlhf",
                "training_method": "ppo",
            }, f)
    
    @staticmethod
    def quantize_model(model, method="int8"):
        if method == "int8":
            from bitsandbytes import quantize_model
            return quantize_model(model)
        elif method == "int4":
            from auto_gptq import quantize
            return quantize(model)
        return model

class InferenceOptimizer:
    @staticmethod
    def setup_vllm(model_path):
        from vllm import LLM
        return LLM(model=model_path)
    
    @staticmethod
    def setup_tensorrt_llm(model_path):
        import tensorrt_llm
        return tensorrt_llm.Runtime(model_path)
    
    @staticmethod
    def batch_generate(model, prompts, max_batch_size=32):
        results = []
        for i in range(0, len(prompts), max_batch_size):
            batch = prompts[i:i + max_batch_size]
            batch_results = model.generate(batch)
            results.extend(batch_results)
        return results