指令微调指南：从 SFT 到 DPO

对齐技术全景

预训练模型 (Base)
    ↓
SFT (Supervised Fine-Tuning)     ← 监督微调，学习指令遵循
    ↓
DPO / RLHF                       ← 偏好对齐，学习人类价值观
    ↓
对齐模型 (Chat/Instruct)

阶段	目标	数据	方法
SFT	学会跟随指令	(指令, 回答) 对	监督学习
RLHF	对齐人类偏好	(prompt, chosen, rejected)	强化学习
DPO	对齐人类偏好	同上	直接优化，无需 RL

SFT：监督微调

数据格式

Alpaca 格式：

{
    "instruction": "将以下句子翻译为英文",
    "input": "今天天气真好",
    "output": "The weather is really nice today."
}

ShareGPT 格式（多轮对话）：

{
    "conversations": [
        {"from": "human", "value": "你好"},
        {"from": "gpt", "value": "你好！有什么可以帮你的？"},
        {"from": "human", "value": "解释一下量子计算"},
        {"from": "gpt", "value": "量子计算利用量子力学原理..."}
    ]
}

数据构造

import json
import random

class SFTDataBuilder:
    def __init__(self, llm_call):
        self.llm = llm_call
    
    def generate_alpaca_style(self, seed_topics: list[str], 
                                n_samples: int = 1000):
        """用 LLM 自动生成 SFT 数据"""
        dataset = []
        
        for topic in seed_topics:
            prompt = f"""生成 10 条多样化的指令-回答对。
主题：{topic}

要求：
1. 指令类型多样（问答、翻译、摘要、代码、推理等）
2. 难度从简单到复杂
3. 回答准确、自然

输出 JSON 数组格式：
[{{"instruction": "", "input": "", "output": ""}}]"""

            result = self.llm(prompt)
            try:
                items = json.loads(result)
                dataset.extend(items)
            except json.JSONDecodeError:
                continue
        
        # 去重 + 打乱
        seen = set()
        unique = []
        for item in dataset:
            key = item["instruction"] + item.get("input", "")
            if key not in seen:
                seen.add(key)
                unique.append(item)
        
        random.shuffle(unique)
        return unique[:n_samples]
    
    def to_sharegpt(self, alpaca_data: list[dict]) -> list[dict]:
        """Alpaca → ShareGPT 格式转换"""
        converted = []
        for item in alpaca_data:
            instruction = item["instruction"]
            if item.get("input"):
                instruction = f"{instruction}\n\n输入: {item['input']}"
            
            converted.append({
                "conversations": [
                    {"from": "human", "value": instruction},
                    {"from": "gpt", "value": item["output"]},
                ]
            })
        return converted

训练配置

from transformers import (
    AutoModelForCausalLM,
    AutoTokenizer,
    TrainingArguments,
    Trainer,
)
from datasets import Dataset
from peft import LoraConfig, get_peft_model, TaskType
import torch

class SFTTrainer:
    def __init__(self, model_path: str, use_lora: bool = True):
        self.tokenizer = AutoTokenizer.from_pretrained(model_path)
        self.tokenizer.pad_token = self.tokenizer.eos_token
        
        self.model = AutoModelForCausalLM.from_pretrained(
            model_path,
            torch_dtype=torch.bfloat16,
            device_map="auto",
        )
        
        if use_lora:
            self._setup_lora()
    
    def _setup_lora(self):
        """配置 LoRA：只训练少量参数"""
        lora_config = LoraConfig(
            task_type=TaskType.CAUSAL_LM,
            r=16,              # LoRA rank
            lora_alpha=32,     # 缩放因子
            lora_dropout=0.05,
            target_modules=[
                "q_proj", "k_proj", "v_proj", "o_proj",
                "gate_proj", "up_proj", "down_proj",
            ],
        )
        self.model = get_peft_model(self.model, lora_config)
        self.model.print_trainable_parameters()
        # 输出: trainable params: 13M || all params: 7B || trainable%: 0.19%
    
    def format_prompt(self, instruction: str, input_text: str = "", 
                      output: str = "") -> str:
        """格式化为 ChatML 模板"""
        user_msg = instruction
        if input_text:
            user_msg += f"\n\n{input_text}"
        
        prompt = f"<|im_start|>user\n{user_msg}<|im_end|>\n"
        if output:
            prompt += f"<|im_start|>assistant\n{output}<|im_end|>"
        return prompt
    
    def train(self, data: list[dict], 
              learning_rate: float = 2e-5,
              num_epochs: int = 3,
              batch_size: int = 4,
              max_length: int = 1024):
        # 格式化
        texts = [
            self.format_prompt(
                d["instruction"], d.get("input", ""), d["output"]
            )
            for d in data
        ]
        
        dataset = Dataset.from_dict({"text": texts})
        
        def tokenize(examples):
            tokenized = self.tokenizer(
                examples["text"],
                truncation=True,
                max_length=max_length,
                padding="max_length",
            )
            tokenized["labels"] = tokenized["input_ids"].copy()
            return tokenized
        
        dataset = dataset.map(tokenize, batched=True)
        
        training_args = TrainingArguments(
            output_dir="./sft_output",
            num_train_epochs=num_epochs,
            per_device_train_batch_size=batch_size,
            gradient_accumulation_steps=4,
            learning_rate=learning_rate,
            warmup_ratio=0.03,
            lr_scheduler_type="cosine",
            logging_steps=10,
            save_strategy="epoch",
            bf16=True,
            optim="adamw_torch",
        )
        
        trainer = Trainer(
            model=self.model,
            args=training_args,
            train_dataset=dataset,
        )
        
        trainer.train()

