关于大模型微调进阶的几点记录

大模型微调 RLHF 指令微调奖励建模

作者: 醉月思发布日期: 2024-07-10 分类: 技术实践

我第一个项目是一个代码助手微调，用 Python 代码片段和注释做训练数据，大概 5 万条。

简单说就是：

输入输出对应关系明确，没有歧义
输出内容完整、准确、有逻辑
没有重复或极度相似的样本
涵盖主要的使用场景，但不要过度细分

为什么指令微调不是终点

去年刚开始接触大模型微调时，我以为把指令微调做完就完事了。毕竟官方文档和各种教程都把 SFT（Supervised Fine-Tuning）说成是必经之路，而且效果确实肉眼可见。

但问题很快就暴露了。

我第一个项目是一个代码助手微调，用 Python 代码片段和注释做训练数据，大概 5 万条。微调后的模型确实能写代码了，但写出来的代码风格很不稳定：有时候注释很详细，有时候完全没注释；有时候用类型提示，有时候直接裸写。

这不是数据质量的问题。我回头检查过，训练数据的风格其实挺一致的。真正的问题是：指令微调只是教会了模型"什么是正确的"，但没有教它"什么是更好的"。

换句话讲，SFT 是一个二分类问题——对或者错；但我们真正需要的是一个排序问题——在多个正确的答案里，哪个更优。

这时候就引入了奖励建模的概念。

指令微调的实践和坑

在聊奖励建模之前，先说说指令微调本身的坑。这些坑后来在我做奖励建模时又踩了一遍。

数据质量大于数据量

这是最痛的教训。

我一开始觉得"越多越好"，拼了命地扩充训练数据。从一个 2 万条的数据集开始，最后搞到 15 万条。结果呢？数据多了，效果反而变差了。

后来分析发现，后期扩充的数据里混进了大量低质量样本：有些示例的输出根本不匹配输入，有些是自动生成的注释但质量很差，还有些是重复的指令只是换了种说法。

我的经验是：宁可要 2 万条高质量数据，也不要 15 万条混合质量数据。什么叫高质量？简单说就是：

输入输出对应关系明确，没有歧义
输出内容完整、准确、有逻辑
没有重复或极度相似的样本
涵盖主要的使用场景，但不要过度细分

一个简单的数据清洗脚本可以帮你过滤掉大部分问题：

import json
from pathlib import Path
from typing import List, Dict
import hashlib

def load_dataset(file_path: str) -> List[Dict]:
    with open(file_path, 'r', encoding='utf-8') as f:
        return [json.loads(line) for line in f]

def calculate_similarity(text1: str, text2: str) -> float:
    # 简单的相似度计算，实际可以用更复杂的算法
    set1 = set(text1.split())
    set2 = set(text2.split())
    intersection = set1 & set2
    union = set1 | set2
    return len(intersection) / len(union) if union else 0

def filter_duplicates(dataset: List[Dict], threshold: float = 0.85) -> List[Dict]:
    seen = set()
    filtered = []

    for item in dataset:
        # 使用输入+输出做唯一性判断
        content = item['input'] + item['output']
        content_hash = hashlib.md5(content.encode()).hexdigest()

        if content_hash not in seen:
            seen.add(content_hash)
            filtered.append(item)

    return filtered

def filter_similar(dataset: List[Dict], threshold: float = 0.9) -> List[Dict]:
    filtered = []
    for item in dataset:
        is_similar = False
        for existing_item in filtered:
            sim = calculate_similarity(item['input'], existing_item['input'])
            if sim > threshold:
                is_similar = True
                break
        if not is_similar:
            filtered.append(item)
    return filtered

def validate_item(item: Dict) -> bool:
    # 基本验证：字段完整、长度合理、非空
    if 'input' not in item or 'output' not in item:
        return False

    if not item['input'].strip() or not item['output'].strip():
        return False

    if len(item['output']) < 10 or len(item['output']) > 4000:
        return False

    return True

def main():
    dataset = load_dataset('raw_data.jsonl')

    # 1. 基本验证
    dataset = [item for item in dataset if validate_item(item)]
    print(f"After validation: {len(dataset)}")

    # 2. 去重
    dataset = filter_duplicates(dataset)
    print(f"After deduplication: {len(dataset)}")

    # 3. 去相似
    dataset = filter_similar(dataset)
    print(f"After similarity filtering: {len(dataset)}")

    # 保存清理后的数据
    with open('cleaned_data.jsonl', 'w', encoding='utf-8') as f:
        for item in dataset:
            f.write(json.dumps(item, ensure_ascii=False) + '\n')

if __name__ == '__main__':
    main()

