目录
一、了解项目
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): # 输入参数为：样本数据、单个批次中的样本条数、设备GPU
self.batch_size = batch_size
self.batches = batches
self.n_batches = len(batches) // batch_size # 计算批次数
self.residue = False # 记录划分后的数据是否存在剩余的数据，初始化为False
if len(batches) % self.n_batches != 0: # 表示有余数
self.residue = True
self.index = 0 # 追踪当前批次的索引
self.device = device
def _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) # 调用上述的私有函数，将样本数据传入，将其转换数据类型tensor
return batches # 返回Tensor数据类型的样本数据
elif self.index > self.n_batches: # 如果迭代使批次索引超过批次总数，终止循环
self.index = 0 # 重置批次索引为1
raise StopIteration # 为了防止迭代永远进行，我们可以使用StopIteration(停止迭代)语句
else: # 当没有读取到最后一个batch批次时:
baches = self.batches[self.index*self.batch_size:(self.index+1)*self.batch_size] # 提取当前批次的数据样本
self.index += 1 # 批次索引+1
baches = self._to_tensor(baches) # 将样本数据转化为张量格式
return baches # 返回张量类型的样本数据
def __iter__(self): # 这个方法使得类的实例能够成为迭代器，返回实例本身。
return self
def __len__(self): # 返回迭代器的长度，即完整批次的数量
if self.residue:
return self.n_batches + 1 # 如果存在剩余数据，长度会增加1。
else:
return self.n_batches
2、定义模型，前向传播函数
将下列代码写入新创建的文件TextRNN.py
import torch.nn as nn
class 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*200
self.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.fc
return out3、定义训练、测试函数
创建一个文件train_eval_test.py，将下列代码写入其中
import torch
import torch.nn.functional as F
import numpy as np
from sklearn import metrics
import time
def evaluate(class_list,model,data_iter,test=False): # 传入参数：种类名称列表、训练好的模型、验证集数据，test表示是否进行测试模式
model.eval() # 模型开始测试
loss_total = 0 # 初始化总损失值为0
predict_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.001
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)) # 第一轮[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*4
loss = 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]表示取索引的值当做预测结果，然后将预测结果传入cpu
train_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 # 每运行一次训练了一个包的文件，对数值加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) # 调用test函数进行测试4、最终文件格式
5、定义主函数
创建一个文件命名为main.py，将下列代码写入文件
import torch
import numpy as np
import load_dataset,TextRNN
from train_eval_test import train
device = "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)运行结果：
