bert模型

发布于 2025-12-31

3.3 Bert模型

学习目标：

1.能够完成数据集的读取及预处理

2.理解dataSet、dataloader与collate_fn关系

3.能够完成Bert分类模型的构建

4.能够完成Bert模型的训练与测试

5.能够完成Bert分类模型的预测以及部署

Tips：

以下是transformer框架以及bert模型的架构。

bert的模型结构：（输入维度 batch_size为2 ）
=========================================================================================================
Layer (type:depth-idx)                                  Output Shape              Param #
=========================================================================================================
BertModel                                               [2, 768]                  --
├─BertEmbeddings: 1-1                                   [2, 128, 768]             --
│    └─Embedding: 2-1                                   [2, 128, 768]             16,226,304
│    └─Embedding: 2-2                                   [2, 128, 768]             1,536
│    └─Embedding: 2-3                                   [1, 128, 768]             393,216
│    └─LayerNorm: 2-4                                   [2, 128, 768]             1,536
│    └─Dropout: 2-5                                     [2, 128, 768]             --
├─BertEncoder: 1-2                                      [2, 128, 768]             --
│    └─ModuleList: 2-6                                  --                        --
│    │    └─BertLayer: 3-1                              [2, 128, 768]             7,087,872
│    │    └─BertLayer: 3-2                              [2, 128, 768]             7,087,872
│    │    └─BertLayer: 3-3                              [2, 128, 768]             7,087,872
│    │    └─BertLayer: 3-4                              [2, 128, 768]             7,087,872
│    │    └─BertLayer: 3-5                              [2, 128, 768]             7,087,872
│    │    └─BertLayer: 3-6                              [2, 128, 768]             7,087,872
│    │    └─BertLayer: 3-7                              [2, 128, 768]             7,087,872
│    │    └─BertLayer: 3-8                              [2, 128, 768]             7,087,872
│    │    └─BertLayer: 3-9                              [2, 128, 768]             7,087,872
│    │    └─BertLayer: 3-10                             [2, 128, 768]             7,087,872
│    │    └─BertLayer: 3-11                             [2, 128, 768]             7,087,872
│    │    └─BertLayer: 3-12                             [2, 128, 768]             7,087,872
├─BertPooler: 1-3                                       [2, 768]                  --
│    └─Linear: 2-7                                      [2, 768]                  590,592
│    └─Tanh: 2-8                                        [2, 768]                  --
=========================================================================================================

BertLayer 是 BERT 模型的核心组件，对应 Transformer 架构中的一个 encoder 层。

(一) 代码结构图

代码位置：

TMFCode\04-bert

基于Bert进行分类建模思路：

①下载bert预训练模型

目前bert的版本脚本，本次使用中文版的bert，bert下载

bert-base-chinese相关文件说明

②构建数据处理脚本 对数据train.txt等相关样本文件进行处理，构建符合模型输入要求的训练数据。处理脚本**：TMFCode\04-bert\src\utils.py**

③构建bert + 线性模块的分类模型

Bert是一个预训练模型，所以我们可以基于bert模型本身加上一个线性分类层来构建我们的下游任务。

模型脚本：TMFCode\04-bert\src\bert_classifer_model.py

④构建训练脚本

⑤构建预测脚本

⑥构建模型部署、api测试脚本

(二) 代码实现

2.1 config配置文件

代码位置：TMFCode\04-bert\src\config.py

import torch
import os
import datetime
from transformers.models import BertModel,BertTokenizer,BertConfig
current_date=datetime.datetime.now().date().strftime("%Y%m%d")