这个脚本在实际项目中帮我过滤掉了大概 40% 的数据，但模型效果提升明显。

训练参数的摸索

指令微调的训练参数也是个坑。一开始我直接用了 LoRA 的默认配置，结果微调出来的模型要么"记住了训练数据但泛化能力差"，要么"完全没学到东西"。

后来摸索出一套相对稳定的参数组合（基于 LLaMA-7B）：

# 训练命令示例
python train.py \
    --model_name_or_path meta-llama/Llama-2-7b-hf \
    --data_path cleaned_data.jsonl \
    --bf16 True \
    --output_dir output/sft_model \
    --num_train_epochs 3 \
    --per_device_train_batch_size 4 \
    --per_device_eval_batch_size 4 \
    --gradient_accumulation_steps 8 \
    --evaluation_strategy "no" \
    --save_strategy "steps" \
    --save_steps 1000 \
    --save_total_limit 2 \
    --learning_rate 2e-5 \
    --weight_decay 0. \
    --warmup_ratio 0.03 \
    --lr_scheduler_type "cosine" \
    --logging_steps 10 \
    --fsdp "full_shard auto_wrap" \
    --fsdp_transformer_layer_cls_to_wrap 'LlamaDecoderLayer' \
    --tf32 True \
    --model_max_length 2048 \
    --gradient_checkpointing True \
    --lora_r 8 \
    --lora_alpha 16 \
    --lora_dropout 0.05 \
    --lora_target_modules q_proj v_proj k_proj o_proj

这里有几个关键点：

学习率：我一开始用 1e-4，结果损失波动很大。调到 2e-5 后就稳定多了。
LoRA 参数：r=8 和 alpha=16 是比较保守的配置，适合任务相对明确的场景。如果任务复杂，可以适当调高。
批次大小：per_device_train_batch_size=4 加上 gradient_accumulation_steps=8，实际批次大小是 32。这个规模在 8 张 A100 上跑起来比较平衡。

过拟合的识别和避免

过拟合是指令微调中最常见的问题。我有一次微调一个客服机器人，训练集上的准确率很高，但在真实用户测试时表现很差。

识别过拟合的方法其实很简单：留一个验证集，观察训练损失和验证损失的变化曲线。

import matplotlib.pyplot as plt
import json

def plot_loss_curves(log_file: str):
    with open(log_file, 'r') as f:
        logs = [json.loads(line) for line in f]

    train_losses = [log['train_loss'] for log in logs if 'train_loss' in log]
    eval_losses = [log['eval_loss'] for log in logs if 'eval_loss' in log]

    plt.figure(figsize=(10, 6))
    plt.plot(train_losses, label='Train Loss')
    plt.plot(eval_losses, label='Eval Loss')
    plt.xlabel('Steps')
    plt.ylabel('Loss')
    plt.title('Training and Evaluation Loss')
    plt.legend()
    plt.savefig('loss_curves.png')

如果训练损失持续下降，但验证损失开始上升或者持平，基本就是过拟合了。

避免过拟合的几个实用技巧：

早停机制：监控验证损失，连续几个 epoch 不下降就停止训练。
数据增强：对输入做一些合理的变换，比如同义词替换、句式调整等。
正则化：适当增加 weight_decay 或者使用 dropout。
增加数据量：如果条件允许，高质量的数据是解决过拟合最有效的方法。

奖励建模的复杂性

解决了指令微调的问题后，我发现模型能完成任务，但完成任务的质量参差不齐。这时候就引入了奖励建模。

奖励模型的数据构造

奖励模型的核心是"偏好数据"——即对同一个输入，给出两个不同的输出，让人标注哪个更好。

一开始我觉得这个很简单，随便找几个同事标注就行了。但现实给了我一巴掌：

标注标准很难统一：有人觉得详细好，有人觉得简洁好
标注成本高：一条偏好数据可能需要 2-3 分钟来仔细对比
标注质量难以保证：疲劳、时间压力都会影响标注质量

后来我们设计了一套标注规范，并且引入了"标注者一致性检查"：

def calculate_agreement(annotations: List[Dict]) -> Dict:
    """计算标注者一致性"""
    # annotations 结构: [{'user_id': 1, 'item_id': 1, 'preference': 'A'}, ...]

    # 计算相同 item_id 的标注一致性
    item_groups = {}
    for ann in annotations:
        if ann['item_id'] not in item_groups:
            item_groups[ann['item_id']] = []
        item_groups[ann['item_id']].append(ann['preference'])

    agreement_scores = {}
    for item_id, preferences in item_groups.items():
        if len(preferences) < 2:
            continue

        # 计算多数类占比
        from collections import Counter
        counts = Counter(preferences)
        majority_ratio = counts.most_common(1)[0][1] / len(preferences)
        agreement_scores[item_id] = majority_ratio

    avg_agreement = sum(agreement_scores.values()) / len(agreement_scores) if agreement_scores else 0

    return {
        'item_agreements': agreement_scores,
        'average_agreement': avg_agreement
    }

