大模型微调完整技术指南

大模型微调体系
│
├── 为什么微调？
│   ├── 预训练模型是通才，业务场景需要专才
│   ├── 提升垂直领域准确率（轨交/医疗/法律）
│   └── 注入私有知识 / 定制输出风格与格式
│
├── 何时选微调 vs RAG？
│   ├── RAG：知识频繁更新、需溯源引用      → 优先 RAG
│   └── 微调：固定领域知识、格式/风格定制   → 优先微调
│
└── 微调方法谱系
    ├── 全量微调（Full Fine-tuning）        成本高，效果最好
    ├── 参数高效微调 PEFT                  主流选择
    │   ├── LoRA / QLoRA                  最常用
    │   ├── Prefix Tuning
    │   └── Prompt Tuning
    └── 基于反馈的微调
        ├── SFT（监督微调）
        ├── RLHF（人类反馈强化学习）
        └── DPO（直接偏好优化）            当前最主流

有多少资源？
      │
      ├── 多卡 A100/H100 集群
      │         └─→  全量微调 或 LoRA（大 rank）
      │
      ├── 单卡 A100（80G）
      │         └─→  LoRA（7B~13B 模型）
      │
      ├── 单卡 3090/4090（24G）
      │         └─→  QLoRA（7B~13B 量化）
      │
      └── 消费级 GPU（< 16G）
                └─→  QLoRA（7B 4bit量化）

任务类型？
      ├── 格式/风格定制      → SFT + LoRA
      ├── 垂直领域知识注入   → SFT + LoRA
      ├── 对齐输出偏好       → DPO
      └── 复杂推理能力提升   → SFT + DPO 组合

原始权重矩阵 W（d×k）冻结不动
         +
低秩矩阵 ΔW = A × B
  A: (d × r)
  B: (r × k)
  r << min(d,k)  ← 秩极小，参数量极少

前向传播：
  h = W·x + ΔW·x
    = W·x + (A·B)·x

训练时：只更新 A 和 B，W 冻结
推理时：将 ΔW 合并回 W，零额外推理开销

以 LLaMA-7B 为例（d=4096，k=4096）

原始单层权重：
  4096 × 4096 = 16,777,216 参数

LoRA r=16：
  A: 4096 × 16 = 65,536
  B: 16 × 4096 = 65,536
  合计：131,072 参数

参数减少：99.2% ！

# SFT 指令微调标准格式（Alpaca 格式）
sft_sample = {
    "instruction": "分析以下轨道交通告警事件，判断等级并给出处置建议",
    "input": "1号线人民广场站上行信号机故障，影响3列车运行",
    "output": "【等级研判】橙色告警\n【影响评估】...\n【处置步骤】\n1.立即通知...\n2.启动备用..."
}

# DPO 偏好对比格式
dpo_sample = {
    "prompt":    "信号机故障如何处置？",
    "chosen":    "规范、详细、步骤清晰的回答...",   # 好回答
    "rejected":  "简单、不专业、步骤缺失的回答..."   # 差回答
}

# 多轮对话格式（ShareGPT 格式）
chat_sample = {
    "conversations": [
        {"from": "human",     "value": "1号线出现信号故障"},
        {"from": "assistant", "value": "请问故障发生在哪个区段？"},
        {"from": "human",     "value": "在人民广场到徐家汇之间"},
        {"from": "assistant", "value": "根据应急预案，建议采取..."},
    ]
}

数据质量要求
│
├── 数量要求
│   ├── 最小可用：500~1000 条高质量样本
│   ├── 良好效果：5000~20000 条
│   └── 最优效果：50000+ 条（配合全量微调）
│
├── 质量要求
│   ├── 准确性：业务专家审核，零错误知识
│   ├── 多样性：覆盖所有业务子场景
│   ├── 一致性：相同问题输出风格统一
│   └── 格式性：严格按目标格式构造
│
└── 数据清洗流程
    ├── 去重（MinHash 相似去重）
    ├── 质量过滤（LLM 打分 > 阈值保留）
    ├── 有害内容过滤
    └── 长度过滤（过短/过长均排除）

