PyTorch搭建LSTM神经网络实现文本情感分析实战（附源码和数据集）

青旅半醒 2023-09-30 16:48 18阅读 0赞

> **需要源码和数据集请点赞关注收藏后评论区留言~~~**

## 一、文本情感分析简介 ##

文本情感分析是指利用自然语言处理和文本挖掘技术，对带有情感色彩的主观性文本进行分析，处理和抽取的过程。

接下来主要实现情感分类，情感分类又称为情感倾向性分析，是指对给定的文本，识别其中主观性文本的倾向是肯定的还是否定的，或者说是正面的还是负面的，这是情感分析领域研究最多的内容。通常，网络中存在大量的主观性文本和客观性文本，客观性文本是对事务的客观性描述，不带有感情色彩和情感倾向。情感分类的对象是带有情感倾向的主观性文本，因此情感分类首先要进行文本的主客观雷芬，以情感词识别为主，利用不同的文本特征表示方法和分类器进行识别研究，对网络文本事先主客观分类，能够提高情感分类的速度和准确度

## 二、数据集简介 ##

本篇博客使用的数据集是IMDB的数据，IMDB数据集包含来自互联网的50000条严重两级分化的评论，该数据被分为用于训练的25000条评论和用于测试的25000条评论

里面给出了数据对应网址

![f17d61f6c91d4c3aacb6220aa49addc8.png][]

## 三、数据预处理 ##

因为这个数据集非常小，所以如果用这个数据集做word embedding有可能过拟合，而且模型没有通用性，所以传入一个已经学好的word embedding，用的是glove的6B 100维的预训练数据

如下图

![1dc9cc1183c34665bbb5ad7e2ddf43b3.png][]

## 四、算法模型 ##

### 1：循环神经网络（RNN） ###

它是一类专门处理时序数据样本的神经网络，它的每一层不仅输出给下一层，而同时还输出一个隐状态，供当前层在处理下一个样本时使用

网络结构图如下

![52289d07b1044a51bd590212713f8c50.png][]

### 2：长短期记忆神经网络（LSTM） ###

然而 RNN在处理长期依赖时会遇到巨大的困难，因此计算距离较远的节点之间的联系会设计雅可比矩阵的多次相乘，这会带来梯度消失或者梯度爆炸。LSTM可以有效的解决这个问题 它的主要思想是

> 门控单元以及线性连接的引入
> 
> 门控单元：有选择性的保存和输出历史信息
> 
> 线性连接：LSTM可以更好的捕捉时序数据中间间隔较大的依赖关系

工作图如下

![c74b146b59ec423e9de8b8d4723902b0.png][]

## 五、模型训练 ##

下面使用基于PyTorch的LSTM模型 效果如下

建议使用GPU或者cuda 单纯用cpu训练时间比较长~~~

![df4cca2d9105428491a538a0f63a3c39.png][]

训练结果图如下

测试集上的损失与精确度的变化

由下图可见 当训练到4-5次左右时模型已经逐渐收敛 不必训练过多次

![887067933b5145b589ab2a63c65529cf.png][]

训练集上的损失与精确度变化

![659c305bb66c4e0287434ed07d1981c1.png][]

## 六、代码 ##

部分源码如下