class Config(object):
    def __init__(self):
        """
        配置类，包含模型和训练所需的各种参数。
        """
        self.model_name = "bert" # 模型名称
        self.data_path = "../../01-data"  #数据集的根路径
        self.train_path = self.data_path + "\\train.txt"  # 训练集
        self.dev_path = self.data_path + "\\dev3.txt"  # 少量验证集，快速验证
        self.test_path = self.data_path + "\\test.txt"  # 测试集

        self.class_path=self.data_path + "\\class.txt" #类别文件

        self.class_list = [line.strip() for line in open(self.class_path, encoding="utf-8")]  # 类别名单

        self.model_save_path = "../save_models/test_bertclassifer_model.pt"  #模型训练结果保存路径

        # 模型训练+预测的时候
        # 训练设备，如果GPU可用，则为cuda，否则为cpu
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

        self.num_classes = len(self.class_list)  # 类别数
        self.num_epochs = 2  # epoch数
        self.batch_size = 2  # mini-batch大小
        self.pad_size = 32  # 每句话处理成的长度(短填长切)
        self.learning_rate = 5e-5  # 学习率
        self.bert_path = "../bert-base-chinese"  # 预训练BERT模型的路径
        self.bert_model=BertModel.from_pretrained(self.bert_path)
        self.tokenizer = BertTokenizer.from_pretrained(self.bert_path) # BERT模型的分词器
        self.bert_config = BertConfig.from_pretrained(self.bert_path) # BERT模型的配置
        self.hidden_size = 768 # BERT模型的隐藏层大小

if __name__ == '__main__':
    conf = Config()
    print(conf.bert_config)
    input_size=conf.tokenizer.convert_tokens_to_ids(["你","好","中国","人"])
    print(input_size)
    print(conf.class_list)

2.2 utils配置文件

代码位置：TMFCode\04-bert\src\utils.py

utils脚本主要实现了train.txt相关数据读取加载切分，以及DataSet、DataLoader的构建，最终输出符合模型需求的数据格式。

input_ids=> [[101, 704, 1290, 1957, 2094, 2110, 7368, 8038, 3315, 4906, 2231, 3613, 788, 122, 683, 689, 2875, 4511, 4495, 102, 0, 0, 0, 0], [101, 697, 1921, 817, 5381, 4991, 5520, 1400, 7028, 7028, 6837, 7443, 8038, 976, 702, 5381, 4991, 4955, 4994, 6206, 1914, 2208, 7178, 102]]
labels=> [3, 4]
attention_mask=> [[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0], [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]]

导入工具和配置信息：

import torch
from torch.utils.data import Dataset, DataLoader
from transformers import BertTokenizer
from tqdm import tqdm
import time
from datetime import timedelta
from config import Config
import time
conf=Config()

(1) 加载原始数据

读取原始数据train.txt等相关文件，进行切分之后，以元组的形式保存样本对到列表中。

train.txt处理为左侧的data列表：

def load_raw_data(file_path):
    """
    读取原始数据文件，解析为文本和标签。

    参数：
        file_path (str): 数据文件路径（如dev2.txt）。

    返回：
        List[Tuple[str, int]]: 包含(文本, 标签)的列表。
    """
    data = []
    with open(file_path, "r", encoding="UTF-8") as f:
        for line in tqdm(f, desc="Loading data"):
            line = line.strip()
            if not line:
                continue
            text, label = line.split("\t")
            data.append((text, int(label)))
    print(data[:5])
    return data

(2) 构建自定义数据集

我们模型构建包括批计算都基于pytorch框架，所以我们数据要构建成符合pytorch模型需求的数据集。

class TextDataset(Dataset):
    def __init__(self, data):
        self.data = data

    def __len__(self):
        return len(self.data)

    def __getitem__(self, idx):
        x=self.data[idx][0]
        y=self.data[idx][1]
        return x, y

(3) 构建dataloader

构建dataloader用于后续建模中batch数据输入到模型中，进行loss的计算以及参数的迭代优化。

