SFT（二）：用Hugging Face Transformers手写监督微调

上一篇 SFT 原理篇已经把监督微调还原成 next-token prediction：

Prompt 与标准答案
  -> 一串 Causal LM token
  -> Prompt 对应 labels 设为 -100
  -> 模型内部进行 one-token shift
  -> 只在 Answer token 上计算交叉熵

这一篇不使用 TRL 的 SFTTrainer，而只部分的 Hugging Face 组件：

Dataset。
AutoTokenizer。
AutoModelForCausalLM。
forward()。
Trainer 与 TrainingArguments。
GenerationConfig 与 generate()。

目标不是重复造一个完善的训练框架，而是让我能亲眼看到 SFT 中最重要的三组张量：

input_ids
attention_mask
labels

等这些张量完全对上后，下一篇再使用 TRL SFTTrainer + LoRA，就不会把自动化误认为魔法。

原生Transformers实现SFT的完整流程

1. 本篇要训练什么

我们使用一个小型 Causal Language Model，让它学习一组中文机器学习问答。

示例：

Instruction:
用一句话解释什么是过拟合。

Response:
过拟合是模型过度记忆训练数据，导致在未见数据上泛化较差的现象。

为了让流程容易检查，训练数据直接写在 Python 文件中。

这份数据只能验证代码链路，不能训练出真正可靠的领域助手。

项目目标是：

加载 Base Causal LM。
构造 Prompt 与 Answer。
只对 Answer 计算 loss。
使用 Trainer 完成优化。
保存模型、Tokenizer 和配置。
使用 GenerationConfig 重新加载并生成。

2. 为什么先不用 Chat Template

真实对话模型通常应使用 Tokenizer 自带的 Chat Template。

但本篇暂时使用一个简单、显式的文本模板：

### Instruction:
{instruction}

### Response:
{answer}<eos>

原因是我们想直接观察：

Prompt token 到哪里结束
Answer token 从哪里开始
哪些 labels 应该设为 -100

第三篇切换到 TRL 时，会换回标准 messages + apply_chat_template() 流程。

3. 环境准备

安装基础依赖：

pip install -U torch transformers datasets accelerate

建议记录实际版本：

python -c "import torch, transformers, datasets, accelerate; print(torch.__version__); print(transformers.__version__); print(datasets.__version__); print(accelerate.__version__)"

Hugging Face API 更新较快。本文代码使用当前文档中的参数名，例如：

eval_strategy。
processing_class=tokenizer。
dtype=...。

如果使用旧版本，应查看对应版本文档，不要混用不同年代的代码片段。

4. 选择 Base Model

教学代码使用：

MODEL_ID = "Qwen/Qwen2.5-0.5B"

它是 Causal Language Model，可以通过：

AutoModelForCausalLM.from_pretrained(MODEL_ID)

加载。

这里选择 Base Model 是为了观察 SFT 前后行为变化，而不是声称 0.5B 参数足以完成复杂任务。

实际选择模型时还应检查：

模型许可证。
支持语言。
参数量和显存需求。
上下文长度。
Tokenizer 与 Chat Template。
transformers 最低版本。
Base 或 Instruct 训练阶段。

5. 构造本地监督数据

先定义若干 instruction / response 对：

TRAIN_EXAMPLES = [
    {
        "instruction": "用一句话解释什么是过拟合。",
        "response": "过拟合是模型过度记忆训练数据，导致在未见数据上泛化较差的现象。",
    },
    {
        "instruction": "梯度下降的作用是什么？",
        "response": "梯度下降通过沿损失函数负梯度方向更新参数，使模型损失逐步减小。",
    },
]

再转换成 Dataset：

from datasets import Dataset, DatasetDict

dataset = DatasetDict(
    {
        "train": Dataset.from_list(TRAIN_EXAMPLES),
        "validation": Dataset.from_list(VALIDATION_EXAMPLES),
    }
)

训练、验证和测试集必须按真实任务划分。

不能把训练问题只改一个标点后放入验证集，否则评估接近记忆测试。

6. 定义 Prompt 模板

