监督微调 SFT：从基础模型到任务专家的关键一步

1
2017: "Fine-Tuning" 概念兴起
2
  │  用于 BERT 等编码器
3
  │  适配下游任务
4
  │
5
2020: GPT-3 in-context learning
6
  │  发现无需微调即可做任务
7
  │  但仍需要 SFT 用于生产
8
  │
9
2021: FLAN / T5 指令微调
10
  │  Instruction Tuning 概念
11
  │  多任务指令提升泛化
12
  │
13
2022: InstructGPT (OpenAI)
14
  │  GPT-3 + SFT + RLHF 流程
15
  │  ChatGPT 的前身
16
  │
17
2023: Alpaca / Vicuna / WizardLM
18
  │  开源指令微调爆发
19
  │
20
2024: 多轮对话、Tool-Use SFT
21
  │  复杂的指令格式
22
  │
23
2025+: 高质量小数据集 SFT

1
def sft_loss(model, batch):
2
    """SFT 的损失函数"""
3
    input_ids = batch['input_ids']
4
    labels = batch['labels']  # -100 表示忽略
5

6
    logits = model(input_ids)
7

8
    # 标准语言建模损失，但只对 response 部分计算
9
    loss = F.cross_entropy(
10
        logits.view(-1, logits.size(-1)),
11
        labels.view(-1),
12
        ignore_index=-100  # 关键！忽略非 response 位置
13
    )
14

15
    return loss

1
DATA_SOURCES = {
2
    'human_annotated': {
3
        'examples': ['', 'HelpSteer', 'OASST'],
4
        'pros': '质量最高',
5
        'cons': '成本高、规模有限',
6
        'best_for': '关键领域',
7
    },
8
    'self_instruct': {
9
        'examples': ['Alpaca', 'WizardLM'],
10
        'pros': '易于扩展',
11
        'cons': '可能继承错误',
12
        'best_for': '通用指令',
13
    },
14
    'synthesized': {
15
        'examples': ['Evol-Instruct', 'OpenHermes'],
16
        'pros': '无限规模',
17
        'cons': '可能单一样本',
18
        'best_for': '特定任务',
19
    },
20
    'distilled': {
21
        'examples': ['从 GPT-4 蒸馏'],
22
        'pros': '质量高',
23
        'cons': '成本/合规',
24
        'best_for': '需要高质量',
25
    },
26
    'real_conversations': {
27
        'examples': ['ShareGPT', 'WildChat'],
28
        'pros': '真实分布',
29
        'cons': '噪声多',
30
        'best_for': '对话微调',
31
    },
32
}

1
# Llama-2 Chat Template
2
LLAMA2_TEMPLATE = """<s>[INST] <<SYS>>
3
{system_message}
4
<</SYS>>
5

6
{user_message_1} [/INST] {assistant_message_1} </s><s>[INST] {user_message_2} [/INST]"""
7

8
# Llama-3 Chat Template
9
LLAMA3_TEMPLATE = """<|begin_of_text|><|start_header_id|>system<|end_header_id|>
10

11
{system}<|eot_id|><|start_header_id|>user<|end_header_id|>
12

13
{user}<|eot_id|><|start_header_id|>assistant<|end_header_id|>
14

15
{assistant}<|eot_id|>"""
16

17
# Mistral / Mixtral Chat Template
18
MISTRAL_TEMPLATE = """<s>[INST] {system}\n\n{user} [/INST]{assistant}</s>"""
19

20
# ChatML Format (通用)
21
CHATML_TEMPLATE = """<|im_start|>system
22
{system}<|im_end|>
23
<|im_start|>user
24
{user}<|im_end|>
25
<|im_start|>assistant
26
{assistant}<|im_end|>"""
27

28
# Qwen
29
QWEN_TEMPLATE = """<|im_start|>system
30
{system}<|im_end|>
31
<|im_start|>user
32
{user}<|im_end|>
33
<|im_start|>assistant
34
{assistant}<|im_end|>"""

1
class DataCleaner:
2
    """数据清洗流程"""
3

4
    def clean(self, dataset):
5
        # 1. 去重
6
        dataset = self.deduplicate(dataset)
7

8
        # 2. 长度过滤
9
        dataset = self.filter_by_length(dataset, min=10, max=2048)
10

11
        # 3. 质量过滤
12
        dataset = self.filter_quality(dataset)
13

14
        # 4. PII 移除
15
        dataset = self.remove_pii(dataset)
16

17
        # 5. 格式验证
18
        dataset = self.validate_format(dataset)
19

20
        # 6. 平衡采样
21
        dataset = self.balance_sampling(dataset)
22

23
        return dataset
24

25
    def filter_quality(self, examples):
26
        """质量过滤规则"""
27
        filtered = []
28
        for ex in examples:
29
            if self._passes_quality(ex):