如果平均一致性低于 0.7，我们就重新组织标注培训，或者调整标注指南。

奖励模型的训练

奖励模型的训练比指令微调更复杂一些。它需要学习的是"相对偏好"而不是"绝对正确性"。

一个典型的奖励模型训练代码框架：

import torch
import torch.nn as nn
from transformers import AutoModelForSequenceClassification, AutoTokenizer

class RewardModel(nn.Module):
    def __init__(self, model_name_or_path: str):
        super().__init__()
        self.model = AutoModelForSequenceClassification.from_pretrained(
            model_name_or_path,
            num_labels=1
        )
        self.tokenizer = AutoTokenizer.from_pretrained(model_name_or_path)

    def forward(self, input_ids, attention_mask):
        outputs = self.model(input_ids=input_ids, attention_mask=attention_mask)
        return outputs.logits.squeeze(-1)

def compute_ranking_loss(reward_scores_A, reward_scores_B, margin=0.1):
    """
    计算排序损失
    如果 A 应该比 B 好，那么 reward_A 应该大于 reward_B + margin
    """
    return torch.clamp(margin + reward_scores_B - reward_scores_A, min=0).mean()

def train_reward_model(train_dataloader, model, optimizer, device):
    model.train()
    total_loss = 0

    for batch in train_dataloader:
        # batch 包含: input_ids_A, attention_mask_A, input_ids_B, attention_mask_B
        input_ids_A = batch['input_ids_A'].to(device)
        attention_mask_A = batch['attention_mask_A'].to(device)
        input_ids_B = batch['input_ids_B'].to(device)
        attention_mask_B = batch['attention_mask_B'].to(device)

        # 前向传播
        reward_A = model(input_ids_A, attention_mask_A)
        reward_B = model(input_ids_B, attention_mask_B)

        # 计算损失
        loss = compute_ranking_loss(reward_A, reward_B)

        # 反向传播
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

        total_loss += loss.item()

    return total_loss / len(train_dataloader)

这里的关键是损失函数的设计。我用的是 margin-based ranking loss，确保模型能学到"相对更好"的概念。

奖励模型的评估

奖励模型的评估比指令微调难。你不能只看准确率，因为你没法用"标准答案"来衡量。

我实际使用的评估方法有几个：

与人类偏好的一致性：保留一部分标注数据作为测试集，看模型的预测是否与人类标注一致。
成对准确率：给定一对 (A, B) 输出，模型能否正确判断哪个更好。
分数分布：检查模型给出的分数分布是否合理，不应该出现极端的分数堆积。

def evaluate_reward_model(model, test_dataloader, device):
    model.eval()
    correct = 0
    total = 0
    score_distribution = []

    with torch.no_grad():
        for batch in test_dataloader:
            input_ids_A = batch['input_ids_A'].to(device)
            attention_mask_A = batch['attention_mask_A'].to(device)
            input_ids_B = batch['input_ids_B'].to(device)
            attention_mask_B = batch['attention_mask_B'].to(device)
            preference = batch['preference'].to(device)  # 1 表示 A 好，0 表示 B 好

            reward_A = model(input_ids_A, attention_mask_A)
            reward_B = model(input_ids_B, attention_mask_B)

            # 模型预测
            predicted_preference = (reward_A > reward_B).float()

            # 统计准确率
            correct += (predicted_preference == preference).sum().item()
            total += preference.size(0)

            # 收集分数用于分布分析
            score_distribution.extend(reward_A.cpu().tolist())
            score_distribution.extend(reward_B.cpu().tolist())

    accuracy = correct / total if total > 0 else 0

    return {
        'accuracy': accuracy,
        'score_distribution': score_distribution
    }

RLHF 的落地挑战

有了指令微调模型和奖励模型，下一步就是 RLHF（Reinforcement Learning from Human Feedback）。这是整个链条中最复杂的一环。

PPO 的实现细节

PPO (Proximal Policy Optimization) 是 RLHF 中最常用的算法。它的实现细节很多，坑也很多。

一个简化版的 PPO 训练循环：

import torch
import torch.nn.functional as F
from transformers import AutoTokenizer

