基于BERT－BiGRU－FFNN情绪分析模型的情绪分析任务_bert+bigru

一、情绪分析模型

为了写作方便，本文将本次实验采用的基于BERT-BASE-UNCASED预训练模型+双向GRU循环神经网络+前馈神经网络模型称为情绪分析模型。

   模型概览：

   由于模型过大，因此只能看到模型的大概信息，详细信息将在后文描述。

输入层

   从模型概览中可以看到首先是模型的输入层，bert-base-uncased预训练模型的输入神经个数为9216，当然这些神经元不会全部用到，由输入序列长度确定。对于输入来说，不是简单的将文本直接输入模型，而是需要根据bert-base-uncased预训练模型的词表，先将文本转换成数字化标记，之后，在文本数字化标记的开头和结尾加上两个特殊标记，［CLS］、［SEP］，分别表示句子的开头和结尾，处理完上述工作之后，将文本的数字化标记输入模型。

bert-base-uncased预训练模型

   bert-base-uncased预训练模型由12个编码器堆叠而成，这里说的编码器就是transformer中的编码器，transformer由堆叠N层的编码器和堆叠N层解码器以及一些前馈神经网络组成，是当前特征提取能力最强的模型。GPT系列模型采用了transformer的解码器，BERT模型则采用transformer的编码器。

   对于编码器来说，首先是一个多头注意力模型，引入残差结构，将多头注意力模型的输出与原始输入做一个加法，这一步主要是为了防止梯度消失问题。之后对输出结果正则化。在多头注意力模型的上面是一个前馈神经网络，前馈神经网络也引入残差结构，之后对输出结果正则化。

   至于为什么要堆叠12层，主要是为了让模型理解文本信息，提取更多的上下文信息，最终目的是为了提取一个好的特征。

双向GRU循环神经网络模型

   双向GRU模型概览

   左边是正向GRU，由一个输入层，两个隐藏层构成，文本序列从前往后传播。右边是反向GRU模型，由一个输入层，两个隐藏层构成，文本序列从后往前传播。

   下面是正向GRU模型与反向GRU模型的具体细节。

正向GRU

   按照时间顺序，从前往后依次从文本序列中取出每个词的词嵌入表示输入正向GRU模型，bert-base-uncased预训练模型提取的词嵌入表示维度是768，也就是说一个词用一个768维的向量表示。那么双向GRU的输入层的维度为768。

反向GRU

   按照时间顺序，从后往前依次从文本序列中取出每个词的词嵌入表示输入反向GRU模型，词嵌入表示的维度和反向GRU的输入层的维度同正向GRU。

前馈神经网络

   文本序列经过双向GRU模型之后，将双向GRU模型的正向GRU的最后一个隐藏层得到的输出与反向GRU模型的最后一个隐藏层的输出拼接为一个向量，维度是原来正向GRU和反向GRU最后一个隐藏层的输出的维度的两倍，即hidden_dim*2,256*2。之后将拼接后的向量作为前馈神经网络的输入。前馈神经网络的输出神经元个数为1，表示该文本序列属于positive的概率。

二、实验代码（Python） 

# -*- codeing = utf-8 -*-
import torch
import random
import numpy as np
 
SEED = 1234
random.seed(SEED)
np.random.seed(SEED)
torch.manual_seed(SEED)
torch.deterministic = True
#torch.backends.deterministic = True
 
from transformers import BertTokenizer
#从预训练模型中引入分词等功能函数
tokenizer = BertTokenizer.from_pretrained('D:\课程\神经网络与深度学习\deep_learning\情感分类\\bert-base-uncased')
print('`tokenizer` 有一个 `vocab` 属性，它包含我们将使用的实际词汇', len(tokenizer.vocab))
 
# 使用`tokenizer.tokenize`方法对字符串进行分词，并统一大小写。
tokens = tokenizer.tokenize('Hello WORLD how ARE yoU?')
print(tokens)#['hello','world','how','are','you','?']
 
