1. LoRA微調

loader：

# -*- coding: utf-8 -*-import json
import re
import os
import torch
import numpy as np
from torch.utils.data import Dataset, DataLoader
from transformers import BertTokenizer
"""
數據加載
"""class DataGenerator:def __init__(self, data_path, config):self.config = configself.path = data_pathself.index_to_label = {0: '家居', 1: '房產', 2: '股票', 3: '社會', 4: '文化',5: '國際', 6: '教育', 7: '軍事', 8: '彩票', 9: '旅游',10: '體育', 11: '科技', 12: '汽車', 13: '健康',14: '娛樂', 15: '財經', 16: '時尚', 17: '游戲'}self.label_to_index = dict((y, x) for x, y in self.index_to_label.items())self.config["class_num"] = len(self.index_to_label)if self.config["model_type"] == "bert":self.tokenizer = BertTokenizer.from_pretrained(config["pretrain_model_path"])self.vocab = load_vocab(config["vocab_path"])self.config["vocab_size"] = len(self.vocab)self.load()def load(self):self.data = []with open(self.path, encoding="utf8") as f:for line in f:line = json.loads(line)tag = line["tag"]label = self.label_to_index[tag]title = line["title"]if self.config["model_type"] == "bert":input_id = self.tokenizer.encode(title, max_length=self.config["max_length"], pad_to_max_length=True)else:input_id = self.encode_sentence(title)input_id = torch.LongTensor(input_id)label_index = torch.LongTensor([label])self.data.append([input_id, label_index])returndef encode_sentence(self, text):input_id = []for char in text:input_id.append(self.vocab.get(char, self.vocab["[UNK]"]))input_id = self.padding(input_id)return input_id#補齊或截斷輸入的序列，使其可以在一個batch內運算def padding(self, input_id):input_id = input_id[:self.config["max_length"]]input_id += [0] * (self.config["max_length"] - len(input_id))return input_iddef __len__(self):return len(self.data)def __getitem__(self, index):return self.data[index]def load_vocab(vocab_path):token_dict = {}with open(vocab_path, encoding="utf8") as f:for index, line in enumerate(f):token = line.strip()token_dict[token] = index + 1  #0留給padding位置，所以從1開始return token_dict#用torch自帶的DataLoader類封裝數據
def load_data(data_path, config, shuffle=True):dg = DataGenerator(data_path, config)dl = DataLoader(dg, batch_size=config["batch_size"], shuffle=shuffle)return dlif __name__ == "__main__":from config import Configdg = DataGenerator("valid_tag_news.json", Config)print(dg[1])

model：

import torch.nn as nn
from config import Config
from transformers import AutoTokenizer, AutoModelForSequenceClassification, AutoModel
from torch.optim import Adam, SGDTorchModel = AutoModelForSequenceClassification.from_pretrained(Config["pretrain_model_path"])def choose_optimizer(config, model):optimizer = config["optimizer"]learning_rate = config["learning_rate"]if optimizer == "adam":return Adam(model.parameters(), lr=learning_rate)elif optimizer == "sgd":return SGD(model.parameters(), lr=learning_rate)

evaluate：

# -*- coding: utf-8 -*-
import torch
from loader import load_data"""
模型效果測試
"""class Evaluator:def __init__(self, config, model, logger):self.config = configself.model = modelself.logger = loggerself.valid_data = load_data(config["valid_data_path"], config, shuffle=False)self.stats_dict = {"correct":0, "wrong":0}  #用于存儲測試結果def eval(self, epoch):self.logger.info("開始測試第%d輪模型效果：" % epoch)self.model.eval()self.stats_dict = {"correct": 0, "wrong": 0}  # 清空上一輪結果for index, batch_data in enumerate(self.valid_data):if torch.cuda.is_available():batch_data = [d.cuda() for d in batch_data]input_ids, labels = batch_data   #輸入變化時這里需要修改，比如多輸入，多輸出的情況with torch.no_grad():pred_results = self.model(input_ids)[0]self.write_stats(labels, pred_results)acc = self.show_stats()return accdef write_stats(self, labels, pred_results):# assert len(labels) == len(pred_results)for true_label, pred_label in zip(labels, pred_results):pred_label = torch.argmax(pred_label)# print(true_label, pred_label)if int(true_label) == int(pred_label):self.stats_dict["correct"] += 1else:self.stats_dict["wrong"] += 1returndef show_stats(self):correct = self.stats_dict["correct"]wrong = self.stats_dict["wrong"]self.logger.info("預測集合條目總量：%d" % (correct +wrong))self.logger.info("預測正確條目：%d，預測錯誤條目：%d" % (correct, wrong))self.logger.info("預測準確率：%f" % (correct / (correct + wrong)))self.logger.info("--------------------")return correct / (correct + wrong)

?main：

