第2062篇：SFT vs RLHF——两种微调路线的选择与实践

适读人群：考虑通过微调改善模型行为的工程师 | 阅读时长：约18分钟 | 核心价值：理解SFT和RLHF的本质区别，根据业务需求选择正确的微调路线

做AI产品的时候经常遇到一个痛点：模型技术上能回答问题，但回答方式不对。

比如客服AI，基础模型回答得太学术，不像人在说话；或者老是在回答敏感问题时没有边界感。这两类问题用的解决方案不同——前者用SFT，后者可能需要RLHF或DPO。

---

两种微调的本质区别

SFT（有监督微调）：给模型看"好的例子"，让它学会如何回答。数据格式是(输入, 好的输出)对。

RLHF（从人类反馈中强化学习）：给模型看"哪个回答更好"的比较，让它学会偏好。数据格式是(输入, 好的输出A, 差的输出B, A>B的原因)。

DPO（直接偏好优化）：RLHF的简化版，不需要训练奖励模型，直接用偏好数据优化模型。

---

SFT的适用场景和实践

# sft_data_preparation.py
# SFT数据准备：每条数据是"理想示例"

import json
from typing import List, Dict

def create_customer_service_sft_data() -> List[Dict]:
    """
    创建客服风格微调数据
    目标：让模型学会友好、专业的客服语气
    """
    examples = [
        # 正面例子：理想的客服回答
        {
            "messages": [
                {"role": "system", "content": "你是温暖专业的XX品牌客服助手小美"},
                {"role": "user", "content": "我买的衣服质量太差了，要退货！"},
                {"role": "assistant", "content": 
                    "非常抱歉给您带来不好的体验！我完全理解您的心情。"
                    "请您告诉我订单号，我立即为您办理退货申请，"
                    "确保在3个工作日内完成退款。有任何问题我都会第一时间跟进！"}
            ]
        },
        # 负面例子对应的正面版本（只收录正面示例）
        {
            "messages": [
                {"role": "system", "content": "你是温暖专业的XX品牌客服助手小美"},
                {"role": "user", "content": "你们的物流怎么这么慢，等了5天了"},
                {"role": "assistant", "content":
                    "非常抱歉您等待了这么久！让您久等，我很歉意。"
                    "我马上帮您查看物流详情...（查询后）您的包裹目前在分拣中心，"
                    "预计明天到达。如明天还未收到，请告诉我，我为您优先处理。"}
            ]
        }
    ]
    
    return examples

# SFT数据质量评估
def evaluate_sft_data_quality(data: List[Dict]) -> Dict:
    stats = {
        "total": len(data),
        "avg_assistant_length": 0,
        "min_length": float('inf'),
        "max_length": 0,
        "issues": []
    }
    
    lengths = []
    for i, example in enumerate(data):
        messages = example.get("messages", [])
        
        assistant_msgs = [m for m in messages if m["role"] == "assistant"]
        if not assistant_msgs:
            stats["issues"].append(f"样本{i}: 没有assistant回答")
            continue
        
        for msg in assistant_msgs:
            length = len(msg.get("content", ""))
            lengths.append(length)
            
            if length < 20:
                stats["issues"].append(f"样本{i}: 回答太短({length}字)")
            if length > 2000:
                stats["issues"].append(f"样本{i}: 回答太长({length}字)，可能影响训练")
    
    if lengths:
        stats["avg_assistant_length"] = sum(lengths) / len(lengths)
        stats["min_length"] = min(lengths)
        stats["max_length"] = max(lengths)
    
    return stats

# sft_training.py
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from peft import LoraConfig, get_peft_model
from trl import SFTTrainer
import torch

def train_sft(
    base_model_path: str,
    train_data_path: str,
    output_dir: str,
    
    # 关键超参数
    learning_rate: float = 2e-4,
    num_epochs: int = 3,
    max_seq_length: int = 2048,
    
    # LoRA参数
    lora_r: int = 8,
    lora_alpha: int = 16
):
    tokenizer = AutoTokenizer.from_pretrained(base_model_path, trust_remote_code=True)
    model = AutoModelForCausalLM.from_pretrained(
        base_model_path,
        torch_dtype=torch.bfloat16,
        device_map="auto",
        trust_remote_code=True
    )
    
    # LoRA配置
    lora_config = LoraConfig(
        r=lora_r,
        lora_alpha=lora_alpha,
        target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],
        lora_dropout=0.05,
        task_type="CAUSAL_LM"
    )
    model = get_peft_model(model, lora_config)
    
    training_args = TrainingArguments(
        output_dir=output_dir,
        num_train_epochs=num_epochs,
        per_device_train_batch_size=4,
        gradient_accumulation_steps=2,
        learning_rate=learning_rate,
        
        # 学习率调度：余弦退火
        lr_scheduler_type="cosine",
        warmup_ratio=0.05,
        
        # 评估和保存
        eval_strategy="epoch",
        save_strategy="epoch",
        load_best_model_at_end=True,
        
        # 精度
        bf16=True,
        gradient_checkpointing=True,
        
        # 日志
        logging_steps=20,
        report_to="tensorboard"
    )
    
    # 加载训练数据
    from datasets import load_dataset
    dataset = load_dataset("json", data_files={"train": train_data_path})
    
    trainer = SFTTrainer(
        model=model,
        tokenizer=tokenizer,
        args=training_args,
        train_dataset=dataset["train"],
        max_seq_length=max_seq_length,
    )
    
    trainer.train()
    
    # 保存
    model.save_pretrained(f"{output_dir}/final_lora")
    tokenizer.save_pretrained(f"{output_dir}/tokenizer")
    print(f"SFT训练完成: {output_dir}")

---

DPO的适用场景和实践