LoRA 超参数选择

参数	推荐值	说明
r (rank)	8-64	越大能力越强，但过拟合风险
lora_alpha	2×r	经验值，平衡缩放
target_modules	all linear	全覆盖效果最好
dropout	0.05	防过拟合
learning_rate	1e-4 ~ 5e-5	LoRA 可用较大学习率

DPO：直接偏好优化

DPO (Rafailov et al., 2023) 是 RLHF 的简化替代——无需训练奖励模型，无需强化学习：

核心原理

RLHF: SFT → 训练奖励模型 → PPO 优化策略 (复杂)
DPO:  SFT → 直接用偏好数据优化 (简单)

DPO 的损失函数直接利用偏好对：

L_DPO = -E[log σ(β · (log π(y_w|x)/π_ref(y_w|x) - log π(y_l|x)/π_ref(y_l|x)))]

其中 y_w 是偏好回答，y_l 是不偏好回答，π_ref 是参考模型。

DPO 实现

import torch
import torch.nn.functional as F
from transformers import AutoModelForCausalLM
from copy import deepcopy

class DPOTrainer:
    def __init__(self, model_path: str, beta: float = 0.1):
        self.beta = beta
        
        # 策略模型（要训练的）
        self.model = AutoModelForCausalLM.from_pretrained(
            model_path, torch_dtype=torch.bfloat16, device_map="auto"
        )
        
        # 参考模型（冻结的 SFT 模型）
        self.ref_model = deepcopy(self.model)
        for p in self.ref_model.parameters():
            p.requires_grad = False
    
    def compute_logprobs(self, model, input_ids, labels):
        """计算序列的 log probabilities"""
        outputs = model(input_ids=input_ids)
        logits = outputs.logits[:, :-1, :]
        labels = labels[:, 1:]
        
        log_probs = F.log_softmax(logits, dim=-1)
        token_log_probs = log_probs.gather(
            -1, labels.unsqueeze(-1)
        ).squeeze(-1)
        
        # 只计算 response 部分的 logprob（忽略 prompt）
        mask = (labels != -100).float()
        return (token_log_probs * mask).sum(dim=-1) / mask.sum(dim=-1)
    
    def dpo_loss(self, policy_chosen_logps, policy_rejected_logps,
                 ref_chosen_logps, ref_rejected_logps):
        """DPO 损失函数"""
        chosen_logratios = policy_chosen_logps - ref_chosen_logps
        rejected_logratios = policy_rejected_logps - ref_rejected_logps
        
        logits = self.beta * (chosen_logratios - rejected_logratios)
        loss = -F.logsigmoid(logits).mean()
        
        # 监控指标
        with torch.no_grad():
            acc = (logits > 0).float().mean()
        
        return loss, acc
    
    def train_step(self, batch, optimizer):
        """单步训练"""
        self.model.train()
        optimizer.zero_grad()
        
        # batch 包含: chosen_input_ids, chosen_labels,
        #             rejected_input_ids, rejected_labels
        
        with torch.no_grad():
            ref_chosen_logps = self.compute_logprobs(
                self.ref_model, batch["chosen_input_ids"], 
                batch["chosen_labels"]
            )
            ref_rejected_logps = self.compute_logprobs(
                self.ref_model, batch["rejected_input_ids"],
                batch["rejected_labels"]
            )
        
        policy_chosen_logps = self.compute_logprobs(
            self.model, batch["chosen_input_ids"], batch["chosen_labels"]
        )
        policy_rejected_logps = self.compute_logprobs(
            self.model, batch["rejected_input_ids"], batch["rejected_labels"]
        )
        
        loss, acc = self.dpo_loss(
            policy_chosen_logps, policy_rejected_logps,
            ref_chosen_logps, ref_rejected_logps
        )
        
        loss.backward()
        optimizer.step()
        
        return {"loss": loss.item(), "accuracy": acc.item()}

