AI大模型微调（Fine-tuning）完全教程

本教程全面讲解大语言模型微调的核心技术与实战方法，帮助开发者掌握大模型微调的完整技术栈。

目录

1. 微调概述与核心概念
2. 参数高效微调方法（PEFT）
3. 训练数据准备与清洗
4. SFT指令微调
5. RLHF与DPO对齐训练
6. 主流微调框架对比
7. 超参数调优
8. 分布式训练
9. 评估与监控
10. 微调后的模型量化与部署
11. 实战案例
12. 最佳实践与常见问题

1. 微调概述与核心概念

1.1 什么是大模型微调

大模型微调（Fine-tuning）是在预训练大语言模型的基础上，使用特定领域或任务的数据对模型进行进一步训练，使其适应特定场景的过程。与从零训练相比，微调成本极低且效果显著。

1.2 为什么需要微调

1.3 微调方法分类

• 全参数微调（Full Fine-tuning）：更新所有模型参数，效果最好但成本最高
• 参数高效微调（PEFT）：只更新少量参数
  • LoRA - 低秩适配
  • QLoRA - 量化低秩适配
  • Prefix Tuning - 前缀微调
  • P-Tuning - 提示微调
  • Adapter - 适配器
• 对齐训练
  • SFT - 监督微调
  • RLHF - 人类反馈强化学习
  • DPO - 直接偏好优化

1.4 微调 vs RAG vs Prompt Engineering

2. 参数高效微调方法（PEFT）

2.1 LoRA（Low-Rank Adaptation）

LoRA是目前最流行的参数高效微调方法，核心思想是将权重更新分解为低秩矩阵：W' = W + BA，其中B和A是低秩矩阵，r远小于原始维度。

LoRA的优势：

• 可训练参数量仅为原始模型的0.1%-1%
• 显存需求大幅降低
• 效果接近全参数微调
• 可以保存多个LoRA适配器，按需切换

import torch
import torch.nn as nn

class LoRALayer(nn.Module):
    def __init__(self, in_features, out_features, rank=8, alpha=16, dropout=0.1):
        super().__init__()
        self.rank = rank
        self.alpha = alpha
        self.scaling = alpha / rank

        # 原始权重（冻结）
        self.linear = nn.Linear(in_features, out_features, bias=False)
        self.linear.weight.requires_grad = False

        # LoRA可训练参数
        self.lora_A = nn.Parameter(torch.randn(rank, in_features) * 0.01)
        self.lora_B = nn.Parameter(torch.zeros(out_features, rank))
        self.dropout = nn.Dropout(dropout)

    def forward(self, x):
        base_output = self.linear(x)
        lora_output = (x @ self.lora_A.T @ self.lora_B.T) * self.scaling
        return base_output + lora_output

使用PEFT库应用LoRA：

from peft import LoraConfig, get_peft_model
from transformers import AutoModelForCausalLM

def apply_lora(model_name):
    model = AutoModelForCausalLM.from_pretrained(
        model_name, torch_dtype=torch.bfloat16, device_map="auto"
    )

    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)

    trainable = sum(p.numel() for p in model.parameters() if p.requires_grad)
    total = sum(p.numel() for p in model.parameters())
    print(f"可训练参数: {trainable} / {total} ({100*trainable/total:.2f}%)")

    return model

2.2 QLoRA（Quantized LoRA）

QLoRA结合4-bit量化和LoRA，在单张消费级GPU上微调大模型。核心创新包括NF4量化、双重量化和分页优化器。

from transformers import BitsAndBytesConfig
from peft import prepare_model_for_kbit_training

def setup_qlora(model_name):
    bnb_config = BitsAndBytesConfig(
        load_in_4bit=True,
        bnb_4bit_quant_type="nf4",
        bnb_4bit_compute_dtype=torch.bfloat16,
        bnb_4bit_use_double_quant=True,
    )

    model = AutoModelForCausalLM.from_pretrained(
        model_name,
        quantization_config=bnb_config,
        device_map="auto",
        trust_remote_code=True,
    )

    model = prepare_model_for_kbit_training(model)
    return model