30
                filtered.append(ex)
31
        return filtered
32

33
    def _passes_quality(self, ex):
34
        # 检查规则:
35
        # 1. 长度合理 (10-2048 字符)
36
        # 2. 没有重复句子
37
        # 3. 没有 placeholder
38
        # 4. 没有乱码
39
        # 5. 不是纯模板
40
        text = ex['output']
41

42
        if len(text) < 10:
43
            return False
44
        if text.count(text[:10]) > 2:  # 重复内容
45
            return False
46
        if '...' in text.lower() and len(text) < 50:  # 过短
47
            return False
48

49
        return True

1
原始数据
2
  ↓
3
[1] 格式统一
4
  ↓
5
[2] 去重
6
  ├── 精确去重
7
  └── 模糊去重 (MinHash)
8
  ↓
9
[3] 质量过滤
10
  ├── 长度过滤
11
  ├── 语言识别
12
  ├── 关键词过滤
13
  └── 质量模型评分
14
  ↓
15
[4] 内容审核
16
  ├── NSFW 过滤
17
  ├── PII 移除
18
  └── 偏差检测
19
  ↓
20
[5] 数据配比
21
  ├── 任务平衡
22
  └── 多样性约束
23
  ↓
24
[6] 分割
25
  ├── Train (90%)
26
  ├── Dev (5%)
27
  └── Test (5%)
28
  ↓
29
最终 SFT 数据集

1
class SFTTrainer:
2
    """SFT 训练器"""
3

4
    def __init__(self, model, tokenizer, config):
5
        self.model = model
6
        self.tokenizer = tokenizer
7
        self.config = config
8

9
    def train(self, dataset):
10
        # 1. 应用 chat template
11
        dataset = self._apply_chat_template(dataset)
12

13
        # 2. 标记 response 部分（用于损失掩码）
14
        dataset = self._mark_response_tokens(dataset)
15

16
        # 3. 创建 DataLoader
17
        dataloader = DataLoader(
18
            dataset,
19
            batch_size=self.config.batch_size,
20
            shuffle=True,
21
            collate_fn=self._collate
22
        )
23

24
        # 4. 训练循环
25
        for epoch in range(self.config.epochs):
26
            for batch in dataloader:
27
                # 计算损失
28
                loss = self._compute_loss(batch)
29

30
                # 反向传播
31
                loss.backward()
32

33
                # 优化器步骤
34
                self.optimizer.step()
35
                self.optimizer.zero_grad()
36

37
                # 学习率调度
38
                self.scheduler.step()
39

40
    def _compute_loss(self, batch):
41
        # 只对 response 计算损失
42
        input_ids = batch['input_ids']
43
        labels = batch['labels']  # response 部分保持原值,其他位置是 -100
44

45
        outputs = self.model(input_ids=input_ids, labels=labels)
46
        return outputs.loss

1
def mask_instruction_tokens(input_ids, tokenizer):
2
    """为损失函数掩码掉 instruction 部分"""
3

4
    # 假设格式: [INST] instruction [/INST] response </s>
5

6
    # 找到 response 开始位置
7
    # 例如，找 [/INST] 之后的位置
8
    response_start = find_token_sequence(input_ids, tokenizer.encode("[/INST]")) + len(tokenizer.encode("[/INST]"))
9

10
    # 创建 labels: -100 表示忽略损失
11
    labels = input_ids.clone()
12
    labels[:response_start] = -100  # 忽略 instruction
13

14
    # 也忽略 padding
15
    labels[labels == tokenizer.pad_token_id] = -100
16

17
    return labels

1
class SFTCollator:
2
    """SFT 数据整理器"""
3

4
    def __init__(self, tokenizer, max_length=2048):
5
        self.tokenizer = tokenizer
6
        self.max_length = max_length
7

8
    def __call__(self, batch):
9
        # 1. 提取文本
10
        texts = [self.format_example(ex) for ex in batch]
11

12
        # 2. Tokenize
13
        tokenized = self.tokenizer(
14
            texts,
15
            truncation=True,
16
            max_length=self.max_length,
17
            padding=True,
18
            return_tensors='pt',
19
        )
20

21
        # 3. 创建损失掩码
22
        labels = tokenized['input_ids'].clone()
23

24
        # 找每个样本的 response 开始位置
25
        for i, text in enumerate(texts):
26
            response_start = self._find_response_start(text)
27
            labels[i, :response_start] = -100
28

29
        # 4. 忽略 padding
30
        labels[labels == self.tokenizer.pad_token_id] = -100
31

32
        return {
33
            'input_ids': tokenized['input_ids'],
34
            'attention_mask': tokenized['attention_mask'],
35
            'labels': labels,
36
        }

