Bert实战
total_num = sum(p.numel() for p in model.parameters()) # 统计模型所有参数的总数trainable_num = sum(p.numel() for p in model.parameters() if p.requires_grad) # 统计可训练参数数量p.numel():返回单个参数张量的元素个数(即参数数量)。该函数返回一个字典,包含
参数量分析和调用分析(预学习)
计算并验证 bert-base-chinese 模型的参数量,包括通过代码自动统计和手动公式计算两种方式,以此来理解 BERT 模型各组件的参数构成

引入transfomers框架的bert有关的包
from transformers import (BertPreTrainedModel, BertConfig,
BertForSequenceClassification, BertTokenizer,BertModel,
)
BertPreaTrainedModel这是所有 BERT 类模型的基类(父类),是一个抽象的基础框架
BertTokenizer核心定位:BERT 的文本分词器,是连接原始文本和模型输入的桥梁。
BertForSequenceClassification:基于 BERT 的序列分类专用模型
BertModel:是 BERT 最核心的部分,包含 Embedding 层和 12 层 Transformer Encoder。
定义参数量统计函数
def get_parameter_number(model):
total_num = sum(p.numel() for p in model.parameters()) # 统计模型所有参数的总数
trainable_num = sum(p.numel() for p in model.parameters() if p.requires_grad) # 统计可训练参数数量
return {'Total': total_num, 'Trainable': trainable_num}
p.numel():返回单个参数张量的元素个数(即参数数量)。
该函数返回一个字典,包含模型总参数量和可训练参数量(默认加载的预训练 BERT 所有参数都可训练)。

数据路径与引用
数据的路径与bert模型的路径
data_path = "jiudian.txt"
bert_path = 'bert-base-chinese'
bert = BertModel.from_pretrained(bert_path)
print(get_parameter_number(bert))
.from_pretrained():BertModel的核心静态方法,作用是加载预训练权重,而非从头初始化随机参数,简单来说就是不仅使用这个模型而且使用这么模型的输入
print(get_parameter_number(bert)):调用你定义的get_parameter_number函数,统计bert模型的参数量,并打印结果
# print(get_parameter_number(model.bert.embeddings))
# print(get_parameter_number(model.bert.encoder.layer[0].attention))
embedding层参数量

self-attention0层参数量

手算参数量
dim = 768
emb_para = 768*2 + 768*512 + 768*21128
self_att = 768*768*3 + 768*768
mlp = 768*3072 + 3072*768
bertlayers_para = 12 * (self_att + mlp)
# print(768*768*3 + 768*768)
print(bertlayers_para+emb_para)




12 层总参数 = 12×708 万 = 84934656(约 8493 万)
这个数值和代码自动统计的102267648略有差异,原因是手动计算未包含 LayerNorm、偏置项等小参数(这些参数数量占比极低,核心量级完全一致)。
数据特征输入(工具类1)
sklearn.model_selection为数据分割,可以分割训练集验证集
from torch.utils.data import DataLoader, Dataset ##数据加载设置包
from sklearn.model_selection import train_test_split #给X,Y 和分割比例, 分割出来一个训练集和验证机的X, Y
import torch ##数据转换为张量
定义读文件的函数
def read_file(path):
# 初始化两个空列表,用于存储提取的文本数据和标签
data = []
label = []
with open(path, "r", encoding="utf-8") as f:
##遍历文件的每一行,enumerate同时获取行号i和行内容line
for i, line in enumerate(f):
if i == 0:
continue ### 跳过第0行(文件的第一行,通常是表头,比如“标签,文本”)
if i > 200 and i< 7500:
continue
line = line.strip("\n")
line = line.split(",", 1) #把这句话,按照,分割, 1表示分割次数
data.append(line[1])
label.append(line[0])
print("读了%d的数据"%len(data))
return data, label
以read形式打开文件,编码方式为utf-8
line = line.strip("\n");line = line.split(",", 1) :出去数据部分的末尾\n换行符 ,且以,分割句子1表示分割次数
把分割后的句子按标签和内容放到了空数组中