显存对比：

• LLaMA-7B 全参数微调: 约60GB (FP16)
• LLaMA-7B LoRA: 约16GB (FP16)
• LLaMA-7B QLoRA: 约6GB (4-bit)

2.3 方法对比

3. 训练数据准备与清洗

3.1 数据格式

SFT指令微调数据格式（Alpaca格式）：

{
    "instruction": "请将以下英文翻译成中文",
    "input": "Machine learning is a subset of AI.",
    "output": "机器学习是人工智能的一个子集。"
}

多轮对话格式（ShareGPT格式）：

{
    "conversations": [
        {"from": "system", "value": "你是一个专业的医疗助手"},
        {"from": "human", "value": "感冒的症状有哪些？"},
        {"from": "gpt", "value": "感冒的常见症状包括发热、咳嗽、流涕、咽痛等。"}
    ]
}

DPO偏好数据格式：

{
    "prompt": "解释什么是量子计算",
    "chosen": "量子计算是利用量子力学原理进行信息处理的计算范式...",
    "rejected": "量子计算就是用量子来计算，很厉害。"
}

3.2 数据清洗流程

import re
import json
import hashlib
from typing import List, Dict

class DataPreprocessor:
    def __init__(self, min_length=10, max_length=4096):
        self.min_length = min_length
        self.max_length = max_length
        self.seen_hashes = set()

    def clean_dataset(self, data: List[Dict]) -> List[Dict]:
        cleaned = []
        for item in data:
            if not self._has_valid_content(item):
                continue
            if not self._check_length(item):
                continue
            item_hash = hashlib.md5(
                json.dumps(item, ensure_ascii=False, sort_keys=True).encode()
            ).hexdigest()
            if item_hash in self.seen_hashes:
                continue
            self.seen_hashes.add(item_hash)
            if self._assess_quality(item) < 0.5:
                continue
            cleaned.append(item)
        return cleaned

    def _has_valid_content(self, item):
        for key in ['instruction', 'output', 'chosen', 'value']:
            if key in item and isinstance(item[key], str) and len(item[key].strip()) > 0:
                return True
        return False

    def _check_length(self, item):
        total_len = sum(len(str(v)) for v in item.values() if isinstance(v, str))
        return self.min_length <= total_len <= self.max_length

    def _assess_quality(self, item):
        score = 0.0
        output = item.get('output', item.get('chosen', ''))
        if 20 <= len(output) <= 2000:
            score += 0.3
        if not re.search(r'[\x00-\x08\x0b\x0c\x0e-\x1f]', output):
            score += 0.3
        if item.get('instruction', '').strip():
            score += 0.2
        return score

3.3 数据增强策略

1. 指令改写：使用LLM将同一条指令用不同方式表达
2. 翻译增强：中英互译扩展数据
3. 输出多样化：对同一指令生成多种风格的回复
4. 负样本构造：构造质量较差的回复用于DPO训练

3.4 数据配比建议

4. SFT指令微调

4.1 完整训练流程

from transformers import (
    AutoModelForCausalLM, AutoTokenizer,
    TrainingArguments, Trainer
)
from peft import LoraConfig, get_peft_model
from datasets import load_dataset

