目录

一、项目概述

二、需要解决的问题

三、数据预处理

1、词汇表构建（vocab_creat.py）

2、数据集加载（load_dataset.py）

四、模型构建（TextRNN.py）

1、嵌入层（Embedding Layer）

2、长短期记忆网络层（LSTM）

3、全连接层（Fully Connected Layer）

五、模型训练与评估（train_eval_test.py）

1、训练过程（train 函数）

2、评估过程（evaluate 函数和 test 函数）

六、项目总结

---

本文将介绍一个基于自然语言处理技术对微博评论文本（simplifyweibo_4_moods.csv）进行情感分析的项目。

一、项目概述

本文将介绍一个基于自然语言处理技术对微博评论文本（simplifyweibo_4_moods.csv）进行情感分析的项目。

本项目旨在构建一个能够对微博评论进行情感分类的模型，将评论分为 “喜悦”、“愤怒”、“厌恶” 和 “低落” 四种情感类别。项目涵盖了从数据预处理、模型构建到训练和评估的完整流程。

项目任务：对微博评论信息的情感分析，建立模型，自动识别评论信息的情绪状态。

二、需要解决的问题

1、目标：将评论内容转换为词向量。

2、每个词/字转换为词向量长度(维度)200

3、每一次传入的词/字的个数是否就是评论的长度?     

应该是固定长度，每次传入数据与图像相似。     

例如选择长度为32。则传入的数据为32*200

4、一条评论如果超过32个词/字怎么处理？     

直接删除后面的内容

5、一条评论如果没有32个词/字怎么处理？     

缺少的内容，统一使用一个数字（非词/字的数字）替代。

6、如果语料库中的词/字太多是否可以压缩？     

可以，某些词/字出现的频率比较低，可能训练不出特征。因此可以选择频率比较高的词来训练。例如选择4760个。

7、被压缩的词/字如何处理？     

可以统一使用一个数字（非词/字的数字）替代。

三、数据预处理

1、词汇表构建（vocab_creat.py）

（1）、首先，通过自定义的分字函数tokenizer将每条评论内容分隔成单个字符。然后，统计每个字符在所有评论中出现的次数，形成一个字典vocab_dic。

（2）、为了控制词汇表的大小，只保留出现频率高于设定阈值（本项目中min_freq = 1）的字符，并按照出现次数从高到低排序，选取前MAX_VOCAB_SIZE = 4760个字符。

（3）、最后，将特殊字符<UNK>（未知字）和<PAD>（填充）添加到词汇表中，并将词汇表保存为pkl文件（simplifyweibo_4_moods.pkl），以便后续使用。

from tqdm import tqdm
import pickle as pkl

MAX_VOCAB_SIZE = 4760
UNK, PAD = '<UNK>', '<PAD>'


def build_vocab(file_path, max_size, min_freq):
    tokenizer = lambda x: [y for y in x]  # 定义一个分字函数
    vocab_dic = {}
    with open(file_path, 'r', encoding='UTF-8') as f:
        i = 0
        for line in tqdm(f):  # 进度条
            if i == 0:
                i += 1
                continue
            lin = line[2:].strip()  # 获取评论内容，删除标签
            if not lin:  # 空行跳过
                continue
            for word in tokenizer(lin):
                vocab_dic[word] = vocab_dic.get(word, 0) + 1  # 以字典保存每个字出现的次数
        vocab_list = sorted([_ for _ in vocab_dic.items() if _[1] > min_freq], key=lambda x: x[1], reverse=True)[
                     :max_size]
        vocab_dic = {word_count[0]: idx for idx, word_count in enumerate(vocab_list)}
        vocab_dic.update({UNK: len(vocab_dic), PAD: len(vocab_dic) + 1})  # UNK:4760, PAD:4761
        print(vocab_dic)
        pkl.dump(vocab_dic, open('simplifyweibo_4_moods.pkl', 'wb'))  # one-hot编码
        print(f"Vocab size:{len(vocab_dic)}")
    return vocab_dic


if __name__ == "__main__":
    vocab = build_vocab('simplifyweibo_4_moods.csv', MAX_VOCAB_SIZE, 1)
    print("vocab")

2、数据集加载（load_dataset.py）

