大模型高效微调：从全参微调到 LoRA/QLoRA 与 DPO

# 显存估算示例：7B 模型全参微调
# 权重 FP16 = 7e9 × 2B = 14GB
# 梯度 FP16 = 14GB
# Adam 状态 = 7e9 × 8B = 56GB（动量 + 方差，FP32）
# 激活值 ≈ 序列长度 × batch × 层数 × hidden / 层
# 合计轻松超过 100GB，单卡 A100-80G 不够

# FSDP 全参微调配置示例（PyTorch 2.x）
from torch.distributed.fsdp import (
    FullyShardedDataParallel as FSDP,
    ShardingStrategy, MixedPrecision
)

fsdp_config = dict(
    sharding_strategy=ShardingStrategy.FULL_SHARD,
    mixed_precision=MixedPrecision(
        param_dtype=torch.bfloat16,
        reduce_dtype=torch.bfloat16,
        buffer_dtype=torch.bfloat16
    ),
)

import torch
import torch.nn as nn

class LoRALinear(nn.Module):
    def __init__(self, in_features, out_features, r=8, alpha=16, dropout=0.05):
        super().__init__()
        self.weight = nn.Parameter(torch.randn(out_features, in_features))
        self.weight.requires_grad = False  # 冻结原权重
        self.lora_A = nn.Parameter(torch.randn(r, in_features) * 0.01)
        self.lora_B = nn.Parameter(torch.zeros(out_features, r))
        self.scaling = alpha / r
        self.dropout = nn.Dropout(dropout)

    def forward(self, x):
        # 原路径 + LoRA 增量路径
        base = nn.functional.linear(x, self.weight)
        lora = self.dropout(x) @ self.lora_A.T @ self.lora_B.T * self.scaling
        return base + lora

from transformers import AutoModelForCausalLM, BitsAndBytesConfig
import torch

bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,                    # 4bit 量化基座
    bnb_4bit_quant_type="nf4",            # NF4 分位量化
    bnb_4bit_use_double_quant=True,       # 嵌套量化（省约 0.4bit/参数）
    bnb_4bit_compute_dtype=torch.bfloat16 # 计算类型保持 BF16
)

model = AutoModelForCausalLM.from_pretrained(
    "Qwen/Qwen2.5-7B-Instruct",
    quantization_config=bnb_config,
    device_map="auto",
    torch_dtype=torch.bfloat16,
)

from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training

# 冻结 + 准备 kbit 训练
model = prepare_model_for_kbit_training(model)

lora_config = LoraConfig(
    r=16, lora_alpha=32,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj"],
    lora_dropout=0.05,
    bias="none", task_type="CAUSAL_LM",
)
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 输出类似：trainable params: 8,388,608 || all params: 7,403,126,784 || trainable%: 0.1133

class BottleneckAdapter(nn.Module):
    def __init__(self, hidden_size, bottleneck_size=64):
        super().__init__()
        self.down = nn.Linear(hidden_size, bottleneck_size)
        self.act = nn.ReLU()
        self.up = nn.Linear(bottleneck_size, hidden_size)

    def forward(self, x):
        # 残差结构：x + Adapter(x)
        return x + self.up(self.act(self.down(x)))

{
  "instruction": "请根据合同条款判断违约责任。",
  "input": "条款 4.2：甲方逾期交货超过 30 天，乙方有权解除合同并索赔违约金，金额为合同总额的 5%。甲方逾期 45 天。",
  "output": "甲方逾期 45 天，超过条款 4.2 约定的 30 天宽限期，乙方有权解除合同，并可索赔合同总额 5% 的违约金。"
}

# 简易数据质量过滤流水线
import pandas as pd

def filter_instruction_data(df: pd.DataFrame) -> pd.DataFrame:
    # 1. 去重：instruction + output 完全一致视为重复
    df = df.drop_duplicates(subset=["instruction", "output"])
    # 2. 长度过滤：过短无效、过长截断
    df = df[df["instruction"].str.len() > 5]
    df = df[df["output"].str.len().between(10, 2048)]
    # 3. 语种过滤：剔除含大量非目标语种的样本（按字符统计）
    return df[df["output"].apply(is_target_language)]

# Hugging Face TRL 训练配置示例
model_name: Qwen/Qwen2.5-7B-Instruct
trainer_type: SFTTrainer
learning_rate: 2.0e-4
per_device_train_batch_size: 4
gradient_accumulation_steps: 8      # 等效 batch = 32
num_train_epochs: 2
max_seq_length: 1024
lr_scheduler_type: cosine
warmup_ratio: 0.1
gradient_checkpointing: true
optim: paged_adamw_8bit             # 8bit 优化器省显存
fp16: true
logging_steps: 10

# 过拟合监测：对比训练集与验证集 loss
import matplotlib.pyplot as plt

# 训练中每个 epoch 记录 train_loss / eval_loss
epochs = [1, 2, 3]
train_loss = [1.2, 0.6, 0.15]   # 训练 loss 持续下降
eval_loss  = [1.1, 0.7, 0.85]   # 验证 loss 在 epoch2 后回升 = 过拟合信号

plt.plot(epochs, train_loss, label="train_loss")
plt.plot(epochs, eval_loss, label="eval_loss")
plt.legend(); plt.title("Early Stopping at Epoch 2")

# DPO 偏好数据格式：chosen（偏好）与 rejected（拒绝）
from datasets import Dataset

dpo_data = Dataset.from_list([
    {
        "prompt": "解释一下过拟合。",
        "chosen": "过拟合是指模型在训练数据上表现好、在未见数据上表现差……",
        "rejected": "过拟合就是训练太好。"   # 简短但信息不足
    },
    # ... 数百条偏好对
])

from peft import PeftModel

# 推理时叠加 LoRA 适配器
model = AutoModelForCausalLM.from_pretrained(base_model_path)
model = PeftModel.from_pretrained(model, "./lora-checkpoint")
merged = model.merge_and_unload()  # 合并权重，推理零额外开销