基于Bert模型的增量微调3-训练
目录
基于Bert模型的增量微调3-训练
一、创建数据集合
使用hugging face平台的数据集作为训练的数据集。
1、创建Mydata.py
from torch.utils.data import Dataset
from datasets import load_from_disk
class MyDataset(Dataset):
#初始化数据集
def __init__(self, split):
# 从磁盘读取数据
dataset_path = r"D:\Test\LLMTrain\day03\data\chn_senti_corp"
self.dataset=load_from_disk(dataset_path)
if split=="train":
self.dataset=self.dataset["train"]
elif split=="test":
self.dataset=self.dataset["test"]
elif split=="validation":
self.dataset=self.dataset["validation"]
else:
print("数据名错误！")
# 返回数据集长度
def __len__(self):
return len(self.dataset)
# 对每条数据单独进行数据处理
def __getitem__(self, idx):
text=self.dataset[idx]["text"]
label=self.dataset[idx]["label"]
return text,label
if __name__== "__main__":
train_dataset=MyDataset("test")
for i in range(10):
print(train_dataset[i])二、处理模型
我们使用2分类任务
创建net.py
import torch
from transformers import BertModel
#定义设备信息
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(DEVICE)
#加载预训练模型
path1=r"D:\Test\LLMTrain\day03\model\bert-base-chinese\models--google-bert--bert-base-chinese\snapshots\c30a6ed22ab4564dc1e3b2ecbf6e766b0611a33f"
pretrained = BertModel.from_pretrained(path1).to(DEVICE)
print(pretrained)
#定义下游任务（增量模型）
class Model(torch.nn.Module):
def __init__(self):
super().__init__()
#设计全连接网络，实现二分类任务
self.fc = torch.nn.Linear(768,2)
#使用模型处理数据（执行前向计算）
def forward(self,input_ids,attention_mask,token_type_ids):
#冻结Bert模型的参数，让其不参与训练
with torch.no_grad():
out = pretrained(input_ids=input_ids,attention_mask=attention_mask,token_type_ids=token_type_ids)
#增量模型参与训练
out = self.fc(out.last_hidden_state[:,0])
return out若是8分类任务，则需要将 self.fc=torch.nn.Liner768,2)中的2改为8 。
我们是对大模型做增量微调训练，所以需要冻结Bert模型的参数，让其不参与训练。所以使用
with torch.no_grad()。
我们定义一个下游任务增量模型Model类，继承 torch.nn.Module。
三、训练的代码
1、创建目录params
存放训练后的结果。
2、写代码
创建train_val.py
#模型训练
import torch
from MyData import MyDataset
from torch.utils.data import DataLoader
from net import Model
from transformers import BertTokenizer,AdamW
#定义设备信息
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
#定义训练的轮次(将整个数据集训练完一次为一轮)
EPOCH = 30000
#加载字典和分词器
token = BertTokenizer.from_pretrained(r"D:\Test\LLMTrain\day03\model\bert-base-chinese\models--google-bert--bert-base-chinese\snapshots\c30a6ed22ab4564dc1e3b2ecbf6e766b0611a33f")
#将传入的字符串进行编码
def collate_fn(data):
sents = [i[0]for i in data]
label = [i[1] for i in data]
#编码
data = token.batch_encode_plus(
batch_text_or_text_pairs=sents,
# 当句子长度大于max_length(上限是model_max_length)时，截断
truncation=True,
max_length=512,
# 一律补0到max_length
padding="max_length",
# 可取值为tf,pt,np,默认为list
return_tensors="pt",
# 返回序列长度
return_length=True
)
input_ids = data["input_ids"]
attention_mask = data["attention_mask"]
token_type_ids = data["token_type_ids"]
label = torch.LongTensor(label)
return input_ids,attention_mask,token_type_ids,label
#创建数据集
train_dataset = MyDataset("train")
train_loader = DataLoader(
dataset=train_dataset,
#训练批次
batch_size=50,
#打乱数据集
shuffle=True,
#舍弃最后一个批次的数据，防止形状出错
drop_last=True,
#对加载的数据进行编码
collate_fn=collate_fn
)
#创建验证数据集
val_dataset = MyDataset("validation")
val_loader = DataLoader(
dataset=val_dataset,
#训练批次
batch_size=50,
#打乱数据集
shuffle=True,
#舍弃最后一个批次的数据，防止形状出错
drop_last=True,
#对加载的数据进行编码
collate_fn=collate_fn
)
if __name__ == '__main__':
#开始训练
print(DEVICE)
model = Model().to(DEVICE)
#定义优化器
optimizer = AdamW(model.parameters())
#定义损失函数
loss_func = torch.nn.CrossEntropyLoss()
#初始化验证最佳准确率
best_val_acc = 0.0
for epoch in range(EPOCH):
for i,(input_ids,attention_mask,token_type_ids,label) in enumerate(train_loader):
#将数据放到DVEVICE上面
input_ids, attention_mask, token_type_ids, label = input_ids.to(DEVICE),attention_mask.to(DEVICE),token_type_ids.to(DEVICE),label.to(DEVICE)
#前向计算（将数据输入模型得到输出）
out = model(input_ids,attention_mask,token_type_ids)
#根据输出计算损失
loss = loss_func(out,label)
#根据误差优化参数
optimizer.zero_grad()
loss.backward()
optimizer.step()
#每隔5个批次输出训练信息
if i%5 ==0:
out = out.argmax(dim=1)
#计算训练精度
acc = (out==label).sum().item()/len(label)
print(f"epoch:{epoch},i:{i},loss:{loss.item()},acc:{acc}")
#验证模型（判断模型是否过拟合）
#设置为评估模型
model.eval()
#不需要模型参与训练
with torch.no_grad():
val_acc = 0.0
val_loss = 0.0
for i, (input_ids, attention_mask, token_type_ids, label) in enumerate(val_loader):
# 将数据放到DVEVICE上面
input_ids, attention_mask, token_type_ids, label = input_ids.to(DEVICE), attention_mask.to(
DEVICE), token_type_ids.to(DEVICE), label.to(DEVICE)
# 前向计算（将数据输入模型得到输出）
out = model(input_ids, attention_mask, token_type_ids)
# 根据输出计算损失
val_loss += loss_func(out, label)
#根据数据，计算验证精度
out = out.argmax(dim=1)
val_acc+=(out==label).sum().item()
val_loss/=len(val_loader)
val_acc/=len(val_loader)
print(f"验证集：loss:{val_loss},acc:{val_acc}")
# #每训练完一轮，保存一次参数
# torch.save(model.state_dict(),f"params/{epoch}_bert.pth")
# print(epoch,"参数保存成功！")
#根据验证准确率保存最优参数
if val_acc > best_val_acc:
best_val_acc = val_acc
torch.save(model.state_dict(),"params/best_bert.pth")
print(f"EPOCH:{epoch}:保存最优参数：acc{best_val_acc}")
#保存最后一轮参数
torch.save(model.state_dict(), "params/last_bert.pth")
print(f"EPOCH:{epoch}:最后一轮参数保存成功！")3、执行代码
这个过程需等待很久，若是使用cuda环境，显存越大，速度越快。
train_loader的训练批次batch_size=50，这个数值是根据电脑的配置来的，数值越大越好，只要不超过显存或者内存的90%即可。
四、使用训练好的模型
我们写一个控制台程序，也可以使用FastAPI。创建run.py文件。
#模型使用接口（主观评估）
#模型训练
import torch
from net import Model
from transformers import BertTokenizer
#定义设备信息
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
#加载字典和分词器
token = BertTokenizer.from_pretrained(r"D:\Test\LLMTrain\day03\model\bert-base-chinese\models--google-bert--bert-base-chinese\snapshots\c30a6ed22ab4564dc1e3b2ecbf6e766b0611a33f")
model = Model().to(DEVICE)
names = ["负向评价","正向评价"]
#将传入的字符串进行编码
def collate_fn(data):
sents = []
sents.append(data)
#编码
data = token.batch_encode_plus(
batch_text_or_text_pairs=sents,
# 当句子长度大于max_length(上限是model_max_length)时，截断
truncation=True,
max_length=512,
# 一律补0到max_length
padding="max_length",
# 可取值为tf,pt,np,默认为list
return_tensors="pt",
# 返回序列长度
return_length=True
)
input_ids = data["input_ids"]
attention_mask = data["attention_mask"]
token_type_ids = data["token_type_ids"]
return input_ids,attention_mask,token_type_ids
def test():
#加载模型训练参数
model.load_state_dict(torch.load("params/best_bert.pth"))
#开启测试模型
model.eval()
while True:
data = input("请输入测试数据（输入‘q’退出）：")
if data=='q':
print("测试结束")
break
input_ids,attention_mask,token_type_ids = collate_fn(data)
input_ids, attention_mask, token_type_ids = input_ids.to(DEVICE),attention_mask.to(DEVICE),token_type_ids.to(DEVICE)
#将数据输入到模型，得到输出
with torch.no_grad():
out = model(input_ids,attention_mask,token_type_ids)
out = out.argmax(dim=1)
print("模型判定：",names[out],"\n")
if __name__ == '__main__':
test()运行程序 ，输入test测试集里的数据进行验证，或许输入其他的文本验证。
正确率还是非常棒的。