# 我们可以使用我们的词汇表使用 `tokenizer.convert_tokens_to_ids` 来数字化标记。
# 下面的tokens是我们之前上面进行了分词和统一大小写之后的list。
indexes = tokenizer.convert_tokens_to_ids(tokens)#将tokens中的单词转换为数字化标记
print(indexes)#[7592,2088,2129,2024,2017,1029]
 
# Transformer还接受了特殊tokens的训练，以标记句子的开头和结尾，[CLS],[SEP]
# 就像我们标准化padding和未知的token一样，我们也可以从`tokenizer`中获取这些。
# **注意**：`tokenizer` 确实有序列开始和序列结束属性（`bos_token` 和 `eos_token`），
# 但我们没有对此进行设置，并且不适用于我们本次训练的transformer。
init_token = tokenizer.cls_token#[CLS]，句子开头
eos_token = tokenizer.sep_token#[SEP]，句子结尾
pad_token = tokenizer.pad_token#[PAD]，padding
unk_token = tokenizer.unk_token#[UNK]
print(init_token, eos_token, pad_token, unk_token)#[CLS] [SEP] [PAD] [UNK]
 
# 我们可以通过反转词汇表来获得特殊tokens的索引
init_token_idx = tokenizer.convert_tokens_to_ids(init_token)#将特殊标记转换为数字化标记
eos_token_idx = tokenizer.convert_tokens_to_ids(eos_token)
pad_token_idx = tokenizer.convert_tokens_to_ids(pad_token)
unk_token_idx = tokenizer.convert_tokens_to_ids(unk_token)
print(init_token_idx, eos_token_idx, pad_token_idx, unk_token_idx)#101 102 0 100
 
# 或者通过tokenizer的方法直接获取它们
init_token_idx = tokenizer.cls_token_id#对于特殊标记，transformer提供了将特殊标记转换为数字化标记的函数
eos_token_idx = tokenizer.sep_token_id
pad_token_idx = tokenizer.pad_token_id
unk_token_idx = tokenizer.unk_token_id
print(init_token_idx, eos_token_idx, pad_token_idx, unk_token_idx)#101 102 0 100
 
# 我们需要处理的另一件事是模型是在具有定义的最大长度的序列上训练的——它不知道如何处理比训练更长的序列。
# 我们可以通过检查我们想要使用的转换器版本的 `max_model_input_sizes` 来获得这些输入大小的最大长度。
max_input_length = tokenizer.max_model_input_sizes['bert-base-uncased']
print(max_input_length)#512
 
# 之前我们使用了 `spaCy` 标记器来标记我们的示例。
# 然而，我们现在需要定义一个函数，我们将把它传递给我们的 `TEXT` 字段，它将为我们处理所有的标记化。
# 它还会将令牌的数量减少到最大长度。 请注意，我们的最大长度比实际最大长度小 2。
# 这是因为我们需要向每个序列附加两个标记，一个在开头，一个在结尾。
def tokenize_and_cut(sentence):
    tokens = tokenizer.tokenize(sentence)
    tokens = tokens[:max_input_length - 2]
    return tokens
 
# 现在我们开始定义我们的字段,transformer期望将batch维度放在第一维上，所以我们设置了 `batch_first = True`。
# 现在我们已经有了文本的词汇数据，由transformer提供，我们设置 `use_vocab = False` 来告诉 torchtext 已经不需要切分数据了。
# 我们将 `tokenize_and_cut` 函数作为标记器传递。 `preprocessing` 参数是一个函数，这是我们将token转换为其索引的地方。
# 最后，我们定义特殊的token——注意我们将它们定义为它们的索引值而不是它们的字符串值，即“100”而不是“[UNK]”这是因为序列已经被转换为索引。
# 我们像以前一样定义标签字段。
from torchtext import data
#获取文本数据集
TEXT = data.Field(batch_first=True,#transformer希望将batch放在第一维
                  use_vocab=False,#不需要切分数据
                  tokenize=tokenize_and_cut,#分词
                  preprocessing=tokenizer.convert_tokens_to_ids,#将词语转换成数字化标记
                  init_token=init_token_idx,#[CLS]的数字化标记，101
                  eos_token=eos_token_idx,#[SEP]的数字化标记
                  pad_token=pad_token_idx,#[PAD]的数字化标记
                  unk_token=unk_token_idx)#[UNK]的数字化标记
