微博评论
思考：向模型中传递数据时，需要提前处理好数据
1、目标：将评论内容转换为词向量。
2、每个词/字转换为词向量长度(维度)200
3、每一次传入的词/字的个数是否就是评论的长度?
应该是固定长度，每次传入数据与图像相似。例如选择长度为70。则传入的数据为70*200
4、一条评论如果超过70个词/字怎么处理？
直接删除后面的内容
5、一条评论如果没有70个词/字怎么处理？
缺少的内容，统一使用一个数字（非词/字的数字）替代。
6、如果语料库中的词/字太多是否可以压缩？
可以，某些词/字出现的频率比较低，可能训练不出特征。因此可以选择频率比较高的词来训练。例如选择4760个。
7、被压缩的词/字如何处理？
可以统一使用一个数字（非词/字的数字）替代。
1.处理文本数据集，将其分为训练集、验证集和测试集
处理文本数据集，将其分为训练集、验证集和测试集
初始化一个空列表，用于存储处理后的数据。
from tqdm import tqdm
import pickle as pkl
import random
import torch
unk,pad='<UNK>','<PAD>'
def load_dataset(path,pad_size=70):
contents=[]
#加载文件：4760个最多的字
vocab=pkl.load(open('simplifyweibo_4_moods.pkl','rb'))
#定义一个函数，加载预训练的词汇表
tokenizer=lambda x:[y for y in x]读取和处理数据遍历每行提取标签，评论内容，每行的字符数目
# 读取文件里面的评论
with open(path,'r',encoding='UTF-8') 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:
#小于70增加pad
if len(token)<pad_size:
token.extend([pad]*(pad_size-len(token)))
# 大于70，去除70后的
else:
token=token[:pad_size]
seq_len=pad_size遍历每个词，使用词汇表将其转换为索引，如果词不在词汇表中，则使用 <UNK> 的索引,将字符串，标签，数目组成一个元组放入列表
for word in token:
# 将在4760里面的字符加入列表，不在里面的定义为unk
words_line.append(vocab.get(word,vocab.get(unk)))
# 将字符串，标签，数目组成一个元组放入列表
contents.append((words_line,int(label),seq_len))随机打乱数据集划分训练集、验证集和测试集
# 随机打乱
random.shuffle(contents)
train_data=contents[:int(len(contents)*0.8)]
dev_data=contents[int(len(contents)*0.8):int(len(contents)*0.9)]
test_data = contents[int(len(contents) * 0.9):]
return vocab,train_data,dev_data,test_data对于生成的训练集、验证集和测试集进行数据转换成LongTensor类型
代码
定义了一个名为 DatasetIterater 的类，它是一个迭代器，用于在深度学习任务中批量处理数据。
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内容，标签，数目转化为LongTensor类型
def _to_tensor(self,datas):
#内容，标签，数目转化为LongTensor类型
x=torch.LongTensor([_[0] for _ in datas]).to(self.device)
y=torch.LongTensor([_[1] for _ in datas]).to(self.device)
seq_len=torch.LongTensor([_[2] for _ in datas]).to(self.device)
return (x,seq_len),y逐个处理数据集中的批次
如果当前索引大于完整批次的数量，处理最后一个不完整的批次。
def __next__(self):
#如果当前索引大于完整批次的数量，处理最后一个不完整的批次。
if self.residue and self.index ==self.n_batches:
#将最后一个包里的数据变换为LongTensor
batches=self.batches[self.index*self.batch_size:len(self.batches)]
self.index+=1
batches=self._to_tensor(batches)
return batches如果当前索引大于完整批次的数量，重置索引并抛出
# 如果当前索引大于完整批次的数量，重置索引并抛出
elif self.index>self.n_batches:
self.index=0
raise StopIteration根据当前索引和批次大小，从self.batches中切片出一个批次的数据，转换为张量
# 否则，根据当前索引和批次大小，从self.batches中切片出一个批次的数据，转换为张量
else:
batches=self.batches[self.index*self.batch_size:(self.index+1)*self.batch_size]
self.index+=1
batches=self._to_tensor(batches)
return batches_iter__ 方法
返回迭代器本身。
def __iter__(self):
return self__len__ 方法
返回批次的总数。
def __len__(self):
if self.residue:
return self.n_batches+1
else:
return self.n_batches2.定义了一个名为 Model 的类，它是一个用于处理序列数据（如文本）的神经网络模型，继承自 PyTorch 的 nn.Module。
定义了模型的嵌入层、LSTM层和全连接层
import torch.nn as nn
import torch.nn.functional as F
import sys
import numpy as np
class Model(nn.Module):
# 预训练的词嵌入矩阵，词汇表数量，词向量的维度，输出类别的数量
def __init__(self,embedding_pertrainde,n_vocab,embed,num_classes):
super(Model,self).__init__()
# 用于将词索引转换为词向量。
if embedding_pertrainde is not None:
self.embedding=nn.Embedding.from_pretrained(embedding_pertrainde,padding_idx=n_vocab-1,freeze=False)
else:
self.embedding=nn.Embedding(n_vocab,embed,padding_idx=n_vocab-1)
# LSTM层，用于处理序列数据，捕捉时间序列上的依赖关系。
self.lstm=nn.LSTM(embed,128,3,bidirectional=True,batch_first=True,dropout=0.3)
# 全连接层，用于将LSTM的输出映射到类别空间。
self.fc=nn.Linear(128*2,num_classes)前向传播使用嵌入层将词索引转换为词向量，LSTM层处理序列数据并捕捉时间序列上的依赖关系，最后使用全连接层将输出映射到类别空间。
def forward(self,x):
x,_=x
# 将输入的词索引通过嵌入层转换为词向量
out=self.embedding(x)
# 将嵌入后的词向量通过LSTM层处理
out,_=self.lstm(out)
# 从LSTM的输出中，只取最后一个输出
out=self.fc(out[:,-1,:])
return out3.evaluate、test 和 train，它们分别用于评估模型性能、测试模型以及训练模型
定义了一个名为 evaluate 的函数，它用于评估一个训练好的模型在特定数据集上的性能
模型前向传播和损失计算，处理预测结果和真实标签
import torch.optim
from sklearn import metrics
import time
import torch
import torch.nn as nn
import torch.nn.functional as F
import sys
import numpy as np
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():
for texts,labels in data_iter:
outputs=model(texts)
# 计算交叉熵损失
loss=F.cross_entropy(outputs,labels)
# 累加损失
loss_total+=loss
labels=labels.data.cpu().numpy()
# 获取预测结果
predict=torch.max(outputs.data,1)[1].cpu().numpy()
# 追加真实标签，追加预测结果
labels_all=np.append(labels_all,labels)
predict_all=np.append(predict_all,predict)
# 计算准确率
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)定义了一个名为 test 的函数，它用于在测试集上评估模型的性能。
def test(model,test_iter,class_list):
# 加载模型
model.load_state_dict(torch.load('TextRNN.ckpt'))
# 评估模式
model.eval()
# 调用评估函数
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)定义了一个名为 train 的函数，它用于训练一个深度学习模型，并在验证集上监测性能以避免过拟合
将模型设置为训练模式，并初始化Adam优化器，初始化训练所需的变量，包括总批次、验证集上的最佳损失、上次改进的批次、早停标志和训练周期。
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("{}/{}".format(epoch+1,epochs))
for i,(trains,labels) in enumerate(train_iter):
# 前向传播
outputs=model(trains)
# 交叉熵损失函数
loss=F.cross_entropy(outputs,labels)
# 梯度清零
model.zero_grad()
# 反向传播
loss.backward()
# 更新模型参数
optimizer.step()
每100个批次评估一次模型在验证集上的性能，如果验证损失降低，则更新最佳验证损失。如果长时间没有改进，则触发早停机制。训练完成后，在测试集上测试模型的性能。
# 每100个批次评估一次
if total_batch%100==0:
# 预测结果
predict=torch.max(outputs.data,1)[1].cpu()
train_acc=metrics.accuracy_score(labels.data.cpu(),predict)
# 调用evaluate函数评估验证集
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
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')
flag=True
if flag:
break
test(model,test_iter,class_list)完整代码
load_dataset.py文件
from tqdm import tqdm
import pickle as pkl
import random
import torch
unk,pad='<UNK>','<PAD>'
def load_dataset(path,pad_size=70):
contents=[]
#加载文件：4760个最多的字
vocab=pkl.load(open('simplifyweibo_4_moods.pkl','rb'))
#定义一个函数，加载预训练的词汇表
tokenizer=lambda x:[y for y in x]
# 读取文件里面的评论
with open(path,'r',encoding='UTF-8') 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:
#小于70增加pad
if len(token)<pad_size:
token.extend([pad]*(pad_size-len(token)))
# 大于70，去除70后的
else:
token=token[:pad_size]
seq_len=pad_size
for word in token:
# 将在4760里面的字符加入列表，不在里面的定义为unk
words_line.append(vocab.get(word,vocab.get(unk)))
# 将字符串，标签，数目组成一个元组放入列表
contents.append((words_line,int(label),seq_len))
# 随机打乱
random.shuffle(contents)
train_data=contents[:int(len(contents)*0.8)]
dev_data=contents[int(len(contents)*0.8):int(len(contents)*0.9)]
test_data = contents[int(len(contents) * 0.9):]
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):
#内容，标签，数目转化为LongTensor类型
x=torch.LongTensor([_[0] for _ in datas]).to(self.device)
y=torch.LongTensor([_[1] for _ in datas]).to(self.device)
seq_len=torch.LongTensor([_[2] for _ in datas]).to(self.device)
return (x,seq_len),y
def __next__(self):
#如果当前索引大于完整批次的数量，处理最后一个不完整的批次。
if self.residue and self.index ==self.n_batches:
#将最后一个包里的数据变换为LongTensor
batches=self.batches[self.index*self.batch_size:len(self.batches)]
self.index+=1
batches=self._to_tensor(batches)
return batches
# 如果当前索引大于完整批次的数量，重置索引并抛出
elif self.index>self.n_batches:
self.index=0
raise StopIteration
# 否则，根据当前索引和批次大小，从self.batches中切片出一个批次的数据，转换为张量
else:
batches=self.batches[self.index*self.batch_size:(self.index+1)*self.batch_size]
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)TextRNN文件
import torch
import torch.nn as nn
import torch.nn.functional as F
import sys
import numpy as np
class Model(nn.Module):
# 预训练的词嵌入矩阵，词汇表数量，词向量的维度，输出类别的数量
def __init__(self,embedding_pertrainde,n_vocab,embed,num_classes):
super(Model,self).__init__()
# 用于将词索引转换为词向量。
if embedding_pertrainde is not None:
self.embedding=nn.Embedding.from_pretrained(embedding_pertrainde,padding_idx=n_vocab-1,freeze=False)
else:
self.embedding=nn.Embedding(n_vocab,embed,padding_idx=n_vocab-1)
# LSTM层，用于处理序列数据，捕捉时间序列上的依赖关系。
self.lstm=nn.LSTM(embed,128,3,bidirectional=True,batch_first=True,dropout=0.3)
# 全连接层，用于将LSTM的输出映射到类别空间。
self.fc=nn.Linear(128*2,num_classes)
def forward(self,x):
x,_=x
# 将输入的词索引通过嵌入层转换为词向量
out=self.embedding(x)
# 将嵌入后的词向量通过LSTM层处理
out,_=self.lstm(out)
# 从LSTM的输出中，只取最后一个输出
out=self.fc(out[:,-1,:])
return outtrain_eval_test文件
import torch.optim
from sklearn import metrics
import time
import torch
import torch.nn as nn
import torch.nn.functional as F
import sys
import numpy as np
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():
for texts,labels in data_iter:
outputs=model(texts)
# 计算交叉熵损失
loss=F.cross_entropy(outputs,labels)
# 累加损失
loss_total+=loss
labels=labels.data.cpu().numpy()
# 获取预测结果
predict=torch.max(outputs.data,1)[1].cpu().numpy()
# 追加真实标签，追加预测结果
labels_all=np.append(labels_all,labels)
predict_all=np.append(predict_all,predict)
# 计算准确率
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.ckpt'))
# 评估模式
model.eval()
# 调用评估函数
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)
total_batch=0
# 初始化验证集上的最佳损失
dev_best_loss=float('inf')
last_improve=0
flag=False
epochs=2
# 批次处理循环
for epoch in range(epochs):
# 打印当前周期
print("{}/{}".format(epoch+1,epochs))
for i,(trains,labels) in enumerate(train_iter):
# 前向传播
outputs=model(trains)
# 交叉熵损失函数
loss=F.cross_entropy(outputs,labels)
# 梯度清零
model.zero_grad()
# 反向传播
loss.backward()
# 更新模型参数
optimizer.step()
# 每100个批次评估一次
if total_batch%100==0:
# 预测结果
predict=torch.max(outputs.data,1)[1].cpu()
train_acc=metrics.accuracy_score(labels.data.cpu(),predict)
# 调用evaluate函数评估验证集
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
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')
flag=True
if flag:
break
test(model,test_iter,class_list)
主文件调用其他文件
import torch
import numpy as np
import pickle as pkl
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'
np.random.seed(1)
torch.manual_seed(1)
torch.cuda.manual_seed_all(1)
torch.backends.cudnn.deterministic=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)
# 加载预训练的词嵌入
embedding_pretrainde=torch.tensor(np.load('embedding_Tencent.npz')['embeddings'].astype('float32'))
# 设置为嵌入的维度
embed=embedding_pretrainde.size(1) if embedding_pretrainde is not None else 200
class_list=["喜悦","愤怒","厌恶","低落"]
num_classes=len(class_list)
# 训练模型
model=TextRNN.Model(embedding_pretrainde,len(vocab),embed,num_classes).to(device)
train(model,train_iter,dev_iter,test_iter,class_list)运行结果