def sft_train(model_name, dataset_path, output_dir,
              num_epochs=3, batch_size=4, learning_rate=2e-5):

    # 1. 加载模型和分词器
    tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
    tokenizer.pad_token = tokenizer.eos_token

    model = AutoModelForCausalLM.from_pretrained(
        model_name,
        torch_dtype=torch.bfloat16,
        device_map="auto",
        trust_remote_code=True,
    )

    # 2. 应用LoRA
    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)

    # 3. 加载和预处理数据
    dataset = load_dataset("json", data_files=dataset_path)

    def preprocess(examples):
        texts = []
        for inst, inp, out in zip(
            examples["instruction"],
            examples.get("input", [""] * len(examples["instruction"])),
            examples["output"]
        ):
            if inp:
                text = f"### Instruction:\n{inst}\n\n### Input:\n{inp}\n\n### Response:\n{out}"
            else:
                text = f"### Instruction:\n{inst}\n\n### Response:\n{out}"
            texts.append(text)

        tokenized = tokenizer(texts, truncation=True, max_length=2048, padding="max_length")
        tokenized["labels"] = tokenized["input_ids"].copy()
        return tokenized

    tokenized_dataset = dataset.map(
        preprocess, batched=True,
        remove_columns=dataset["train"].column_names
    )

    # 4. 训练参数
    training_args = TrainingArguments(
        output_dir=output_dir,
        num_train_epochs=num_epochs,
        per_device_train_batch_size=batch_size,
        gradient_accumulation_steps=4,
        learning_rate=learning_rate,
        lr_scheduler_type="cosine",
        warmup_ratio=0.1,
        weight_decay=0.01,
        logging_steps=10,
        save_steps=100,
        save_total_limit=3,
        bf16=True,
        gradient_checkpointing=True,
        report_to="none",
    )

    # 5. 开始训练
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=tokenized_dataset["train"],
        tokenizer=tokenizer,
    )

    trainer.train()
    trainer.save_model(output_dir)
    tokenizer.save_pretrained(output_dir)

    return model, tokenizer

4.2 ChatML格式训练

对于对话模型，推荐使用ChatML格式进行训练。每个对话以system消息开始，然后是多轮user/assistant交替。使用特殊标记来标识消息边界。

5. RLHF与DPO对齐训练

5.1 RLHF流程

RLHF（Reinforcement Learning from Human Feedback）包含三个阶段：

1. SFT阶段：使用监督数据微调基础模型
2. 奖励模型训练：训练一个模型来评估回复质量
3. PPO强化学习：使用奖励模型优化策略模型

# 奖励模型训练
from transformers import AutoModelForSequenceClassification

class RewardModel(nn.Module):
    def __init__(self, model_name):
        super().__init__()
        self.model = AutoModelForSequenceClassification.from_pretrained(
            model_name, num_labels=1, torch_dtype=torch.bfloat16
        )

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

def train_reward_model(model, dataset, tokenizer):
    """训练奖励模型"""
    optimizer = torch.optim.AdamW(model.parameters(), lr=1e-5)

    for batch in dataset:
        chosen_scores = model(batch["chosen_ids"], batch["chosen_mask"])
        rejected_scores = model(batch["rejected_ids"], batch["rejected_mask"])

        # Bradley-Terry损失
        loss = -torch.log(torch.sigmoid(chosen_scores - rejected_scores)).mean()

        loss.backward()
        optimizer.step()
        optimizer.zero_grad()

5.2 DPO（Direct Preference Optimization）

DPO是RLHF的简化替代方案，无需训练奖励模型和强化学习：

import torch.nn.functional as F

class DPOTrainer:
    def __init__(self, model, ref_model, tokenizer, beta=0.1):
        self.model = model
        self.ref_model = ref_model
        self.tokenizer = tokenizer
        self.beta = beta

    def dpo_loss(self, chosen_logps, rejected_logps,
                 ref_chosen_logps, ref_rejected_logps):
        """DPO损失函数"""
        chosen_rewards = self.beta * (chosen_logps - ref_chosen_logps)
        rejected_rewards = self.beta * (rejected_logps - ref_rejected_logps)

        loss = -F.logsigmoid(chosen_rewards - rejected_rewards).mean()
        return loss

    def compute_logprobs(self, model, input_ids, attention_mask, labels):
        """计算序列的对数概率"""
        outputs = model(input_ids=input_ids, attention_mask=attention_mask)
        logits = outputs.logits

        # 计算每个token的对数概率
        log_probs = F.log_softmax(logits, dim=-1)
        token_logps = torch.gather(log_probs, 2, labels.unsqueeze(2)).squeeze(2)

        # 只计算response部分的损失
        return token_logps.sum(dim=-1)

    def train_step(self, batch):
        """单步训练"""
        chosen_logps = self.compute_logprobs(
            self.model, batch["chosen_ids"], batch["chosen_mask"], batch["chosen_labels"]
        )
        rejected_logps = self.compute_logprobs(
            self.model, batch["rejected_ids"], batch["rejected_mask"], batch["rejected_labels"]
        )

        with torch.no_grad():
            ref_chosen_logps = self.compute_logprobs(
                self.ref_model, batch["chosen_ids"], batch["chosen_mask"], batch["chosen_labels"]
            )
            ref_rejected_logps = self.compute_logprobs(
                self.ref_model, batch["rejected_ids"], batch["rejected_mask"], batch["rejected_labels"]
            )

        loss = self.dpo_loss(chosen_logps, rejected_logps, ref_chosen_logps, ref_rejected_logps)
        return loss

