继S1.2的实战后，我们来做一个基于Bert的文本主题与情感识别。虽然我们还不知道Bert的相关原理，但是本节我们主要来了解一下大模型的下游任务。

数据集是来自真实的汽车行业用户观点数据，（提取码：M2Hm）数据示例：

其中标签有两类，分别是主题和情感。训练集数据中主题被分为10类，包括：动力、价格、内饰、配置、安全性、外观、操控、油耗、空间、舒适性；情感分为3类，分别用0，1，-1表示中立、正向、负向

项目需求：

情感识别是三分类(互斥，single-label)，主题识别是多标签分类(multi-label)

指标采用accuracy、precision、recall、F1，绘制训练和测试的loss、accuracy曲线

如果有多个标签，情感采用数字加和的方式，sum大于0为正向，小于0为负向，0为中性

多标签分类的准确率和召回率指标采用macro平均

注：我们不需要完全陷入数学公式，具体原理我们会在后续的章节详细刨析，代码中几乎所有部分我已注释，大家先掌握大模型处理文本的流程，但是一定要手写代码感受一下。

# 导入了所需的库和模块。
# 包括PyTorch及其子模块、transformers库中的BERT模型和分词器、数据处理库Pandas和NumPy、绘图库matplotlib、评估分类性能的库sklearn等。
import torch
import torch.nn as nn#导入 PyTorch 的神经网络模块，包含常用的层（如线性层、激活函数）和损失函数等，用于定义模型结构
from torch.utils.data import Dataset, DataLoader
#导入数据加载工具，Dataset用于自定义数据集格式，DataLoader用于批量加载数据并支持并行处理，是高效训练的基础
from transformers import BertTokenizer, BertModel
#导入 BERT 模型相关工具。BertTokenizer用于文本预处理（如分词、转换为词向量索引），BertModel是预训练的 BERT 模型，可用于特征提取或微调任务（如文本分类、情感分析等）
import pandas as pd
#导入 Pandas，用于处理结构化数据（如 CSV 表格），方便读取、清洗和转换数据集。
import numpy as np
import matplotlib
matplotlib.use('Agg')  # 设置后端为 Agg，适合无 GUI 环境，适合在无图形界面的环境（如服务器）中生成图片文件
import matplotlib.pyplot as plt
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, classification_report
from sklearn.preprocessing import MultiLabelBinarizer#用于多标签分类任务中的标签预处理（将多类别标签转换为二进制矩阵）
from sklearn.model_selection import train_test_split
import os#导入操作系统接口，用于文件路径处理、目录创建等操作（如保存模型、图片时指定路径）

# 这个CarOpinionDataset类继承自PyTorch的Dataset，用于封装数据集。它将文本数据进行编码，并将主题和情感标签转换为张量
class CarOpinionDataset(Dataset):
    def __init__(self, texts, topics, sentiments, tokenizer, max_length):
        self.texts = texts
        self.topics = topics
        self.sentiments = sentiments
        self.tokenizer = tokenizer#BERT 分词器（如BertTokenizer），用于文本预处理
        self.max_length = max_length#文本最大长度（超过则截断，不足则补全），保证输入模型的序列长度一致

    def __len__(self):
        return len(self.texts)#返回数据集的样本总数，是Dataset类的强制实现方法

    def __getitem__(self, idx):#根据索引 idx 获取单条样本，并完成预处理（文本编码 + 标签转张量），是Dataset类的核心方法
        text = str(self.texts[idx])#通过索引 idx 提取单条文本、主题标签、情感标签
        topic = self.topics[idx]
        sentiment = self.sentiments[idx]

        encoding = self.tokenizer.encode_plus(
            text,
            add_special_tokens=True,#添加 BERT 要求的特殊符号（如句首[CLS]、句尾[SEP]）
            max_length=self.max_length,#超过max_length的文本截断
            return_token_type_ids=False,
            padding='max_length',#不足max_length的文本用[PAD]补全，保证序列长度统一
            truncation=True,
            return_attention_mask=True,#返回注意力掩码（标记哪些是真实文本、哪些是补全的[PAD]，模型会忽略[PAD]部分）
            return_tensors='pt'#直接返回 PyTorch 张量（避免后续手动转换）
        )
        return {
            'input_ids': encoding['input_ids'].flatten(),
            'attention_mask': encoding['attention_mask'].flatten(),
            'topics': torch.tensor(topic, dtype=torch.float),  # 因主题可能是多标签（如一条文本同时涉及 “价格” 和 “油耗”，标签为[1,1,0,...]），用浮点型适配多标签分类的损失计算
            'sentiments': torch.tensor(sentiment, dtype=torch.long)#情感是单标签分类（如 0 = 负面、1 = 中性、2 = 正面），用长整型适配分类任务的标签格式
        }