def format_prompt(instruction: str) -> str:
    return (
        "### Instruction:\n"
        f"{instruction.strip()}\n\n"
        "### Response:\n"
    )

注意返回文本停在 Response: 后面，不包含标准答案。

例如：

prompt = format_prompt("梯度下降的作用是什么？")
print(prompt)

结果：

### Instruction:
梯度下降的作用是什么？

### Response:

标准答案单独构造：

answer = response.strip() + tokenizer.eos_token

这样 Prompt 与 Answer 的边界在代码中是明确的。

7. 加载 Tokenizer

from transformers import AutoTokenizer

tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
tokenizer.padding_side = "right"

if tokenizer.pad_token_id is None:
    tokenizer.pad_token = tokenizer.eos_token

7.1 为什么训练时使用右 Padding

右 Padding 后：

真实 token ... EOS PAD PAD PAD

这与多数训练 collator 的处理方式一致，也便于观察标签。

生成时批量处理不同长度 Prompt，有些 Decoder-only 模型更适合左 Padding。训练和生成可根据模型要求分别设置，不要只凭一个固定习惯。

7.2 为什么要确认 EOS

如果答案末尾没有结束 token，模型可能只学习“怎样开始回答”，却没有明确学习“何时结束”。

assert tokenizer.eos_token_id is not None

8. Prompt 与 Answer 为什么分开 Tokenize

我们需要知道回答 token 从哪里开始。

prompt_ids = tokenizer(
    prompt,
    add_special_tokens=True,
    truncation=False,
)["input_ids"]

answer_ids = tokenizer(
    answer,
    add_special_tokens=False,
    truncation=False,
)["input_ids"]

只在 Prompt 侧增加模型默认起始特殊 token，Answer 侧不重复增加。

然后拼接：

input_ids = prompt_ids + answer_ids

如果把 Prompt 与 Answer 各自都设置 add_special_tokens=True，中间可能重复插入 BOS 或其他边界 token。

8.1 为什么不直接用 len(tokenizer(prompt)) 切完整序列

某些子词 Tokenizer 的结果受字符串边界影响。

下面两种方法未必严格等价：

tokenizer(prompt + answer)

tokenizer(prompt) + tokenizer(answer)

本篇模板在 Prompt 末尾保留清晰换行，并显式分段编码，目的是得到稳定边界。

生产系统如果必须保持“整串文本一次分词”的精确结果，可以使用 offset mapping、模型 Chat Template 返回的 assistant mask，或 TRL 已实现的数据处理路径。

9. 构造 Completion-only Labels

Prompt 不进入 loss：

labels = [-100] * len(prompt_ids) + answer_ids.copy()

因此：

input_ids:
[Prompt tokens........................][Answer tokens......]

labels:
[-100, -100, -100, ...................][Answer token ids...]

input_ids 和 labels 必须等长：

assert len(input_ids) == len(labels)

模型仍能看到 Prompt，只是不会因“没有预测好 Prompt 自身”受到惩罚。

Prompt与Answer如何变成SFT张量

10. 截断时优先保留回答

最简单的：

input_ids = input_ids[:max_length]
labels = labels[:max_length]

可能把回答完全截掉。

本篇采用一个明确策略：

为答案预留空间。
Prompt 太长时从左侧截断 Prompt。
答案仍太长时截断答案。
尽量让最后一个答案 token 保持为 EOS。

示例实现：

def truncate_prompt_and_answer(prompt_ids, answer_ids, max_length):
    if len(answer_ids) >= max_length:
        answer_ids = answer_ids[:max_length]
        answer_ids[-1] = tokenizer.eos_token_id
        return [], answer_ids

    prompt_budget = max_length - len(answer_ids)
    prompt_ids = prompt_ids[-prompt_budget:]
    return prompt_ids, answer_ids

这适合“回答是主要监督目标”的教学问答。

但它并不是所有任务的通用答案。如果 Prompt 中最前面的 system rule 不能删除，就要采用更精细的分段截断。

11. 完整预处理函数

MAX_LENGTH = 256