5.3 SimPO和其他方法

SimPO（Simple Preference Optimization）是DPO的进一步简化版本，使用平均对数概率作为隐式奖励，无需参考模型，训练效率更高。

6. 主流微调框架对比

6.1 框架对比表

6.2 LLaMA-Factory使用

# config.yaml
model_name_or_path: Qwen/Qwen2.5-7B-Instruct
stage: sft
do_train: true
finetuning_type: lora
lora_rank: 16
lora_alpha: 32
lora_target: all
dataset: alpaca_zh
template: qwen
cutoff_len: 2048
per_device_train_batch_size: 2
gradient_accumulation_steps: 8
num_train_epochs: 3
learning_rate: 2.0e-5
lr_scheduler_type: cosine
warmup_ratio: 0.1
bf16: true
output_dir: output/qwen2.5-7b-sft

# 启动训练
llamafactory-cli train config.yaml

# WebUI模式
llamafactory-cli webui

6.3 Unsloth使用

from unsloth import FastLanguageModel

def unsloth_train():
    # 加载模型（自动优化）
    model, tokenizer = FastLanguageModel.from_pretrained(
        model_name="unsloth/Qwen2.5-7B-Instruct-bnb-4bit",
        max_seq_length=2048,
        dtype=None,
        load_in_4bit=True,
    )

    # 应用LoRA（优化版本）
    model = FastLanguageModel.get_peft_model(
        model,
        r=16,
        target_modules=["q_proj", "k_proj", "v_proj", "o_proj",
                        "gate_proj", "up_proj", "down_proj"],
        lora_alpha=16,
        lora_dropout=0,
        bias="none",
        use_gradient_checkpointing="unsloth",
        random_state=3407,
    )

    # Unsloth训练速度提升2倍，显存减少60%

7. 超参数调优

7.1 关键超参数

7.2 调优策略

def hyperparameter_search(model_name, dataset, output_dir):
    """超参数搜索"""
    import optuna

    def objective(trial):
        lr = trial.suggest_float("lr", 1e-5, 5e-5, log=True)
        lora_rank = trial.suggest_categorical("lora_rank", [8, 16, 32, 64])
        batch_size = trial.suggest_categorical("batch_size", [2, 4, 8])
        warmup_ratio = trial.suggest_float("warmup_ratio", 0.05, 0.15)

        # 训练并评估
        metrics = train_and_evaluate(
            model_name, dataset, output_dir,
            lr=lr, lora_rank=lora_rank,
            batch_size=batch_size, warmup_ratio=warmup_ratio
        )

        return metrics["eval_loss"]

    study = optuna.create_study(direction="minimize")
    study.optimize(objective, n_trials=20)

    print(f"最佳参数: {study.best_params}")
    return study.best_params

7.3 学习率调度

from transformers import get_cosine_schedule_with_warmup

def get_scheduler(optimizer, num_training_steps, warmup_ratio=0.1):
    num_warmup_steps = int(num_training_steps * warmup_ratio)

    return get_cosine_schedule_with_warmup(
        optimizer,
        num_warmup_steps=num_warmup_steps,
        num_training_steps=num_training_steps
    )

8. 分布式训练