#这段代码定义了一个基于 BERT 预训练模型的多任务分类器 BertClassifier，继承自 PyTorch 的 nn.Module，核心功能是同时完成主题分类和情感分类两个任务。模型通过 BERT 提取文本特征，再分别通过两个独立的线性层输出分类结果，适合处理需要同时预测文本主题和情感倾向的场景（如汽车评论的主题识别 + 情感分析）
class BertClassifier(nn.Module):
    def __init__(self, bert_model_name, num_topics, num_sentiments):#定义模型的核心组件，包括预训练 BERT、dropout 层和两个分类头
        super(BertClassifier, self).__init__()
        self.bert = BertModel.from_pretrained(bert_model_name)#加载预训练 BERT 模型，作为文本特征提取器。BERT 能将输入文本转换为包含上下文信息的高维特征向量
        self.dropout = nn.Dropout(0.5)#训练时随机让 50% 的神经元失活，用于减轻过拟合
        self.topic_classifier = nn.Linear(self.bert.config.hidden_size, num_topics)#主题分类头，是一个线性层（nn.Linear），输入维度为 BERT 的隐藏层维度（self.bert.config.hidden_size，通常为 768），输出维度为num_topics（主题类别数），负责输出主题分类的 logits；
        self.sentiment_classifier = nn.Linear(self.bert.config.hidden_size, num_sentiments)

    def forward(self, input_ids, attention_mask):#input_ids：文本经分词器编码后的数字序列（BERT 的输入格式）；
attention_mask：注意力掩码（标记哪些位置是真实文本，哪些是填充的[PAD]，BERT 会忽略填充部分）。
        outputs = self.bert(input_ids=input_ids, attention_mask=attention_mask)
        pooled_output = outputs.pooler_output
        pooled_output = self.dropout(pooled_output)#提取句子级特征向量，应用 dropout 层，增强模型泛化能力调用 BERT 模型处理输入，返回的outputs包含多个信息，其中 pooler_output 是 BERT 对整个句子的全局特征表示（从[CLS] token 的隐藏状态转换而来
        topic_logits = self.topic_classifier(pooled_output)#将处理后的特征输入主题分类头，输出主题分类的 logits（未经过 softmax 的原始分数）
        sentiment_logits = self.sentiment_classifier(pooled_output)
        return topic_logits, sentiment_logits#后续可通过损失函数（如主题用 BCEWithLogitsLoss，情感用 CrossEntropyLoss）计算损失并反向传播更新参数

# 这个函数从文件中读取数据，将文本、主题和情感标签分别存储在列表中。主题和情感标签通过映射转换为数字。
def read_data(file_path):
    texts = []
    topics = []
    sentiments = []
    topic_mapping = {'动力': 0, '价格': 1, '内饰': 2, '配置': 3, '安全性': 4, '外观': 5, '操控': 6, '油耗': 7, '空间': 8, '舒适性': 9}  # 主题标签映射
    sentiment_mapping = {'0': 0, '1': 1, '-1': 2}  # 情感标签映射，将 -1 映射为 2
    with open(file_path, 'r', encoding='utf-8') as file:
        for line in file:
            parts = line.strip().split('\t')
            text = parts[0]#假设文件每行数据用制表符\t分隔为两部分（第一部分是文本，第二部分是标签组合）；text = parts[0]提取第一部分作为原始文本，存入texts列表
            topic_labels = []
            sentiment_label = 0
            for label in parts[1].split(' '):
#假设标签部分用空格分隔多个 “主题 #情感” 组合（如 “动力 #1 价格 #-1” 表示文本涉及 “动力” 主题且情感为正面，涉及 “价格” 主题且情感为负面）
#循环拆分每个 “主题 #情感” 标签：topic_label, sentiment = label.split('#')
#将主题标签转换为数字：topic_mapping[topic_label]，存入topic_labels列表（一条文本可能对应多个主题，因此是列表形式）；
将情感标签转换为数字：sentiment_mapping[sentiment]，并累加得到sentiment_label（这里逻辑是将多个情感标签的数值相加，可能用于简化为单标签，具体需结合数据实际场景）
                topic_label, sentiment = label.split('#')
                topic_labels.append(topic_mapping[topic_label])
                sentiment_label += sentiment_mapping[sentiment]
            texts.append(text)
            topics.append(topic_labels)
            sentiments.append(sentiment_label)
    return texts, topics, sentiments