def tokenize_example(example):
    prompt = format_prompt(example["instruction"])
    answer = example["response"].strip() + tokenizer.eos_token

    prompt_ids = tokenizer(
        prompt,
        add_special_tokens=True,
        truncation=False,
    )["input_ids"]
    answer_ids = tokenizer(
        answer,
        add_special_tokens=False,
        truncation=False,
    )["input_ids"]

    prompt_ids, answer_ids = truncate_prompt_and_answer(
        prompt_ids,
        answer_ids,
        MAX_LENGTH,
    )

    input_ids = prompt_ids + answer_ids
    labels = [-100] * len(prompt_ids) + answer_ids.copy()

    return {
        "input_ids": input_ids,
        "attention_mask": [1] * len(input_ids),
        "labels": labels,
    }

调用 map()：

tokenized_dataset = dataset.map(
    tokenize_example,
    remove_columns=dataset["train"].column_names,
)

由于每条样本长度不同，此时先不 Padding。

12. 必须检查一条 Tokenized 样本

在启动训练前，至少打印一条样本：

sample = tokenized_dataset["train"][0]

print("length:", len(sample["input_ids"]))
print("input:", tokenizer.decode(sample["input_ids"]))

target_ids = [
    token_id
    for token_id, label in zip(sample["input_ids"], sample["labels"])
    if label != -100
]
print("target:", tokenizer.decode(target_ids))

预期：

input:
### Instruction:
用一句话解释什么是过拟合。

### Response:
过拟合是……<eos>

target:
过拟合是……<eos>

还要检查：

assert len(sample["input_ids"]) == len(sample["labels"])
assert any(label != -100 for label in sample["labels"])

如果整条 labels 都是 -100，交叉熵没有有效目标，训练会得到无意义 loss 或 NaN。

13. 为什么不在 map() 时 Padding 到 max_length

假设三条样本长度是：

61, 78, 93

直接全部 Padding 到 256 会浪费大量计算。

更合理的是在每个 batch 中动态补到当前最长长度：

Batch 1: [61, 78] -> Padding 到 78
Batch 2: [93, 70] -> Padding 到 93

因此 map() 只负责：

Tokenize。
截断。
构造 labels。

Data Collator 再负责：

当前 batch 的 Padding。
列表转 PyTorch 张量。

14. 手写 Completion-only Data Collator

from dataclasses import dataclass

import torch