#获取文本数据集的标签
LABEL = data.LabelField(dtype=torch.float)#标签的类型是torch.float
 
# 加载数据，拆分成训练集和验证集
from torchtext import datasets
train_data, test_data = datasets.IMDB.splits(TEXT, LABEL)#将数据集划分为训练集和测试集
train_data, valid_data = train_data.split(random_state=random.seed(SEED))#将训练集划分为训练集和验证集，引入随机种子
print(f"Number of training examples: {len(train_data)}")#17500
print(f"Number of validation examples: {len(valid_data)}")#7500
print(f"Number of testing examples: {len(test_data)}")#25000
 
# 随便看一个例子，看下具体效果如何,输出其中一个句子的one-hot向量。
print(f"文本的数字化标记{vars(train_data.examples[6])}")#输出数字化标记
# 我们可以使用 `convert_ids_to_tokens` 将这些索引转换回可读的tokens。
tokens = tokenizer.convert_ids_to_tokens(vars(train_data.examples[6])['text'])#将数字化标记转换成词语
print(f"数字化标记转换成文本{tokens}")
 
# 尽管我们已经处理了文本的词汇表，当然也需要为标签构建词汇表。
LABEL.build_vocab(train_data)
print(LABEL.vocab.stoi)#defaultdict(None, {'neg': 0, 'pos': 1})
 
# 像之前一样，我们创建迭代器。根据以往经验，使用最大的batch size可以使transformer获得最好的效果，
# 当然，你也可以尝试一下使用其他的batch size，如果你的显卡比较好的话。
BATCH_SIZE = 16#批处理大小
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')#利用cuda进行训练
train_iterator, valid_iterator, test_iterator = data.BucketIterator.splits(
    (train_data, valid_data, test_data),
    batch_size=BATCH_SIZE,
    device=device)#将训练数据、验证数据、测试数据按照GPU的标准划分批次，16个数据为一组
 
#构建模型
#导入预训练模型
from transformers import BertTokenizer, BertModel
bert = BertModel.from_pretrained('D:\\课程\\神经网络与深度学习\\deep_learning\\情感分类\\bert-base-uncased\\')
import torch.nn as nn
 
class BERTGRUSentiment(nn.Module):
    def __init__(self,bert,hidden_dim,output_dim,n_layers,bidirectional,dropout):
        super().__init__()
        self.bert = bert#预训练模型
        embedding_dim = bert.config.to_dict()['hidden_size']#[batch_size,sentence_length,hidden_size],[1,7,768]
        #构建双向GRU模型，一个输入层，两个隐藏层，采用双向训练，引入dropout
        #输入层的神经元数量为bert模型的提取的词嵌入的维度，即768
        #隐藏层的神经元数量为256
        self.rnn = nn.GRU(embedding_dim,hidden_dim,num_layers=n_layers,bidirectional=bidirectional,batch_first=True,dropout=0 if n_layers < 2 else dropout)
        #前馈神经网络，包含一个输入层和一个输出层，输入层的神经元个数为hidden_dim*2，输出层的神经元数量为1，类型为float，表示属于positive的概率
        self.out = nn.Linear(hidden_dim * 2 if bidirectional else hidden_dim, output_dim)
        #定义dropout
        self.dropout = nn.Dropout(dropout)
 
    def forward(self, text):
        #text = [batch_size, sent_len]
        #使用静态微调
        with torch.no_grad():
            embedded = self.bert(text)[0]#embedded=[batch_size,sent_len,emb_dim],[1,7,768]，从[CLS]获得训练文本数据的词嵌入表示
        #将提取的词嵌入向量输入GRU网络
        _, hidden = self.rnn(embedded)
        #hidden=[n_layers*n_directions,batch_size,emb_dim],[2*2,1,7]
        #一共有2层隐藏层*2（双向GRU）*每批次1个数据*每个数据7个单词个隐藏层
        if self.rnn.bidirectional:
            hidden = self.dropout(torch.cat((hidden[-2, :, :], hidden[-1, :, :]), dim=1))#取最后两个隐藏层的数据，连接，正则化
        else:
            hidden = self.dropout(hidden[-1, :, :])#如果没有采用双向GRU，则只需要取最后一个隐藏层的数据即可
        # hidden = [batch_size, 2*hid_dim] or [batch_size,hid_dim]
        output = self.out(hidden)#将隐藏层的输出数据作为前馈神经网络的输入
        # output = [batch_size, out_dim]
        return output
 
