【Pytorch & Transformers Fine-tune】使用BERT进行情感分类任务微调

在这篇教程中，将带你了解如何对预训练模型进行微调，这是一种强大的技术，可以让你将最先进的模型应用到你的特定任务上。微调相比从头训练模型有显著优势：它减少了计算成本，降低了碳足迹，并允许你无需从零开始就能使用先进模型。

预训练模型已经在大量数据上学习了通用表示，只需要在特定任务上进行少量训练即可适应新任务。🤗 Transformers 库提供了数千个预训练模型，涵盖了各种任务。

开始前的准备

首先，需要将所需要的 Python 库安装好：

pip install "peft>=0.4.0" "accelerate>=0.27.2" "bitsandbytes>=0.41.3" "trl>=0.4.7" "safetensors>=0.3.1" "tiktoken"
pip install "torch>=2.1.1" -U
pip install "datasets" -U
pip install transformers

另外，由于某些原因，建议使用huggingface的国内镜像网站，在终端使用命令手动下载数据集及模型：

$env:HF_ENDPOINT = "https://hf-mirror.com"

huggingface-cli download bert-base-chinese --local-dir .\models\bert-base-chinese

使用Pytorch 和 Transformers进行微调

下面将介绍如何使用PyTorch与Transformers库进行BERT模型的微调，以完成文本分类任务。我们将以情感分析（正面或负面情感）为例，演示从数据处理到模型微调的全过程。

1. 数据准备

首先，我们需要准备并处理数据集。我们将自定义一个数据集类CustomDataset，该类会将文本数据标记化并生成对应的输入格式。

from torch.utils.data import Dataset, DataLoader
from transformers import BertTokenizer
import torch

model_path = './models/bert-base-chinese

class CustomDataset(Dataset):
    def __init__(self, texts, labels, tokenizer, max_length=128):
        self.tokenizer = tokenizer
        self.texts = texts
        self.labels = labels
        self.max_length = max_length

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

    def __getitem__(self, idx):
        text = self.texts[idx]
        label = self.labels[idx]

        encoding = self.tokenizer(
            text,
            add_special_tokens=True,
            max_length=self.max_length,
            padding='max_length',
            truncation=True,
            return_attention_mask=True,
            return_tensors='pt'
        )

        return {
            'input_ids': encoding['input_ids'].flatten(),
            'attention_mask': encoding['attention_mask'].flatten(),
            'labels': torch.tensor(label, dtype=torch.long)
        }

解释：

• CustomDataset类继承自PyTorch的Dataset，接收文本和标签，并使用BERT的BertTokenizer对文本进行标记化处理。
• max_length指定了输入序列的最大长度，所有文本都将被填充（padding）或截断（truncation）为该长度。
• __getitem__方法负责返回一个字典，包含标记化后的input_ids、attention_mask以及标签labels。

2. 加载预训练BERT模型

接下来，我们需要加载一个预训练的BERT模型，并设置它用于文本分类任务。以下函数会加载BERT模型，并指定标签的数量。

from transformers import BertForSequenceClassification

def load_model(num_labels):
    # 加载预训练的BERT模型
    model = BertForSequenceClassification.from_pretrained(
        model_path,  # 或其他预训练模型
        num_labels=num_labels,
        output_attentions=False,
        output_hidden_states=False,
    )
    return model

解释：

• BertForSequenceClassification是用于分类任务的BERT模型。num_labels指定分类任务的标签数（本例为二分类任务）。
• output_attentions和output_hidden_states参数用于控制是否输出注意力权重和隐藏状态，通常在分类任务中可以不需要。

3. 设置训练过程

我们将定义一个train函数，包含训练循环、损失计算、反向传播和评估过程。

from transformers import AdamW, get_linear_schedule_with_warmup
import torch
import numpy as np