def build_dataloader():
    """
    构建DataLoader，整合数据加载、Dataset和collate_fn。

    参数：
        file_path (str): 数据文件路径。
        batch_size (int): 批次大小。
        padding_size (int): 统一padding长度（默认28）。
        device (str): 设备（"cpu"或"cuda"）。

    返回：
        DataLoader: 用于训练的DataLoader。
    """
    # 加载原始数据
    train_data = load_raw_data(conf.train_path)
    test_data = load_raw_data(conf.test_path)
    dev_data = load_raw_data(conf.dev_path)

    # 创建 Dataset
    train_dataset = TextDataset(train_data)
    dev_dataset = TextDataset(dev_data)
    test_dataset = TextDataset(test_data)

    # 创建 DataLoader
    train_dataloader = DataLoader(train_dataset,batch_size=conf.batch_size,shuffle=False,collate_fn=collate_fn)
    test_dataloader = DataLoader(test_dataset, batch_size=conf.batch_size, shuffle=False, collate_fn=collate_fn)
    dev_dataloader = DataLoader(dev_dataset, batch_size=conf.batch_size, shuffle=False, collate_fn=collate_fn)

    return train_dataloader,test_dataloader,dev_dataloader

(4) 构建collate_fn

collate_fn是dataloader为了解决进入模型训练的数据不符合要求的进一步处理，例如batch级别数据处理长度、数据数值化等。

def collate_fn(batch):
    """
    DataLoader的collate_fn，处理分词、统一padding、mask生成和Tensor转换。

    参数：
        batch (List[Tuple[str, int]]): 批次数据，包含(文本, 标签)。
        tokenizer (BertTokenizer): BERT分词器。
        padding_size (int): 统一padding长度（默认28，基于文本长度统计）。
        device (str): 设备（"cpu"或"cuda"）。

    返回：
        Tuple[torch.Tensor, ...]: (input_ids, seq_len, attention_mask, labels) 的Tensor格式。
    """
    # 提取文本和标签
    texts = [item[0] for item in batch]
    labels = [item[1] for item in batch]

    # 批量分词，自动添加 [CLS] 和 [SEP]  add_special_tokens  # padding，统一处理
    text_tokens = conf.tokenizer.batch_encode_plus(texts,padding=True)
    token_ids_list = text_tokens["input_ids"]
    token_attention_mask_list = text_tokens["attention_mask"]
    # 转为 Tensor
    input_ids = torch.tensor(token_ids_list)
    attention_mask = torch.tensor(token_attention_mask_list)
    labels = torch.tensor(labels)
    #
    # print("================================")
    # print(labels)
    # print(attention_mask)
    # print(input_ids)
    return input_ids, attention_mask, labels

(5) 验证数据处理完整逻辑

if __name__ == "__main__":
    # 构建 DataLoader
    train_dataloader,test_dataloader,dev_dataloader = build_dataloader()
    # #遍历 DataLoader
    for batch in train_dataloader:
        input_ids, attention_mask, labels = batch
        print("input_ids=>",input_ids.tolist())
        print("labels=>",labels.tolist())
        print("attention_mask=>",attention_mask.tolist())
        breakpoint()

输入日志：

input_ids=> [[101, 704, 1290, 1957, 2094, 2110, 7368, 8038, 3315, 4906, 2231, 3613, 788, 122, 683, 689, 2875, 4511, 4495, 102, 0, 0, 0, 0], [101, 697, 1921, 817, 5381, 4991, 5520, 1400, 7028, 7028, 6837, 7443, 8038, 976, 702, 5381, 4991, 4955, 4994, 6206, 1914, 2208, 7178, 102]]
labels=> [3, 4]
attention_mask=> [[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0], [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]]

2.3 模型构建

代码位置：TMFCode\04-bert\src\bert_classifer_model.py

import torch
import torch.nn as nn
from transformers import BertModel
from config import Config
from utils import build_dataloader