HIDDEN_DIM = 256#隐藏层大小
OUTPUT_DIM = 1#前馈神经网络输出层大小
N_LAYERS = 2#循环神经网络隐藏层层数
BIDIRECTIONAL = True#是否使用双向GRU
DROPOUT = 0.25#正则化
model = BERTGRUSentiment(bert,HIDDEN_DIM,OUTPUT_DIM,N_LAYERS,BIDIRECTIONAL,DROPOUT)#定义模型
 
def count_parameters(model):
    return sum(p.numel() for p in model.parameters() if p.requires_grad)
#没有冻结bert模型参数时，模型可训练参数量，112241409
print(f'The model has {count_parameters(model):,} trainable parameters')
 
#冻结bert模型参数
for name, param in model.named_parameters():#遍历模型的所有参数
    if name.startswith('bert'):#如果模型参数名以bert开头，说明该参数是bert模型的参数
        param.requires_grad = False#冻结bert模型的参数
 
def count_parameters(model):
    return sum(p.numel() for p in model.parameters() if p.requires_grad)
#冻结bert模型参数后，输出模型可训练参数，2759169
print(f'The model has {count_parameters(model):,} trainable parameters')
#输出模型可训练参数名
for name, param in model.named_parameters():
    if param.requires_grad:
        print(name)
 
#构建模型优化器和损失函数
import torch.optim as optim
optimizer = optim.Adam(model.parameters())#优化器
criterion = nn.BCEWithLogitsLoss()#采用带有sigmiod()函数的交叉熵损失函数
#将模型和损失函数放在GPU上，如果你有GPU的话
model = model.to(device)
criterion = criterion.to(device)
 
#定义函数用于：计算准确度、定义train、evaluate函数以及计算训练/评估时期每一个epoch所需的时间。
#计算模型的准确度
def binary_accuracy(preds, y):
    """
    Returns accuracy per batch, i.e. if you get 8/10 right, this returns 0.8, NOT 8
    """
    # round predictions to the closest integer
    rounded_preds = torch.round(torch.sigmoid(preds))
    correct = (rounded_preds == y).float()  # convert into float for division
    acc = correct.sum() / len(correct)
    return acc
 
#模型训练函数
def train(model, iterator, optimizer, criterion):
    epoch_loss = 0
    epoch_acc = 0
    model.train()
    for batch in iterator:
        optimizer.zero_grad()
        predictions = model(batch.text).squeeze(1)
        loss = criterion(predictions, batch.label)
        acc = binary_accuracy(predictions, batch.label)
        loss.backward()
        optimizer.step()
        epoch_loss += loss.item()
        epoch_acc += acc.item()
    return epoch_loss / len(iterator), epoch_acc / len(iterator)
 
#模型评估函数
def evaluate(model, iterator, criterion):
    epoch_loss = 0
    epoch_acc = 0
 
    model.eval()
    with torch.no_grad():
        for batch in iterator:
            predictions = model(batch.text).squeeze(1)
            loss = criterion(predictions, batch.label)
            acc = binary_accuracy(predictions, batch.label)
            epoch_loss += loss.item()
            epoch_acc += acc.item()
    return epoch_loss / len(iterator), epoch_acc / len(iterator)
 
import time
#计算一个epoch所用时间
def epoch_time(start_time, end_time):
    elapsed_time = end_time - start_time
    elapsed_mins = int(elapsed_time / 60)
    elapsed_secs = int(elapsed_time - (elapsed_mins * 60))
    return elapsed_mins, elapsed_secs
 