@dataclass
class CompletionOnlyCollator:
    pad_token_id: int
    pad_to_multiple_of: int | None = 8

    def __call__(self, features):
        max_length = max(len(item["input_ids"]) for item in features)

        if self.pad_to_multiple_of:
            multiple = self.pad_to_multiple_of
            max_length = ((max_length + multiple - 1) // multiple) * multiple

        batch_input_ids = []
        batch_attention_mask = []
        batch_labels = []

        for item in features:
            pad_length = max_length - len(item["input_ids"])

            batch_input_ids.append(
                item["input_ids"] + [self.pad_token_id] * pad_length
            )
            batch_attention_mask.append(
                item["attention_mask"] + [0] * pad_length
            )
            batch_labels.append(
                item["labels"] + [-100] * pad_length
            )

        return {
            "input_ids": torch.tensor(batch_input_ids, dtype=torch.long),
            "attention_mask": torch.tensor(batch_attention_mask, dtype=torch.long),
            "labels": torch.tensor(batch_labels, dtype=torch.long),
        }

三种补充值不同：

张量	Padding 值	含义
`input_ids`	`pad_token_id`	合法词表 id
`attention_mask`	`0`	不作为有效上下文
`labels`	`-100`	不进入交叉熵

SFT动态Padding后的Batch张量

15. 检查 Collator 输出

collator = CompletionOnlyCollator(
    pad_token_id=tokenizer.pad_token_id,
)

batch = collator(
    [
        tokenized_dataset["train"][0],
        tokenized_dataset["train"][1],
    ]
)

for name, tensor in batch.items():
    print(name, tensor.shape)

可能得到：

input_ids      torch.Size([2, 80])
attention_mask torch.Size([2, 80])
labels         torch.Size([2, 80])

三个张量形状必须一致。

还可以统计有效目标 token：

print((batch["labels"] != -100).sum(dim=1))

输出例如：

tensor([31, 28])

表示两条样本分别有 31 和 28 个 token 参与 loss。

16. 加载 AutoModelForCausalLM

根据设备选择 dtype：

def choose_dtype():
    if not torch.cuda.is_available():
        return torch.float32
    if torch.cuda.is_bf16_supported():
        return torch.bfloat16
    return torch.float16

加载模型：

dtype = choose_dtype()

model = AutoModelForCausalLM.from_pretrained(
    MODEL_ID,
    dtype=dtype,
)

本篇执行 Full Fine-tuning，所以默认所有参数都可训练：

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: {trainable:,}")
print(f"total: {total:,}")

对 0.5B 模型进行全参数训练仍需要明显多于推理的显存。教学代码小，不代表可以在任意硬件上完整训练。

17. 先手动执行一次 forward()

在交给 Trainer 前，先验证最底层路径：

device = next(model.parameters()).device
batch = {name: tensor.to(device) for name, tensor in batch.items()}

outputs = model(**batch)

print(outputs.logits.shape)
print(outputs.loss)

如果 batch 形状为 [2, 80]，词表大小为 $V$：

outputs.logits.shape = [2, 80, V]
outputs.loss.shape   = []

loss 是标量。

17.1 模型内部做了什么

概念上：

shift_logits = logits[:, :-1, :]
shift_labels = labels[:, 1:]

loss = cross_entropy(
    shift_logits.reshape(-1, vocab_size),
    shift_labels.reshape(-1),
    ignore_index=-100,
)

因此外部传入的 labels 与 input_ids 等长即可。

18. 为什么训练时关闭 use_cache

KV Cache 用于加速自回归生成。

训练时我们需要一次并行处理完整序列并进行反向传播，通常不需要保留生成缓存：

model.config.use_cache = False

使用 Gradient Checkpointing 时，use_cache=True 还可能产生不兼容警告。

训练完成、准备推理前再恢复：

model.config.use_cache = True

19. TrainingArguments

from transformers import TrainingArguments

training_args = TrainingArguments(
    output_dir="outputs/sft-transformers/checkpoints",
    overwrite_output_dir=True,
    num_train_epochs=5,
    per_device_train_batch_size=2,
    per_device_eval_batch_size=2,
    gradient_accumulation_steps=4,
    learning_rate=2e-5,
    weight_decay=0.01,
    warmup_ratio=0.1,
    lr_scheduler_type="cosine",
    logging_steps=1,
    eval_strategy="epoch",
    save_strategy="epoch",
    load_best_model_at_end=True,
    metric_for_best_model="eval_loss",
    greater_is_better=False,
    gradient_checkpointing=True,
    bf16=dtype == torch.bfloat16,
    fp16=dtype == torch.float16,
    report_to="none",
    seed=42,
)

19.1 有效 Batch Size

单卡时：

$$
B_{effective}
=B_{device}\times N_{accumulation}
$$

这里：

$$
B_{effective}=2\times4=8
$$

多卡数据并行时还要乘设备数。

19.2 为什么 SFT 学习率通常较小

模型已经有预训练参数。全参数微调如果学习率过大，可能快速破坏已有表示。

2e-5 只是常见起点，不是固定答案。需要结合：

模型大小。
数据量。
Full Fine-tuning 或 LoRA。
Batch Size。
训练步数。
验证集表现。

20. 创建 Trainer

from transformers import Trainer

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=tokenized_dataset["train"],
    eval_dataset=tokenized_dataset["validation"],
    data_collator=collator,
    processing_class=tokenizer,
)

这里的 Trainer 并不知道“SFT”这个名字。

它只看到：

Dataset 提供 input_ids / attention_mask / labels
Collator 组成 batch
Model forward 返回 loss
Trainer 对 loss 反向传播

SFT 语义主要由我们的数据构造方式决定。

Trainer如何执行一次SFT更新

21. 启动训练

train_result = trainer.train()
print(train_result.metrics)