conf = Config()
class BertClassifier(nn.Module):
    """
    BERT + 全连接层的分类模型。
    """
    def __init__(self):
        """
        初始化模型，包括BERT和全连接层。
        """
        super(BertClassifier, self).__init__()
        self.bert = BertModel.from_pretrained(conf.bert_path)
        self.fc = nn.Linear(conf.hidden_size, conf.num_classes)

    def forward(self, input_ids, attention_mask):
        _, pooled = self.bert(input_ids=input_ids, attention_mask=attention_mask, return_dict=False)
        out = self.fc(pooled)
        return out

2.4 模型训练及验证

代码位置：

TMFCode\04-bert\src\train.py

导包以及相关配置：

import torch
import torch.nn as nn
import torch.optim as optim
from torch.optim import AdamW
from sklearn.metrics import classification_report, f1_score, accuracy_score, precision_score
from tqdm import tqdm
import os
from config import Config
from utils import build_dataloader, get_time_diff
from bert_classifer_model import BertClassifier
import time
# 加载配置对象，包含模型参数、路径等
conf = Config()
# 忽略的警告信息
import warnings
warnings.filterwarnings("ignore")

(1) 训练函数实现

def model2train():
    """
    训练 BERT 分类模型并在验证集上评估，保存最佳模型。

    参数：
        无显式参数，所有配置通过全局 conf 对象获取。

    返回：
        无返回值，训练过程中保存最佳模型到指定路径。
    """
    # 1. 加载训练、测试和验证数据集的 DataLoader
    train_loader, test_loader, dev_loader = build_dataloader()

    # 2. 定义训练参数，从配置对象中获取
    device = conf.device  # 设备（"cuda" 或 "cpu"）
    num_epochs = conf.num_epochs  # 训练轮数

    # 3. 初始化 BERT 分类模型
    model = BertClassifier().to(device)

    # 4. 定义优化器（AdamW，适合 Transformer 模型）和损失函数（交叉熵）
    optimizer = AdamW(model.parameters(), lr=conf.learning_rate)
    criterion = nn.CrossEntropyLoss()

    # 5. 初始化最佳验证 F1 分数，用于保存性能最好的模型
    best_dev_f1 = 0.0

    # 6. 遍历每个训练轮次（epoch）
    for epoch in range(num_epochs):
        # 设置模型为训练模式（启用 dropout 和 batch norm）
        model.train()
        total_loss = 0  # 累计训练损失
        train_preds, train_labels = [], []  # 存储训练集预测和真实标签

        # 7. 遍历训练 DataLoader 进行模型训练
        for batch in tqdm(train_loader, desc=f"Bert Classifier Training Epoch {epoch + 1}/{num_epochs}...."):
            # 7.0 提取批次数据并移动到设备
            input_ids, attention_mask, labels = batch
            input_ids, attention_mask, labels = input_ids.to(device), attention_mask.to(device), labels.to(device)

            # 7.1 前向传播：模型预测
            logits = model(input_ids, attention_mask)
            # 7.2 损失计算
            loss = criterion(logits, labels)
            # 7.3 梯度归零
            optimizer.zero_grad()
            # 7.4 反向传播
            loss.backward()
            # 7.5 参数更新
            optimizer.step()

            # 7.6 累计损失
            total_loss += loss.item()
            # 7.7 获取预测结果（最大 logits 对应的类别）
            preds = torch.argmax(logits, dim=1)
            # 7.8 存储预测和真实标签，用于计算训练集指标
            train_preds.extend(preds.tolist())
            train_labels.extend(labels.tolist())

            # 7.9 打印训练信息并评估验证集
            print(f"Epoch {epoch + 1}/{num_epochs}")
            print(f"Train Loss: {total_loss / len(train_loader):.4f}")
            # 在验证集上评估模型
            report, f1score, accuracy, precision = model2dev(model, dev_loader, device)
            print(f"Dev F1: {f1score:.4f}")
            print(f"Dev Accuracy: {accuracy:.4f}")

            # 7.10 如果验证 F1 分数优于历史最佳，保存模型
            if f1score > best_dev_f1:
                best_dev_f1 = f1score
                torch.save(model.state_dict(), conf.model_save_path)
                print("模型保存！！")

        # 7.11 计算并打印训练集的分类报告
        train_report = classification_report(train_labels, train_preds,
                                             target_names=conf.class_list, output_dict=True)
        print(train_report)