#训练模型。由于transformer的尺寸的原因，这比以前的任何型号都要长得多。
#即使我们没有训练任何transformer的参数，我们仍然需要通过模型传递数据，这在标准GPU上需要花费大量时间。
N_EPOCHS = 1
best_valid_loss = float('inf')#定义一个变量，初始化为一个极大值，目的是为了找到模型损失函数值最小的那一次epoch，保存模型
for epoch in range(N_EPOCHS):
    start_time = time.time()#获取开始时间
    train_loss, train_acc = train(model, train_iterator, optimizer, criterion)#用训练集训练模型
    valid_loss, valid_acc = evaluate(model, valid_iterator, criterion)#用验证集评估模型
    end_time = time.time()#获取结束时间
    epoch_mins, epoch_secs = epoch_time(start_time, end_time)#计算模型训练和评估所用的时间
    if valid_loss < best_valid_loss:#判断是否是损失函数值最小的模型
        best_valid_loss = valid_loss
        torch.save(model.state_dict(), 'tut6-model.pt')
    print(f'Epoch: {epoch + 1:02} | Epoch Time: {epoch_mins}m {epoch_secs}s')#迭代次数，时间
    print(f'\tTrain Loss: {train_loss:.3f} | Train Acc: {train_acc * 100:.2f}%')#训练集损失函数值，训练集准确度
    print(f'\t Val. Loss: {valid_loss:.3f} |  Val. Acc: {valid_acc * 100:.2f}%')#验证集损失函数值，验证集准确度
 
# 加载在验证集上损失函数值最小的模型参数，用测试集测试
model.load_state_dict(torch.load('tut6-model.pt'))
test_loss, test_acc = evaluate(model, test_iterator, criterion)
print(f'Test Loss: {test_loss:.3f} | Test Acc: {test_acc * 100:.2f}%')#测试集损失函数值，测试集准确度
 
#模型测试
def predict_sentiment(model, tokenizer, sentence):
    model.eval()
    tokens = tokenizer.tokenize(sentence)
    tokens = tokens[:max_input_length - 2]
    indexed = [init_token_idx] + tokenizer.convert_tokens_to_ids(tokens) + [eos_token_idx]
    tensor = torch.LongTensor(indexed).to(device)
    tensor = tensor.unsqueeze(0)
    prediction = torch.sigmoid(model(tensor))
    return prediction.item()
 
text1=("recently watched a highly anticipated film, and it was a major letdown. The plot was confusing,"
       " with shallow character development, making the overall story dull and uninteresting. The actors' "
       "performances were lackluster, lacking emotional engagement and failing to resonate with the audience. "
       "The music in the film was a disaster, lacking any emotional tension and instead creating an awkward "
       "atmosphere. Overall, this movie left me feeling disappointed and disheartened, questioning the director "
       "and actors' creative abilities.")
text2=("recently viewed a controversial film, and the production quality was quite decent. While the storyline "
       "was somewhat bland, the director managed to present a unique perspective through subtle camera techniques "
       "and distinctive narrative methods. The actors' performances, although not evoking strong emotional "
       "resonance, were passable. The music design in the film, while not outstanding, contributed reasonably "
       "well to the overall atmosphere. In summary, this movie, while not stirring, is a moderately standard "
       "piece of work.")
text3=("recently enjoyed a breathtaking film, undoubtedly a delightful audio-visual journey. The plot not only "
       "gripped the audience but also featured well-developed characters, each with unforgettable unique charm. "
       "The actors delivered outstanding performances, adding significant depth to the overall story, and every "
       "emotional fluctuation touched the audience's heartstrings. The music in the film seamlessly integrated "
       "with the plot, infusing more emotional nuances into each scene. Overall, this movie is enchanting, "
       "allowing the audience to experience the true magic of cinematic art.")
 
y_pred1=predict_sentiment(model, tokenizer, "This film is terrible")
y_pred2=predict_sentiment(model, tokenizer, "This film is great")
 
y_pred3=predict_sentiment(model, tokenizer, text1)#消极
y_pred4=predict_sentiment(model, tokenizer, text2)#中性
y_pred5=predict_sentiment(model, tokenizer, text3)#积极
 
print(y_pred1)
print(y_pred2)
print(y_pred3)
print(y_pred4)
print(y_pred5)