def train(model, train_dataloader, val_dataloader, epochs=3):
    device = torch.device('cuda') if torch.cuda.is_available() else torch.device('cpu')
    print(f"使用设备: {device}")

    model.to(device)

    # 定义优化器
    optimizer = AdamW(model.parameters(), lr=2e-5, eps=1e-8)

    # 设置学习率调度器
    total_steps = len(train_dataloader) * epochs
    scheduler = get_linear_schedule_with_warmup(
        optimizer,
        num_warmup_steps=0,
        num_training_steps=total_steps
    )

    # 记录训练过程
    training_stats = []

    # 开始训练循环
    for epoch in range(epochs):
        print(f'Epoch {epoch + 1}/{epochs}')
        print('-' * 10)

        # 训练模式
        model.train()
        total_loss = 0

        for batch in train_dataloader:
            # 将数据加载到设备
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            labels = batch['labels'].to(device)

            # 清除之前的梯度
            model.zero_grad()

            # 前向传播
            outputs = model(
                input_ids=input_ids,
                attention_mask=attention_mask,
                labels=labels
            )

            loss = outputs.loss
            total_loss += loss.item()

            # 反向传播
            loss.backward()

            # 梯度裁剪，防止梯度爆炸
            torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)

            # 更新参数
            optimizer.step()

            # 更新学习率
            scheduler.step()

        # 计算平均损失
        avg_train_loss = total_loss / len(train_dataloader)
        print(f'平均训练损失: {avg_train_loss}')

        # 评估模式
        model.eval()
        val_accuracy = []
        val_loss = []

        for batch in val_dataloader:
            # 将数据加载到设备
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            labels = batch['labels'].to(device)

            # 不计算梯度
            with torch.no_grad():
                outputs = model(
                    input_ids=input_ids,
                    attention_mask=attention_mask,
                    labels=labels
                )

            loss = outputs.loss
            logits = outputs.logits

            val_loss.append(loss.item())

            # 计算准确率
            preds = torch.argmax(logits, dim=1).flatten()
            accuracy = (preds == labels).cpu().numpy().mean() * 100
            val_accuracy.append(accuracy)

        # 计算平均验证损失和准确率
        avg_val_loss = np.mean(val_loss)
        avg_val_accuracy = np.mean(val_accuracy)

        print(f'验证损失: {avg_val_loss}')
        print(f'验证准确率: {avg_val_accuracy}%')

        # 保存训练统计信息
        training_stats.append({
            'epoch': epoch + 1,
            'train_loss': avg_train_loss,
            'val_loss': avg_val_loss,
            'val_accuracy': avg_val_accuracy
        })

    return model, training_stats

解释：

• train函数负责进行多个周期的训练。每个epoch开始时，模型进入训练模式，计算损失并进行反向传播，使用优化器更新模型参数。
• 每个batch处理完后，我们会对验证集进行评估，计算验证损失和准确率。

4. 保存微调后的模型

训练完成后，我们需要将微调后的模型和分词器保存，以便后续使用。

def save_model(model, tokenizer, output_dir):
    import os

    # 创建输出目录（如果不存在）
    if not os.path.exists(output_dir):
        os.makedirs(output_dir)

    print(f"保存模型到 {output_dir}")

    # 保存模型
    model.save_pretrained(output_dir)

    # 保存分词器
    tokenizer.save_pretrained(output_dir)

解释：

• save_model函数将保存微调后的BERT模型和对应的分词器（Tokenizer）。output_dir指定保存路径。

5. 主函数

我们在main函数中执行整个流程，包括数据准备、模型加载、训练和保存。

def main():
    # 示例数据
    texts = [
        "这个产品非常好用，我很满意",
        "质量太差了，很快就坏了",
        "价格合理，物有所值",
        # 添加更多文本...
    ]
    labels = [1, 0, 1]  # 1表示正面情感，0表示负面情感

    # 加载分词器
    tokenizer = BertTokenizer.from_pretrained(model_path)

    # 创建数据集
    dataset = CustomDataset(texts, labels, tokenizer)

    # 划分训练集和验证集（这里简化处理）
    train_size = int(0.8 * len(dataset))
    val_size = len(dataset) - train_size
    train_dataset, val_dataset = torch.utils.data.random_split(dataset, [train_size, val_size])

    # 创建数据加载器
    train_dataloader = DataLoader(train_dataset, batch_size=16, shuffle=True)
    val_dataloader = DataLoader(val_dataset, batch_size=16)

    # 加载模型
    model = load_model(num_labels=2)  # 二分类任务

    # 训练模型
    model, stats = train(model, train_dataloader, val_dataloader, epochs=3)

    # 保存模型
    save_model(model, tokenizer, './fine_tuned_model')

    print("微调完成！")