from datasets import Dataset
import hashlib

class DatasetCleaner:

    def clean(self, raw_data: list[dict]) -> list[dict]:
        """完整数据清洗流水线"""
        data = raw_data
        data = self._remove_duplicates(data)      # 去重
        data = self._filter_by_length(data)       # 长度过滤
        data = self._filter_by_quality(data)      # 质量过滤
        data = self._normalize_format(data)       # 格式归一
        print(f"清洗后数据量: {len(raw_data)} → {len(data)}")
        return data

    def _remove_duplicates(self, data: list[dict]) -> list[dict]:
        """基于内容哈希去重"""
        seen = set()
        result = []
        for item in data:
            key = hashlib.md5(
                (item["instruction"] + item["output"]).encode()
            ).hexdigest()
            if key not in seen:
                seen.add(key)
                result.append(item)
        return result

    def _filter_by_length(self, data: list[dict],
                           min_len=50, max_len=2048) -> list[dict]:
        """过滤过短/过长的样本"""
        return [
            item for item in data
            if min_len <= len(item["output"]) <= max_len
        ]

    def _filter_by_quality(self, data: list[dict],
                            threshold: float = 4.0) -> list[dict]:
        """用 LLM 对每条数据打分，过滤低质量样本"""
        result = []
        for item in data:
            score = self._llm_score(item)    # LLM 打 1~5 分
            if score >= threshold:
                result.append(item)
        return result

    def _llm_score(self, item: dict) -> float:
        prompt = f"""
        评估以下问答对的质量（1-5分）：
        问题：{item['instruction']}
        回答：{item['output']}
        
        评分标准：5=准确专业完整，1=错误/无用
        只输出数字分数：
        """
        resp = scorer_llm.invoke(prompt)
        return float(resp.content.strip())

pip install transformers==4.40.0
pip install peft==0.10.0
pip install trl==0.8.6
pip install bitsandbytes==0.43.0   # QLoRA 量化
pip install datasets accelerate
pip install deepspeed               # 多卡训练

import torch
from transformers import (
    AutoTokenizer, AutoModelForCausalLM,
    BitsAndBytesConfig, TrainingArguments
)
from peft import LoraConfig, get_peft_model, TaskType
from trl import SFTTrainer
from datasets import load_dataset

# ── 1. 模型配置 ────────────────────────────────────────────────────────
MODEL_NAME = "Qwen/Qwen2.5-7B-Instruct"   # 基座模型
OUTPUT_DIR = "./output/rail-transit-sft"

# ── 2. 4-bit 量化配置（QLoRA 核心）───────────────────────────────────
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,                         # 4bit 量化加载
    bnb_4bit_use_double_quant=True,            # 双重量化，进一步压缩
    bnb_4bit_quant_type="nf4",                 # NormalFloat4 量化类型
    bnb_4bit_compute_dtype=torch.bfloat16,     # 计算时用 BF16
)

# ── 3. 加载基座模型 ────────────────────────────────────────────────────
model = AutoModelForCausalLM.from_pretrained(
    MODEL_NAME,
    quantization_config=bnb_config,
    device_map="auto",                         # 自动分配 GPU/CPU
    trust_remote_code=True,
)
model.config.use_cache = False                 # 训练时关闭 KV Cache
model.config.pretraining_tp = 1

tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME, trust_remote_code=True)
tokenizer.pad_token = tokenizer.eos_token
tokenizer.padding_side = "right"

