Sentiment classification is a foundational task in natural language processing, aiming to determine the emotional tone of text—typically as positive or negative. In this implementation, we build a neural network using LSTM (Long Short-Term Memory), a variant of RNN, to classify movie reviews from the IMDB dataset. The model learns to distinguish between phrases like "This film is terrible" (Negative) and "This film is great" (Positive).

Data Preparation

We use the IMDB movie review dataset, which contains 50,000 labeled reviews (25,000 for training, 25,000 for testing), evenly split between positive and negative sentiments. Each review is a raw text string, requiring preprocessing before model ingestion.

To efficiently retrieve the dataset and pre-trained word embeddings, we design a reusable download utility using requests and tqdm for progress tracking. Temporary files are downloaded to a cache directory to avoid redundant network calls.

import os
import shutil
import requests
import tempfile
from tqdm import tqdm
from pathlib import Path

cache_dir = Path.home() / '.mindspore_examples'

def http_get(url: str, temp_file: tempfile._TemporaryFileWrapper):
    req = requests.get(url, stream=True)
    total = int(req.headers.get('Content-Length', 0))
    with tqdm(total=total, unit='B', unit_scale=True) as pbar:
        for chunk in req.iter_content(1024):
            if chunk:
                temp_file.write(chunk)
                pbar.update(len(chunk))

def download_file(filename: str, url: str) -> str:
    if not cache_dir.exists():
        cache_dir.mkdir(parents=True)
    cache_path = cache_dir / filename
    if not cache_path.exists():
        with tempfile.NamedTemporaryFile(delete=False) as tmp:
            http_get(url, tmp)
            shutil.copy2(tmp.name, cache_path)
    return str(cache_path)

We download the IMDB dataset and GloVe word embeddings from Huawei Cloud mirrors:

imdb_archive = download_file('aclImdb_v1.tar.gz', 'https://mindspore-website.obs.myhuaweicloud.com/notebook/datasets/aclImdb_v1.tar.gz')
glove_archive = download_file('glove.6B.zip', 'https://mindspore-website.obs.myhuaweicloud.com/notebook/datasets/glove.6B.zip')

Loading and Preprocessing IMDB Data

The IMDB archive contains structured directories: train/pos/, train/neg/, etc. We extract reviews and labels using Python’s tarfile module, applying basic text cleaning: converting to lowercase, removing punctuation, and splitting into tokens.

import tarfile
import re
import string
import six

class IMDBDataset:
    label_map = {'pos': 1, 'neg': 0}
    
    def __init__(self, archive_path: str, split: str = 'train'):
        self.split = split
        self.reviews = []
        self.labels = []
        pattern = re.compile(rf"aclImdb/{split}/(pos|neg)/.*\.txt$")
        
        with tarfile.open(archive_path) as tar:
            member = tar.next()
            while member:
                if pattern.match(member.name):
                    content = tar.extractfile(member).read().decode('utf-8').strip()
                    # Clean: lowercase, remove punctuation, split
                    cleaned = content.lower().translate(str.maketrans('', '', string.punctuation)).split()
                    self.reviews.append(cleaned)
                    self.labels.append([self.label_map[member.name.split('/')[2]]])
                member = tar.next()

    def __getitem__(self, idx):
        return self.reviews[idx], self.labels[idx]
    
    def __len__(self):
        return len(self.reviews)

We wrap the dataset using MindSpore’s GeneratorDataset for integration with its data pipeline:

import mindspore.dataset as ds

def load_datasets(archive_path: str):
    train_ds = ds.GeneratorDataset(IMDBDataset(archive_path, 'train'), 
                                   column_names=['text', 'label'], shuffle=True)
    test_ds = ds.GeneratorDataset(IMDBDataset(archive_path, 'test'), 
                                  column_names=['text', 'label'], shuffle=False)
    return train_ds, test_ds

train_dataset, test_dataset = load_datasets(imdb_archive)

Loading GloVe Word Embeddings

We use GloVe 6B 100d embeddings—100-dimensional vectors trained on 6 billion tokens. These are loaded into a vocabulary and a weight matrix for the embedding layer.

import zipfile
import numpy as np

