我們一起聊聊大模型 SFT 有監(jiān)督微調(diào)教程
開源地址：???https://github.com/JieShenAI/csdn/tree/main/25/02/SFT???
??train.ipynb???：模型有監(jiān)督微調(diào)的代碼??infer.ipynb??: 模型訓(xùn)練完成后，進(jìn)行推理的代碼\
{
'instruct': '請(qǐng)你給敖丙寫一首詩(shī)：',
'input': '碧海生龍子，云中舞雪霜。',
'label': '恩仇難兩忘，何處是家鄉(xiāng)？'
}預(yù)訓(xùn)練與有監(jiān)督微調(diào)對(duì)比
兩者的訓(xùn)練數(shù)據(jù)，大部分都一模一樣，維度在 label 部分，SFT 需要把指令部分的 label 設(shè)置為-100。
import json
from typing import List, Dict, Sequence
import torch
from torch.nn.utils.rnn import pad_sequence
import transformers
from transformers import TrainingArguments, Trainer, AutoModelForCausalLM, AutoTokenizer
from torch.utils.data import Dataset
from dataclasses import dataclass
IGNORE_INDEX = -100
device = "cuda:0"if torch.cuda.is_available() else"cpu"
model_dir = r"Qwen/Qwen2.5-0.5B"
model = AutoModelForCausalLM.from_pretrained(model_dir)
model = model.to("cuda:0")
tokenizer = AutoTokenizer.from_pretrained(model_dir, padding_side="right")
tokenizer.add_special_tokens({
"pad_token": "[PAD]"
})
# 數(shù)據(jù)加載
with open("data.json.demo", "r") as f:
data = json.load(f)自定義數(shù)據(jù)集
class PreTrainDataset(Dataset):
def __init__(self, data: List):
super().__init__()
self.data = data
def __len__(self):
return len(self.data)
def __getitem__(self, idx) -> List[Dict]:
item = self.data[idx]
text = item["instruct"] + item["input"] + item["label"] + tokenizer.eos_token
text_token = tokenizer(
text,
return_tensors="pt",
padding="longest",
max_length=tokenizer.model_max_length,
truncatinotallow=True,
)
label = text_token["input_ids"].clone()
instruct = item["instruct"] + item["input"]
instruct_token = tokenizer(
instruct,
return_tensors="pt",
padding="longest",
max_length=tokenizer.model_max_length,
truncatinotallow=True,
)
instruct_len = instruct_token["input_ids"].size(-1)
label[:, :instruct_len] = -100
text_token["labels"] = label
return text_token
dataset = PreTrainDataset(data)
dataset[0]因?yàn)?tokenizer 對(duì)文本進(jìn)行encode的時(shí)候，并不是一個(gè)詞一個(gè)token，會(huì)出現(xiàn)多個(gè)詞對(duì)應(yīng)一個(gè)token的情況。為了確定指令部分的token長(zhǎng)度，單獨(dú)對(duì)指令部分的文本計(jì)算一次的encode。然后使用切片 ??label[:, :instruct_len] = -100?? 把指令部分的 label 設(shè)置為 -100 不計(jì)算 loss。
查看第一個(gè)數(shù)據(jù):
# 查看第一個(gè)原始數(shù)據(jù)
data[0]輸出：
{'instruct': '請(qǐng)你給哪吒寫一首詩(shī)：',
'input': '哪吒降世，意氣飛揚(yáng)。\n逆天改命，破障沖霄。',
'label': '紅綾纏腕，風(fēng)火踏浪。\n不屈不悔，笑傲蒼茫。'}
# 查看需要計(jì)算loss的文本
test_label = dataset[0][0]["label"]
test_label = test_label[test_label != -100]
tokenizer.decode(test_label)輸出:
'紅綾纏腕，風(fēng)火踏浪。\n不屈不悔，笑傲蒼茫。<|endoftext|>'
# 查看label -100位置對(duì)應(yīng)的input_ids的文本
test_input_ids = dataset[0][0]["input_ids"]
test_label = dataset[0][0]["labels"]
test_input_ids = test_input_ids[test_label == -100]
tokenizer.decode(test_input_ids)
# label -100 位置的都是用戶的指令不參與 loss 計(jì)算輸出：
'請(qǐng)你給哪吒寫一首詩(shī)：哪吒降世，意氣飛揚(yáng)。\n逆天改命，破障沖霄。'DataCollatorForSFTDataset
下面是使用 ??pad_sequence?? 對(duì) tensor 進(jìn)行填充的一個(gè)示例。batch 放在第一個(gè)維度，用 0 進(jìn)行填充，在右邊進(jìn)行填充。
pad_sequence(
[torch.randn(2), torch.randn(3), torch.randn(4)],
batch_first=True,
padding_value=0,
padding_side="right",
)輸出：
tensor([[-0.3421, 0.4131, 0.0000, 0.0000],
[-0.1345, 1.2843, 1.0892, 0.0000],
[-0.0567, -0.6993, -0.9386, 1.1316]])使用 ??pad_sequence?? 在 DataCollatorForSFTDataset中，對(duì) tensor 進(jìn)行拼接與填充。
@dataclass
class DataCollatorForSFTDataset(object):
tokenizer: transformers.PreTrainedTokenizer
def __call__(self, items: Sequence) -> Dict[str, torch.Tensor]:
# pad_sequence 不支持多維tensor，進(jìn)行維度壓縮 squeeze
# input_ids, attention_mask = [
# [item.squeeze(0) for item in tokens[k]]
# for k in ["input_ids", "attention_mask"]
# ]
input_ids = [item["input_ids"].squeeze(0) for item in items]
attention_mask = [item["attention_mask"].squeeze(0) for item in items]
label = [item["label"].squeeze(0) for item in items]
input_ids = pad_sequence(
input_ids,
batch_first=True,
padding_value=tokenizer.pad_token_id,
padding_side="right",
)
attention_mask = pad_sequence(
attention_mask,
batch_first=True,
padding_value=0,
padding_side="right",
)
label = pad_sequence(
label,
batch_first=True,
padding_value=-100,
padding_side="right",
)
return {
"input_ids": input_ids,
"attention_mask": attention_mask,
"labels": label,
}注意: 在返回的字典中，要用 ??labels??? 而不是 ??label??。
驗(yàn)證一下，??DataCollatorForSFTDataset?? 的效果：
DataCollatorForSFTDataset(tokenizer=tokenizer)([dataset[0], dataset[1], dataset[2]])模型訓(xùn)練
args = TrainingArguments(
output_dir=r"C:\Users\1\Desktop\train_model_output\Qwen2.5-0.5B\SFT_output",
num_train_epochs=10,
per_device_train_batch_size=2,
save_safetensors=True,
logging_strategy="epoch",
)??processing_class?? 是新參數(shù)名，使用舊參數(shù)名也可以：
trainer = Trainer(
model=model,
processing_class=tokenizer,
args=args,
train_dataset=dataset,
eval_dataset=None,
data_collator=DataCollatorForSFTDataset(tokenizer=tokenizer),
)
train_result = trainer.train()
查看模型訓(xùn)練的結(jié)果：
train_result.metrics保存訓(xùn)練完成的模型：
trainer.save_state()
trainer.save_model(output_dir=args.output_dir)
tokenizer.save_pretrained(args.output_dir)模型推理
看一下模型有監(jiān)督微調(diào)的效果。對(duì)比一下，預(yù)訓(xùn)練與有監(jiān)督微調(diào)，模型在進(jìn)行推理的時(shí)候的區(qū)別：
- 預(yù)訓(xùn)練的模型，對(duì)于輸入的文本都可以繼續(xù)續(xù)寫出原文；
- 有監(jiān)督微調(diào)，只能根據(jù)指令寫出對(duì)應(yīng)的答案；無(wú)法根據(jù)指令的前半部分，寫出指令的后半部分：
instruct + label 作為指令部分，label 是指令的答案。若SFT微調(diào)后的大模型，輸入 instruct + label 能得到 label，說(shuō)明模型微調(diào)有效。當(dāng)給SFT微調(diào)后的大模型輸入instruct，模型應(yīng)該輸出label中的文本，但不能輸出input的文本，就能說(shuō)明label設(shè)置為-100，沒有計(jì)算指令部分loss。
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
device = "cuda:0"if torch.cuda.is_available() else"cpu"
train_model = r"C:\Users\1\Desktop\train_model_output\Qwen2.5-0.5B\SFT_output"
model = AutoModelForCausalLM.from_pretrained(train_model)
model = model.to(device)
tokenizer = AutoTokenizer.from_pretrained(train_model, padding_side="right")
tokenizer.add_special_tokens({"pad_token": "[PAD]"})
import json
with open("data.json", "r") as f:
data =json.load(f)
data
def infer(text):
input_ids = tokenizer(text, return_tensors="pt").to(model.device)
generated_ids = model.generate(**input_ids)
generated_ids = [
output_ids[len(input_ids) :]
for input_ids, output_ids in zip(input_ids.input_ids, generated_ids)
]
response = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]
return response
print("=" * 50 + "instruct" + "=" * 50)
for item in data:
# instruct + input -> label
instruct, input, label = item["instruct"], item["input"], item["label"]
print(f"text_input: {instruct + input}")
print(f"predict: {infer(instruct + input)}")
print(f"label: {label}")
print("-" * 101)部分輸出結(jié)果：
text_input: 請(qǐng)你給哪吒寫一首詩(shī)：哪吒降世，意氣飛揚(yáng)。
逆天改命，破障沖霄。
predict: 紅綾纏腕，風(fēng)火踏浪。
不屈不悔，笑傲蒼茫。
label: 紅綾纏腕，風(fēng)火踏浪。
不屈不悔，笑傲蒼茫。模型能夠根據(jù)指令，完成詩(shī)歌下半部分的寫作。
print("=" * 50 + "instruct" + "=" * 50)
for item in data:
# instruct + input -> label
instruct, input, label = item["instruct"], item["input"], item["label"]
print(f"text_input: {instruct }")
print(f"predict: {infer(instruct)}")
print(f"label: {label}")
print("-" * 101)部分輸出：
text_input: 請(qǐng)你給哪吒寫一首詩(shī)：
predict: 紅綾纏腕，風(fēng)火踏浪。不屈不悔，笑傲蒼茫。
label: 紅綾纏腕，風(fēng)火踏浪。
不屈不悔，笑傲蒼茫。大模型只能輸出 label中的文本，模型不能輸出 input中的詩(shī)歌: ??哪吒降世，意氣飛揚(yáng)。逆天改命，破障沖霄。??這說(shuō)明模型沒有學(xué)到用戶指令部分的文本，這符合我們的預(yù)期。
本文轉(zhuǎn)載自??AI悠閑區(qū)??，作者：AI悠閑區(qū)