eval_metrics = trainer.evaluate()
print(eval_metrics)

一次优化步骤可以展开为：

DataLoader 取样本
  -> Collator 动态 Padding
  -> model(**batch)
  -> Causal LM logits
  -> shift + ignore_index=-100
  -> loss
  -> backward
  -> 梯度累积
  -> 梯度裁剪
  -> AdamW step
  -> scheduler step

22. 为什么评估时先看 eval_loss

本篇数据很小，不适合构造复杂自动指标。

eval_loss 至少能回答：

模型对未参与更新的标准 Answer token，平均负对数似然是否下降？

但它不能完整回答：

回答是否事实正确。
是否严格遵循指令。
是否产生多余内容。
是否满足 JSON 等结构约束。

真正项目还要增加生成评估和人工检查。

23. 保存最终模型

FINAL_DIR = "outputs/sft-transformers/final"

trainer.model.config.use_cache = True
trainer.save_model(FINAL_DIR)
tokenizer.save_pretrained(FINAL_DIR)

目录中通常包括：

final/
  config.json
  generation_config.json
  model.safetensors
  tokenizer.json / vocab assets
  tokenizer_config.json
  special_tokens_map.json
  training_args.bin

实际文件名取决于模型和 Tokenizer 类型。

Full Fine-tuning 保存的是完整模型，因此目录体积接近 Base Model 权重大小。

24. 用 GenerationConfig 推理

重新加载：

from transformers import GenerationConfig

tokenizer = AutoTokenizer.from_pretrained(FINAL_DIR)
model = AutoModelForCausalLM.from_pretrained(
    FINAL_DIR,
    dtype=dtype,
)
model.eval()

构造生成配置：

generation_config = GenerationConfig(
    max_new_tokens=96,
    do_sample=False,
    pad_token_id=tokenizer.pad_token_id,
    eos_token_id=tokenizer.eos_token_id,
)

生成：

prompt = format_prompt("什么是学习率？")
inputs = tokenizer(prompt, return_tensors="pt").to(model.device)

with torch.no_grad():
    output_ids = model.generate(
        **inputs,
        generation_config=generation_config,
    )

new_tokens = output_ids[0, inputs["input_ids"].shape[1]:]
answer = tokenizer.decode(new_tokens, skip_special_tokens=True)
print(answer)

切片非常重要：

output_ids[0, prompt_length:]

否则 decode() 会同时返回 Prompt 和模型新生成的文本。

25. Greedy 与 Sampling 怎样选择

比较微调前后时，优先使用确定性的 Greedy：

GenerationConfig(
    do_sample=False,
    max_new_tokens=96,
)

这样相同 Prompt 的结果更容易比较。

需要开放式表达时再使用：

GenerationConfig(
    do_sample=True,
    temperature=0.8,
    top_p=0.9,
    max_new_tokens=96,
)

不要把 Sampling 的随机差异当作 SFT 效果。

26. 微调前后应该怎样对比

保存一组固定测试 Prompt：

EVAL_PROMPTS = [
    "用一句话解释什么是验证集。",
    "为什么学习率太大会导致训练不稳定？",
    "给出两种缓解过拟合的方法。",
]

固定：

相同 Prompt 模板。
相同 Tokenizer。
相同 GenerationConfig。
相同随机种子，或直接使用 Greedy。
相同最大生成长度。

记录：

Base Model 输出
SFT Model 输出
目标行为
错误类型

27. 完整代码

完整项目代码位于：

source/code/nlp-blog/sft_transformers.py

运行：

python source/code/nlp-blog/sft_transformers.py

脚本会依次执行：

创建本地 Dataset
  -> 加载 Tokenizer
  -> 构造 Completion-only labels
  -> 检查一条样本
  -> 检查一个 batch
  -> 加载 Base Model
  -> 训练与验证
  -> 保存完整模型
  -> 重新加载并生成

默认示例用于学习流程。实际下载约 0.5B 模型并进行 Full Fine-tuning 仍需要合适的网络、内存和 GPU 环境。

28. 常见错误