(2) 验证函数实现

def model2dev(model, data_loader, device):
    """
    在验证或测试集上评估 BERT 分类模型的性能。

    参数：
        model (nn.Module): BERT 分类模型。
        data_loader (DataLoader): 数据加载器（验证或测试集）。
        device (str): 设备（"cuda" 或 "cpu"）。

    返回：
        tuple: (分类报告, F1 分数, 准确度, 精确度)
            - report: 分类报告（包含每个类别的精确度、召回率、F1 分数等）。
            - f1score: 微平均 F1 分数。
            - accuracy: 准确度。
            - precision: 微平均精确度。
    """
    # 1. 设置模型为评估模式（禁用 dropout 和 batch norm）
    model.eval()

    # 2. 初始化列表，存储预测结果和真实标签
    preds, true_labels = [], []

    # 3. 禁用梯度计算以提高效率并减少内存占用
    with torch.no_grad():
        # 4. 遍历数据加载器，逐批次进行预测
        for batch in tqdm(data_loader, desc="Bert Classifier Evaluating ......"):
            # 4.1 提取批次数据并移动到设备
            input_ids, attention_mask, labels = batch
            input_ids, attention_mask, labels = input_ids.to(device), attention_mask.to(device), labels.to(device)

            # 4.2 前向传播：模型预测
            logits = model(input_ids, attention_mask)

            # 4.3 获取预测结果（最大 logits 对应的类别）
            batch_preds = torch.argmax(logits, dim=1)

            # 4.4 存储预测和真实标签
            preds.extend(batch_preds.cpu().numpy())
            true_labels.extend(labels.cpu().numpy())

    # 5. 计算分类报告、F1 分数、准确度和精确度
    report = classification_report(true_labels, preds)
    f1score = f1_score(true_labels, preds, average='micro')  # 使用微平均计算 F1 分数
    accuracy = accuracy_score(true_labels, preds)  # 计算准确度
    precision = precision_score(true_labels, preds, average='micro')  # 使用微平均计算精确度

    # 6. 返回评估结果
    return report, f1score, accuracy, precision

(3) 日志效果

if __name__ == '__main__':
    # 主程序入口
    model2train()

    # 1. 加载测试集数据
    train_dataloader, test_dataloader, dev_dataloader = build_dataloader()
    # 2. 初始化 BERT 分类模型
    model = BertClassifier()
    # 3. 加载预训练模型权重
    model.load_state_dict(torch.load("../save_models/bert20250521_.pt"))
    # 4. 将模型移动到指定设备
    model.to(conf.device)
    # 5. 在测试集上评估模型
    test_report, f1score, accuracy, precision = model2dev(model, test_dataloader, conf.device)
    # 6. 打印测试集评估结果
    print("Test Set Evaluation:")
    print(f"Test F1: {f1score:.4f}")
    print("Test Classification Report:")
    print(test_report)

输出日志：

Bert Classifier Evaluating ......:   0%|          | 0/2 [00:00<?, ?it/s]Epoch 2/2
Train Loss: 0.0107

Bert Classifier Evaluating ......:  50%|█████     | 1/2 [00:00<00:00,  9.22it/s]
Bert Classifier Evaluating ......: 100%|██████████| 2/2 [00:00<00:00,  9.09it/s]
Bert Classifier Training Epoch 2/2....:   5%|▌         | 38/704 [00:18<05:32,  2.01it/s]Dev F1: 0.9276
Dev Accuracy: 0.9276

..........
Epoch 2/2
Train Loss: 0.0042