数据处理函数
继承PyTorch的Dataset基类,自定义数据集的标准写法
class jdDataset(Dataset):
# 1. 初始化方法:接收数据和标签,做预处理
def __init__(self, data, label):
# 把文本数据(比如["房间宽敞", "卫生差"])赋值给实例变量self.X
self.X = data
# 把标签处理成PyTorch的LongTensor类型(整数张量),赋值给self.Y
self.Y = torch.LongTensor([int(i) for i in label])
# 2. 核心方法:按索引item获取单条数据
def __getitem__(self, item):
# 返回第item条的文本和对应的标签(一一对应)
return self.X[item], self.Y[item]
# 3. 核心方法:返回数据集的总长度
def __len__(self):
# 标签列表的长度就是数据集的总条数(data和label长度一致)
return len(self.Y)
分割数据与调用
传入path数据地址,通过batchsize规定每轮训练次数
训练集x,y与验证集x,y通过train_test_split函数分割,val_size=0.2表示分割给验证集20%的数据和对应的真值y
def get_data_loader(path, batchsize, val_size=0.2): #读入数据,分割数据。
data, label = read_file(path)
train_x, val_x, train_y, val_y = train_test_split(data, label, test_size=val_size, shuffle=True, stratify=label)
train_set = jdDataset(train_x, train_y)
val_set = jdDataset(val_x, val_y)
train_loader = DataLoader(train_set, batchsize, shuffle=True)
val_loader = DataLoader(val_set, batchsize, shuffle=True)
return train_loader, val_loader
得到train_set和val_set后调用jdDataset函数将val_y处理成Y张量标签,X处理成输入shuffle表示随机打乱
如果是main文件运行则获取
# 脚本入口判断:只有当脚本被直接运行时,才执行下面的代码
if __name__ == "__main__":
# 调用get_data_loader函数,传入文件路径和批量大小2
get_data_loader("../jiudian.txt", 2)
自定义文本分类模型bert设置(工具类二)
import torch # PyTorch核心库(基础框架)
import torch.nn as nn # PyTorch的神经网络模块(构建模型层)
from transformers import BertModel, BertTokenizer, BertConfig # BERT专用工具
# 继承PyTorch的nn.Module(所有自定义模型必须继承)
class myBertModel(nn.Module):
# 1. 初始化方法:定义模型结构和参数
def __init__(self, bert_path, num_class, device):
# 调用父类nn.Module的初始化(必须写)
super(myBertModel, self).__init__()
# 加载预训练的BERT基础模型(核心:复用预训练权重)
self.bert = BertModel.from_pretrained(bert_path)
注释掉的代码:我们自己设置bert的输入参数
# config = BertConfig.from_pretrained(bert_path)
# self.bert = BertModel(config)
# 保存设备(CPU/GPU),后续把数据移到对应设备
self.device = device
# 定义分类头:线性层,把BERT输出的768维向量→num_class维(分类类别数)
self.cls_head = nn.Linear(768, num_class)
# 加载BERT分词器,内置到模型中(方便直接处理原始文本)
self.tokenizer = BertTokenizer.from_pretrained(bert_path)
# 2. 前向传播方法:定义模型的计算流程(输入→输出)
def forward(self, text):
# 第一步:用内置分词器处理原始文本,转成BERT能识别的输入格式
input = self.tokenizer(
text, # 输入的原始文本(列表/字符串)
return_tensors="pt", # 返回PyTorch张量(而非列表)
truncation=True, # 超过max_length的文本截断
padding="max_length", # 不足max_length的文本补0到指定长度
max_length=128 # 文本最大长度设为128(BERT最大支持512)
)
# 第二步:把分词后的张量移到指定设备(CPU/GPU),避免设备不匹配报错
input_ids = input["input_ids"].to(self.device) # token的数字ID
token_type_ids = input['token_type_ids'].to(self.device)# 句子类型ID(单句为0)
attention_mask = input['attention_mask'].to(self.device)# 注意力掩码(区分真实token和补0的token)
# 第三步:将输入传入BERT模型,获取语义特征
sequence_out, pooler_out = self.bert(
input_ids=input_ids, # token ID
token_type_ids=token_type_ids,# 句子类型ID
attention_mask=attention_mask,# 注意力掩码
return_dict=False # 返回元组(而非字典),方便解包
) #return_dict
# 第四步:用分类头处理BERT的pooler_out([CLS] token的768维向量),得到分类预测
pred = self.cls_head(pooler_out)
# 返回预测结果(num_class维的向量,每个维度对应一个类别的得分)
return pred
self.bert = BertModel.from_pretrained(bert_path)这句话的意思是我们要bert大佬预训练模型的输入参数
self.cls_head = nn.Linear(768, num_class)将768维全连接成num_class维(类似于分类任务
self.tokenizer = BertTokenizer.from_pretrained(bert_path) 加载BERT分词器,内置到模型中(方便直接处理原始文本)
# 脚本入口判断:只有直接运行当前脚本时,才执行下面的代码
if __name__ == "__main__":
model = myBertModel("../bert-base-chinese", 2)
pred = model("今天天气真好")
num_class参数分类任务的类别数为 2(比如二分类:正面 / 负面、0/1)。
模型训练类
import torch # PyTorch核心库(基础计算)
import time # Python内置时间库(统计训练耗时)
import matplotlib.pyplot as plt # 绘图库(可视化训练曲线)
import numpy as np # 数值计算库(处理数组/矩阵)
from tqdm import tqdm # 进度条库(显示训练/验证进度)
此处的参数都会通过主函数调用传入,后面的代码都为同一类中
def train_val(para):
model = para['model'] # 待训练的模型(如CNN、Transformer等)
train_loader =para['train_loader'] # 训练集数据加载器(批处理数据)
val_loader = para['val_loader'] # 验证集数据加载器
scheduler = para['scheduler'] # 学习率调度器(调整学习率)
optimizer = para['optimizer'] # 优化器(如Adam、SGD,更新模型参数)
loss = para['loss'] # 损失函数(如CrossEntropyLoss)
epoch = para['epoch'] # 总训练轮次
device = para['device'] # 训练设备(cpu/cuda)
save_path = para['save_path'] # 模型保存路径
max_acc = para['max_acc'] # 初始化的最高验证准确率(用于对比保存最优模型)
val_epoch = para['val_epoch'] # 验证间隔(每val_epoch轮做一次验证)
plt_train_loss = [] # 记录每轮训练损失
plt_train_acc = [] # 记录每轮训练准确率
plt_val_loss = [] # 记录每轮验证损失
plt_val_acc = [] # 记录每轮验证准确率
val_rel = [] # 记录验证集的预测结果(代码中未后续使用,可能是预留)
for i in range(epoch): # 遍历每一个训练轮次(epoch)
start_time = time.time() # 记录本轮开始时间(计算单轮耗时)
model.train() # 将模型设为训练模式(启用Dropout、BN等训练特性)
train_loss = 0.0 # 初始化本轮训练总损失
train_acc = 0.0 # 初始化本轮训练总准确率
val_acc = 0.0 # 初始化验证准确率(本轮若验证则更新)
val_loss = 0.0 # 初始化验证损失(本轮若验证则更新)
for batch in tqdm(train_loader): # tqdm是进度条工具,可视化批处理进度
model.zero_grad() # 清空模型参数的梯度(避免累积)
text, labels = batch[0], batch[1].to(device) # 取出数据和标签,移到指定设备
pred = model(text) # 模型前向传播,得到预测结果
bat_loss = loss(pred, labels) # 计算当前批次的损失
bat_loss.backward() # 损失反向传播,计算梯度
optimizer.step() # 优化器更新模型参数
scheduler.step() # 学习率调度器调整学习率(注意:此处位置可能有问题)
optimizer.zero_grad() # 再次清空梯度(冗余,建议移到循环开头)
torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0) # 梯度裁剪,防止梯度爆炸
train_loss += bat_loss.item() # 累加批次损失(item()取出张量数值)
# 计算批次准确率:预测类别(argmax)与真实标签对比,求和
train_acc += np.sum(np.argmax(pred.cpu().data.numpy(),axis=1)==
labels.cpu().numpy())