28.1 Prompt 与 Answer 都参与 loss

如果目标是 Completion-only，应检查 Prompt labels 是否为 -100。

28.2 把 input_ids 中的 Prompt 也改成 -100

input_ids 必须是合法词表 id。-100 只应出现在 labels 中。

28.3 Padding labels 使用 pad_token_id

这会训练模型预测大量 Padding。Padding labels 应设为 -100。

28.4 手动提前 shift labels

AutoModelForCausalLM 通常已经内部 shift。外部再次 shift 会错位。

28.5 截断后没有 Answer token

检查每条样本：

sum(label != -100 for label in labels) > 0

28.6 重复添加特殊 token

Prompt 与 Answer 分开编码时，应明确哪一段添加 BOS；使用 Chat Template 后更要避免再次增加重复 special token。

28.7 忘记 EOS

模型可能学不会在回答末尾停止。

28.8 开启 Gradient Checkpointing 却保留 use_cache

训练前设置：

model.config.use_cache = False

推理前再恢复。

28.9 CPU 运行看似卡住

0.5B 模型的全参数前向、反向和优化器状态对 CPU 很重。教学代码不是轻量性能基准。

28.10 用十几条数据判断模型能力

小数据只验证代码和过拟合能力，不能证明真实泛化。

29. Full Fine-tuning 的显存都花在哪里

训练显存不只有模型权重：

模型参数
梯度
优化器状态
前向激活
临时 logits
CUDA kernel workspace

特别是 Causal LM logits：

$$
[B,T,V]
$$

当词表 $V$ 很大时，也会占用明显显存。

常见优化包括：

BF16 / FP16。
Gradient Accumulation。
Gradient Checkpointing。
缩短 max_length。
动态 Padding。
Packing。
Flash Attention / SDPA。
FSDP / DeepSpeed。
LoRA / QLoRA。

下一篇重点使用 LoRA，减少可训练参数和优化器状态。

30. 怎样把本篇改成 Chat Template 训练

本篇手写文本模板是为了学习边界。

真实模型可改为：

messages = [
    {"role": "user", "content": instruction},
    {"role": "assistant", "content": response},
]

然后：

tokenizer.apply_chat_template(
    messages,
    tokenize=True,
    add_generation_prompt=False,
)

难点转变为：

怎样准确得到 assistant token 对应的 mask？

可以选择：

使用模板提供的 assistant mask。
使用 Prompt 与 Completion 分离格式。
使用 TRL SFTTrainer 的 completion_only_loss 或 assistant_only_loss。

第三篇采用第 3 种方式。

31. Trainer 与 SFTTrainer 的关系

原生 Trainer：

负责训练循环
但要求我提前准备好 input_ids / labels

TRL SFTTrainer：

继承并扩展 Trainer
额外理解 text / prompt-completion / messages 数据
自动应用 Chat Template
自动构造 completion 或 assistant mask
支持 Packing 和 PEFT

所以学习顺序应是：

先理解 Trainer 接收到什么
  -> 再理解 SFTTrainer 替我生成了什么

32. 总结

本篇使用原生 Hugging Face Transformers 完成了 SFT 的完整闭环：

instruction / response
  -> format_prompt
  -> Prompt 与 Answer 分开 Tokenize
  -> input_ids
  -> Prompt labels = -100
  -> Answer labels = token ids
  -> Collator 动态 Padding
  -> AutoModelForCausalLM.forward
  -> 内部 label shift + CrossEntropyLoss
  -> Trainer 反向传播
  -> save_pretrained
  -> GenerationConfig + generate

最值得记住的并不是 Trainer(...) 的参数，而是这组不变量：

input_ids、attention_mask、labels 形状一致
Prompt 保留在 input_ids 中
Prompt 对应 labels 使用 -100
Answer 至少包含一个有效目标 token
Padding 对应 attention_mask=0、labels=-100
训练时不需要 Sampling
推理时不再传 labels

下一篇将用 TRL SFTTrainer 和 PEFT LoRA 重写同一条流程，并解释它自动处理了哪些步骤、哪些责任仍然必须由我承担。