（1）、从保存的词汇表文件中读取词汇信息。然后，对原始的微博评论文本数据进行处理。

（2）、对于每条评论，提取其情感标签（0，1，2，3），并将评论内容进行分字处理。根据设定的最大长度pad_size = 70，如果字符数少于 70，则用<PAD>填充；如果多于 70，则只取前 70 个字。

（3）、将每个字符转换为词汇表中的对应索引，最终将处理后的每条评论信息（包括字符索引列表、情感标签和实际长度）以元组形式存储在列表中。

（4）、接着，将整个数据集随机打乱，并按照 80%、10%、10% 的比例划分为训练集、验证集和测试集。

注：__next__方法用于按顺序获取数据

      __getitem__方法通过索引获取数据

from tqdm import tqdm
import pickle as pkl
import random
import torch

UNK, PAD = '<UNK>', '<PAD>'  # 未知字，padding符号


def load_dataset(path, pad_size=70):  # path为文件地址，pad_size为单条评论字符的最大长度
    contents = []  # 用来存储转换为数值标号的句子
    vocab = pkl.load(open('simplifyweibo_4_moods.pkl', 'rb'))  # 读取vocab词库文件，rb二进制只读
    tokenizer = lambda x: [y for y in x]  # 自定义函数用来将字符串分隔成单个字符并存入列表
    with open(path, 'r', encoding='utf8') as f:
        i = 0
        for line in tqdm(f):  # 遍历文件内容的每一行，同时展示进度条
            if i == 0:  # 此处循环目的为了跳过第一行的无用内容
                i += 1
                continue
            if not line:  # 筛选是不是空行，空行则跳过
                continue
            label = int(line[0])  # 返回当前行的标签
            content = line[2:].strip('\n')  # 取出标签和逗号后的所有内容，同时去除前后的换行符(评论内容)
            words_line = []
            token = tokenizer(content)  # 将每一行的内容进行分字，返回一个列表
            seq_len = len(token)  # 获取一行实际内容的长度
            if pad_size:
                if len(token) < pad_size:  # 如果一行的字符数少于70，则填充字符<PAD>，填充个数为少于的部分的个数
                    token.extend([PAD] * (pad_size - len(token)))
                else:  # 如果一行的字大于70，则只取前70个字
                    token = token[:pad_size]  # 如果一条评论种的宁大于或等于70个字，索引的切分
                    seq_len = pad_size  # 当前评论的长度
            # word to id
            for word in token:  # 遍历实际内容的每一个字符
                words_line.append(vocab.get(word, vocab.get(UNK)))  # vocab为词库，其中为字典形式，
                # 使用get去获取遍历出来的字符的值，值可表示索引值，
                # 如果该字符不在词库中则将其值增加为字典中键UNK对应的值，
                # words_line中存放的是每一行的每一个字符对应的索引值
            contents.append((words_line, int(label), seq_len))  # 将每一行评论的字符对应的索引以及这一行评论的类别，还有当前评论的实际内容的长度，以元组的形式存入列表
        random.shuffle(contents)  # 随机打乱每一行内容的顺序
        """切分80%训练集、10%验证集、10%测试集"""
        train_data = contents[: int(len(contents) * 0.8)]  # 前80%的评论数据作为训练集
        dev_data = contents[int(len(contents) * 0.8):int(len(contents) * 0.9)]  # 把80%~90%的评论数据集作为验证数热
        test_data = contents[int(len(contents) * 0.9):]  # 90%~最后的数据作为测试数据集
    return vocab, train_data, dev_data, test_data  # 返回词库、训练集、验证集、测试集，数据集为列表中的元组形式


