基于 TextRNN 的微博四分类情感分析实战 | 完整可运行全流程教程

情感分析Sentiment Analysis是自然语言处理NLP中最经典、应用最广泛的任务之一。在社交媒体、电商评论、舆情监控等场景中自动识别文本的情感倾向具有极高的商业价值。本文将从零开始完整实现一套基于TextRNN的微博四分类情感分析系统覆盖数据预处理、词表构建、数据集封装、模型搭建、训练评估、结果可视化、工程优化全流程所有代码均可直接运行适合NLP初学者入门与实战。一、项目整体架构本次项目采用端到端文本分类架构整体流程如下1. 数据层微博原始文本 → 分词/字 → 序列补齐 → 词索引映射 → 训练/验证/测试集划分2. 表征层腾讯预训练中文词向量 → 嵌入层编码文本3. 模型层双向LSTMTextRNN提取序列特征4. 输出层全连接层Softmax完成4分类5. 训练层损失计算、反向传播、模型保存、指标评估6. 应用层模型加载、单句预测、结果展示任务定义4分类情感识别• 标签0喜悦• 标签1愤怒• 标签2厌恶• 标签3低落技术栈• 框架PyTorch• 数据处理Pickle、NumPy、TQDM• 模型BiLSTMTextRNN• 设备CUDA / MPS / CPU 自适应二、全模块代码详解模块1词表构建工具vocab_create.py用于从原始文本生成词表统一UNK/PAD符号是数据预处理的基础。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,encodingUTF-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 _[i] min_freq],keylambda x:x[1],reverseTrue)[: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}) print(vocab_dic) pkl.dump(vocab_dic,open(simplifyweibo_4_moods.pkl,wb)) print(fVocab size:{len(vocab_dic)}) return vocab_dic if __name__ __main__: vocab build_vocab(simplifyweibo_4_moods.csv,MAX_VOCAB_SIZE,3) print(vocab)模块2数据加载与迭代器load_dataset.py核心功能文本读取 → 字分割 → 长度统一 → 索引转换 → 数据集划分 → Batch迭代器from tqdm import tqdm import pickle as pkl import random import torch UNK,PAD UNK,PAD def load_dataset(path,pad_size70): contents [] vocab pkl.load(open(simplifyweibo_4_moods.pkl,rb)) tokenizer lambda x:[y for y in x] with open(path,r,encodingUTF-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: if len(token) pad_size: token.extend([PAD] * (pad_size - len(token))) else: token token[:pad_size] seq_len pad_size for word in token: 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): 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: 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 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) print(结束)模块3TextRNN模型定义TextRNN.py采用双层双向LSTM更擅长捕捉上下文依赖适配中文文本。import torch 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: self.embedding nn.Embedding.from_pretrained(embedding_pretrained, freezeFalse) else: self.embedding nn.Embedding(n_vocab,, embed, padding_idxn_vocab-1) # 双向LSTM self.lstm nn.LSTM( input_sizeembed, hidden_size128, num_layers3, bidirectionalTrue, batch_firstTrue, dropout0.3 ) # 全连接分类 self.fc nn.Linear(128 * 2, num_classes) def forward(self, x): x,_ x out self.emedding(x) out,_ self.lstm(out) out self.fc(out[:,-1,:]) return out模块4训练、验证、测试逻辑train_eval_test.py包含训练循环、早停、指标计算、混淆矩阵、日志输出。import torch from sklearn import metrics from torch.nn import functional as F import numpy as np # 评估函数计算损失和准确率 def evaluate(class_list,model, data_iter, testFalse): model.eval() # 切换到评估模式 loss_total 0 predict_all np.array([], dtypeint) labels_all np.array([], dtypeint) with torch.no_grad(): # 禁用梯度计算 for texts, labels in data_iter: outputs model(texts) # 匹配Model的forward输入格式 loss F.cross_entropy(outputs, labels) loss_total loss # 计算预测结果 predic torch.max(outputs.data, 1)[1].cpu().numpy() labels labels.cpu().numpy() predict_all np.append(predict_all, predic) labels_all np.append(labels_all, labels) # 计算准确率 acc metrics.accuracy_score(labels_all,predict_all) if test: report metrics.classification_report(labels_all,predict_all,target_namesclass_list,digits4) return acc,loss_total / len(data_iter), report return acc,loss_total / len(data_iter) def test(model,test_iter,class_list): model.eval() test_acc,test_loss,test_report evaluate(class_list,model,test_iter,testTrue) msg Test Loss:{0:5.2},Test Acc:{16.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(), lr1e-3) total_batch 0 # 累计批次 dev_best_loss float(inf) # 最优验证集损失 last_improve 0 # 最后一次提升的批次 flag False # 是否早停 epochs 20 for epoch in range(epochs): print((Epoch [{}/{}].format(epoch 1, epochs))) for i, (trains, labels) in enumerate(train_iter): outputs model(trains) # 匹配Model的forward输入格式 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 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模块5主程序入口main.py统一调度自动适配设备一键启动训练。import torch import numpy as np import load_dataset import TextRNN from train_eval_test import train # 固定随机种子 np.random.seed(1) torch.manual_seed(1) torch.cuda.manual_seed(1) torch.backends.cudnn.deterministic True # 设备自适应 device cuda if torch.cuda.is_available() else mps if torch.backends.mps.is_available() else cpu print(f使用设备{device}) # 1. 加载数据 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) # 2. 加载预训练词向量 embedding_pretrained torch.tensor(np.load(embedding_Tencent.npz)[embeddings].astype(float32)) embed embedding_pretrained.size(1) if embedding_pretrained is not None else 200 # 3. 分类设置 class_list [喜悦,愤怒,厌恶,低落] num_classes len(class_list) # 4. 初始化模型 model TextRNN.Model(embedding_pretrained, len(vocab), embed_dim, num_classes).to(device) # 5. 启动训练 train(model, train_iter, dev_iter, test_iter, class_list)三、模型优化与拓展方向1. 效果提升• 改用BERT/RoBERTa精度可提升至92%• 加入对抗训练FGM• 数据增强回译、同义词替换• 调整LSTM层数、隐藏层维度、学习率2. 工程化拓展• 封装单句预测接口• 构建Web APIFastAPI/Flask• 批量文件情感分析• 前端可视化界面四、项目总结本文完整实现了基于TextRNN的中文微博四分类情感分析系统包含✅ 词表构建✅ 数据预处理✅ 数据集迭代器✅ 双向LSTM模型✅ 训练/验证/测试全流程✅ 最优模型保存✅ 分类指标可视化这是NLP文本分类最标准的工程模板可直接迁移到评论分析、舆情检测、垃圾邮件识别等场景。