基于Bert模型的增量微調3-使用csv文件訓練

我們使用weibo評價數據，8分類的csv格式數據集。

一、創建數據集合

使用csv格式的數據作為數據集。

1、創建MydataCSV.py

from  torch.utils.data import Dataset
from datasets import load_datasetclass MyDataset(Dataset):#初始化數據集def __init__(self, split):# 加載csv數據self.dataset=load_dataset(path="csv",data_files=f"D:\Test\LLMTrain\day03\data\Weibo/{split}.csv", split= "train")# 返回數據集長度def __len__(self):return len(self.dataset)# 對每條數據單獨進行數據處理def __getitem__(self, idx):text=self.dataset[idx]["text"]label=self.dataset[idx]["label"]return  text,labelif __name__== "__main__":train_dataset=MyDataset("test")for i in range(10):print(train_dataset[i])

二、處理模型

我們使用8分類任務

創建netCSV.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__()#設計全連接網絡，實現8分類任務self.fc = torch.nn.Linear(768,8)#使用模型處理數據（執行前向計算）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,8) 。

我們是對大模型做增量微調訓練，所以需要凍結Bert模型的參數，讓其不參與訓練。所以使用?

with torch.no_grad()。

我們定義一個下游任務增量模型Model類，繼承 torch.nn.Module。

三、訓練的代碼

1、創建目錄params

存放訓練后的結果。

2、寫代碼

創建train_val_csv.py

#模型訓練
import torch
from MyDataCSV import MyDataset
from torch.utils.data import DataLoader
from netCSV 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_lengthpadding="max_length",# 可取值為tf,pt,np,默認為listreturn_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.0for 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.0val_loss = 0.0for 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_acctorch.save(model.state_dict(),"params1/best_bert.pth")print(f"EPOCH:{epoch}:保存最優參數：acc{best_val_acc}")#保存最后一輪參數torch.save(model.state_dict(), "params1/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_lengthpadding="max_length",# 可取值為tf,pt,np,默認為listreturn_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_idsdef test():#加載模型訓練參數model.load_state_dict(torch.load("params/best_bert.pth"))#開啟測試模型model.eval()while True:data = input("請輸入測試數據（輸入‘q’退出）：")if data=='q':print("測試結束")breakinput_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測試集里的數據進行驗證，或許輸入其他的文本驗證。