情感分类是自然语言处理（NLP）中的一个经典任务，广泛应用于社交媒体分析、市场调研和客户反馈等领域。本篇博客将带领大家使用MindSpore框架，基于RNN（循环神经网络）实现一个情感分类模型。我们将详细介绍数据准备、模型构建、训练与评估等步骤，并最终实现对自定义输入的情感预测。
数据准备
下载和加载数据集
- 为什么要使用
requests和tqdm库？requests库提供了简洁的HTTP请求接口，方便我们从网络上下载数据。tqdm库则用于显示下载进度条，帮助我们实时了解下载进度，提高用户体验。 - 为什么要使用临时文件和
shutil库？ 使用临时文件可以确保下载的数据在出现意外中断时不会影响最终保存的文件。shutil库提供了高效的文件操作方法，确保数据能正确地从临时文件复制到最终保存路径。
我们使用IMDB影评数据集，这是一个经典的情感分类数据集，包含积极（Positive）和消极（Negative）两类影评。为了方便数据下载和处理，我们首先设计一个数据下载模块。
import os
import shutil
import requests
import tempfile
from tqdm import tqdm
from typing import IO
from pathlib import Path
# 指定保存路径
cache_dir = Path.home() / '.mindspore_examples'
def http_get(url: str, temp_file: IO):
req = requests.get(url, stream=True)
content_length = req.headers.get('Content-Length')
total = int(content_length) if content_length is not None else None
progress = tqdm(unit='B', total=total)
for chunk in req.iter_content(chunk_size=1024):
if chunk:
progress.update(len(chunk))
temp_file.write(chunk)
progress.close()
def download(file_name: str, url: str):
if not os.path.exists(cache_dir):
os.makedirs(cache_dir)
cache_path = os.path.join(cache_dir, file_name)
cache_exist = os.path.exists(cache_path)
if not cache_exist:
with tempfile.NamedTemporaryFile() as temp_file:
http_get(url, temp_file)
temp_file.flush()
temp_file.seek(0)
with open(cache_path, 'wb') as cache_file:
shutil.copyfileobj(temp_file, cache_file)
return cache_path
下载IMDB数据集并进行解压和加载：
import re
import six
import string
import tarfile
class IMDBData():
label_map = {"pos": 1, "neg": 0}
def __init__(self, path, mode="train"):
self.mode = mode
self.path = path
self.docs, self.labels = [], []
self._load("pos")
self._load("neg")
def _load(self, label):
pattern = re.compile(r"aclImdb/{}/{}/.*\.txt$".format(self.mode, label))
with tarfile.open(self.path) as tarf:
tf = tarf.next()
while tf is not None:
if bool(pattern.match(tf.name)):
self.docs.append(str(tarf.extractfile(tf).read().rstrip(six.b("\n\r"))
.translate(None, six.b(string.punctuation)).lower()).split())
self.labels.append([self.label_map[label]])
tf = tarf.next()
def __getitem__(self, idx):
return self.docs[idx], self.labels[idx]
def __len__(self):
return len(self.docs)
加载训练数据集进行测试，输出数据集数量：
imdb_path = download('aclImdb_v1.tar.gz', 'https://mindspore-website.obs.myhuaweicloud.com/notebook/datasets/aclImdb_v1.tar.gz')
imdb_train = IMDBData(imdb_path, 'train')
print(len(imdb_train))
加载预训练词向量
为什么要使用预训练词向量？ 预训练词向量（如Glove）是基于大规模语料库训练得到的，能够捕捉到丰富的语义信息。使用预训练词向量可以提高模型的性能，尤其是在数据量较小的情况下。
- 为什么要添加
<unk>和<pad>标记符？<unk>标记符用于处理词汇表中未出现的单词，避免模型在遇到新词时无法处理。<pad>标记符用于填充不同长度的文本序列，使其能够被打包为一个batch进行并行计算，提高训练效率。
我们使用Glove预训练词向量对文本进行编码。以下是加载Glove词向量的代码：
import zipfile
import numpy as np
def load_glove(glove_path):
glove_100d_path = os.path.join(cache_dir, 'glove.6B.100d.txt')
if not os.path.exists(glove_100d_path):
glove_zip = zipfile.ZipFile(glove_path)
glove_zip.extractall(cache_dir)
embeddings = []
tokens = []
with open(glove_100d_path, encoding='utf-8') as gf:
for glove in gf:
word, embedding = glove.split(maxsplit=1)
tokens.append(word)
embeddings.append(np.fromstring(embedding, dtype=np.float32, sep=' '))
embeddings.append(np.random.rand(100))
embeddings.append(np.zeros((100,), np.float32))
vocab = ds.text.Vocab.from_list(tokens, special_tokens=["<unk>", "<pad>"], special_first=False)
embeddings = np.array(embeddings).astype(np.float32)
return vocab, embeddings
glove_path = download('glove.6B.zip', 'https://mindspore-website.obs.myhuaweicloud.com/notebook/datasets/glove.6B.zip')
vocab, embeddings = load_glove(glove_path)
print(len(vocab.vocab()))
数据集预处理
为什么要对文本进行分词和去除标点？ 分词和去除标点是文本预处理的基础步骤，有助于模型更好地理解文本的语义。将文本转为小写可以减少词汇表的大小，提高模型的泛化能力。
对加载的数据集进行预处理，包括将文本转为index id序列，并进行序列填充。
import mindspore as ms
import mindspore.dataset as ds
lookup_op = ds.text.Lookup(vocab, unknown_token='<unk>')
pad_op = ds.transforms.PadEnd([500], pad_value=vocab.tokens_to_ids('<pad>'))
type_cast_op = ds.transforms.TypeCast(ms.float32)
imdb_train = imdb_train.map(operations=[lookup_op, pad_op], input_columns=['text'])
imdb_train = imdb_train.map(operations=[type_cast_op], input_columns=['label'])
imdb_test = imdb_test.map(operations=[lookup_op, pad_op], input_columns=['text'])
imdb_test = imdb_test.map(operations=[type_cast_op], input_columns=['label'])
imdb_train, imdb_valid = imdb_train.split([0.7, 0.3])
imdb_valid = imdb_valid.batch(64, drop_remainder=True)
模型构建
接下来，我们构建用于情感分类的RNN模型。模型主要包括以下几层：
- Embedding层：将单词的index id转为词向量。
- RNN层：使用LSTM进行特征提取。
- 全连接层：将LSTM的输出特征映射到分类结果。
为什么选择LSTM而不是经典RNN？ LSTM通过引入门控机制，有效地缓解了经典RNN中存在的梯度消失问题，能够更好地捕捉长距离依赖信息，提高模型的效果。
import math
import mindspore as ms
import mindspore.nn as nn
import mindspore.ops as ops
from mindspore.common.initializer import Uniform, HeUniform
class RNN(nn.Cell):
def __init__(self, embeddings, hidden_dim, output_dim, n_layers, bidirectional, pad_idx):
super().__init__()
vocab_size, embedding_dim = embeddings.shape
self.embedding = nn.Embedding(vocab_size, embedding_dim, embedding_table=ms.Tensor(embeddings), padding_idx=pad_idx)
self.rnn = nn.LSTM(embedding_dim, hidden_dim, num_layers=n_layers, bidirectional=bidirectional, batch_first=True)
weight_init = HeUniform(math.sqrt(5))
bias_init = Uniform(1 / math.sqrt(hidden_dim * 2))
self.fc = nn.Dense(hidden_dim * 2, output_dim, weight_init=weight_init, bias_init=bias_init)
def construct(self, inputs):
embedded = self.embedding(inputs)
_, (hidden, _) = self.rnn(embedded)
hidden = ops.concat((hidden[-2, :, :], hidden[-1, :, :]), axis=1)
output = self.fc(hidden)
return output
损失函数与优化器
我们使用二分类交叉熵损失函数nn.BCEWithLogitsLoss和Adam优化器。
为什么使用二分类交叉熵损失函数和Adam优化器？ 二分类交叉熵损失函数适用于二分类任务，能够衡量模型预测结果与真实标签之间的差异。Adam优化器结合了动量和自适应学习率的优点，具有较快的收敛速度和较好的效果，广泛应用于深度学习模型的训练。
hidden_size = 256
output_size = 1
num_layers = 2
bidirectional = True
lr = 0.001
pad_idx = vocab.tokens_to_ids('<pad>')
model = RNN(embeddings, hidden_size, output_size, num_layers, bidirectional, pad_idx)
loss_fn = nn.BCEWithLogitsLoss(reduction='mean')
optimizer = nn.Adam(model.trainable_params(), learning_rate=lr)
训练逻辑
设计训练一个epoch的函数，用于训练过程和loss的可视化。
def forward_fn(data, label):
logits = model(data)
loss = loss_fn(logits, label)
return loss
grad_fn = ms.value_and_grad(forward_fn, None, optimizer.parameters)
def train_step(data, label):
loss, grads = grad_fn(data, label)
optimizer(grads)
return loss
def train_one_epoch(model, train_dataset, epoch=0):
model.set_train()
total = train_dataset.get_dataset_size()
loss_total = 0
step_total = 0
with tqdm(total=total) as t:
t.set_description('Epoch %i' % epoch)
for i in train_dataset.create_tuple_iterator():
loss = train_step(*i)
loss_total += loss.asnumpy()
step_total += 1
t.set_postfix(loss=loss_total/step_total)
t.update(1)
评估指标和逻辑
设计评估逻辑，计算模型在验证集上的准确率。
def binary_accuracy(preds, y):
rounded_preds = np.around(ops.sigmoid(preds).asnumpy())
correct = (rounded_preds == y).astype(np.float32)
acc = correct.sum() / len(correct)
return acc
def evaluate(model, test_dataset, criterion, epoch=0):
total = test_dataset.get_dataset_size()
epoch_loss = 0
epoch_acc = 0
step_total = 0
model.set_train(False)
with tqdm(total=total) as t:
t.set_description('Epoch %i' % epoch)
for i in test_dataset.create_tuple_iterator():
predictions = model(i[0])
loss = criterion(predictions, i[1])
epoch_loss += loss.asnumpy()
acc = binary_accuracy(predictions, i[1])
epoch_acc += acc
step_total += 1
t.set_postfix(loss=epoch_loss/step_total, acc=epoch_acc/step_total)
t.update(1)
return epoch_loss / total
模型训练与保存
进行模型训练，并保存最优模型。
num_epochs = 2
best_valid_loss = float('inf')
ckpt_file_name = os.path.join(cache_dir, 'sentiment-analysis.ckpt')
for epoch in range(num_epochs):
train_one_epoch(model, imdb_train, epoch)
valid_loss = evaluate(model, imdb_valid, loss_fn, epoch)
if valid_loss < best_valid_loss:
best_valid_loss = valid_loss
ms.save_checkpoint(model, ckpt_file_name)
模型加载与测试
加载已保存的最优模型，并在测试集上进行评估。
param_dict = ms.load_checkpoint(ckpt_file_name)
ms.load_param_into_net(model, param_dict)
imdb_test = imdb_test.batch(64)
evaluate(model, imdb_test, loss_fn)
自定义输入测试
设计一个预测函数，实现对自定义输入的情感预测。
score_map = {
1: "Positive",
0: "Negative"
}
def predict_sentiment(model, vocab, sentence):
model.set_train(False)
tokenized = sentence.lower().split()
indexed = vocab.tokens_to_ids(tokenized)
tensor = ms.Tensor(indexed, ms.int32)
tensor = tensor.expand_dims(0)
prediction = model(tensor)
通过本文的学习，我们成功地使用MindSpore框架实现了一个基于RNN的情感分类模型。我们从数据准备开始，详细讲解了如何加载和处理IMDB影评数据集，以及使用预训练的Glove词向量对文本进行编码。然后，我们构建了一个包含Embedding层、LSTM层和全连接层的情感分类模型，并使用二分类交叉熵损失函数和Adam优化器进行训练。最后，我们评估了模型在测试集上的性能，并实现了对自定义输入的情感预测。希望这篇博客能帮助你更好地理解RNN在自然语言处理中的应用，并激发你在NLP领域的更多探索和实践。