1
HYPERPARAMETER_SETTINGS = {
2
    'learning_rate': {
3
        'full_finetuning': '1e-5 到 5e-5',
4
        'LoRA': '1e-4 到 5e-4',
5
        'common': '比预训练小 10-100 倍',
6
        'best_practice': '2e-5 (lr), cosine schedule',
7
    },
8
    'epochs': {
9
        'common': '2-3 个 epoch',
10
        'warning': '太多 epoch 容易过拟合',
11
        'best_practice': '在 val loss 上升时停止',
12
    },
13
    'batch_size': {
14
        'small_model': '32-128',
15
        'large_model': '1-16 (per GPU)',
16
        'gradient_accumulation': '通常 4-16',
17
        'best_practice': '等效 batch size 128-512',
18
    },
19
    'max_length': {
20
        'short': '512-1024',
21
        'medium': '2048',
22
        'long': '4096-8192',
23
        'best_practice': '根据数据分布决定',
24
    },
25
    'warmup_ratio': {
26
        'common': '0.03-0.1',
27
        'best_practice': '0.03',
28
    },
29
    'weight_decay': {
30
        'common': '0.0-0.1',
31
        'best_practice': '0.0 (与预训练相同)',
32
    },
33
}

1
def get_lr_scheduler(optimizer, num_training_steps, warmup_ratio=0.03):
2
    """SFT 常用学习率调度"""
3

4
    warmup_steps = int(num_training_steps * warmup_ratio)
5

6
    # Cosine 退火 (主流)
7
    scheduler = get_cosine_schedule_with_warmup(
8
        optimizer,
9
        num_warmup_steps=warmup_steps,
10
        num_training_steps=num_training_steps,
11
    )
12

13
    return scheduler
14

15
# 学习率曲线示例
16
LR_CURVE = """
17
LR
18
^
19
|     ___________
20
|    /           \___
21
|   /                \____
22
|  /                      \___
23
| /                            \____
24
+--------------------------→ Steps
25
   ^
26
   warmup
27
"""

1
BATCH_SIZE_GUIDELINES = """
2
总 batch_size = per_device_batch × grad_accum × n_gpus
3

4
7B 模型 (单 GPU 24GB):
5
  per_device_batch=2-4, grad_accum=16, total=32-64
6

7
70B 模型 (8 GPU):
8
  per_device_batch=1, grad_accum=8-16, total=8-16
9

10
推荐: total batch size = 64-128
11

12
经验公式:
13
  GPU 内存 ≈ 2 × per_device_batch × seq_len × 4 bytes × num_layers
14
  (粗略估算，FP16/BF16)
15
"""

1
class MultiTurnCollator:
2
    """多轮对话 SFT 数据整理器"""
3

4
    def format_conversation(self, messages, tokenizer):
5
        # 累积对话
6
        text = ""
7
        assistant_ranges = []  # (start, end) of assistant tokens
8

9
        for msg in messages:
10
            # 添加对话块
11
            block_start = len(tokenizer.encode(text))
12
            text += f"{msg['role']}: {msg['content']}\n"
13
            block_end = len(tokenizer.encode(text))
14

15
            # 记录 assistant 消息范围
16
            if msg['role'] == 'assistant':
17
                assistant_ranges.append((block_start, block_end))
18

19
        # Tokenize
20
        tokens = tokenizer.encode(text)
21

22
        # 创建 labels: assistant 部分保留,其他部分 -100
23
        labels = [-100] * len(tokens)
24
        for start, end in assistant_ranges:
25
            labels[start:end] = tokens[start:end]
26

27
        return tokens, labels
28

29
# 注意：使用 mask 而不是简单的"只看最后一句"
30
# 这样可以让模型学会所有轮的生成

1
class QLoRALayer:
2
    """QLoRA: 4-bit 量化 + LoRA"""
3

4
    def __init__(self):
5
        # 1. 将基础模型量化为 4-bit (NF4)
6
        self.base_weight = quantize_4bit(original_weight)  # ~4 GB
7

8
        # 2. LoRA 参数保持 FP16
9
        self.lora_A = nn.Parameter(torch.randn(8, 4096))
10
        self.lora_B = nn.Parameter(torch.randn(4096, 8))
11

12
        # 3. 分页优化器 (Paged Optimizer)
13
        # 4. 双量化
14
        # 5. 嵌套量化
15

16
    def forward(self, x):
17
        # 反量化 base 做计算
18
        base_out = x @ dequantize(self.base_weight).T
19
        lora_out = lora_computation(x, self.lora_A, self.lora_B)
20
        return base_out + lora_out
21

22
# 内存对比 (65B 模型):
23
# FP16 全量: 130 GB
24
# LoRA: 130 GB base + 100 MB LoRA = 130 GB
25
# QLoRA: 65 GB (4-bit) + 100 MB LoRA = 65 GB (省一半!)

1
from trl import SFTTrainer, SFTConfig
2