# 训练损失:总损失 / 训练集样本总数
plt_train_loss.append(train_loss/train_loader.dataset.__len__())
# 训练准确率:总正确数 / 训练集样本总数
plt_train_acc.append(train_acc/train_loader.dataset.__len__())
if i % val_epoch == 0: # 每val_epoch轮做一次验证
model.eval() # 模型设为验证模式(关闭Dropout、固定BN参数)
with torch.no_grad(): # 禁用梯度计算,节省内存、加速验证
for batch in tqdm(val_loader): # 遍历验证集批次
val_text, val_labels = batch[0], batch[1].to(device)
val_pred = model(val_text) # 验证集前向传播
val_bat_loss = loss(val_pred, val_labels) # 计算验证批次损失
val_loss += val_bat_loss.cpu().item() # 累加验证损失
# 计算当前批次中模型预测正确的样本数,并累加到val_acc(验证集总正确数)。
val_acc += np.sum(np.argmax(val_pred.cpu().data.numpy(), axis=1) == val_labels.cpu().numpy())
val_rel.append(val_pred) # 记录验证预测结果
# 保存最优模型:如果当前验证准确率超过历史最高
if val_acc > max_acc:
torch.save(model, save_path+str(epoch)+"ckpt")
max_acc = val_acc # 更新最高准确率
# 验证指标归一化
plt_val_loss.append(val_loss/val_loader.dataset.__len__())
plt_val_acc.append(val_acc/val_loader.dataset.__len__())
# 打印训练+验证结果
print('[%03d/%03d] %2.2f sec(s) TrainAcc : %3.6f TrainLoss : %3.6f | valAcc: %3.6f valLoss: %3.6f ' % \
(i, epoch, time.time()-start_time, plt_train_acc[-1], plt_train_loss[-1], plt_val_acc[-1], plt_val_loss[-1])
)
# 每50轮保存一次模型快照
if i % 50 == 0:
torch.save(model, save_path+'-epoch:'+str(i)+ '-%.2f'%plt_val_acc[-1])
model.eval() # 模型设为验证模式(关闭Dropout、固定BN参数)