# ── 4. LoRA 配置 ────────────────────────────────────────────────────────
lora_config = LoraConfig(
    r=16,                                      # 低秩矩阵的秩
    lora_alpha=32,                             # 缩放系数
    lora_dropout=0.05,                         # dropout 防过拟合
    bias="none",
    task_type=TaskType.CAUSAL_LM,
    # 应用 LoRA 的目标层（Qwen2 架构）
    target_modules=[
        "q_proj", "k_proj", "v_proj",          # 注意力层（最重要）
        "o_proj",                              # 输出投影
        "gate_proj", "up_proj", "down_proj",   # FFN 层
    ],
)

model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 输出示例: trainable params: 20,185,088 || all params: 7,721,795,584
#           trainable%: 0.2613%

# ── 5. 数据集加载 ──────────────────────────────────────────────────────
def format_sample(sample: dict) -> str:
    """将数据格式化为模型训练格式"""
    return f"""<|im_start|>system
你是轨道交通运营专家智能助手。<|im_end|>
<|im_start|>user
{sample['instruction']}
{sample.get('input', '')}<|im_end|>
<|im_start|>assistant
{sample['output']}<|im_end|>"""

dataset = load_dataset("json", data_files={
    "train": "./data/train.jsonl",
    "eval":  "./data/eval.jsonl"
})

# ── 6. 训练参数 ────────────────────────────────────────────────────────
training_args = TrainingArguments(
    output_dir=OUTPUT_DIR,
    num_train_epochs=3,
    per_device_train_batch_size=4,
    per_device_eval_batch_size=4,
    gradient_accumulation_steps=8,     # 等效 batch_size=32
    gradient_checkpointing=True,       # 节省显存（以时间换空间）
    optim="paged_adamw_32bit",         # QLoRA 专用优化器
    learning_rate=2e-4,
    lr_scheduler_type="cosine",        # 余弦学习率衰减
    warmup_ratio=0.05,
    fp16=False,
    bf16=True,                         # A100/H100 用 BF16
    logging_steps=10,
    save_strategy="epoch",
    evaluation_strategy="epoch",
    load_best_model_at_end=True,
    report_to="wandb",                 # 实验追踪
)

# ── 7. 启动训练 ────────────────────────────────────────────────────────
trainer = SFTTrainer(
    model=model,
    args=training_args,
    train_dataset=dataset["train"],
    eval_dataset=dataset["eval"],
    tokenizer=tokenizer,
    formatting_func=format_sample,     # 数据格式化函数
    max_seq_length=2048,
    dataset_num_proc=4,
)

trainer.train()
trainer.save_model(OUTPUT_DIR)

from peft import PeftModel
from transformers import AutoModelForCausalLM, AutoTokenizer

def merge_lora_weights(base_model_name: str,
                        lora_adapter_path: str,
                        output_path: str):
    """将 LoRA 权重合并回基座模型，推理零额外开销"""

    print("加载基座模型...")
    base_model = AutoModelForCausalLM.from_pretrained(
        base_model_name,
        torch_dtype=torch.bfloat16,
        device_map="cpu",              # CPU 合并，节省 GPU 显存
    )

    print("加载 LoRA 适配器...")
    model = PeftModel.from_pretrained(base_model, lora_adapter_path)

    print("合并权重...")
    model = model.merge_and_unload()   # 将 ΔW 合并回 W

    print("保存合并后模型...")
    model.save_pretrained(output_path, safe_serialization=True)

    tokenizer = AutoTokenizer.from_pretrained(base_model_name)
    tokenizer.save_pretrained(output_path)
    print(f"完成！合并模型已保存至: {output_path}")

merge_lora_weights(
    base_model_name="Qwen/Qwen2.5-7B-Instruct",
    lora_adapter_path="./output/rail-transit-sft",
    output_path="./output/rail-transit-merged"
)

RLHF 流程（复杂）：
  SFT → 训练 Reward Model → PPO 强化学习
  问题：三阶段、不稳定、计算成本高

DPO 流程（简洁）：
  SFT → 直接在偏好数据上优化策略
  优点：稳定、高效、无需 Reward Model
  