3
# 配置 SFT
4
config = SFTConfig(
5
    output_dir="./output",
6
    num_train_epochs=3,
7
    per_device_train_batch_size=4,
8
    gradient_accumulation_steps=4,
9
    learning_rate=2e-5,
10
    warmup_ratio=0.03,
11
    lr_scheduler_type="cosine",
12
    max_seq_length=2048,
13
    logging_steps=10,
14
    save_steps=500,
15
    bf16=True,
16
    gradient_checkpointing=True,
17
)
18

19
# 初始化 Trainer
20
trainer = SFTTrainer(
21
    model=model,
22
    args=config,
23
    train_dataset=dataset,
24
    processing_class=tokenizer,
25
)
26

27
# 开始训练
28
trainer.train()

1
from transformers import Trainer, TrainingArguments
2

3
training_args = TrainingArguments(
4
    output_dir="./output",
5
    num_train_epochs=3,
6
    per_device_train_batch_size=2,
7
    gradient_accumulation_steps=8,
8
    learning_rate=2e-5,
9
    bf16=True,
10
    gradient_checkpointing=True,
11
    logging_steps=10,
12
    save_strategy="steps",
13
    save_steps=500,
14
)
15

16
trainer = Trainer(
17
    model=model,
18
    args=training_args,
19
    train_dataset=train_dataset,
20
    eval_dataset=eval_dataset,
21
    data_collator=collator,
22
)
23

24
trainer.train()

1
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
2

3
# 准备模型 (可选 QLoRA)
4
model = prepare_model_for_kbit_training(
5
    model,
6
    use_gradient_checkpointing=True,
7
)
8

9
# LoRA 配置
10
lora_config = LoraConfig(
11
    r=16,
12
    lora_alpha=32,
13
    target_modules=[
14
        "q_proj", "k_proj", "v_proj", "o_proj",
15
        "gate_proj", "up_proj", "down_proj",
16
    ],
17
    lora_dropout=0.05,
18
    bias="none",
19
    task_type="CAUSAL_LM",
20
)
21

22
# 包装模型
23
model = get_peft_model(model, lora_config)
24

25
# 打印可训练参数
26
model.print_trainable_parameters()
27
# trainable params: 16,384,000 || all params: 6,738,415,616 || trainable%: 0.24%
28

29
# 正常训练
30
trainer = SFTTrainer(
31
    model=model,
32
    args=training_args,
33
    train_dataset=dataset,
34
)

1
# 一站式 SFT 框架
2
# 支持 100+ 模型 + 多种 SFT 模式
3

4
# 命令行
5
llamafactory-cli train \
6
    --model_name_or_path meta-llama/Llama-3-8B-Instruct \
7
    --template llama3 \
8
    --dataset alpaca_zh \
9
    --output_dir ./output \
10
    --finetuning_type lora \
11
    --lora_rank 16 \
12
    --per_device_train_batch_size 2 \
13
    --gradient_accumulation_steps 8 \
14
    --learning_rate 2e-4 \
15
    --num_train_epochs 3

1
MEMORY_OPTIMIZATIONS = {
2
    'mixed_precision': {
3
        'method': 'BF16 混合精度',
4
        'saves': '~50%',
5
        'code': 'bf16=True',
6
    },
7
    'gradient_checkpointing': {
8
        'method': '激活值不保存, 反向时重算',
9
        'saves': '~30-50%',
10
        'code': 'gradient_checkpointing=True',
11
        'cost': '+30% 计算',
12
    },
13
    'lora': {
14
        'method': '参数高效微调',
15
        'saves': '~70% (LLM 部分)',
16
    },
17
    'qlora': {
18
        'method': '4-bit 量化 + LoRA',
19
        'saves': '~80%',
20
    },
21
    'deepspeed_zeRO3': {
22
        'method': '完全分片数据并行',
23
        'saves': '线性扩展',
24
        'code': 'deepspeed_stage=3',
25
    },
26
    'flash_attention': {
27
        'method': 'IO 优化的注意力',
28
        'saves': '长序列显著',
29
        'code': 'attn_implementation="flash_attention_2"',
30
    },
31
}

1
STABILITY_TIPS = """
2
1. 学习率不要太大
3
   - 推荐 2e-5 (全量) / 2e-4 (LoRA)
4
   - 比预训练小 10-100 倍
5

6
2. 监控 loss 和 grad_norm
7
   - grad_norm > 5: 警告
8
   - grad_norm > 10: 干预
9

10
3. 检查 NaN
11
   - 添加: torch.autograd.set_detect_anomaly(True)
12
   - 跳过 NaN 批次
13

14
4. 慢启动
15
   - Warmup ratio 0.03
16
   - 不要从高 LR 开始
17

18
5. 定期评估
19
   - val loss 上升时停止
20
   - 不要等到计划结束
21
"""