8.1 DeepSpeed ZeRO

{
    "zero_optimization": {
        "stage": 2,
        "offload_optimizer": {"device": "cpu"},
        "offload_param": {"device": "none"},
        "overlap_comm": true,
        "contiguous_gradients": true
    },
    "bf16": {"enabled": true},
    "train_batch_size": 32,
    "train_micro_batch_size_per_gpu": 4,
    "gradient_accumulation_steps": 8,
    "gradient_clipping": 1.0,
    "optimizer": {
        "type": "AdamW",
        "params": {"lr": 2e-5, "weight_decay": 0.01}
    }
}

8.2 多GPU训练

# 使用torchrun启动多GPU训练
torchrun --nproc_per_node=4 train.py \
    --model_name Qwen/Qwen2.5-7B-Instruct \
    --dataset data/train.json \
    --output_dir output/multi-gpu \
    --deepspeed ds_config.json

# 使用Accelerate
accelerate launch --multi_gpu train.py

8.3 FSDP训练

from torch.distributed.fsdp import FullyShardedDataParallel as FSDP

def setup_fsdp(model):
    return FSDP(
        model,
        sharding_strategy=ShardingStrategy.FULL_SHARD,
        mixed_precision=MixedPrecision(
            param_dtype=torch.bfloat16,
            reduce_dtype=torch.bfloat16,
            buffer_dtype=torch.bfloat16,
        ),
    )

9. 评估与监控

9.1 评估指标

def evaluate_model(model, tokenizer, eval_dataset):
    """模型评估"""
    results = {}

    # 1. 困惑度（Perplexity）
    perplexity = compute_perplexity(model, tokenizer, eval_dataset)
    results["perplexity"] = perplexity

    # 2. 任务特定指标
    accuracy = compute_accuracy(model, tokenizer, eval_dataset)
    results["accuracy"] = accuracy

    # 3. 生成质量
    bleu = compute_bleu(model, tokenizer, eval_dataset)
    results["bleu"] = bleu

    # 4. 安全性评估
    safety_score = evaluate_safety(model, tokenizer)
    results["safety_score"] = safety_score

    return results

def compute_perplexity(model, tokenizer, dataset):
    """计算困惑度"""
    import math

    total_loss = 0
    total_tokens = 0

    for batch in dataset:
        inputs = tokenizer(batch["text"], return_tensors="pt", truncation=True)
        with torch.no_grad():
            outputs = model(**inputs, labels=inputs["input_ids"])
        total_loss += outputs.loss.item() * inputs["input_ids"].numel()
        total_tokens += inputs["input_ids"].numel()

    return math.exp(total_loss / total_tokens)

9.2 训练监控

import wandb

def setup_monitoring(project_name, config):
    """设置训练监控"""
    wandb.init(project=project_name, config=config)

    # 记录关键指标
    wandb.define_metric("train/loss", summary="min")
    wandb.define_metric("eval/loss", summary="min")
    wandb.define_metric("eval/perplexity", summary="min")

def log_metrics(trainer, step):
    """记录训练指标"""
    metrics = {
        "train/loss": trainer.state.log_history[-1].get("loss"),
        "train/learning_rate": trainer.state.log_history[-1].get("learning_rate"),
        "train/epoch": trainer.state.epoch,
    }
    wandb.log(metrics, step=step)

10. 微调后的模型量化与部署

10.1 模型合并

def merge_lora_to_base(model_name, lora_path, output_path):
    """将LoRA权重合并到基础模型"""
    from peft import PeftModel

    # 加载基础模型
    base_model = AutoModelForCausalLM.from_pretrained(
        model_name, torch_dtype=torch.bfloat16, device_map="auto"
    )

    # 加载LoRA
    model = PeftModel.from_pretrained(base_model, lora_path)

    # 合并权重
    merged_model = model.merge_and_unload()

    # 保存
    merged_model.save_pretrained(output_path)
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    tokenizer.save_pretrained(output_path)

    return merged_model

10.2 量化部署