with torch.no_grad(): # 禁用梯度计算,节省内存、加速验证
验证当前轮次模型验证时的准确率不需要计算梯度
val_loader 是 PyTorch 的 DataLoader 对象,已经将验证集数据分成了若干批次(比如每批 32/64 个样本),遍历它会逐批返回数据,避免一次性加载所有数据导致内存不足。
batch[0]:验证集的输入数据(这里是val_text,比如文本的 token 化张量、图像的像素张量等);
batch[1]:验证集的真实标签(比如分类任务的类别索引)
np.argmax(..., axis=1):对预测结果按「样本维度」(axis=1)取最大值的索引,也就是模型预测的类别(比如预测得分[0.1, 0.8, 0.1],argmax 后得到索引 1,即第 2 类)
绘图
plt.plot(plt_train_loss) # 绘制训练损失曲线
plt.plot(plt_val_loss) # 绘制验证损失曲线
plt.title('loss') # 设置图表标题
plt.legend(['train', 'val'])# 添加图例,区分两条曲线
plt.show() # 显示绘制的图表
plt.plot(plt_train_acc) # 绘制训练准确率曲线
plt.plot(plt_val_acc) # 绘制验证准确率曲线
plt.title('Accuracy') # 设置图表标题
plt.legend(['train', 'val'])# 添加图例
plt.savefig('acc.png') # 将图表保存为图片文件
plt.show() # 显示图表
主函数参数设置与调用
引入其他包和其他类
import random
import torch
import torch.nn as nn
import numpy as np
import os
from model_utils.data import get_data_loader
from model_utils.model import myBertModel
from model_utils.train import train_val
随机数函数
def seed_everything(seed):
torch.manual_seed(seed)
torch.cuda.manual_seed(seed)
torch.cuda.manual_seed_all(seed)
torch.backends.cudnn.benchmark = False
torch.backends.cudnn.deterministic = True
random.seed(seed)
np.random.seed(seed)
os.environ['PYTHONHASHSEED'] = str(seed)
#################################################################
seed_everything(0)
###############################################