Bert Classifier Evaluating ......:   0%|          | 0/2 [00:00<?, ?it/s]
Bert Classifier Evaluating ......:  50%|█████     | 1/2 [00:00<00:00,  9.75it/s]
Bert Classifier Evaluating ......: 100%|██████████| 2/2 [00:00<00:00,  9.62it/s]
Bert Classifier Training Epoch 2/2....:   100%|██████████| 704/704
Dev F1: 0.9364
Dev Accuracy: 0.9364

结论: 采用BERT预训练模型的验证集测试 F1和Acc能达到93.17%，效果显著。

2.5 预测脚本

代码位置：

TMFCode\04-bert\src\predict_fun.py

导入工具及相关配置、加载模型：

import torch
from transformers import BertTokenizer
from bert_classifer_model import BertClassifier
from config import Config
# 初始化配置
conf = Config()

device = conf.device
tokenizer = conf.tokenizer
model = BertClassifier().to(device)
model.load_state_dict(torch.load("../save_models/bert20250521_.pt"))
model.eval()

以下为预测函数：

#预测函数
def predict(data):
    # 1.获取数据 data['text']
    text = data['text']

    # 2.预处理数据----text===> input_ids,attention_mask <=== tokenizer.encode_plus
    tokenize = tokenizer.encode_plus(text, return_tensors='pt')
    input_ids = tokenize['input_ids'].to(device)
    attention_mask = tokenize['attention_mask'].to(device)

    # 3.模型预测
    ## 3.1 关闭梯度计算
    with torch.no_grad():
        ## 3.2 前向推理 model()
        pred_logits = model(input_ids, attention_mask)
        ## 3.3 获取预测结果---softmax  argmax(最大概率值所对应的索引)  class_name
        pred_prob = torch.softmax(pred_logits, dim=1)
        pred_ids = torch.argmax(pred_prob, dim=1)
        pred_class = conf.class_list[pred_ids]

    return {"text": text, "pred_class": pred_class}

输入日志：

预测结果：
{'text': '中华女子学院：本科层次仅1专业招男生', 'pred_class': 'education'}

2.6 模型部署

(1) 服务端

# 模型部署
import fasttext
import jieba
from predict_fun import predict
from flask import Flask, request,jsonify

app = Flask(__name__)

@app.route('/predict', methods=['POST'])
def predict_api():
    # 获取请求数据
    data = request.get_json()
    #预测
    print("-------------预测结果------------")
    result=predict(data)
    print(result)
    return jsonify(result)

if __name__ == '__main__':
    app.run(host='0.0.0.0',port=8003)

启动日志：

C:\python.exe C:\TMFCode\04-bert\src\api.py
 * Serving Flask app 'api'
 * Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on all addresses (0.0.0.0)
 * Running on http://127.0.0.1:8003
 * Running on http://26.26.26.1:8003

(2) 客户端

import requests
import time
# 定义预测接口地址
url = 'http://127.0.0.1:8003/predict'

# 构造请求数据
data = {'text': "中华女子学院：本科层次仅1专业招男生"}
# 记录开始时间
start_time = time.time()
# 发送 POST 请求
try:
    response = requests.post(url, json=data)
    # 计算耗时（毫秒）
    elapsed_time = (time.time() - start_time) * 1000
    print(f"请求耗时: {elapsed_time:.2f} ms")

    # 检查响应状态
    if response.status_code == 200:
        result = response.json()
        print(f"预测结果: {result['pred_class']}")
    else:
        print(f"请求失败: {response.status_code}, {response.json()['error']}")
except Exception as e:
    print(f"请求出错: {str(e)}")

输出日志：

C:\python.exe C:\TMFCode\04-bert\src\api_test.py
请求耗时: 60.96 ms
预测结果: education

2.7 前端预测

代码位置：TMFCode\04-bert\src\app.py

启动命令：

streamlit run app.py --server.port 8504

发布于 2025-12-31