# -*- coding: utf-8 -*-import torch
import os
import random
import os
import numpy as np
import torch.nn as nn
import logging
from config import Config
from model import TorchModel, choose_optimizer
from evaluate import Evaluator
from loader import load_data
from peft import get_peft_model, LoraConfig, \PromptTuningConfig, PrefixTuningConfig, PromptEncoderConfig #[DEBUG, INFO, WARNING, ERROR, CRITICAL]
logging.basicConfig(level=logging.INFO, format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)"""
模型訓練主程序
"""seed = Config["seed"]
random.seed(seed)
np.random.seed(seed)
torch.manual_seed(seed)
torch.cuda.manual_seed_all(seed)def main(config):#創建保存模型的目錄if not os.path.isdir(config["model_path"]):os.mkdir(config["model_path"])#加載訓練數據train_data = load_data(config["train_data_path"], config)#加載模型model = TorchModel#大模型微調策略tuning_tactics = config["tuning_tactics"]if tuning_tactics == "lora_tuning":peft_config = LoraConfig(r=8,lora_alpha=32,lora_dropout=0.1,target_modules=["query", "key", "value"])elif tuning_tactics == "p_tuning":peft_config = PromptEncoderConfig(task_type="SEQ_CLS", num_virtual_tokens=10)elif tuning_tactics == "prompt_tuning":peft_config = PromptTuningConfig(task_type="SEQ_CLS", num_virtual_tokens=10)elif tuning_tactics == "prefix_tuning":peft_config = PrefixTuningConfig(task_type="SEQ_CLS", num_virtual_tokens=10)model = get_peft_model(model, peft_config)# print(model.state_dict().keys())if tuning_tactics == "lora_tuning":# lora配置會凍結原始模型中的所有層的權重，不允許其反傳梯度# 但是事實上我們希望最后一個線性層照常訓練，只是bert部分被凍結，所以需要手動設置for param in model.get_submodule("model").get_submodule("classifier").parameters():param.requires_grad = True# 標識是否使用gpucuda_flag = torch.cuda.is_available()if cuda_flag:logger.info("gpu可以使用，遷移模型至gpu")model = model.cuda()#加載優化器optimizer = choose_optimizer(config, model)#加載效果測試類evaluator = Evaluator(config, model, logger)#訓練for epoch in range(config["epoch"]):epoch += 1model.train()logger.info("epoch %d begin" % epoch)train_loss = []for index, batch_data in enumerate(train_data):if cuda_flag:batch_data = [d.cuda() for d in batch_data]optimizer.zero_grad()input_ids, labels = batch_data   #輸入變化時這里需要修改，比如多輸入，多輸出的情況output = model(input_ids)[0]loss = nn.CrossEntropyLoss()(output, labels.view(-1))loss.backward()optimizer.step()train_loss.append(loss.item())if index % int(len(train_data) / 2) == 0:logger.info("batch loss %f" % loss)logger.info("epoch average loss: %f" % np.mean(train_loss))acc = evaluator.eval(epoch)model_path = os.path.join(config["model_path"], "%s.pth" % tuning_tactics)save_tunable_parameters(model, model_path)  #保存模型權重return accdef save_tunable_parameters(model, path):saved_params = {k: v.to("cpu")for k, v in model.named_parameters()if v.requires_grad}torch.save(saved_params, path)if __name__ == "__main__":main(Config)

pred：

import torch
import logging
from model import TorchModel
from peft import get_peft_model, LoraConfig, PromptTuningConfig, PrefixTuningConfig, PromptEncoderConfigfrom evaluate import Evaluator
from config import Configlogging.basicConfig(level=logging.INFO, format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)#大模型微調策略
tuning_tactics = Config["tuning_tactics"]print("正在使用 %s"%tuning_tactics)if tuning_tactics == "lora_tuning":peft_config = LoraConfig(r=8,lora_alpha=32,lora_dropout=0.1,target_modules=["query", "key", "value"])
elif tuning_tactics == "p_tuning":peft_config = PromptEncoderConfig(task_type="SEQ_CLS", num_virtual_tokens=10)
elif tuning_tactics == "prompt_tuning":peft_config = PromptTuningConfig(task_type="SEQ_CLS", num_virtual_tokens=10)
elif tuning_tactics == "prefix_tuning":peft_config = PrefixTuningConfig(task_type="SEQ_CLS", num_virtual_tokens=10)#重建模型
model = TorchModel
# print(model.state_dict().keys())
# print("====================")model = get_peft_model(model, peft_config)
# print(model.state_dict().keys())
# print("====================")state_dict = model.state_dict()#將微調部分權重加載
if tuning_tactics == "lora_tuning":loaded_weight = torch.load('output/lora_tuning.pth')
elif tuning_tactics == "p_tuning":loaded_weight = torch.load('output/p_tuning.pth')
elif tuning_tactics == "prompt_tuning":loaded_weight = torch.load('output/prompt_tuning.pth')
elif tuning_tactics == "prefix_tuning":loaded_weight = torch.load('output/prefix_tuning.pth')print(loaded_weight.keys())
state_dict.update(loaded_weight)#權重更新后重新加載到模型
model.load_state_dict(state_dict)#進行一次測試
model = model.cuda()
evaluator = Evaluator(Config, model, logger)
evaluator.eval(0)