DPO 偏好数据格式

{
    "prompt": "解释什么是递归",
    "chosen": "递归是一种编程技巧，函数调用自身来解决更小的同类问题。例如阶乘：f(n) = n * f(n-1)，基准情况 f(0) = 1。",
    "rejected": "递归就是函数调用自己。"
}

使用 TRL 库的完整 DPO 训练

from trl import DPOTrainer as TRLDPOTrainer, DPOConfig
from datasets import Dataset
from transformers import AutoModelForCausalLM, AutoTokenizer

# 加载模型
model = AutoModelForCausalLM.from_pretrained(
    "./sft_model", torch_dtype=torch.bfloat16, device_map="auto"
)
tokenizer = AutoTokenizer.from_pretrained("./sft_model")

# 加载偏好数据
dataset = Dataset.from_json("preference_data.json")

# DPO 配置
config = DPOConfig(
    output_dir="./dpo_output",
    beta=0.1,
    learning_rate=5e-7,      # DPO 需要很小的学习率
    per_device_train_batch_size=2,
    gradient_accumulation_steps=8,
    num_train_epochs=1,
    warmup_ratio=0.1,
    logging_steps=10,
    save_strategy="epoch",
    bf16=True,
)

# 训练
trainer = TRLDPOTrainer(
    model=model,
    args=config,
    train_dataset=dataset,
    tokenizer=tokenizer,
)
trainer.train()

DPO vs RLHF 对比

维度	RLHF (PPO)	DPO
训练流程	SFT → RM → PPO	SFT → DPO
模型数量	4个（policy, ref, RM, value）	2个（policy, ref）
超参数	多且敏感	少且稳定
训练稳定性	差（PPO 易崩溃）	好
效果	略优（理论上限高）	接近 RLHF
显存	高	中
实现难度	高	中

DPO 超参数

参数	推荐值	说明
beta	0.1	控制偏离参考模型的程度，越大越保守
learning_rate	5e-7	比SFT小10-100倍
epochs	1	DPO 极易过拟合，通常只训1轮
batch_size	16-32	有效 batch size (考虑梯度累积)

实战建议

数据质量 > 数据量：1万条高质量 SFT 数据 > 10万条低质量
SFT 不要过训练：1-3 epoch 足够，过训练损害多样性
DPO 学习率要小：5e-7 到 5e-6 之间，大了直接崩
DPO 只训 1 epoch：DPO 极易过拟合，多了效果反而变差
评估不可少：用 MT-Bench / AlpacaEval 做自动评估，人工评估抽检
保留 SFT checkpoint：DPO 是从 SFT 开始的，出问题可以回滚

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。

对齐技术全景#

SFT：监督微调#

数据格式#

数据构造#

训练配置#

LoRA 超参数选择#

DPO：直接偏好优化#

核心原理#

DPO 实现#

DPO 偏好数据格式#

使用 TRL 库的完整 DPO 训练#

DPO vs RLHF 对比#

DPO 超参数#

实战建议#

加入讨论#

📌 相关推荐

对齐技术全景

SFT：监督微调

数据格式

数据构造

训练配置

LoRA 超参数选择

DPO：直接偏好优化

核心原理

DPO 实现

DPO 偏好数据格式

使用 TRL 库的完整 DPO 训练

DPO vs RLHF 对比

DPO 超参数

实战建议

加入讨论