Recurrent Neural Networks and Sequential Learning

import torch
import torch.nn as nn
import numpy as np
import matplotlib.pyplot as plt

# Reproducible notebook results
torch.manual_seed(42)
np.random.seed(42)

print("PyTorch environment ready!")

# Create a sine wave and convert it into past-to-future pairs.
x_axis = np.linspace(0, 100, 1000)
wave_data = np.sin(x_axis)

def create_sequences(data, seq_length):
    xs, ys = [], []
    for i in range(len(data) - seq_length):
        xs.append(data[i:i + seq_length])
        ys.append(data[i + seq_length])
    return np.array(xs), np.array(ys)

seq_length = 1
X_raw, Y_raw = create_sequences(wave_data, seq_length)

# Add the feature axis expected by nn.RNN.
X = torch.tensor(X_raw, dtype=torch.float32).unsqueeze(-1)
Y = torch.tensor(Y_raw, dtype=torch.float32).unsqueeze(-1)

print("X shape:", tuple(X.shape))
print("Y shape:", tuple(Y.shape))
print("First input:", np.round(X_raw[0], 3))
print("First target:", round(float(Y_raw[0]), 3))

# Optional Colab visualization of one sliding window.
start = 30
end = start + seq_length
plt.figure(figsize=(10, 4))
plt.plot(wave_data[:120], label="Sine wave")
plt.scatter(range(start, end), wave_data[start:end], label="Input window")
plt.scatter(end, wave_data[end], marker="*", s=160, label="Next-value target")
plt.xlabel("Time step")
plt.ylabel("Wave amplitude")
plt.legend()
plt.grid(True)
plt.show()

class WaveRNN(nn.Module):
    def __init__(self, input_size=1, hidden_size=32, output_size=1):
        super().__init__()
        self.rnn = nn.RNN(input_size, hidden_size, batch_first=True)
        self.fc = nn.Linear(hidden_size, output_size)

    def forward(self, x):
        all_hidden_states, final_hidden = self.rnn(x)
        prediction = self.fc(final_hidden[-1])
        return prediction

model = WaveRNN()
one_sequence = X[0:1]  # (one example, one time step, one feature)

with torch.no_grad():
    all_hidden_states, final_hidden = model.rnn(one_sequence)

print(model)
print("Input shape:", tuple(one_sequence.shape))
print("All hidden states:", tuple(all_hidden_states.shape))
print("Final hidden state:", tuple(final_hidden.shape))

criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
loss_history = []

for epoch in range(150):
    model.train()
    predictions = model(X)
    loss = criterion(predictions, Y)

    optimizer.zero_grad()
    loss.backward()  # Autograd performs BPTT.
    optimizer.step()

    loss_history.append(loss.item())
    if epoch % 25 == 0:
        print(f"Epoch {epoch:3d} | Loss: {loss.item():.5f}")

model.eval()
with torch.no_grad():
    wave_predictions = model(X).numpy()

# Plot training loss.
plt.figure(figsize=(8, 4))
plt.plot(loss_history)
plt.xlabel("Epoch")
plt.ylabel("MSE loss")
plt.grid(True)
plt.show()

# Plot the first 200 targets and predictions.
plt.figure(figsize=(10, 4))
plt.plot(Y.numpy()[:200], label="Actual wave", linewidth=3, alpha=0.5)
plt.plot(wave_predictions[:200], "r--", label="RNN prediction")
plt.xlabel("Time step")
plt.ylabel("Wave amplitude")
plt.legend()
plt.grid(True)
plt.show()

# Colab note: this downloads the NLTK Movie Reviews dataset when run.
import nltk
from nltk.corpus import movie_reviews
from collections import Counter
import random

nltk.download("movie_reviews", quiet=True)

documents = []
for label in movie_reviews.categories():
    for file_id in movie_reviews.fileids(label):
        words = [w.lower() for w in movie_reviews.words(file_id) if w.isalpha()]
        documents.append((words, 1 if label == "pos" else 0))

random.seed(42)
random.shuffle(documents)

word_counts = Counter(word for words, _ in documents for word in words)
vocab = {"<PAD>": 0, "<UNK>": 1}
for word, _ in word_counts.most_common(9_998):
    vocab[word] = len(vocab)

max_len = 100

def encode_review(words):
    ids = [vocab.get(word, vocab["<UNK>"]) for word in words[:max_len]]
    return ids + [vocab["<PAD>"]] * (max_len - len(ids))

X_text = torch.tensor([encode_review(words) for words, _ in documents])
Y_text = torch.tensor([label for _, label in documents], dtype=torch.float32).unsqueeze(1)

split = int(0.8 * len(X_text))
X_train, X_test = X_text[:split], X_text[split:]
Y_train, Y_test = Y_text[:split], Y_text[split:]

print("Reviews:", len(documents))
print("Encoded input shape:", tuple(X_text.shape))

class SentimentRNN(nn.Module):
    def __init__(self, vocab_size, embedding_dim=64, hidden_size=64):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embedding_dim, padding_idx=0)
        self.rnn = nn.RNN(embedding_dim, hidden_size, batch_first=True)
        self.fc = nn.Linear(hidden_size, 1)

    def forward(self, token_ids):
        embedded = self.embedding(token_ids)
        _, final_hidden = self.rnn(embedded)
        return self.fc(final_hidden[-1])

sentiment_model = SentimentRNN(len(vocab))
criterion = nn.BCEWithLogitsLoss()
optimizer = torch.optim.Adam(sentiment_model.parameters(), lr=0.001)

for epoch in range(5):
    sentiment_model.train()
    permutation = torch.randperm(X_train.size(0))

    for start in range(0, X_train.size(0), 64):
        indices = permutation[start:start + 64]
        logits = sentiment_model(X_train[indices])
        loss = criterion(logits, Y_train[indices])

        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

def predict_sentiment(sentence):
    words = [word.lower() for word in sentence.split() if word.isalpha()]
    token_ids = torch.tensor([encode_review(words)])

    sentiment_model.eval()
    with torch.no_grad():
        probability = torch.sigmoid(sentiment_model(token_ids)).item()

    label = "positive" if probability >= 0.5 else "negative"
    return label, probability

print(predict_sentiment("the movie was great and good"))
print(predict_sentiment("the movie was long and boring"))