DPO 损失函数核心思想：
  最大化 chosen 相对 rejected 的对数似然比
  同时用参考模型（ref_model）防止偏离太远

from trl import DPOTrainer, DPOConfig
from datasets import Dataset

# ── DPO 数据格式 ──────────────────────────────────────────────────────
dpo_data = [
    {
        "prompt":    "信号机故障应该如何处置？",
        "chosen":    "根据《轨道交通应急预案》第三章...",  # 专家认可的好回答
        "rejected":  "重启一下信号机试试。",              # 不规范的差回答
    },
    # ... 更多偏好对
]

dataset = Dataset.from_list(dpo_data)
train_test = dataset.train_test_split(test_size=0.1)

# ── DPO 训练配置 ──────────────────────────────────────────────────────
dpo_config = DPOConfig(
    output_dir="./output/rail-transit-dpo",
    num_train_epochs=2,
    per_device_train_batch_size=2,
    gradient_accumulation_steps=8,
    learning_rate=5e-5,
    beta=0.1,                          # KL 散度惩罚系数（越大越保守）
    max_prompt_length=512,
    max_length=1024,
    bf16=True,
    logging_steps=10,
)

# ── 加载 SFT 后的模型作为起点 ─────────────────────────────────────────
model = AutoModelForCausalLM.from_pretrained(
    "./output/rail-transit-merged",    # SFT 之后的模型
    torch_dtype=torch.bfloat16,
    device_map="auto"
)

# ── 启动 DPO 训练 ─────────────────────────────────────────────────────
dpo_trainer = DPOTrainer(
    model=model,
    ref_model=None,                    # None = 自动使用 SFT 模型副本
    args=dpo_config,
    train_dataset=train_test["train"],
    eval_dataset=train_test["test"],
    tokenizer=tokenizer,
)

dpo_trainer.train()

// ds_config.json
{
  "zero_optimization": {
    "stage": 3,
    "offload_optimizer": {"device": "cpu"},   // 优化器状态卸载到 CPU
    "offload_param": {"device": "cpu"},       // 模型参数卸载到 CPU
    "overlap_comm": true,
    "contiguous_gradients": true,
    "sub_group_size": 1e9,
    "reduce_bucket_size": "auto",
    "stage3_prefetch_bucket_size": "auto",
    "stage3_param_persistence_threshold": "auto"
  },
  "fp16": {"enabled": false},
  "bf16": {"enabled": true},
  "gradient_clipping": 1.0,
  "train_micro_batch_size_per_gpu": 4,
  "gradient_accumulation_steps": 4
}

# 4卡训练启动命令
deepspeed --num_gpus=4 train.py \
    --deepspeed ds_config.json \
    --model_name Qwen/Qwen2.5-7B-Instruct \
    --output_dir ./output/rail-transit-sft

模型显存（BF16）：参数量 × 2 bytes
  7B  模型 ≈ 14 GB
  13B 模型 ≈ 26 GB
  70B 模型 ≈ 140 GB

训练额外开销（全量微调）：
  梯度：× 2 bytes = 同上
  优化器（AdamW）：× 8 bytes = 4倍

微调方案选择：
┌──────────────────┬──────────┬───────────┬────────────┐
│ 方案              │  7B 需求  │  13B 需求  │  70B 需求  │
├──────────────────┼──────────┼───────────┼────────────┤
│ 全量微调          │  ~80GB   │  ~160GB   │  ~800GB    │
│ LoRA (BF16)      │  ~20GB   │  ~40GB    │  ~160GB    │
│ QLoRA (4bit)     │  ~10GB   │  ~18GB    │  ~48GB     │
└──────────────────┴──────────┴───────────┴────────────┘

from evaluate import load
import json