def load_glove_embeddings(glove_zip_path: str):
    extract_path = cache_dir / 'glove.6B.100d.txt'
    if not extract_path.exists():
        with zipfile.ZipFile(glove_zip_path) as zf:
            zf.extract('glove.6B.100d.txt', cache_dir)
    
    vocab_words = []
    embeddings = []
    
    with open(extract_path, encoding='utf-8') as f:
        for line in f:
            parts = line.strip().split(' ', 1)
            word = parts[0]
            vec = np.fromstring(parts[1], sep=' ', dtype=np.float32)
            vocab_words.append(word)
            embeddings.append(vec)
    
    # Add special tokens: unknown and padding
    embeddings.append(np.random.randn(100).astype(np.float32))  # <unk>
    embeddings.append(np.zeros(100, dtype=np.float32))          # <pad>
    
    vocab = ds.text.Vocab.from_list(vocab_words, 
                                    special_tokens=['<unk>', '<pad>'], 
                                    special_first=False)
    embedding_matrix = np.stack(embeddings).astype(np.float32)
    return vocab, embedding_matrix

vocab, embedding_weights = load_glove_embeddings(glove_archive)
print(f"Vocabulary size: {len(vocab)}")  # 400002
</pad></unk>

Data Pipeline Preprocessing

Text must be converted from tokens to indices and padded to uniform length. We use MindSpore’s data transformation operators:

• Lookup: Maps tokens to their index IDs using the vocabulary.
• PadEnd: Pads sequences to length 500 using the <pad> token ID.
• TypeCast: Converts labels from int to float32 for loss computation.

import mindspore as ms
from mindspore.dataset import text, transforms

lookup_op = text.Lookup(vocab, unknown_token='<unk>')
pad_op = transforms.PadEnd([500], pad_value=vocab.tokens_to_ids('<pad>'))
cast_op = transforms.TypeCast(ms.float32)

train_dataset = train_dataset.map(operations=[lookup_op, pad_op], input_columns=['text'])
train_dataset = train_dataset.map(operations=[cast_op], input_columns=['label'])

test_dataset = test_dataset.map(operations=[lookup_op, pad_op], input_columns=['text'])
test_dataset = test_dataset.map(operations=[cast_op], input_columns=['label'])

# Split training set into train/validation (70%/30%)
train_dataset, val_dataset = train_dataset.split([0.7, 0.3])

# Batch data and drop incomplete batches
train_dataset = train_dataset.batch(64, drop_remainder=True)
val_dataset = val_dataset.batch(64, drop_remainder=True)
test_dataset = test_dataset.batch(64)

Model Architecture

The network consists of three components:

1. Embedding Layer: Uses pre-trained GloVe weights to convert token indices into dense vectors.
2. LSTM Layer: A bidirectional LSTM with 2 layers extracts sequential features. We use the final hidden states from both directions.
3. Linear Layer: Projects the concatenated hidden states to a single output for binary classification.

import mindspore.nn as nn
import mindspore.ops as ops
from mindspore.common.initializer import HeUniform, Uniform
import math

class SentimentClassifier(nn.Cell):
    def __init__(self, embedding_matrix, hidden_size, num_layers, bidirectional, pad_idx):
        super().__init__()
        vocab_size, embed_dim = embedding_matrix.shape
        self.embedding = nn.Embedding(vocab_size, embed_dim, 
                                      embedding_table=ms.Tensor(embedding_matrix), 
                                      padding_idx=pad_idx)
        self.lstm = nn.LSTM(embed_dim, hidden_size, num_layers=num_layers, 
                            bidirectional=bidirectional, batch_first=True)
        
        # Dense layer: concatenate forward and backward final states
        fc_input_dim = hidden_size * 2 if bidirectional else hidden_size
        self.fc = nn.Dense(fc_input_dim, 1, 
                           weight_init=HeUniform(math.sqrt(5)), 
                           bias_init=Uniform(1 / math.sqrt(fc_input_dim)))
    
    def construct(self, x):
        embedded = self.embedding(x)  # (batch, seq_len, embed_dim)
        _, (hidden, _) = self.lstm(embedded)  # hidden: (num_layers * 2, batch, hidden_size)
        
        # Concatenate last forward and backward hidden states
        forward_hidden = hidden[-2]  # last forward
        backward_hidden = hidden[-1] # last backward
        combined = ops.concat((forward_hidden, backward_hidden), axis=1)
        
        output = self.fc(combined)   # (batch, 1)
        return output

