基于LSTM的情感分析

导包

根据代码需要进行相关的导包

import os import torch import random import numpy as np from sklearn.model_selection import train_test_split from torch.utils.data import TensorDataset, DataLoader import torch.nn as nn from tqdm import tqdm 相关路径文件的准备 # 原始数据路径 data_file = os.path.relpath(os.path.join(os.path.dirname(__file__),'data/hotel_discuss2.csv')) #判断路径是否存在 if not os.path.exists(data_file): print(f"{data_file}数据文件不存在！") exit() # 字典路径 dict_file = os.path.relpath(os.path.join(os.path.dirname(__file__),'data/dict.txt')) # 编码后存放的路径 encoding_file = os.path.relpath(os.path.join(os.path.dirname(__file__),'data/encoding.txt')) 设备的切换 # 设备切换 device = "cuda" if torch.cuda.is_available() else "cpu" 数据准备（数据清洗） # 需要过滤的字符（例如标点符号） puncts = '。,' # 词向量编码 with open(data_file, 'r', encoding='utf-8-sig') as f: data = f.readlines() for line in data: # line = line.strip() 是一个在编程中常见的操作，主要用于去除字符串开头和结尾的空白字符。 # 这里的空白字符包括空格、制表符（tab）、换行符等。 line = line.strip() # line字符串可进行遍历循环 for word in line: # 先过滤掉不需要的字符 if word not in puncts: # 判断遍历的词是否在字典中，不在，则加入字典中 # 达到一个词去重的目的 if word not in mydict: # 从1开始编码 mydict[word] = code code += 1 # 未知字符的编码 mydict['<unk>'] = code # 将数据字典存到本地 with open(dict_file, 'w', encoding='utf-8-sig') as f: f.write(str(mydict)) 加载词向量文件 #加载字典内容 def load_dict(dict_file): with open(dict_file, 'r', encoding='utf-8-sig') as file: """ file.read()：这个方法会读取文件中的所有内容，并将其作为一个字符串返回。如果文件很大，这样做可能会消耗大量内存。 eval()：这是一个内置函数，它会解析传入的字符串表达式，并在当前上下文中执行它。 这意味着如果文件中包含的是有效的Python表达式或语句，eval() 将执行这些代码并返回执行的结果。 重点 例如，如果文件中包含字典、列表或任何其他Python表达式，eval() 可以将它们转换成相应的对象。 """ return eval(file.read()) new_dict = load_dict(dict_file) 定义词向量字典的长度 # 定义获取字典大小的函数 def get_dict_size(new_dict): return len(new_dict) 打开原始数据，构建训练数据集 # 打开原始数据 with open(data_file, 'r', encoding='utf-8-sig') as f1: # 写入编码文件 with open(encoding_file, 'w', encoding='utf-8-sig') as f2: #将原始数据的每行数据进行编码 for line in f1.readlines(): # 去字符串两头的空格 line = line.strip() # 获取标签和内容 label = line[0] # 从2取是因为索引1的位置是， content = line[2:] # 对内容进行遍历 for ch in content: # 剔除在puncts中的字符 if ch not in puncts: # 写入每个词对应的编码 f2.write(str(new_dict[ch])) # 每个词之间用逗号隔开 f2.write(',') # 写入标签：用空格和内容进行分隔，并对下一行1数据进行换行 f2.write('\t'+str(label)+'\n') 数据长度的填充 #存放训练数据 values = [] #存放真实数据 labels = [] #预设每次输入的长度是256，不够就补0，超过的部分则不要 max_lens = 256 # 读取编码数据 with open(encoding_file, 'r', encoding='utf-8-sig') as f: lines = f.readlines() # random.shuffle(lines) 是 Python 中用于将列表 lines 的元素随机打乱顺序的方法。 random.shuffle(lines) for line in lines: # 根据空格来划分训练数据和标签 data, label = line.split('\t') # 根据逗号进行分割，并转换为int类型，存入列表中 val = [int(i) for i in data[0:-1].split(',')] # 如果i 大于 max_lens，则将val列表中的元素填充为0，直到 val 的长度等于 max_lens。 for i in range(max_lens - len(val)): val.append(0) val = val[:max_lens] values.append(val) labels.append(int(label)) # 将列表转换为numpy数组，并转换为int64类型 values = np.array(values, dtype='int64') labels = np.array(labels, dtype='int64') 数据集的划分 # 数据集的划分 train_x, test_x, train_y, test_y = train_test_split(values, labels, test_size=0.2, random_state=1) 转为Tensor数据集