def quantize_model(model_path, output_path, bits=4):
    """量化模型"""
    from transformers import BitsAndBytesConfig
    import onnxruntime as ort

    if bits == 4:
        config = BitsAndBytesConfig(
            load_in_4bit=True,
            bnb_4bit_quant_type="nf4",
            bnb_4bit_compute_dtype=torch.bfloat16,
        )
    elif bits == 8:
        config = BitsAndBytesConfig(load_in_8bit=True)

    model = AutoModelForCausalLM.from_pretrained(
        model_path, quantization_config=config, device_map="auto"
    )

    # 使用GPTQ量化（更高效）
    from optimum.gptq import GPTQQuantizer

    quantizer = GPTQQuantizer(bits=4, dataset="c4", group_size=128)
    quantized_model = quantizer.quantize_model(model, tokenizer)

    quantized_model.save_pretrained(output_path)

10.3 vLLM部署

from vllm import LLM, SamplingParams

def deploy_with_vllm(model_path):
    """使用vLLM部署微调后的模型"""
    llm = LLM(
        model=model_path,
        tensor_parallel_size=1,
        gpu_memory_utilization=0.9,
        max_model_len=2048,
    )

    sampling_params = SamplingParams(
        temperature=0.7,
        top_p=0.9,
        max_tokens=512,
    )

    # 批量推理
    prompts = ["你好，请介绍一下自己。"]
    outputs = llm.generate(prompts, sampling_params)

    for output in outputs:
        print(output.outputs[0].text)

    return llm

11. 实战案例

11.1 医疗领域微调

# 步骤1: 准备医疗数据
medical_data = [
    {
        "instruction": "根据症状诊断可能的疾病",
        "input": "患者出现发热、咳嗽、乏力，持续3天",
        "output": "根据描述的症状，可能的诊断包括：\n1. 流行性感冒 - 最常见，伴有全身症状\n2. 新冠病毒感染 - 需结合流行病学史\n3. 普通感冒 - 症状相对较轻\n建议进行血常规和病原学检测以明确诊断。"
    }
]

# 步骤2: 微调
model, tokenizer = sft_train(
    model_name="Qwen/Qwen2.5-7B-Instruct",
    dataset_path="medical_data.json",
    output_dir="output/medical-7b",
    num_epochs=3,
    learning_rate=2e-5
)

# 步骤3: 评估
metrics = evaluate_model(model, tokenizer, eval_dataset)
print(f"评估结果: {metrics}")

11.2 代码生成微调

code_data = [
    {
        "instruction": "编写Python函数实现快速排序",
        "input": "",
        "output": "def quicksort(arr):\n    if len(arr) <= 1:\n        return arr\n    pivot = arr[len(arr) // 2]\n    left = [x for x in arr if x < pivot]\n    middle = [x for x in arr if x == pivot]\n    right = [x for x in arr if x > pivot]\n    return quicksort(left) + middle + quicksort(right)"
    }
]

12. 最佳实践与常见问题

12.1 最佳实践

1. 数据质量优先：少量高质量数据优于大量低质量数据
2. 从小模型开始：先在7B模型上验证，再扩展到更大模型
3. 逐步迭代：先SFT，再DPO，逐步提升效果
4. 充分评估：训练后在多个维度评估模型
5. 版本管理：记录每次训练的配置和数据版本

12.2 常见问题

12.3 安全注意事项

1. 安全对齐：确保微调后的模型不会生成有害内容
2. 数据脱敏：训练数据中的敏感信息需要脱敏处理
3. 红队测试：在部署前进行安全性测试
4. 监控与回滚：部署后持续监控，准备回滚方案

总结

本教程详细讲解了大模型微调的完整技术栈，从LoRA/QLoRA等参数高效方法到SFT/DPO训练流程，再到框架选型和部署实践。

关键要点：

• LoRA/QLoRA是当前最推荐的微调方法
• 数据质量比数量更重要
• 先验证再扩展，逐步迭代
• 做好评估和监控

本教程内容原创，仅供参考学习使用。