参数设置
lr = 0.0001 #学习率
batchsize = 16
loss = nn.CrossEntropyLoss()
bert_path = "bert-base-chinese"
num_class = 2 #分类个数
data_path = "jiudian.txt"
max_acc= 0.6 #最低准确率
device = "cuda" if torch.cuda.is_available() else "cpu"
model = myBertModel(bert_path, num_class, device).to(device)
optimizer = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=0.00001)
train_loader, val_loader = get_data_loader(data_path, batchsize)
epochs = 5
save_path = "model_save/best_model.pth"
scheduler = torch.optim.lr_scheduler.CosineAnnealingWarmRestarts(optimizer, T_0=20, eta_min=1e-9) #改变学习率
val_epoch = 1

利用权重思维防止过拟合

传入参数para数组调用训练类
para = {
"model": model,
"train_loader": train_loader,
"val_loader": val_loader,
"scheduler" :scheduler,
"optimizer": optimizer,
"loss": loss,
"epoch": epochs,
"device": device,
"save_path": save_path,
"max_acc": max_acc,
"val_epoch": val_epoch #训练多少论验证一次
}
train_val(para)
总结
-
预训练(Pretraining):BERT模型在大量的文本数据上进行过预训练,它学习了很多通用的语言表示。比如,BERT会学习到词语之间的关系、上下文的语法结构等。
-
微调(Fine-tuning):尽管BERT已经预训练好,但对于特定任务(比如情感分析、文本分类等),它的通用特征表示并不足以直接应用。微调的过程就是在特定任务的数据集上,对BERT的参数进行再次优化,让它更好地适应这个特定任务。比如,增加一个全连接层(
cls_head),用来进行分类任务。
目标:
-
文本分类任务:
-
输入是中文文本,目标是将文本分到不同的类别中(比如正面或负面情感分类)。
-
该模型的任务是根据文本内容预测其类别(通过对训练数据的学习来预测标签)。
-
-
使用BERT进行特征提取和分类:
-
利用预训练的 BERT模型 来提取文本的深层特征。BERT通过上下文关系的学习能够为文本生成更丰富的表示,适用于各种NLP任务。
-
在BERT模型的基础上添加一个 全连接层(
cls_head) 进行分类任务,输出文本的类别。
-
-
训练过程:
-
训练集 用来训练模型,不断调整模型的参数以最小化损失函数。
-
验证集 用来验证模型在训练过程中的表现,防止过拟合。
-
通过 梯度裁剪(
torch.nn.utils.clip_grad_norm_)来避免梯度爆炸,保证训练稳定。 -
使用 CosineAnnealingWarmRestarts 学习率调度器来动态调整学习率,以便在训练过程中更好地收敛。
-
-
保存最好的模型:
-
在每次验证后,如果验证集的准确率提高,模型会被保存,确保最终得到的模型是最优的。
-
-
训练与验证可视化:
-
通过绘制训练过程中的 损失 和 准确率 曲线,观察训练过程的收敛情况,帮助调试和优化模型。
-

更多推荐


所有评论(0)