解释：

• 在main函数中，我们提供了示例数据（文本和标签）。我们使用CustomDataset类将数据加载并分割为训练集和验证集。
• 使用train函数进行训练，并最终保存微调后的模型。

通过上述步骤，你可以实现BERT模型的微调，并用于文本分类任务。

测试微调后的模型

微调后的模型验证是确保模型性能符合预期的重要步骤。以下是验证微调模型的几种方法：

import torch
from transformers import BertTokenizer, BertForSequenceClassification
import numpy as np
from torch.utils.data import Dataset, DataLoader

# 1. 加载微调后的模型和分词器
def load_fine_tuned_model(model_path):
    model = BertForSequenceClassification.from_pretrained(model_path)
    tokenizer = BertTokenizer.from_pretrained(model_path)
    return model, tokenizer

# 2. 创建测试数据集
class TestDataset(Dataset):
    def __init__(self, texts, labels, tokenizer, max_length=128):
        self.tokenizer = tokenizer
        self.texts = texts
        self.labels = labels
        self.max_length = max_length
        
    def __len__(self):
        return len(self.texts)
    
    def __getitem__(self, idx):
        text = self.texts[idx]
        label = self.labels[idx]
        
        encoding = self.tokenizer(
            text,
            add_special_tokens=True,
            max_length=self.max_length,
            padding='max_length',
            truncation=True,
            return_attention_mask=True,
            return_tensors='pt'
        )
        
        return {
            'input_ids': encoding['input_ids'].flatten(),
            'attention_mask': encoding['attention_mask'].flatten(),
            'labels': torch.tensor(label, dtype=torch.long)
        }

# 3. 模型评估函数
def evaluate_model(model, test_dataloader, device):
    model.eval()
    
    total_accuracy = 0
    total_loss = 0
    all_preds = []
    all_labels = []
    
    # 不计算梯度
    with torch.no_grad():
        for batch in test_dataloader:
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            labels = batch['labels'].to(device)
            
            outputs = model(
                input_ids=input_ids,
                attention_mask=attention_mask,
                labels=labels
            )
            
            loss = outputs.loss
            logits = outputs.logits
            
            total_loss += loss.item()
            
            # 计算准确率
            preds = torch.argmax(logits, dim=1).flatten()
            accuracy = (preds == labels).cpu().numpy().mean() * 100
            total_accuracy += accuracy
            
            # 收集预测和标签用于计算其他指标
            all_preds.extend(preds.cpu().numpy())
            all_labels.extend(labels.cpu().numpy())
    
    avg_accuracy = total_accuracy / len(test_dataloader)
    avg_loss = total_loss / len(test_dataloader)
    
    return avg_accuracy, avg_loss, all_preds, all_labels

# 4. 计算更多评估指标
def calculate_metrics(true_labels, predictions):
    from sklearn.metrics import precision_recall_fscore_support, confusion_matrix
    
    precision, recall, f1, _ = precision_recall_fscore_support(
        true_labels, predictions, average='weighted'
    )
    
    conf_matrix = confusion_matrix(true_labels, predictions)
    
    return {
        'precision': precision,
        'recall': recall,
        'f1_score': f1,
        'confusion_matrix': conf_matrix
    }

# 5. 对单个文本进行预测
def predict_text(text, model, tokenizer, device):
    # 准备输入
    encoding = tokenizer(
        text,
        add_special_tokens=True,
        max_length=128,
        padding='max_length',
        truncation=True,
        return_attention_mask=True,
        return_tensors='pt'
    )
    
    input_ids = encoding['input_ids'].to(device)
    attention_mask = encoding['attention_mask'].to(device)
    
    # 预测
    model.eval()
    with torch.no_grad():
        outputs = model(
            input_ids=input_ids,
            attention_mask=attention_mask
        )
    
    logits = outputs.logits
    prediction = torch.argmax(logits, dim=1).item()
    
    # 计算概率
    probabilities = torch.nn.functional.softmax(logits, dim=1)
    probability = probabilities[0][prediction].item()
    
    return prediction, probability