Training Configuration

For binary classification, we use BCEWithLogitsLoss (logits + sigmoid + binary cross-entropy). The optimizer is Adam with a learning rate of 0.001.

hidden_dim = 256
num_layers = 2
bidirectional = True
learning_rate = 0.001
pad_idx = vocab.tokens_to_ids('<pad>')

model = SentimentClassifier(embedding_weights, hidden_dim, num_layers, bidirectional, pad_idx)
loss_fn = nn.BCEWithLogitsLoss(reduction='mean')
optimizer = nn.Adam(model.trainable_params(), learning_rate=learning_rate)

Training and Evaluation Loops

We define forward and training steps using MindSpore’s automatic differentiation:

def forward_fn(inputs, labels):
    logits = model(inputs)
    loss = loss_fn(logits, labels)
    return loss

grad_fn = ms.value_and_grad(forward_fn, None, optimizer.parameters)

def train_step(inputs, labels):
    loss, grads = grad_fn(inputs, labels)
    optimizer(grads)
    return loss

def train_epoch(dataset, epoch_num):
    model.set_train()
    total_loss = 0
    steps = 0
    with tqdm(total=dataset.get_dataset_size(), desc=f"Epoch {epoch_num}") as pbar:
        for batch in dataset.create_tuple_iterator():
            loss = train_step(*batch)
            total_loss += loss.asnumpy()
            steps += 1
            pbar.set_postfix(avg_loss=total_loss / steps)
            pbar.update(1)
    return total_loss / steps

For evaluation, we compute accuracy by thresholding sigmoid outputs at 0.5:

import numpy as np

def compute_accuracy(predictions, labels):
    probs = ops.sigmoid(predictions).asnumpy()
    preds = np.round(probs)
    correct = (preds == labels.asnumpy()).astype(np.float32)
    return np.mean(correct)

def evaluate_epoch(dataset, epoch_num):
    model.set_train(False)
    total_loss = 0
    total_acc = 0
    steps = 0
    with tqdm(total=dataset.get_dataset_size(), desc=f"Eval {epoch_num}") as pbar:
        for batch in dataset.create_tuple_iterator():
            inputs, labels = batch
            logits = model(inputs)
            loss = loss_fn(logits, labels)
            acc = compute_accuracy(logits, labels)
            
            total_loss += loss.asnumpy()
            total_acc += acc
            steps += 1
            
            pbar.set_postfix(avg_loss=total_loss/steps, avg_acc=total_acc/steps)
            pbar.update(1)
    return total_loss / steps

Training Loop and Model Saving

We train for 5 epochs, saving the model when validation loss improves:

import os

checkpoint_path = cache_dir / 'sentiment_model.ckpt'
best_val_loss = float('inf')

for epoch in range(5):
    train_loss = train_epoch(train_dataset, epoch)
    val_loss = evaluate_epoch(val_dataset, epoch)
    
    if val_loss < best_val_loss:
        best_val_loss = val_loss
        ms.save_checkpoint(model, str(checkpoint_path))
        print(f"✅ Saved model at epoch {epoch} with val_loss: {val_loss:.4f}")

Model Inference and Prediction

After training, we load the best checkpoint and create a prediction function that accepts raw text:

param_dict = ms.load_checkpoint(str(checkpoint_path))
ms.load_param_into_net(model, param_dict)

def predict_sentiment(text: str) -> str:
    model.set_train(False)
    tokens = text.lower().split()
    indices = [vocab.tokens_to_ids(t) if t in vocab else vocab.tokens_to_ids('<unk>') for t in tokens]
    tensor = ms.Tensor(indices, ms.int32).expand_dims(0)
    logits = model(tensor)
    prob = ops.sigmoid(logits).asnumpy()[0][0]
    return "Positive" if prob > 0.5 else "Negative"

# Test predictions
predict_sentiment("This film is terrible")  # Output: 'Negative'
predict_sentiment("This film is great")    # Output: 'Positive'