Recurrent Neural Networks (RNN)

The Story That Explains Recurrent Neural Networks

An RNN is a type of neural network designed for sequential data — data where the order and history of inputs matters. Unlike a standard network that maps one input to one output independently, an RNN loops: it feeds its output (a hidden state) back into itself at the next time step, building a running memory of the sequence it has seen so far.

Why a Standard Neural Network Can't Do This

The RNN Architecture — Animated

The magic of an RNN is its recurrent connection: the hidden state h at each time step depends on both the current input and the previous hidden state. Below is an animated diagram of information flowing through an RNN.

The Mathematics — Animated Equations

The equations of a vanilla RNN are surprisingly compact. Two lines define everything that happens inside a single cell at time step t.

The Four RNN Input/Output Patterns

RNNs are flexible about how they consume and produce data. The same core architecture supports four distinct sequence patterns, each suited to different tasks.

The Vanishing Gradient Problem

LSTM — Long Short-Term Memory

GRU — Gated Recurrent Unit

Introduced by Cho et al. in 2014, the GRU is a streamlined LSTM. It merges the forget and input gates into a single update gate, and eliminates the separate cell state — the hidden state is the memory. The result: fewer parameters, faster training, often comparable performance.

Backpropagation Through Time (BPTT)

Training an RNN requires computing gradients across time. Because the same weights W are used at every step, the chain rule must unroll backward through all the time steps — this is Backpropagation Through Time (BPTT).

Python — Vanilla RNN from Scratch (NumPy)

Before using PyTorch's built-in RNN, let's build a one-step RNN cell entirely in NumPy to demystify the maths. This is a character-level model trained on a tiny corpus.

import numpy as np

# ── Hyperparameters ────────────────────────────────────────────
hidden_size  = 64    # Number of hidden units
seq_length   = 25    # Truncated BPTT length
learning_rate = 1e-1

# ── Tiny corpus ────────────────────────────────────────────────
data = "hello world, this is a recurrent neural network tutorial"
chars = list(set(data))
vocab_size = len(chars)
char_to_ix = {ch: i for i, ch in enumerate(chars)}
ix_to_char = {i: ch for i, ch in enumerate(chars)}

# ── Weight initialisation (Xavier-ish) ────────────────────────
Wx = np.random.randn(hidden_size, vocab_size) * 0.01   # Input → hidden
Wh = np.random.randn(hidden_size, hidden_size) * 0.01  # Hidden → hidden
Wy = np.random.randn(vocab_size, hidden_size) * 0.01   # Hidden → output
bh = np.zeros((hidden_size, 1))                          # Hidden bias
by = np.zeros((vocab_size, 1))                           # Output bias

def forward(inputs, targets, hprev):
    xs, hs, ys, ps = {}, {}, {}, {}
    hs[-1] = np.copy(hprev)
    loss = 0

    # ── Forward pass ──────────────────────────────────────────────
    for t in range(len(inputs)):
        xs[t] = np.zeros((vocab_size, 1))
        xs[t][inputs[t]] = 1                          # One-hot encode
        hs[t] = np.tanh(Wx @ xs[t] + Wh @ hs[t-1] + bh)  # Hidden state
        ys[t] = Wy @ hs[t] + by                       # Logits
        ps[t] = np.exp(ys[t]) / np.sum(np.exp(ys[t])) # Softmax
        loss += -np.log(ps[t][targets[t], 0])        # Cross-entropy

    # ── Backward pass (BPTT) ──────────────────────────────────────
    dWx, dWh, dWy = np.zeros_like(Wx), np.zeros_like(Wh), np.zeros_like(Wy)
    dbh, dby = np.zeros_like(bh), np.zeros_like(by)
    dhnext = np.zeros_like(hs[0])

    for t in reversed(range(len(inputs))):
        dy = np.copy(ps[t])
        dy[targets[t]] -= 1            # dL/dy (softmax + CE gradient)
        dWy += dy @ hs[t].T
        dby += dy
        dh = Wy.T @ dy + dhnext         # Backprop into hidden state
        dhraw = (1 - hs[t] ** 2) * dh  # tanh derivative
        dbh += dhraw
        dWx += dhraw @ xs[t].T
        dWh += dhraw @ hs[t-1].T
        dhnext = Wh.T @ dhraw

    # Gradient clipping — prevents exploding gradients
    for dparam in [dWx, dWh, dWy, dbh, dby]:
        np.clip(dparam, -5, 5, out=dparam)

    return loss, dWx, dWh, dWy, dbh, dby, hs[len(inputs)-1]

Python — LSTM for Sentiment Analysis (PyTorch)

import torch
import torch.nn as nn
from torch.utils.data import DataLoader, Dataset
import numpy as np

# ── 1. Define the LSTM Model ───────────────────────────────────
class SentimentLSTM(nn.Module):
    def __init__(self, vocab_size, embed_dim, hidden_dim, n_layers, dropout=0.3):
        super().__init__()
        # Word → dense vector
        self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0)
        # Stacked LSTM layers
        self.lstm = nn.LSTM(
            input_size=embed_dim,
            hidden_size=hidden_dim,
            num_layers=n_layers,
            batch_first=True,       # (batch, seq, feature)
            dropout=dropout if n_layers > 1 else 0.0,
            bidirectional=False     # Use True for Bi-LSTM
        )
        self.dropout = nn.Dropout(dropout)
        self.fc      = nn.Linear(hidden_dim, 1)   # Binary output
        self.sigmoid = nn.Sigmoid()

    def forward(self, x):
        emb = self.dropout(self.embedding(x))      # (B, T, E)
        out, (hn, cn) = self.lstm(emb)             # out: (B, T, H)
        # Use ONLY the last time step's hidden state
        last_hidden = self.dropout(hn[-1])          # (B, H)
        logit = self.fc(last_hidden)                # (B, 1)
        return self.sigmoid(logit).squeeze(1)      # (B,)