# 6. 主函数
def main():
    # 加载微调后的模型
    model_path = './fine_tuned_model'
    model, tokenizer = load_fine_tuned_model(model_path)
    
    # 设置设备
    device = torch.device('cuda') if torch.cuda.is_available() else torch.device('cpu')
    model.to(device)
    
    # 准备测试数据
    test_texts = [
        "这个产品真的很不错，质量很好",
        "太令人失望了，完全不值这个价",
        "一般般吧，没有特别惊艳",
        "非常满意这次购买",
        "这是我买过的最糟糕的产品"
    ]
    test_labels = [1, 0, 1, 1, 0]  # 1表示正面情感，0表示负面情感
    
    # 创建测试数据集
    test_dataset = TestDataset(test_texts, test_labels, tokenizer)
    test_dataloader = DataLoader(test_dataset, batch_size=8)
    
    # 评估模型
    print("开始模型评估...")
    avg_accuracy, avg_loss, all_preds, all_labels = evaluate_model(model, test_dataloader, device)
    
    print(f"测试准确率: {avg_accuracy:.2f}%")
    print(f"测试损失: {avg_loss:.4f}")
    
    # 计算更多指标
    metrics = calculate_metrics(all_labels, all_preds)
    print(f"精确率: {metrics['precision']:.4f}")
    print(f"召回率: {metrics['recall']:.4f}")
    print(f"F1分数: {metrics['f1_score']:.4f}")
    print("混淆矩阵:")
    print(metrics['confusion_matrix'])
    
    # 对单个样本进行预测示例
    print("\n单个文本预测示例:")
    
    sample_text = "这个产品设计很精美，但是使用起来不太方便"
    prediction, probability = predict_text(sample_text, model, tokenizer, device)
    
    label_map = {0: "负面", 1: "正面"}
    print(f"文本: '{sample_text}'")
    print(f"预测标签: {label_map[prediction]} (概率: {probability:.4f})")

if __name__ == "__main__":
    main()

验证微调模型的主要方法包括：

1. 量化评估指标

• 准确率（Accuracy）: 正确预测的比例
• 精确率（Precision）: 预测为正例中真正例的比例
• 召回率（Recall）: 真正例中被正确预测的比例
• F1分数: 精确率和召回率的调和平均
• 混淆矩阵: 展示各类别预测的详细分布

2. 数据拆分验证

• 测试集: 使用完全未见过的数据评估模型性能
• 交叉验证: 多次使用不同的训练/验证拆分来获得更稳健的评估
• K折交叉验证: 将数据分成K份，轮流使用其中一份作为测试集

3. 实例分析

• 单样本预测: 分析单个预测及其概率分布
• 错误分析: 重点研究模型预测错误的案例
• 边界案例测试: 测试模型在难以分类的样本上的表现

4. 可视化分析

• 混淆矩阵热图: 直观显示分类错误的分布
• ROC曲线: 评估二分类模型的性能
• 特征重要性: 分析哪些特征对模型预测影响最大

5. 实际应用测试

• 真实场景测试: 在实际应用环境中评估
• A/B测试: 与其他模型或原始模型进行对比测试
• 用户反馈收集: 收集最终用户对模型输出的评价

进阶验证技术

1. 模型鲁棒性测试:

   • 添加噪声数据测试模型稳定性
   • 对抗样本测试，检验模型是否容易被欺骗

2. 多样性数据测试:

   • 使用不同领域或来源的数据验证
   • 检测模型在各种数据分布上的表现

3. 模型解释性分析:

   • 使用SHAP值或LIME等工具解释模型决策
   • 分析模型关注的特征或文本部分

通过以上方法的组合应用，可以全面验证微调模型的性能，确保它在实际应用中能够发挥预期效果，并了解其潜在局限性。