class PPOTrainer:
    def __init__(self, policy_model, ref_model, reward_model, tokenizer, kl_coef=0.1):
        self.policy_model = policy_model
        self.ref_model = ref_model
        self.reward_model = reward_model
        self.tokenizer = tokenizer
        self.kl_coef = kl_coef

    def generate_responses(self, prompts):
        """使用策略模型生成响应"""
        responses = []
        for prompt in prompts:
            inputs = self.tokenizer(prompt, return_tensors='pt')
            outputs = self.policy_model.generate(
                **inputs,
                max_length=512,
                do_sample=True,
                temperature=0.7,
                top_p=0.9
            )
            response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
            responses.append(response[len(prompt):])  # 去掉 prompt 部分
        return responses

    def compute_rewards(self, prompts, responses):
        """使用奖励模型计算奖励"""
        rewards = []
        for prompt, response in zip(prompts, responses):
            text = prompt + response
            inputs = self.tokenizer(text, return_tensors='pt')
            with torch.no_grad():
                reward = self.reward_model(**inputs)
            rewards.append(reward.item())
        return rewards

    def compute_kl_penalty(self, prompts, responses):
        """计算 KL 散度惩罚"""
        kl_penalties = []
        for prompt, response in zip(prompts, responses):
            text = prompt + response
            inputs = self.tokenizer(text, return_tensors='pt')

            # 策略模型的 logits
            policy_logits = self.policy_model(**inputs).logits
            # 参考模型的 logits
            with torch.no_grad():
                ref_logits = self.ref_model(**inputs).logits

            # 计算 KL 散度
            policy_probs = F.softmax(policy_logits, dim=-1)
            ref_probs = F.softmax(ref_logits, dim=-1)
            kl = F.kl_div(
                F.log_softmax(policy_logits, dim=-1),
                ref_probs,
                reduction='batchmean'
            )
            kl_penalties.append(kl.item())
        return kl_penalties

    def train_step(self, prompts, old_responses, old_log_probs, advantages):
        """PPO 训练步骤"""
        # 生成新的响应
        new_responses = self.generate_responses(prompts)

        # 计算新的 log probs
        new_log_probs = []
        for prompt, response in zip(prompts, new_responses):
            text = prompt + response
            inputs = self.tokenizer(text, return_tensors='pt')
            logits = self.policy_model(**inputs).logits
            probs = F.softmax(logits, dim=-1)
            log_probs = torch.log(probs).squeeze()
            new_log_probs.append(log_probs[-1].item())  # 简化处理，实际需要完整序列

        # 计算比率
        ratio = torch.exp(torch.tensor(new_log_probs) - torch.tensor(old_log_probs))

        # PPO 损失
        surr1 = ratio * advantages
        surr2 = torch.clamp(ratio, 1 - 0.2, 1 + 0.2) * advantages
        policy_loss = -torch.min(surr1, surr2).mean()

        # KL 惩罚
        kl_penalties = self.compute_kl_penalty(prompts, new_responses)
        kl_loss = self.kl_coef * torch.tensor(kl_penalties).mean()

        # 总损失
        total_loss = policy_loss + kl_loss

        return total_loss

训练稳定性的问题

RLHF 的最大问题是训练不稳定。我遇到过几次明明训练损失在下降，但模型效果反而变差的情况。

解决稳定性问题的几个经验：

学习率要很小：我用的是 1e-6 到 1e-7 这个级别，比指令微调小两个数量级。
梯度裁剪很重要：防止梯度爆炸，我用的裁剪阈值是 1.0。
KL 散度要监控：如果 KL 散度增长太快，说明模型偏离参考模型太远，需要调整 kl_coef。
奖励归一化：对奖励做标准化处理，避免奖励值过大导致训练不稳定。

# 奖励归一化的实现
def normalize_rewards(rewards):
    rewards = torch.tensor(rewards)
    mean = rewards.mean()
    std = rewards.std()
    normalized = (rewards - mean) / (std + 1e-8)
    return normalized.tolist()

计算资源的权衡

RLHF 的计算开销非常大。相比指令微调，RLHF 需要同时维护三个模型：策略模型、参考模型、奖励模型。

我的资源分配经验：

策略模型：需要梯度更新，占用 GPU 显存最多
参考模型：不需要梯度，可以放在 CPU 或低显存 GPU 上
奖励模型：需要频繁推理，最好放在 GPU 上

一个合理的资源分配方案：

# 在多 GPU 环境下分配模型
device_policy = 'cuda:0'  # 主 GPU，训练策略模型
device_ref = 'cuda:1'     # 次要 GPU，参考模型
device_reward = 'cuda:0'  # 与策略模型共享 GPU，轮流使用

policy_model = policy_model.to(device_policy)
ref_model = ref_model.to(device_ref)
reward_model = reward_model.to(device_reward)