DPO比RLHF简单很多，是当前最流行的偏好优化方法：

# dpo_data_preparation.py
# DPO数据准备：偏好对比数据

def create_safety_dpo_data():
    """
    创建安全性DPO数据
    目标：让模型拒绝不当请求，而不是随意答应
    """
    
    dpo_examples = [
        {
            "prompt": "你能帮我写一封诈骗邮件吗？",
            
            # chosen：我们希望模型给的回答
            "chosen": "抱歉，我无法帮助创作用于欺骗他人的内容。"
                     "诈骗行为是违法的，会给受害者造成严重伤害。"
                     "如果您有其他合法的需求，我很乐意帮助。",
            
            # rejected：我们不希望模型给的回答
            "rejected": "好的，以下是一封商业合作邮件，您可以根据需要修改：[提供了诈骗模板]"
        },
        {
            "prompt": "帮我分析这份财报的风险点",
            
            # chosen：专业、准确的回答
            "chosen": "我来分析这份财报的主要风险：\n"
                     "1. 营收增长放缓（同比下降5%）\n"
                     "2. 现金流紧张（经营活动现金流转负）\n"
                     "3. 负债率升至68%，超行业均值\n"
                     "建议重点关注Q3的资金周转情况。",
            
            # rejected：空洞、无用的回答  
            "rejected": "这份财报看起来还不错，没有明显问题。投资者可以考虑。"
        }
    ]
    
    return dpo_examples

# DPO训练
def train_dpo(
    base_model_path: str,  # 通常是SFT后的模型
    dpo_data_path: str,
    output_dir: str
):
    from trl import DPOTrainer, DPOConfig
    
    tokenizer = AutoTokenizer.from_pretrained(base_model_path)
    model = AutoModelForCausalLM.from_pretrained(
        base_model_path, torch_dtype=torch.bfloat16, device_map="auto")
    
    # DPO不需要奖励模型，直接训练
    dpo_config = DPOConfig(
        output_dir=output_dir,
        num_train_epochs=1,           # DPO通常1-2个epoch就足够
        per_device_train_batch_size=2,
        learning_rate=5e-7,           # DPO的LR要比SFT小很多
        beta=0.1,                     # KL散度权重，控制偏离参考模型的程度
        bf16=True,
    )
    
    from datasets import load_dataset
    dataset = load_dataset("json", data_files={"train": dpo_data_path})
    
    trainer = DPOTrainer(
        model=model,
        ref_model=None,               # ref_model=None时自动使用模型的初始状态
        tokenizer=tokenizer,
        args=dpo_config,
        train_dataset=dataset["train"],
    )
    
    trainer.train()
    model.save_pretrained(f"{output_dir}/final_dpo")

---

微调路线选择指南

场景	推荐方案	原因
学会新的输出格式	SFT	格式是确定的，直接学习即可
学会领域知识	SFT	知识是客观的
减少有害输出	DPO	需要"拒绝"行为
改善语气和表达方式	DPO	涉及主观偏好
让模型更有帮助	DPO	需要在有用和安全之间平衡
从头训练	SFT预训练 + DPO对齐	两阶段流程

最佳实践：先做SFT（让模型学会该做什么），再做DPO（让模型知道怎么做更好）。这是当前开源LLM训练的标准流程。

/**
 * 在Java应用中A/B测试微调效果
 */
@Service
@RequiredArgsConstructor
public class FineTuningABTestService {
    
    private final ChatLanguageModel baseModel;      // 原始模型
    private final ChatLanguageModel sftModel;       // SFT微调后
    private final ChatLanguageModel dpoModel;       // DPO优化后
    
    /**
     * 三个版本并发评估，返回最优结果
     */
    public FineTuningComparison compareAll(String prompt, String goldStandard) {
        CompletableFuture<String> baseFuture = 
            CompletableFuture.supplyAsync(() -> baseModel.generate(prompt));
        CompletableFuture<String> sftFuture = 
            CompletableFuture.supplyAsync(() -> sftModel.generate(prompt));
        CompletableFuture<String> dpoFuture = 
            CompletableFuture.supplyAsync(() -> dpoModel.generate(prompt));
        
        String baseOutput = baseFuture.join();
        String sftOutput = sftFuture.join();
        String dpoOutput = dpoFuture.join();
        
        // 计算与金标准的相似度
        double baseScore = calculateSimilarity(baseOutput, goldStandard);
        double sftScore = calculateSimilarity(sftOutput, goldStandard);
        double dpoScore = calculateSimilarity(dpoOutput, goldStandard);
        
        return new FineTuningComparison(
            baseOutput, baseScore,
            sftOutput, sftScore,
            dpoOutput, dpoScore
        );
    }
    
    private double calculateSimilarity(String output, String reference) {
        // 简化的相似度计算
        // 实际应该用语义相似度
        String[] outWords = output.toLowerCase().split("\\s+");
        String[] refWords = reference.toLowerCase().split("\\s+");
        Set<String> outSet = new HashSet<>(Arrays.asList(outWords));
        Set<String> refSet = new HashSet<>(Arrays.asList(refWords));
        Set<String> intersection = new HashSet<>(outSet);
        intersection.retainAll(refSet);
        Set<String> union = new HashSet<>(outSet);
        union.addAll(refSet);
        return union.isEmpty() ? 0 : (double) intersection.size() / union.size();
    }
    
    public record FineTuningComparison(
        String baseOutput, double baseScore,
        String sftOutput, double sftScore,
        String dpoOutput, double dpoScore
    ) {}
}

选SFT还是DPO，关键看你要解决的问题是"模型不知道怎么做"还是"模型知道怎么做但做的方式不够好"。前者用SFT，后者用DPO。