class DatasetIterater(object):
    def __init__(self, batches, batch_size, device):
        self.batch_size = batch_size
        self.batches = batches
        self.n_batches = len(batches) // batch_size
        self.residue = False
        if len(batches) % self.n_batches != 0:  # 表示有余数
            self.residue = True
        self.index = 0
        self.device = device

    def _to_tensor(self, datas):
        x = torch.LongTensor([_[0] for _ in datas]).to(self.device)  # 评论内容
        y = torch.LongTensor([_[1] for _ in datas]).to(self.device)  # 评论情感

        # pad前的长度(超过pad_size的设为pad_size)
        seq_len = torch.LongTensor([_[2] for _ in datas]).to(self.device)
        return (x, seq_len), y  # (([23,34,..13],70),2)
        # __getitem__:是通过索引的方式获取数据对象中的内容。__next__ 是使用 for i in trgin iter:

    def __next__(self):  # 用于定义迭代器对象的下一个元素。当一个对象实现了__next_方法时，它可以被用于创建迭代器对象。
        if self.residue and self.index == self.n_batches:  # 当读取到数据的最后一个batch:
            batches = self.batches[self.index * self.batch_size: len(self.batches)]
            self.index += 1
            batches = self._to_tensor(batches)  # 转换数据类型tensor
            return batches
        elif self.index > self.n_batches:  # 当读取完最后一个batch时:
            self.index = 0
            raise StopIteration  # 为了防止迭代永远进行，我们可以使用StopIteration(停止迭代)语句
        else:  # 当没有读取到最后一个batch时:
            batches = self.batches[self.index * self.batch_size:(self.index + 1) * self.batch_size]  # 提取当前bathsize的数据量
            self.index += 1
            batches = self._to_tensor(batches)
            return batches

    def __iter__(self):
        return self

    def __len__(self):
        if self.residue:
            return self.n_batches + 1
        else:
            return self.n_batches



    if __name__ == '__main__':
        vocab, train_data, dev_data, test_data = load_dataset('simplifyweibo_4_moods.csv')
        print(train_data, dev_data, test_data)
        print('结束')

    # 将train_data、dev_data、test_data数据内容保存为pkl文件，分别后面直接读取。
    # #当我们自己写一个函数的时候，调试，函数调试好，

四、模型构建（TextRNN.py）

本项目采用了 TextRNN 模型，它是一种基于循环神经网络（RNN）的架构，具有以下特点：

1、嵌入层（Embedding Layer）

（1）、如果有预训练的词嵌入向量（本项目中从embedding_Tencent.npz文件加载），则使用nn.Embedding.from_pretrained方法创建嵌入层；否则，使用nn.Embedding随机初始化词嵌入矩阵。

（2）、嵌入层的作用是将输入的字符索引转换为低维的向量表示，捕捉字符之间的语义关系。

2、长短期记忆网络层（LSTM）

（1）、采用多层双向 LSTM 结构（nn.LSTM），其中每层有 128 个隐藏单元，共 3 层。双向 LSTM 能够同时捕捉文本的正向和反向信息，更好地理解文本的语义。

（2）、通过设置batch_first = True，使输入数据的维度顺序符合批量优先的原则，方便后续的计算。同时，使用dropout = 0.3来防止过拟合。

3、全连接层（Linear）

最后一层是全连接层（nn.Linear），它将 LSTM 层输出的特征向量映射到情感类别数量的维度上（本项目中为 4 种情感类别），从而得到每个类别的预测概率。

import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
import sys


class Model(nn.Module):
    def __init__(self, embedding_pretrained, n_vocab, embed, num_classes):
        super(Model, self).__init__()
        if embedding_pretrained is not None:
            self.embedding = nn.Embedding.from_pretrained(embedding_pretrained, padding_idx=n_vocab - 1)
        else:
            self.embedding = nn.Embedding(n_vocab, embed, padding_idx=n_vocab - 1)
        self.lstm = nn.LSTM(embed, 128, 3, bidirectional=True, batch_first=True, dropout=0.3)
        self.fc = nn.Linear(128 * 2, num_classes)

    def forward(self, x):
        x, _ = x
        out = self.embedding(x)
        out, _ = self.lstm(out)
        out = self.fc(out[:, -1, :])
        return out

五、模型训练与评估（train_eval_test.py）

1、训练过程（train 函数）

（1）、在训练过程中，使用torch.optim.Adam优化器对模型参数进行更新，学习率设置为1e - 3。

（2）、每个训练轮次（epoch）中，遍历训练集的每个批次（batch）。对于每个批次的数据，计算模型的输出和损失（使用交叉熵损失函数F.cross_entropy），然后进行反向传播更新模型参数。