# 读取训练和测试数据，使用MultiLabelBinarizer将多标签数据转换为二进制形式，并创建数据加载器。
def prepare_data(tokenizer, max_length, train_path, test_path):
    train_texts, train_topics, train_sentiments = read_data(train_path)
    test_texts, test_topics, test_sentiments = read_data(test_path)
    mlb = MultiLabelBinarizer(classes=list(range(10)))  # 10 个主题类别
    train_topics = mlb.fit_transform(train_topics)
    test_topics = mlb.transform(test_topics)
    train_dataset = CarOpinionDataset(train_texts, train_topics, train_sentiments, tokenizer, max_length)
    test_dataset = CarOpinionDataset(test_texts, test_topics, test_sentiments, tokenizer, max_length)
    train_loader = DataLoader(train_dataset, batch_size=8, shuffle=True)
    test_loader = DataLoader(test_dataset, batch_size=8)
    return train_loader, test_loader, mlb

def train(model, train_loader, test_loader, optimizer, device, num_epochs, criterion_topic, criterion_sentiment):
    train_losses = []
    test_losses = []
    train_accuracies = []
    test_accuracies = []
    # scheduler = torch.optim.lr_scheduler.StepLR(optimizer, step_size=5, gamma=0.1)
    for epoch in range(num_epochs):
        model.train()
        total_loss = 0
        all_preds_topic = []
        all_labels_topic = []
        all_preds_sentiment = []
        all_labels_sentiment = []
        for batch in train_loader:
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            topics = batch['topics'].to(device)
            sentiments = batch['sentiments'].to(device)
            optimizer.zero_grad()
            topic_logits, sentiment_logits = model(input_ids, attention_mask)
            loss_topic = criterion_topic(topic_logits, topics)
            loss_sentiment = criterion_sentiment(sentiment_logits, sentiments)
            loss = loss_topic + loss_sentiment
            total_loss += loss.item()
            loss.backward()
            optimizer.step()
            # 主题分类的预测
            topic_preds = (torch.sigmoid(topic_logits) > 0.5).float()
            all_preds_topic.extend(topic_preds.cpu().detach().numpy())
            all_labels_topic.extend(topics.cpu().numpy())
            # 情感分类的预测
            _, sentiment_preds = torch.max(sentiment_logits, dim=1)
            all_preds_sentiment.extend(sentiment_preds.cpu().detach().numpy())
            all_labels_sentiment.extend(sentiments.cpu().numpy())
        train_losses.append(total_loss / len(train_loader))
        train_accuracy_topic = accuracy_score(all_labels_topic, all_preds_topic)
        train_accuracy_sentiment = accuracy_score(all_labels_sentiment, all_preds_sentiment)
        train_accuracies.append((train_accuracy_topic + train_accuracy_sentiment) / 2)
        print(f"Epoch {epoch+1}/{num_epochs}, Train Loss: {train_losses[-1]}, Train Accuracy: {train_accuracies[-1]}")
        model.eval()
        total_loss = 0
        all_preds_topic = []
        all_labels_topic = []
        all_preds_sentiment = []
        all_labels_sentiment = []
        with torch.no_grad():
            for batch in test_loader:
                input_ids = batch['input_ids'].to(device)
                attention_mask = batch['attention_mask'].to(device)
                topics = batch['topics'].to(device)
                sentiments = batch['sentiments'].to(device)
                topic_logits, sentiment_logits = model(input_ids, attention_mask)
                loss_topic = criterion_topic(topic_logits, topics)
                loss_sentiment = criterion_sentiment(sentiment_logits, sentiments)
                loss = loss_topic + loss_sentiment
                total_loss += loss.item()
                # 主题分类的预测
                topic_preds = (torch.sigmoid(topic_logits) > 0.5).float()
                all_preds_topic.extend(topic_preds.cpu().numpy())
                all_labels_topic.extend(topics.cpu().numpy())
                # 情感分类的预测
                _, sentiment_preds = torch.max(sentiment_logits, dim=1)
                all_preds_sentiment.extend(sentiment_preds.cpu().numpy())
                all_labels_sentiment.extend(sentiments.cpu().numpy())
        test_losses.append(total_loss / len(test_loader))
        test_accuracy_topic = accuracy_score(all_labels_topic, all_preds_topic)
        test_accuracy_sentiment = accuracy_score(all_labels_sentiment, all_preds_sentiment)
        test_accuracies.append((test_accuracy_topic + test_accuracy_sentiment) / 2)
        print(f"Epoch {epoch+1}/{num_epochs}, Test Loss: {test_losses[-1]}, Test Accuracy: {test_accuracies[-1]}")
    return train_losses, test_losses, train_accuracies, test_accuracies