本项目使用的pytorch的模型框架进行训练，所以数据集最终要转为tensor类型的数据集

# 将数据转换为tensor train_data = TensorDataset(torch.from_numpy(train_x), torch.from_numpy(train_y)) test_data = TensorDataset(torch.from_numpy(test_x), torch.from_numpy(test_y)) 数据集批次的划分 train_data = DataLoader(dataset=train_data, batch_size=64, shuffle=True) test_data = DataLoader(dataset=test_data, batch_size=64, shuffle=True) 模型搭建 # 模型搭建 class LSTM(nn.Module): def __init__(self, input_size, output_size, embedding_dim=512,hidden_size=512): # 继承父类 super(LSTM, self).__init__() # 定义隐藏层大小 self.hidden_size = hidden_size # 词嵌入 # input_size：字典的大小 # embedding_dim：词向量的维度 # device：设备 self.embedding = nn.Embedding(input_size, embedding_dim, device=device) # 定义LSTM层 # embedding_dim：词向量的维度 # hidden_size：隐藏层的维度 # num_layers：LSTM层的数量 self.lstm = nn.LSTM(embedding_dim, hidden_size, num_layers=4) # 定义全连接层，输出层 # hidden_size：隐藏层的维度 # output_size：输出维度 self.fc = nn.Linear(hidden_size, output_size) def forward(self, x): # 词嵌入 embed = self.embedding(x) # 数据形状：(seq_len, batch, input_size)，因为batch_first=True，所以需要转置一下 permute_ = embed.permute(1, 0, 2) # h0 = torch.zeros(1, permute_.size(0), self.hidden_size).to(device) # c0 = torch.zeros(1, permute_.size(0), self.hidden_size).to(device) # LSTM层，输出：(seq_len, batch, hidden_size) # out: 输出，h_n：最后一个隐藏层的状态，c_n：最后一个cell的状态 out, _ = self.lstm(permute_) # 输出层 """ out[-1, :, :]：这个索引操作选择了out张量的最后一个时间步的所有批次和所有隐藏单元的数据。具体来说： -1 表示选择序列中的最后一个时间步的输出。 : 表示选择该时间步下所有批次的数据。 第二个 : 表示选择这些批次数据中的所有隐藏单元。 """ out = self.fc(out[-1, :, :]) return out 训练参数定义 # 学习率 lr = 1e-4 # 训练轮次 epochs = 20 模型、损坏函数、优化器的创建 model = LSTM(get_dict_size(new_dict),2) # model.train() model.to(device) optimizer = torch.optim.AdamW(model.parameters(), lr=0.001) loss_fc = nn.CrossEntropyLoss() loss_old = 100 模型训练与测试 # 初始化保存训练和测试结果的列表 train_result = [] test_result = [] # 初始化保存训练和测试损失、精度的列表 train_losses = [] train_accuracies = [] test_losses = [] test_accuracies = [] # 循环训练每个周期 for epoch in range(1, epochs + 1): # 创建一个进度条以显示训练数据的处理进度 pbar = tqdm(train_data) # 初始化每个周期的总损失和总准确度 loss_all = 0 acc_all = 0 # 遍历训练数据集的每一个批次 for step, (x, y) in enumerate(pbar): # 将标签数据移动到指定的设备（如GPU） y = y.to(device) # 将输入数据移动到指定的设备（如GPU） x = x.to(device) # 计算每个序列的实际长度（非零元素的数量） # lengths = torch.sum(x != 0, dim=-1).cpu().long() # 使用模型进行前向传播，得到输出 out = model(x) # 计算损失 loss = loss_fc(out, y) # 反向传播计算梯度 loss.backward() # 更新模型参数 optimizer.step() # 将梯度归零 optimizer.zero_grad() # 累加损失 loss_all += loss.item() # 计算当前批次的平均损失 loss_time = loss_all / (step + 1) # 计算当前批次的准确度 # 使用 torch.mean() 计算这些 1.0 和 0.0 值的平均数，也就是所有预测正确的样本占总样本的比例，这实际上就是模型的准确率。 acc = torch.mean((y == torch.argmax(out, dim=-1)).float()) # 累加准确度 acc_all += acc # 计算当前批次的平均准确度 acc_time = acc_all / (step + 1) # 构建进度条的描述信息 s = "train => epoch:{} - step:{} - loss:{:.4f} - loss_time:{:.4f} - acc:{:.4f} - acc_time:{:.4f}".format(epoch, step, loss, loss_time, acc, acc_time) # 设置进度条的描述 pbar.set_description(s) # 将描述信息添加到训练结果列表中 train_result.append(s + "\n") # 记录当前epoch的训练损失和精度 train_losses.append(loss_time) train_accuracies.append(acc_time.item()) # 禁用梯度计算，因为我们只需要前向传播进行预测 with torch.no_grad(): # 创建一个进度条以显示测试数据的处理进度 pbar = tqdm(test_data) # 初始化每个周期的总损失和总准确度 loss_all = 0 acc_all = 0 # 遍历测试数据集的每一个批次 for step, (x, y) in enumerate(pbar): # 将标签数据移动到指定的设备（如GPU） y = y.to(device) # 将输入数据移动到指定的设备（如GPU） x = x.to(device) # 计算每个序列的实际长度（非零元素的数量） lengths = torch.sum(x != 0, dim=-1).cpu().long() # 使用模型进行前向传播，得到输出 out = model(x) # 计算损失 loss = loss_fc(out, y) # 累加损失 loss_all += loss.item() # 计算当前批次的平均损失 test_loss_time = loss_all / (step + 1) # 计算当前批次的准确度 acc = torch.mean((y == torch.argmax(out, dim=-1)).float()) # 累加准确度 acc_all += acc # 计算当前批次的平均准确度 acc_time = acc_all / (step + 1) # 构建进度条的描述信息 s = "test => epoch:{} - step:{} - loss:{:.4f} - loss_time:{:.4f} -acc:{:.4f} - acc_time:{:.4f}".format(epoch, step, loss, test_loss_time, acc, acc_time) # 设置进度条的描述 pbar.set_description(s) # 将描述信息添加到测试结果列表中 test_result.append(s + "\n") # 记录当前epoch的测试损失和精度 test_losses.append(test_loss_time) test_accuracies.append(acc_time.item()) # 将训练结果写入文件 with open(f"train_result.txt", "w") as f: f.writelines(train_result) # 将测试结果写入文件 with open(f"test_result.txt", "w") as f: f.writelines(test_result) # 保存模型，如果当前测试损失小于之前记录的最小损失 if loss_old > test_loss_time: loss_old = test_loss_time torch.save(model.state_dict(), f"model.pkl") 应用 import torch from data_deal import * from model import LSTM # 加载字典 new_dict = load_dict(dict_file) # 设备切换 device = "cuda" if torch.cuda.is_available() else "cpu" # 模型加载 model = LSTM(get_dict_size(new_dict), 2) model.load_state_dict(torch.load("model.pkl")) model.to(device) model.eval() # 设置为评估模式 # 用户输入 while True: text = input("请输入文本（输入'退出'以结束）：") if text == "退出": break # 词编码 text = [new_dict[ch] for ch in sen] text = torch.tensor(text).unsqueeze(0).to(device) prediction = model(text) prediction = torch.argmax(prediction, dim=1).item() print(f"预测结果: {'正面' if prediction == 1 else '负面'}")