class FineTuneEvaluator:
    """微调效果多维评估"""

    def evaluate(self, model, test_dataset: list[dict]) -> dict:
        results = {
            "业务准确率": self._business_accuracy(model, test_dataset),
            "格式合规率": self._format_compliance(model, test_dataset),
            "ROUGE-L":   self._rouge_score(model, test_dataset),
            "LLM评分":   self._llm_judge_score(model, test_dataset),
            "推理延迟":  self._latency_benchmark(model),
        }
        return results

    def _business_accuracy(self, model, dataset) -> float:
        """领域专家核心指标：业务回答准确率"""
        correct = 0
        for item in dataset:
            pred = model.generate(item["instruction"])
            # 关键信息点提取后比较
            if self._key_points_match(pred, item["output"]):
                correct += 1
        return correct / len(dataset)

    def _llm_judge_score(self, model, dataset) -> float:
        """用强模型（GPT-4o）评判微调模型输出质量"""
        scores = []
        for item in dataset:
            pred = model.generate(item["instruction"])
            prompt = f"""
            问题：{item['instruction']}
            参考答案：{item['output']}
            模型回答：{pred}
            
            请从准确性/完整性/专业性三个维度打分（1-10），
            输出JSON：{{"accuracy": x, "completeness": x, "professionalism": x}}
            """
            score = json.loads(judge_llm.invoke(prompt).content)
            scores.append(sum(score.values()) / 3)
        return sum(scores) / len(scores)

# 在训练数据中混入通用数据，保持基础能力
def build_mixed_dataset(domain_data: list, ratio: float = 0.1):
    """
    领域数据 + 少量通用数据混合
    防止微调后模型丧失通用对话能力
    """
    general_data = load_general_dataset()   # 通用指令数据
    n_general = int(len(domain_data) * ratio)
    mixed = domain_data + random.sample(general_data, n_general)
    random.shuffle(mixed)
    return mixed

微调实施路径
│
├── 阶段一：数据准备（最重要，占总工作 60%）
│   ├── □ 业务场景梳理，确定微调目标
│   ├── □ 构建种子数据（50~200 条专家标注）
│   ├── □ Self-Instruct 数据扩充
│   ├── □ 数据清洗（去重/质检/格式化）
│   └── □ 划分训练集/验证集（9:1）
│
├── 阶段二：基座模型选择
│   ├── □ 中文能力：Qwen2.5 / Baichuan2 / DeepSeek
│   ├── □ 英文能力：LLaMA3 / Mistral
│   └── □ 参数规模：≥7B（业务复杂度决定）
│
├── 阶段三：微调训练
│   ├── □ 先小数据验证 Pipeline 可跑通
│   ├── □ 超参调优（lr / rank / batch_size）
│   ├── □ Loss 曲线监控（防欠/过拟合）
│   └── □ 定期 Checkpoint 保存
│
├── 阶段四：评估与迭代
│   ├── □ 业务指标评估（准确率 / 格式合规率）
│   ├── □ LLM Judge 自动评分
│   ├── □ 与基座模型对比（不能退步）
│   └── □ Bad Case 分析 → 补充数据 → 再训练
│
└── 阶段五：部署上线
    ├── □ LoRA 权重合并（推理零开销）
    ├── □ 量化压缩（GPTQ/AWQ，推理加速）
    ├── □ vLLM 部署（高吞吐推理框架）
    └── □ A/B 测试上线，监控线上效果

┌─────────────────────────────────────────────────────┐
│              大模型微调技术栈                         │
│                                                     │
│  框架      Transformers + PEFT + TRL                │
│  训练优化  DeepSpeed ZeRO-3 / FSDP                  │
│  量化      BitsAndBytes（训练）/ GPTQ/AWQ（推理）     │
│  数据处理  datasets + 自定义清洗流水线               │
│  实验追踪  Wandb / TensorBoard                      │
│  评估框架  RAGAS / LLM Judge / ROUGE                │
│  推理部署  vLLM / TGI（Text Generation Inference）   │
│  基座模型  Qwen2.5 / LLaMA3 / DeepSeek-V2          │
└─────────────────────────────────────────────────────┘