《深度学习》【项目】自然语言处理——情感分析 ＜下＞

目录

一、了解项目

1、任务

2、文件内容

二、续接上篇内容

1、打包数据，转化Tensor类型

2、定义模型，前向传播函数

3、定义训练、测试函数

4、最终文件格式

5、定义主函数

运行结果：

---

一、了解项目

1、任务

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

2、文件内容

二、续接上篇内容

《深度学习》【项目】自然语言处理——情感分析 ＜上＞-CSDN博客文章浏览阅读537次，点赞23次，收藏16次。对微博评论信息的情感分析，建立模型，自动识别评论信息的情绪状态。https://ahao1004.blog.csdn.net/article/details/142926591?fromshare=blogdetail&sharetype=blogdetail&sharerId=142926591&sharerefer=PC&sharesource=qq_64603703&sharefrom=from_link

        上篇博客，我们已经实现了评论固定长度输出、词表的生成、数据集的切分等操作，接下来需要对数据进行打包、放入模型进行训练等操作

1、打包数据，转化Tensor类型

        将下列代码写入上述创建的load_dataset.py文件内，写成下列格式

class DatasetIterater(object):   # 定义一个类，用于迭代地处理数据集，将其分割成指定大小的批次（batch），并能够在GPU或其他设备上运行# 将数据batches   切分为batch_size的包。def __init__(self,batches,batch_size,device):  # 输入参数为：样本数据、单个批次中的样本条数、设备GPUself.batch_size = batch_sizeself.batches = batchesself.n_batches = len(batches) // batch_size   # 计算批次数self.residue = False   # 记录划分后的数据是否存在剩余的数据，初始化为Falseif len(batches) % self.n_batches != 0:  # 表示有余数self.residue = Trueself.index = 0   # 追踪当前批次的索引self.device = devicedef _to_tensor(self,datas):   # 自己定义的一个函数私有方法，将一批数据转换为PyTorch张量，并发送到指定设备（GPU），datas为一个列表x = torch.LongTensor([_[0] for _ in datas]).to(self.device)  # _[0]为评论内容y = torch.LongTensor([_[1] for _ in datas]).to(self.device)  # _[1]评论情感#  pad前的长度(超过pad_size的设为pad_size)seq_len = torch.LongTensor([_[2] for _ in datas]).to(self.device)  # _[2]为评论长度return (x,seq_len),y   # 返回评论的内容、评论长度、标签# getitem__: 是通过索引的方式获取数据对象中的内容。-_next_是使用 for i in trgin_iter:def __next__(self):  # 用于定义迭代器对象的下一个元素。当一个对象实现了__next__方法时，它可以被用于创建迭代器对象。# 处理打包后的剩余数据的批次数据if self.residue and self.index == self.n_batches:   # 存在剩余数据 并且 当前批次索引等于批次数，此时即最后一部分打包不齐而剩余的数据batches = self.batches[self.index*self.batch_size:len(self.batches)]  # self.index*self.batch_size为批次索引*批次内样本个数，即当前批次中的样本所在总样本内的位置 到 总样本数据长度，此条表示取出打包剩余的样本self.index += 1  # 批次数加一batches = self._to_tensor(batches)  # 调用上述的私有函数，将样本数据传入，将其转换数据类型tensorreturn batches   # 返回Tensor数据类型的样本数据elif self.index > self.n_batches:  # 如果迭代使批次索引超过批次总数，终止循环self.index = 0  # 重置批次索引为1raise StopIteration    # 为了防止迭代永远进行，我们可以使用StopIteration(停止迭代)语句else:    # 当没有读取到最后一个batch批次时:baches = self.batches[self.index*self.batch_size:(self.index+1)*self.batch_size]   # 提取当前批次的数据样本self.index += 1   # 批次索引+1baches = self._to_tensor(baches)   # 将样本数据转化为张量格式return baches   # 返回张量类型的样本数据def __iter__(self):  # 这个方法使得类的实例能够成为迭代器，返回实例本身。return selfdef __len__(self):  # 返回迭代器的长度，即完整批次的数量if self.residue:return self.n_batches + 1  # 如果存在剩余数据，长度会增加1。else:return self.n_batches

2、定义模型，前向传播函数

        将下列代码写入新创建的文件TextRNN.py