# coding: utf-8
    
    # In[1]:
    
    
    import torch.autograd as autograd
    import torchtext.vocab as torchvocab
    from torch.autograd import Variable
    import tqdm
    import os
    import time
    import re
    import pandas as pd
    import string
    import gensim
    import time
    import random
    import snowballstemmer
    import collections
    from collections import Counter
    from nltk.corpus import stopwords
    from itertools import chain
    from sklearn.metrics import accuracy_score
    
    from gensim.test.utils import datapath, get_tmpfile
    from gensim.models import KeyedVectors
    
    
    # In[2]:
    
    
    def clean_text(text):
        ## Remove puncuation
        text = text.translate(string.punctuation)
    
        ## Convert words to lower case and split them
        text = text.lower().split()
    
        # Remove stop words
        stops = set(stopwords.words("english"))
        text = [w for w in text if not w in stops and len(w) >= 3]
    
        text = " ".join(text)
        ## Clean the text
        text = re.sub(r"[^A-Za-z0-9^,!.\/'+-=]", " ", text)
        text = re.sub(r"what's", "what is ", text)
        text = re.sub(r"\'s", " ", text)
        text = re.sub(r"\'ve", " have ", text)
        text = re.sub(r"n't", " not ", text)
        text = re.sub(r"i'm", "i am ", text)
        text = re.sub(r"\'re", " are ", text)
        text = re.sub(r"\'d", " would ", text)
        text = re.sub(r"\'ll", " will ", text)
        text = re.sub(r",", " ", text)
        text = re.sub(r"\.", " ", text)
        text = re.sub(r"!", " ! ", text)
        text = re.sub(r"\/", " ", text)
        text = re.sub(r"\^", " ^ ", text)
        text = re.sub(r"\+", " + ", text)
        text = re.sub(r"\-", " - ", text)
        text = re.sub(r"\=", " = ", text)
        text = re.sub(r"'", " ", text)
        text = re.sub(r"(\d+)(k)", r"\g<1>000", text)
        text = re.sub(r":", " : ", text)
        tan ", text)
        text = re.sub(r"\0s", "0", text)
        text = re.sub(r" 9 11 ", "911", text)
        text = re.sub(r"e - mail", "email", text)
        text = re.sub(r"j k", "jk", text)
        text = re.sub(r"\s{2,}", " ", text)
        ## Stemming
        text = text.split()
        stemmer = snowballstemmer.stemmer('english')
        stemmed_words = [stemmer.stemWord(word) for word in text]
        text = " ".join(stemmed_words)
        print(text)
        return text
    
    
    # In[3]:
    
    
    def readIMDB(path, seg='train'):
        pos_or_neg = ['pos', 'neg']
        data = []
        for label in pos_or_neg:
            files = os.listdir(os.path.join(path, seg, label))
            for file in files:
                with open(os.path.join(path, seg, label, file), 'r', encoding='utf8') as rf:
                    review = rf.read().replace('\n', '')
                    if label == 'pos':
                        data.append([review, 1])
                    elif label == 'neg':
                        data.append([review, 0])
        return data
    
    
    # In[3]:
    
    
    root = r'C:\Users\Admin\Desktop\aclImdb\aclImdb'
    train_data = readIMDB(root)
    test_data = readIMDB(root, 'test')
    
    
    # In[4]:
    
    
    def tokenizer(text):
        return [tok.lower() for tok in text.split(' ')]
    
    
    train_tokenized = []
    test_tokenized = []
    for review, score in train_data:
        train_tokenized.append(tokenizer(review))
    for review, score in test_data:
        test_tokenized.append(tokenizer(review))
    
    # In[5]:
    
    
    vocab = set(chain(*train_tokenized))
    vocab_size = len(vocab)
    
    # In[6]:
    
    
    # 输入文件
    glove_file = datapath(r'C:\Users\Admin\Desktop\glove.6B.100d.txt')
    # 输出文件
    tmp_file = get_tmpfile(r'C:\Users\Admin\Desktop\wv.6B.100d.txt')
    
    # call glove2word2vec script
    # default way (through CLI): python -m gensim.scripts.glove2word2vec --input <glove_file> --output <w2v_file>
    
    # 开始转换
    from gensim.scripts.glove2word2vec import glove2word2vec
    
    glove2word2vec(glove_file, tmp_file)
    
    # 加载转化后的文件
    wvmodel = KeyedVectors.load_word2vec_format(tmp_file)
    
    # In[7]:
    
    
    word_to_idx = {word: i + 1 for i, word in enumerate(vocab)}
    word_to_idx['<unk>'] = 0
    idx_to_word = {i + 1: word for i, word in enumerate(vocab)}
    idx_to_word[0] = '<unk>'
    
    
    # In[8]:
    
    
    def encode_samples(tokenized_samples, vocab):
        features = []
        for sample in tokenized_samples:
            feature = []
            for token in sample:
                if token in word_to_idx:
                    feature.append(word_to_idx[token])
                else:
                    feature.append(0)
            features.append(feature)
        return features
    
    
    # In[9]:
    
    
    def pad_samples(features, maxlen=500, PAD=0):
        padded_features = []
        for feature in features:
            if len(feature) >= maxlen:
                padded_feature = feature[:maxlen]
            else:
                padded_feature = feature
                while (len(padded_feature) < maxlen):
                    padded_feature.append(PAD)
            padded_features.append(padded_feature)
        return padded_features
    
    
    # In[10]:
    
    
    train_features = torch.tensor(pad_samples(encode_samples(train_tokenized, vocab)))
    train_labels = torch.tensor([score for _, score in train_data])
    test_features = torch.tensor(pad_samples(encode_samples(test_tokenized, vocab)))
    test_labels = torch.tensor([score for _, score in test_data])
    
    
    # In[13]:
    
    
    class SentimentNet(nn.Module):
        def __init__(self, vocab_size, embed_size, num_hiddens, num_layers,
                     bidirectional, weight, labels, use_gpu, **kwargs):
            super(SentimentNet, self).__init__(**kwargs)
            self.num_hiddens = num_hiddens
            self.num_layers = num_layers
            self.use_gpu = use_gpu
            self.bidirectional = bidirectional
            self.embedding = nn.Embedding.from_pretrained(weight)
            self.embedding.weight.requires_grad = False
            self.encoder = nn.LSTM(input_size=embed_size, hidden_size=self.num_hiddens,
                                   num_layers=num_layers, bidirectional=self.bidirectional,
                                   dropout=0)
            if self.bidirectional:
                self.decoder = nn.Linear(num_hiddens * 4, labels)
            else:
                self.decoder = nn.Linear(num_hiddens * 2, labels)
    
        def forward(self, inputs):
            embeddings = self.embedding(inputs)
            states, hidden = self.encoder(embeddings.permute([1, 0, 2]))
            encoding = torch.cat([states[0], states[-1]], dim=1)
            outputs = self.decoder(encoding)
            return outputs
    
    
    # In[16]:
    
    
    num_epochs = 5
    embed_size = 100
    num_hiddens = 100
    num_layers = 2
    bidirectional = True
    batch_size = 64
    labels = 2
    lr = 0.8
    
    device = torch.device('cpu')
    
    use_gpu = True
    
    weight = torch.zeros(vocab_size + 1, embed_size)
    
    for i in range(len(wvmodel.index_to_key)):
        try:
            index = word_to_idx[wvmodel.index_to_key[i]]
        except:
            continue
        weight[index, :] = torch.from_numpy(wvmodel.get_vector(
            idx_to_word[word_to_idx[wvmodel.index_to_key[i]]]))
    
    # In[17]:
    
    
    net = SentimentNet(vocab_size=(vocab_size + 1), embed_size=embed_size,
                       num_hiddens=num_hiddens, num_layers=num_layers,
                       bidirectional=bidirectional, weight=weight,
                       labels=labels, use_gpu=use_gpu)
    net.to(device)
    loss_function = nn.CrossEntropyLoss()
    optimizer = optim.SGD(net.parameters(), lr=lr)
    
    # In[18]:
    
    
    train_set = torch.utils.data.TensorDataset(train_features, train_labels)
    test_set = torch.utils.data.TensorDataset(test_features, test_labels)
    
    train_iter = torch.utils.data.DataLoader(train_set, batch_size=batch_size,
                                             shuffle=True)
    test_iter = torch.utils.data.DataLoader(test_set, batch_size=batch_size,
                                            shuffle=False)
    
    # In[20]:
    
    
    num_epochs = 20
    
    # In[ ]:
    
    
    for epoch in range(num_epochs):
        start = time.time()
        train_loss, test_losses = 0, 0
        train_acc, test_acc = 0, 0
        n, m = 0, 0
        for feature, label in train_iter:
            n += 1
            net.zero_grad()
            feature = Variable(feature.cpu())
            label = Variable(label.cpu())
            score = net(feature)
            loss = loss_function(score, label)
            loss.backward()
            optimizer.step()
            train_acc += accuracy_score(torch.argmax(score.cpu().data,
                                                     dim=1), label.cpu())
            train_loss += loss
        with torch.no_grad():
            for test_feature, test_label in test_iter:
                m += 1
                test_feature = test_feature.cpu()
                net(test_feature)
                test_loss = loss_function(test_score, test_label)
                test_acc += accuracy_score(torch.argmax(test_score.cpu().data,
                                                        dim=1), test_label.cpu())
                test_losses += test_loss
        end = time.time()
        runtime = end - start
        epoch: %d, train loss: %.4f, train acc: %.2f, test loss: %.4f, test acc: %.2f, time: %.2f' %
              (epoch, train_loss.data / n, train_acc / n, test_losses.data / m, test_acc / m, runtime))
    
    # In[ ]:

**创作不易 觉得有帮助请点赞关注收藏~~~**

[f17d61f6c91d4c3aacb6220aa49addc8.png]: https://img-blog.csdnimg.cn/f17d61f6c91d4c3aacb6220aa49addc8.png
[1dc9cc1183c34665bbb5ad7e2ddf43b3.png]: https://img-blog.csdnimg.cn/1dc9cc1183c34665bbb5ad7e2ddf43b3.png
[52289d07b1044a51bd590212713f8c50.png]: https://img-blog.csdnimg.cn/52289d07b1044a51bd590212713f8c50.png
[c74b146b59ec423e9de8b8d4723902b0.png]: https://img-blog.csdnimg.cn/c74b146b59ec423e9de8b8d4723902b0.png
[df4cca2d9105428491a538a0f63a3c39.png]: https://img-blog.csdnimg.cn/df4cca2d9105428491a538a0f63a3c39.png
[887067933b5145b589ab2a63c65529cf.png]: https://img-blog.csdnimg.cn/887067933b5145b589ab2a63c65529cf.png
[659c305bb66c4e0287434ed07d1981c1.png]: https://img-blog.csdnimg.cn/659c305bb66c4e0287434ed07d1981c1.png