（3）、每 100 个批次，在训练集和验证集上评估模型的性能。计算准确率（accuracy）作为评估指标，如果验证集上的损失小于之前的最优损失，则保存当前模型为最优模型，并记录对应的批次编号。

（4）、如果在连续 10000 个批次中模型没有得到优化，则自动停止训练。

2、评估过程（evaluate 函数和 test 函数）

（1）、在评估阶段，关闭梯度计算（with torch.no_grad()），以节省内存和计算资源。

（2）、对于测试集或验证集的数据，计算模型的输出和损失，并将预测结果与真实标签进行比较。计算准确率和其他评估指标（如分类报告metrics.classification_report），以全面评估模型的性能。
 

import torch
import torch.nn.functional as F
import torch.nn as nn
import numpy as np
from sklearn import metrics
import time


def evaluate(class_list,model,data_iter,test=False):
    model.eval()
    loss_total=0
    predict_all = np.array([],dtype=int)
    labels_all = np.array([],dtype=int)
    with torch.no_grad():  # 一个上下文管理器，关闭梯度计算。当你确认不会调用Tensor.backward()的
        for texts,labels in data_iter:
            outputs = model(texts)
            loss = F.cross_entropy(outputs,labels)
            loss_total += loss
            labels = labels.data.cpu().numpy()
            predic = torch.max(outputs.data,1)[1].cpu().numpy()
            labels_all = np.append(labels_all,labels)
            predict_all = np.append(predict_all,predic)

    acc = metrics.accuracy_score(labels_all,predict_all)
    if test:
        report = metrics.classification_report(labels_all,predict_all,target_names=class_list,digits=4)
        return acc,loss_total/len(data_iter),report
    return acc,loss_total/len(data_iter)


def test(model,test_iter,class_list):
    model.load_state_dict(torch.load('TextRNN.skpt'))
    model.eval()
    start_time = time.time()
    test_acc,test_loss,test_report = evaluate(class_list,model,test_iter,test=True)
    msg = "Test Loss:{0:>5.2},Test Acc:{1:6.2%}"
    print(msg.format(test_loss,test_acc))
    print(test_report)
    pass


def train(model,train_iter,dev_iter,test_iter,class_list):
    model.train()
    optimizer = torch.optim.Adam(model.parameters(),lr=1e-3)

    total_batch = 0
    dev_best_loss = float('inf')
    last_improve = 0
    flag = False
    epochs = 2
    for epoch in range(epochs):
        print('Epoch [{}/{}]'.format(epoch+1,epochs))
        for i,(trains,labels) in enumerate(train_iter):
            # 经过DatasetIterater中的to_tensor  返回的数据格式为:(x，seq_len)，y
            outputs = model(trains)
            loss = F.cross_entropy(outputs,labels)
            model.zero_grad()
            loss.backward()
            optimizer.step()
            if total_batch % 100 == 0:   # 每多少轮输出在训练集和验证集上的效果
                predic = torch.max(outputs.data,1)[1].cpu()
                train_acc = metrics.accuracy_score(labels.data.cpu(),predic)
                dev_acc,dev_loss = evaluate(class_list,model,dev_iter)    # 将验证数据集传入模型，获得验证结果
                if dev_loss < dev_best_loss:
                    dev_best_loss = dev_loss  # 保存最优模型
                    torch.save(model.state_dict(),'TextRNN.ckpt')
                    last_improve = total_batch   # 保存最优模型的batch值 800batchs 21000

                msg = 'Iter:{0:>6},Train Loss:{1:>5.2},Train Acc:{2:>6.2%},Val Loss:{3:>5.2},Val Acc:{4:>6.2%}'
                print(msg.format(total_batch,loss.item(), train_acc, dev_loss, dev_acc))

                model.train()
            total_batch += 1
            if total_batch - last_improve > 10000:
                print("No optimization for a long time,auto-stopping...")
                flag = True

        if flag:
            break

    # test(model,test_iter,class_list)

六、项目总结

通过本项目，我们展示了一个完整的自然语言处理情感分析流程，从数据预处理到模型构建和训练评估。TextRNN 模型在微博评论文本的情感分类任务上取得了一定的效果。然而，自然语言处理是一个复杂的领域，还有很多可以改进和优化的地方。