import torch.nn as nnclass Model(nn.Module):   # 定义一个类，继承神经网络的基类，参数管理、模型保存加载...def __init__(self,embedding_pretrained,n_vocab,embed,num_classes):   # 传入参数表示为：预训练的词向量(当前项目导入腾讯训练好的词向量)、词汇表的长度、词向量维度、分类标签的数量super(Model,self).__init__()if embedding_pretrained is not None:  # 如果有预训练模型# 创建一个词嵌入层，用与接收预训练的嵌入层权重作为输入，指定填充词在词汇表中的索引为n_vocab-1，freeze:指定是否冻结embedinq层的权重,False表示可以更新预训练模型的权重参数self.embedding = nn.Embedding.from_pretrained(embedding_pretrained,padding_idx=n_vocab-1,freeze=False)else:  # 如果没有预训练模型，则初始化一个随机嵌入层，维度为n_vocab*embed   项目内是4762*200self.embedding = nn.Embedding(n_vocab,embed,padding_idx=n_vocab-1)# 建立LSTM网络层，输入维度为embed，有128个隐藏单元共三层，bidirectional=True表示双向LSTM，所以输出为128*2，batch_first=True表示输入张量第一个维度是批次数，dropout = 0.3表示LSTM层使用的dropout比例self.lstm = nn.LSTM(embed,128,3,bidirectional=True,batch_first=True,dropout=0.3)# 128为每一层中每个隐状态中的U、W、V的神经元个数，# 3为隐藏层的层数,batch_first=True表示输入和输出张量将以(batch，seq，feature)而不是(seq,betch,featur)。# bidirectiongl = True: 指定LSTM是双向的。网络会同时从前向后和从后向前处理输入序列，两个方向的# dropout = 0.3: 这指定了在LSTM层中使用的dropout比例。Dropout是一种正则化技术，用于防止网络在训练过程中过拟?self.fc = nn.Linear(128*2,num_classes)  # 设置全连接层，在每个时间步的最后一个状态的输出映射到类别数上def forward(self,x):   # 定义前向传播函数，输入的参数x为batch_size批次数以及sequence_length单词样本数x,_ = x    # 返回新的x值为批次数out = self.embedding(x)   # 将批次数传入词嵌入层，将整数索引转换为连续的、密集的词向量out,_ = self.lstm(out)   # 将词向量传入LSTM网络层out = self.fc(out[:,-1,:])   # 只选择LSTM输出序列的最后一个时间步的隐藏状态传递给self.fcreturn out

3、定义训练、测试函数

        创建一个文件train_eval_test.py，将下列代码写入其中

import torch
import torch.nn.functional as F
import numpy as np
from sklearn import metrics
import timedef evaluate(class_list,model,data_iter,test=False):   # 传入参数：种类名称列表、训练好的模型、验证集数据，test表示是否进行测试模式model.eval()   # 模型开始测试loss_total = 0  # 初始化总损失值为0predict_all = np.array([],dtype=int)   # 定义一个数组用于存放预测结果的标签labels_all = np.array([],dtype=int)    # 存放所有样本的真实标签with torch.no_grad():  # 一个上下文管理器，关闭梯度计算for texts,labels in data_iter:   # 遍历出来 每128条评价的包 的独热编码及长度 和标签outputs = model(texts)   # 输出模型进行测试，返回每个包中每条评论的测试结果loss = F.cross_entropy(outputs,labels)   # 计算交叉熵损失值loss_total += loss   # 损失值叠加labels = labels.data.cpu().numpy()   # 将真实标签转化为numpy数组predic = torch.max(outputs.data,1)[1].cpu().numpy()   # 计算预测值的标签并转化为numpy数组labels_all = np.append(labels_all,labels)    # 将真实标签增加到labels_all数组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)   # 如果不是测试模式，那么打印分类报告，target_names用于识别每个类别的名称class_list,digits表示打印报告中浮点数的位数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.ckpt'))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)def train(model,train_iter,dev_iter,test_iter,class_list):  # 传入模型结构、训练集、验证集、测试集、标签类别model.train()   # 开始训练optimizer = torch.optim.Adam(model.parameters(),lr=1e-3)   # 优化器，用于更新模型权重，学习率为0.001total_batch = 0dev_best_loss = float('inf')  # 初始化设置最大损失值为正无穷大last_improve = 0flag = Falseepochs = 2   # 设置训练次数for epoch in range(epochs):print('Epoch [{}/{}]'.format(epoch+1,epochs))  # 第一轮[1/2],第二轮[2/2]for i,(trains,labels) in enumerate(train_iter):  # 遍历训练集的索引和数据，数据存放的是 每128条评价的包 的字在词表中的索引信息、标签信息、评价长度# 经过DatasetIterater中的 to_tensor 返回的数据格式为:(x，seq_len)，y，即独热编码、长度、标签outputs = model(trains)   # 将数据放入模型进行训练，得到预测输出值，这里的forward没有展示，即传入模型进行前向传播，返回预测结果格式为128*4loss = F.cross_entropy(outputs,labels)   # 将输出值与标签放入交叉熵损失计算损失值，多分类计算损失值model.zero_grad()   # 对模型进行梯度清0，为下一轮训练做准备loss.backward()   # 根据损失计算梯度optimizer.step()   # 根据梯度更新模型参数if total_batch % 100 == 0:   # 每100轮 输出 在训练集和验证集上的效果，每一百个批次的包打印出来一次predic = torch.max(outputs.data,1)[1].cpu()  # outputs.data为 每128条评价的包 的预测大小状态128*4，因为都在GPU中，所以为Tensor类型，torch.max返回第二个维度的最大值及索引，1表示第二个维度,[1]表示取索引的值当做预测结果，然后将预测结果传入cputrain_acc = metrics.accuracy_score(labels.data.cpu(),predic)   # 将真实值的标签结果与预测结果输入函数计算准确率dev_acc,dev_loss = evaluate(class_list,model,dev_iter)    # 将种类名、模型、验证集数据传入evaluate函数，获得验证结果，返回准确率和损失值if dev_loss < dev_best_loss:   # 判断当前损失值是否小于历史损失值dev_best_loss = dev_loss  # 如果损失值比前面的小，那么更新之前的损失值，然后保存这个模型的权重信息torch.save(model.state_dict(),'TextRNN.ckpt')  # 保存最优模型last_improve = total_batch   # 保存最优模型的batch值，整数赋值给last_improve# 打印模型的轮数右对齐字符宽为6、训练集的损失值长度为5保留2个小数、训练集的准确率、验证集的损失值和准确率，其中的0,1,2,3,4表示序号第一个参数第二个参数...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()  # 因为上述使用了evaluate将模型设置了测试模式，所以此处再次设置为训练模式total_batch += 1   # 每运行一次训练了一个包的文件，对数值加1if total_batch - last_improve > 10000:print("No optimization for a long time,auto-stopping...")flag = Trueif flag:breaktest(model,test_iter,class_list)  # 调用test函数进行测试