如果资源有限，可以考虑这几个优化方向：

使用更小的批次大小，但增加梯度累积步数
使用混合精度训练 (FP16/BF16)
使用模型并行或梯度检查点
考虑使用 DPO (Direct Preference Optimization) 替代 PPO，计算开销更小

对齐技术的实用判断

从指令微调到奖励建模再到 RLHF，整个链条走下来，你会发现"对齐"这个概念比想象中复杂。

什么时候需要 RLHF

不是所有场景都需要 RLHF。我的判断标准是：

任务的主观性强：比如创意写作、对话生成，“好坏"标准不明确
需要细粒度优化：不是对错问题，而是更好更坏的问题
有足够的标注资源：RLHF 需要大量高质量的人工标注

如果只是让模型完成一些明确的任务（比如代码生成、信息抽取），指令微调往往就足够了。

DPO 作为简化方案

最近 DPO (Direct Preference Optimization) 越来越受欢迎，因为它省略了显式的奖励模型，直接用偏好数据训练策略模型。

DPO 的实现比 PPO 简单很多：

def compute_dpo_loss(policy_logits, ref_logits, chosen_mask, rejected_mask, beta=0.1):
    """
    DPO 损失计算
    policy_logits: 策略模型对 (chosen, rejected) 的 log probs
    ref_logits: 参考模型对 (chosen, rejected) 的 log probs
    """
    # 计算 log ratio
    log_ratio = policy_logits - ref_logits

    # 提取 chosen 和 rejected 的 log probs
    chosen_logps = log_ratio[chosen_mask]
    rejected_logps = log_ratio[rejected_mask]

    # 计算 DPO 损失
    loss = -F.logsigmoid(beta * (chosen_logps - rejected_logps)).mean()

    return loss

DPO 的优势在于：

不需要单独训练奖励模型
训练更稳定，不涉及复杂的 RL 超参数调优
计算开销更小

但 DPO 也有局限：它更依赖于高质量的偏好数据，而且对数据的分布比较敏感。

实际项目的选择建议

基于我的实践经验，给一个简单的选择建议：

小团队、资源有限：从指令微调开始，如果效果不够好，再考虑 DPO
有足够标注资源、追求最优效果：可以考虑完整的 RLHF 流程
任务相对明确：指令微调 + 一些后处理规则可能就够了
任务高度主观、需要精细控制：RLHF 是值得投入的方向

实践中的最终判断

走完这一整套流程，我对"微调"有了更实际的理解。

微调不是魔法，它更多是在数据上做精细的工程。好的微调效果，往往来自于：

对任务场景的深刻理解
高质量、有针对性的数据准备
合理的模型选择和参数调优
持续的监控和迭代

我也学会了不迷信"最新最好"的技术。有时候一个简单但扎实实现的指令微调，比一个复杂但不稳定的 RLHF 更有实用价值。

技术选型要看场景和资源，而不是跟着论文走。这个道理在项目实践中一次又一次地被验证。

路还在往前延伸，从 SFT 到 RLHF，再到现在的各种对齐技术。但回头想想，核心问题其实一直没变：如何让模型真正理解我们要它做什么，并且做得好。

剩下的就是在实践中慢慢摸索，找到适合自己场景的那条路。

版权声明: 本文首发于指尖魔法屋-关于大模型微调进阶的几点记录（https://blog.thinkmoon.cn/post/216-llm-finetuning-from-sft-to-rhf/）转载或引用必须申明原指尖魔法屋来源及源地址！

AI人类反馈强化学习实践笔记

在这次实践之前，我的模型训练经验主要停留在监督学习阶段：准备数据、训练、评估。

📁 技术实践

📅 2023-11-11

迁移学习：这次怎么落地的

很多人一上来就讲迁移学习的全景图；我更想先把这次卡住的点说清楚。

📁 技术实践

📅 2023-04-04

AI 迁移学习踩坑记录

去年接了一个中文医疗问答分类的项目，数据量只有 3000 条，标注质量一般，模型效果一直上不去。

📁 技术实践

📅 2023-07-18

关于大模型微调进阶的几点记录

为什么指令微调不是终点

指令微调的实践和坑

数据质量大于数据量

训练参数的摸索

过拟合的识别和避免

奖励建模的复杂性

奖励模型的数据构造

奖励模型的训练

奖励模型的评估

RLHF 的落地挑战

PPO 的实现细节

训练稳定性的问题

计算资源的权衡

对齐技术的实用判断

什么时候需要 RLHF

DPO 作为简化方案

实际项目的选择建议

实践中的最终判断

评论

相关阅读