# 这个函数实现了模型的训练过程。它在每个epoch中计算训练和测试损失，并评估准确率。
# 使用了二元交叉熵损失函数（BCEWithLogitsLoss）进行多标签分类，使用交叉熵损失函数（CrossEntropyLoss）进行情感分类。

def evaluate(model, test_loader, device, mlb):
    model.eval()
    all_preds_topic = []
    all_labels_topic = []
    all_preds_sentiment = []
    all_labels_sentiment = []
    with torch.no_grad():
        for batch in test_loader:
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            topics = batch['topics'].to(device)
            sentiments = batch['sentiments'].to(device)
            topic_logits, sentiment_logits = model(input_ids, attention_mask)
            # 主题分类的预测
            topic_preds = (torch.sigmoid(topic_logits) > 0.5).float()
            all_preds_topic.extend(topic_preds.cpu().numpy())
            all_labels_topic.extend(topics.cpu().numpy())
            # 情感分类的预测
            _, sentiment_preds = torch.max(sentiment_logits, dim=1)
            all_preds_sentiment.extend(sentiment_preds.cpu().numpy())
            all_labels_sentiment.extend(sentiments.cpu().numpy())
    print("Topic Classification Report:")
    print(classification_report(all_labels_topic, all_preds_topic, target_names=[str(i) for i in range(10)], zero_division=0))
    print("Sentiment Classification Report:")
    print(classification_report(all_labels_sentiment, all_preds_sentiment, target_names=['neutral', 'positive', 'negative']))
    return all_preds_topic, all_labels_topic, all_preds_sentiment, all_labels_sentiment

def main():
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    tokenizer_path = '/root/bert-base-uncased'
    tokenizer = BertTokenizer.from_pretrained(tokenizer_path)
    model = BertClassifier(tokenizer_path, num_topics=10, num_sentiments=3).to(device)
    optimizer = torch.optim.AdamW(model.parameters(), lr=1e-5,weight_decay=0.01)
    criterion_topic = nn.BCEWithLogitsLoss()  # 多标签分类使用 BCEWithLogitsLoss
    criterion_sentiment = nn.CrossEntropyLoss()  # 情感分类使用 CrossEntropyLoss
    train_path = '/root/train.txt'  # 修改为远程服务器上的训练集路径
    test_path = '/root/test.txt'    # 修改为远程服务器上的测试集路径
    train_loader, test_loader, mlb = prepare_data(tokenizer, max_length=128, train_path=train_path, test_path=test_path)
    num_epochs = 50
    train_losses, test_losses, train_accuracies, test_accuracies = train(model, train_loader, test_loader, optimizer, device, num_epochs, criterion_topic, criterion_sentiment)
    all_preds_topic, all_labels_topic, all_preds_sentiment, all_labels_sentiment = evaluate(model, test_loader, device, mlb)
    # 绘制 loss 曲线和 accuracy 曲线
    plt.figure(figsize=(12, 4))
    plt.subplot(1, 2, 1)
    plt.plot(train_losses, label='Train Loss')
    plt.plot(test_losses, label='Test Loss')
    plt.title('Loss Curve')
    plt.xlabel='Epoch'
    plt.ylabel='Loss'
    plt.legend()
    plt.subplot(1, 2, 2)
    plt.plot(train_accuracies, label='Train Accuracy')
    plt.plot(test_accuracies, label='Test Accuracy')
    plt.title='Accuracy Curve'
    plt.xlabel='Epoch'
    plt.ylabel='Accuracy'
    plt.legend()
    plt.savefig('/tmp/loss_accuracy_curves.png')  # 保存图像到文件，而不是显示图像
    # plt.show()


if __name__ == "__main__":
    main()