一、前言

        在深度学习领域，Bert 是基于 Transformer 的一个经典的预训练模型，是现代大语言模型发展的一个里程碑。本文主要记录对Bert的一个简单下游任务记录。

        对于Bert，我的上一篇博客有简单的介绍：

https://blog.csdn.net/Wu_Deng_Sheng/article/details/158040820?spm=1001.2014.3001.5502

        本项目的核心：基于酒店的评论文本，训练一个二分类模型（0-负面评论，1-正面评论）

        注意，本项目选用的 Bert 版本的 Transformer 的 Encoder 层数为12，维度为768。

---

二、实现步骤

2.1 数据处理模块

2.1.1 数据读取

功能：读取文本，并分离标签和文本

        比如一行是 1,酒店很干净，服务很好，分割后变成 ["1", "酒店很干净，服务很好"]，并依次接入 data[ ] 和 label[ ] 并返回。

def read_file(path):
    data = []
    label = []
    with open(path, "r", encoding="utf-8") as f:
        for i, line in tqdm(enumerate(f)):
            # 跳过第一行（表头）
            if i == 0:
                continue
            line = line.strip("\n")
            # 按逗号分割1次，分离标签和文本（格式：标签,评论内容）
            line = line.split(",", 1)
            data.append(line[1])
            label.append(line[0])
    print("读了%d的数据" % len(data))
    return data, label

2.1.2 自定义数据类

class jdDataset(Dataset):
    def __init__(self, data, label):
        self.X = data  
        # 分类任务中，标签是长整型。字符型要变成长整型，适配交叉熵损失
        self.Y = torch.LongTensor([int(i) for i in label])

    def __getitem__(self, item):
        # 返回单条文本+标签
        return self.X[item], self.Y[item]  

    def __len__(self):
        # 数据集长度
        return len(self.Y)  

2.1.3 数据加载器

其中，stratify = label分层抽样，避免验证集标签分布不均

返回训练集和验证集，以完成后续训练

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)
    # 生成DataLoader（批量加载数据，打乱训练集）
    train_loader = DataLoader(train_set, batchsize, shuffle=True)
    val_loader = DataLoader(val_set, batchsize, shuffle=False)
    return train_loader, val_loader

2.2 模型模块

 核心模块，调用预训练bert模型，返回序列输出和[CLS]，用于分类任务。

import torch
import torch.nn as nn
from transformers import BertModel, BertTokenizer, BertConfig

class myBertModel(nn.Module):
    def __init__(self, bert_path, num_class, device):
        super(myBertModel, self).__init__()
        # 加载预训练BERT模型（中文）
        self.bert = BertModel.from_pretrained(bert_path)
        self.device = device
        # 分类头：BERT输出维度768 → 分类数num_class（768 → 2）
        self.cls_head = nn.Linear(768, num_class)
        # 加载BERT分词器（用于文本转token）
        self.tokenizer = BertTokenizer.from_pretrained(bert_path)

    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        # 文本最大长度
        )
        # 转到同一设备
        input_ids = input["input_ids"].to(self.device)
        token_type_ids = input['token_type_ids'].to(self.device)
        attention_mask = input['attention_mask'].to(self.device)

        # BERT前向传播：返回序列输出和[CLS] token输出（pooler_out）
        sequence_out, pooler_out = self.bert(
            input_ids=input_ids,
            token_type_ids=token_type_ids,
            attention_mask=attention_mask,
            return_dict=False  # 不返回字典，方便获取输出
        )
        # 分类头预测：使用[CLS] token的输出做分类
        pred = self.cls_head(pooler_out)
        return pred

2.3 训练验证模块

核心中的核心：训练正式开始。

这里用到了梯度裁剪，作用是防止梯度爆炸，把梯度限制在1.0以内

还用到了调整学习率调度器CosineAnnealingWarmRestarts，让学习率周期变化，用到的核心技术是：余弦退火（好帅的名字）+ 热重启

def train_val(para):
    # 解析参数
    model = para['model']
    train_loader =para['train_loader']
    val_loader = para['val_loader']
    scheduler = para['scheduler']
    optimizer = para['optimizer']
    loss = para['loss']
    epoch = para['epoch']
    device = para['device']
    save_path = para['save_path']
    max_acc = para['max_acc']
    val_epoch = para['val_epoch']

    # 初始可视化列表
    plt_train_loss = []
    plt_train_acc = []
    plt_val_loss = []
    plt_val_acc = []
    val_rel = []

    for i in range(epoch):
        start_time = time.time()
        # 训练模式
        model.train() 
        train_loss = 0.0
        train_acc = 0.0
        val_acc = 0.0
        val_loss = 0.0

        # 训练阶段
        for batch in tqdm(train_loader):
            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()
            # 再次清空梯度  
            # 梯度裁剪：防止梯度爆炸（阈值1.0）
            torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
            # 累计损失和精度
            train_loss += bat_loss.item()
            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__())

        # 验证阶段（每val_epoch轮验证一次）
        if i % val_epoch == 0:
            # 验证模式
            model.eval()  
            # 禁用梯度计算，节省显存
            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 += 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(i)+"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])
        else:
            # 未验证时，复用上一次验证结果（保证可视化曲线长度一致）
            plt_val_loss.append(plt_val_loss[-1])
            plt_val_acc.append(plt_val_acc[-1])
            print('[%03d/%03d] %2.2f sec(s) TrainAcc : %3.6f TrainLoss : %3.6f   ' % \
                  (i, epoch, time.time()-start_time, plt_train_acc[-1], plt_train_loss[-1])
                  )
    
    # 绘制损失曲线
    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()

2.4 主程序

import random
import torch
import torch.nn as nn
import numpy as np
import os

# 固定随机种子（保证实验可复现）
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)

# 固定种子为0
seed_everything(0)

# 超参数配置
lr = 0.0001                     # 学习率
batchsize =                  # 批次大小（文本任务显存占用高，建议小批量）
loss = nn.CrossEntropyLoss()    # 交叉熵损失（
bert_path = "# 预训练BERT路径" 
num_class = 2                   # 分类数
data_path = " # 数据文件路径"     
max_acc =                  # 初始最优验证精度

# 设备选择（优先GPU）
device = "cuda" if torch.cuda.is_available() else "cpu"

# 初始化模型
model = myBertModel(bert_path, num_class, device).to(device)

# 优化器：AdamW（带权重衰减，适合BERT微调）
optimizer = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=0.00001)

# 获取数据加载器
train_loader, val_loader = get_data_loader(data_path, batchsize)

# 训练轮数（文本任务收敛快）
epochs = 
save_path = "# 保存路径"

# 学习率调度器
scheduler = torch.optim.lr_scheduler.CosineAnnealingWarmRestarts(optimizer, T_0=20, eta_min=1e-9)
# 每轮验证
val_epoch = 1  

# 封装训练参数——字典
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)

---

三、总结

        本文主要记录我在对Bert下游任务学习中的收货和代码，加油加油！继续加油！