4、最终文件格式

5、定义主函数

        创建一个文件命名为main.py，将下列代码写入文件

import torch
import numpy as np
import load_dataset,TextRNN
from train_eval_test import traindevice = "cuda" if torch.cuda.is_available() else "mps" if torch.backends.mps.is_available() else "cpu"   # 判断当前使用的是GPU还是CPU或者是mac
np.random.seed(1)   # 设置numpy的随机种子为1，使用numpy生成的随机数序列都是相同的
torch.manual_seed(1)  # 设置PyTorch全局随机种子为1
torch.cuda.manual_seed_all(1)   # 为cuda设备设置随机种子，确保使用多个GPU时，PyTorch生成的随机数是可重复的
torch.backends.cudnn.deterministic = True  # 启用了CuDNN的确定性模式，优化PyTorch等框架在GPU上的性能，CuDNN的某些操作（如卷积和池化）可能是非确定性的，即它们可能会在不同的运行之间产生略微不同的结果，即使输入和随机种子都是相同的，设置为True表示可以牺牲一些性能换取结果的一致性#
# 调用之前写的文件中定义的函数，并输入参数：文件地址,返回词库、训练集、验证集、测试集 元组形式存放
vocab,train_data,dev_data,test_data = load_dataset.load_dataset('simplifyweibo_4_moods.csv')"""获取数据集"""
train_iter = load_dataset.DatasetIterater(train_data,128,device)
dev_iter = load_dataset.DatasetIterater(dev_data,128,device)
test_iter = load_dataset.DatasetIterater(test_data,128,device)"""调用腾讯训练好的的词嵌入模型"""
# 读取预先训练好的模型，将其转化为张量格式传入GPU进行运算，模型格式为4760*200
embedding_pretrained = torch.tensor(np.load('embedding_Tencent.npz')['embeddings'].astype('float32'))
# embedding_pretrained =None   # 不使用外部训练的词向量，则使用随机初始化的词嵌入# 使用if语句，表示如果导入了预先训练好的模型，那么使用模型的第二个维度，即嵌入向量的维度，否则定义词向量维度为200
embed = embedding_pretrained.size(1) if embedding_pretrained is not None else 200
class_list = ['喜悦','愤怒','厌恶','低落']   # 定义情感分析的标签类别
num_classes = len(class_list)   # 返回类别种类个数# 传入参数定义模型
model = TextRNN.Model(embedding_pretrained,len(vocab),embed,num_classes).to(device)
#
train(model,train_iter,dev_iter,test_iter,class_list)

        运行结果：