# ── 2. Instantiate and inspect ─────────────────────────────────
VOCAB_SIZE  = 10_000
EMBED_DIM   = 128
HIDDEN_DIM  = 256
N_LAYERS    = 2
BATCH_SIZE  = 64

model = SentimentLSTM(VOCAB_SIZE, EMBED_DIM, HIDDEN_DIM, N_LAYERS)
print(model)

# Count trainable parameters
total_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
print(f"Trainable parameters: {total_params:,}")

# ── 3. Training loop ───────────────────────────────────────────
criterion = nn.BCELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)

def train_epoch(model, loader, optimizer, criterion, device):
    model.train()
    total_loss, correct = 0, 0
    for texts, labels in loader:
        texts, labels = texts.to(device), labels.float().to(device)
        optimizer.zero_grad()
        preds = model(texts)
        loss  = criterion(preds, labels)
        loss.backward()
        # ── Gradient clipping — critical for RNN/LSTM stability
        nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
        optimizer.step()
        total_loss += loss.item()
        correct += ((preds > 0.5).float() == labels).sum().item()
    n = len(loader.dataset)
    return total_loss / len(loader), correct / n

# ── 4. Inference on new text ───────────────────────────────────
def predict_sentiment(model, tokenized_text, word2idx, device, max_len=200):
    model.eval()
    idxs = [word2idx.get(w, 1) for w in tokenized_text[:max_len]]
    idxs += [0] * (max_len - len(idxs))   # Pad to fixed length
    x = torch.tensor([idxs], dtype=torch.long).to(device)
    with torch.no_grad():
        prob = model(x).item()
    return "Positive 😊" if prob > 0.5 else "Negative 😞", prob

Python — GRU for Stock Price Forecasting

import torch
import torch.nn as nn
import numpy as np
from sklearn.preprocessing import MinMaxScaler

# ── GRU Model ──────────────────────────────────────────────────
class StockGRU(nn.Module):
    def __init__(self, input_features, hidden_dim, n_layers, output_size=1):
        super().__init__()
        self.gru = nn.GRU(
            input_size=input_features,
            hidden_size=hidden_dim,
            num_layers=n_layers,
            batch_first=True,
            dropout=0.2 if n_layers > 1 else 0
        )
        self.fc = nn.Linear(hidden_dim, output_size)

    def forward(self, x):
        # x shape: (batch, seq_len=60, features=5)
        gru_out, hn = self.gru(x)           # gru_out: (B, 60, H)
        last = gru_out[:, -1, :]            # Take last time step (B, H)
        return self.fc(last)               # (B, 1) — tomorrow's price

# ── Data preparation: sliding window ──────────────────────────
def create_sequences(data, seq_len=60):
    """Slide a window of length seq_len over the data."""
    X, y = [], []
    for i in range(len(data) - seq_len):
        X.append(data[i : i + seq_len])
        y.append(data[i + seq_len, 0])   # Predict closing price (col 0)
    return np.array(X), np.array(y)

# ── Simulate some stock data (5 features per day) ─────────────
np.random.seed(42)
raw_data = np.random.randn(500, 5)  # close, open, high, low, volume

scaler = MinMaxScaler()
scaled = scaler.fit_transform(raw_data)
X, y   = create_sequences(scaled, seq_len=60)

# ── Train/test split (no shuffle — time series!) ──────────────
split = int(len(X) * 0.8)
X_train, X_test = X[:split], X[split:]
y_train, y_test = y[:split], y[split:]

X_train = torch.FloatTensor(X_train)
X_test  = torch.FloatTensor(X_test)
y_train = torch.FloatTensor(y_train).unsqueeze(1)
y_test  = torch.FloatTensor(y_test).unsqueeze(1)

# ── Instantiate and train ──────────────────────────────────────
model     = StockGRU(input_features=5, hidden_dim=128, n_layers=2)
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)

for epoch in range(50):
    model.train()
    optimizer.zero_grad()
    pred = model(X_train)
    loss = criterion(pred, y_train)
    loss.backward()
    nn.utils.clip_grad_norm_(model.parameters(), 1.0)
    optimizer.step()
    if epoch % 10 == 0:
        print(f"Epoch {epoch:3d} | Train Loss: {loss.item():.6f}")

Bidirectional RNNs

# Bidirectional LSTM — a 2-line change from the standard LSTM
class BiLSTMClassifier(nn.Module):
    def __init__(self, vocab_size, embed_dim, hidden_dim, n_classes):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.lstm = nn.LSTM(
            embed_dim, hidden_dim,
            num_layers=2,
            batch_first=True,
            bidirectional=True   # ← key change
        )
        # hidden_dim * 2 because forward + backward states are concatenated
        self.fc = nn.Linear(hidden_dim * 2, n_classes)

    def forward(self, x):
        emb = self.embedding(x)            # (B, T, E)
        out, (hn, _) = self.lstm(emb)      # hn: (2*layers, B, H)
        # Concatenate last forward and last backward hidden state
        fwd = hn[-2, :, :]                 # Forward last layer
        bwd = hn[-1, :, :]                 # Backward last layer
        combined = torch.cat([fwd, bwd], dim=1)  # (B, 2H)
        return self.fc(combined)           # (B, n_classes)

Real-World Applications of RNNs

RNN vs. Transformer — What Changed

Since 2017 and the "Attention Is All You Need" paper, Transformers have largely replaced RNNs for NLP tasks. Understanding why helps you choose the right tool.

Golden Rules