PyTorch RNN Tutorial: Build from Scratch
Recurrent neural networks (RNNs) have revolutionized how we process sequential data in deep learning. From natural language processing to time series forecasting, these powerful architectures enable machines to understand patterns across time. In this comprehensive PyTorch RNN tutorial, you’ll learn how to build RNN models from the ground up, understand their architecture, and implement advanced variants like LSTM and GRU.
1. Understanding recurrent neural networks
What is RNN?
A recurrent neural network is a type of artificial neural network designed to recognize patterns in sequences of data. Unlike traditional feedforward networks that process inputs independently, RNNs maintain an internal state (memory) that allows them to process sequences of inputs. This makes them ideal for tasks like language modeling, speech recognition, and time series prediction.
The key innovation of RNNs lies in their ability to use information from previous time steps. Each neuron in an RNN receives input from both the current data point and the hidden state from the previous time step, creating a feedback loop that captures temporal dependencies.
RNN architecture basics
The fundamental RNN architecture consists of three main components:
- Input layer: Receives the current time step’s data
- Hidden layer: Maintains the network’s memory state
- Output layer: Produces predictions based on the hidden state
The mathematical formulation of a basic RNN cell is:
$$h_t = \tanh(W_{hh}h_{t-1} + W_{xh}x_t + b_h)$$
$$y_t = W_{hy}h_t + b_y$$
Where:
- \(h_t\) is the hidden state at time (t)
- \(x_t\) is the input at time (t)
- \(y_t\) is the output at time (t)
- \(W_{hh}\), \(W_{xh}\), and \(W_{hy}\) are weight matrices
- \(b_h\) and \(b_y\) are bias vectors
Why use PyTorch for RNN implementation?
PyTorch has become one of the most popular frameworks for implementing RNNs due to its intuitive API, dynamic computation graphs, and excellent debugging capabilities. The framework provides built-in RNN modules while also allowing you to create custom implementations, making it perfect for both beginners and researchers.
2. Setting up your environment
Before diving into RNN implementation, let’s set up the necessary tools and libraries:
import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import matplotlib.pyplot as plt
# Set random seed for reproducibility
torch.manual_seed(42)
np.random.seed(42)
# Check if GPU is available
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(f"Using device: {device}")
You’ll need PyTorch installed in your environment. The code above imports essential modules and sets up device configuration for GPU acceleration if available.
3. Building a basic RNN from scratch
Creating the RNN cell
Let’s implement a basic RNN cell from scratch to understand its inner workings:
class SimpleRNNCell(nn.Module):
def __init__(self, input_size, hidden_size):
super(SimpleRNNCell, self).__init__()
self.hidden_size = hidden_size
# Weight matrices
self.W_xh = nn.Linear(input_size, hidden_size)
self.W_hh = nn.Linear(hidden_size, hidden_size)
def forward(self, x, hidden):
"""
x: input at current time step (batch_size, input_size)
hidden: previous hidden state (batch_size, hidden_size)
"""
hidden = torch.tanh(self.W_xh(x) + self.W_hh(hidden))
return hidden
def init_hidden(self, batch_size):
"""Initialize hidden state with zeros"""
return torch.zeros(batch_size, self.hidden_size).to(device)
This implementation creates a single RNN cell that can process one time step. The forward method implements the core RNN equation we discussed earlier.
Building a complete RNN model
Now let’s create a full RNN model that processes entire sequences:
class BasicRNN(nn.Module):
def __init__(self, input_size, hidden_size, output_size, num_layers=1):
super(BasicRNN, self).__init__()
self.hidden_size = hidden_size
self.num_layers = num_layers
# RNN layer
self.rnn = nn.RNN(input_size, hidden_size, num_layers, batch_first=True)
# Output layer
self.fc = nn.Linear(hidden_size, output_size)
def forward(self, x, hidden=None):
"""
x: input sequence (batch_size, seq_length, input_size)
hidden: initial hidden state
"""
# If no hidden state provided, initialize with zeros
if hidden is None:
hidden = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device)
# Forward propagate through RNN
out, hidden = self.rnn(x, hidden)
# Decode the hidden state of the last time step
out = self.fc(out[:, -1, :])
return out, hidden
Training a simple sequence prediction task
Let’s train our RNN model on a simple task: predicting the next number in a sine wave sequence.
# Generate synthetic data
def generate_sine_wave(seq_length, num_samples):
"""Generate sine wave sequences for training"""
X = []
y = []
for _ in range(num_samples):
start = np.random.uniform(0, 2 * np.pi)
x_seq = np.sin(np.linspace(start, start + 2 * np.pi, seq_length + 1))
X.append(x_seq[:-1])
y.append(x_seq[-1])
return torch.FloatTensor(X).unsqueeze(-1), torch.FloatTensor(y)
# Prepare data
seq_length = 20
num_samples = 1000
X_train, y_train = generate_sine_wave(seq_length, num_samples)
X_train, y_train = X_train.to(device), y_train.to(device)
# Initialize model
input_size = 1
hidden_size = 32
output_size = 1
model = BasicRNN(input_size, hidden_size, output_size).to(device)
# Loss and optimizer
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)
# Training loop
num_epochs = 100
losses = []
for epoch in range(num_epochs):
model.train()
optimizer.zero_grad()
# Forward pass
outputs, _ = model(X_train)
loss = criterion(outputs.squeeze(), y_train)
# Backward pass and optimization
loss.backward()
optimizer.step()
losses.append(loss.item())
if (epoch + 1) % 10 == 0:
print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {loss.item():.4f}')
This example demonstrates the complete workflow of training an RNN model: data generation, model initialization, and the training loop with forward and backward passes.
4. Advanced RNN architectures
LSTM: Long Short-Term Memory networks
While basic RNNs are powerful, they struggle with long-term dependencies due to vanishing gradients. LSTM networks solve this problem with a more complex cell structure that includes gates to control information flow.
The LSTM cell equations are:
$$f_t = \sigma(W_f \cdot [h_{t-1}, x_t] + b_f)$$
$$i_t = \sigma(W_i \cdot [h_{t-1}, x_t] + b_i)$$
$$\tilde{C}_t = \tanh(W_C \cdot [h_{t-1}, x_t] + b_C)$$
$$C_t = f_t \cdot C_{t-1} + i_t \cdot \tilde{C}_t$$
$$o_t = \sigma(W_o \cdot [h_{t-1}, x_t] + b_o)$$
$$h_t = o_t \cdot \tanh(C_t)$$
Here’s how to implement an RNN LSTM model in PyTorch:
class LSTMModel(nn.Module):
def __init__(self, input_size, hidden_size, output_size, num_layers=1):
super(LSTMModel, self).__init__()
self.hidden_size = hidden_size
self.num_layers = num_layers
# LSTM layer
self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
# Output layer
self.fc = nn.Linear(hidden_size, output_size)
def forward(self, x):
# Initialize hidden and cell states
h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device)
c0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device)
# Forward propagate LSTM
out, _ = self.lstm(x, (h0, c0))
# Decode the hidden state of the last time step
out = self.fc(out[:, -1, :])
return out
GRU: Gated Recurrent Unit
The Gated Recurrent Unit is a simplified version of LSTM that uses fewer gates while maintaining similar performance. GRU combines the forget and input gates into a single update gate and merges the cell state with the hidden state.
class GRUModel(nn.Module):
def __init__(self, input_size, hidden_size, output_size, num_layers=1):
super(GRUModel, self).__init__()
self.hidden_size = hidden_size
self.num_layers = num_layers
# GRU layer
self.gru = nn.GRU(input_size, hidden_size, num_layers, batch_first=True)
# Output layer
self.fc = nn.Linear(hidden_size, output_size)
def forward(self, x):
# Initialize hidden state
h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device)
# Forward propagate GRU
out, _ = self.gru(x, h0)
# Decode the hidden state
out = self.fc(out[:, -1, :])
return out
Bidirectional RNN
A bidirectional RNN processes sequences in both forward and backward directions, allowing the network to have access to both past and future context. This is particularly useful for tasks like named entity recognition or machine translation.
class BidirectionalRNN(nn.Module):
def __init__(self, input_size, hidden_size, output_size, num_layers=1):
super(BidirectionalRNN, self).__init__()
self.hidden_size = hidden_size
self.num_layers = num_layers
# Bidirectional LSTM
self.lstm = nn.LSTM(input_size, hidden_size, num_layers,
batch_first=True, bidirectional=True)
# Output layer (note: hidden_size * 2 because of bidirectional)
self.fc = nn.Linear(hidden_size * 2, output_size)
def forward(self, x):
# Initialize hidden states for both directions
h0 = torch.zeros(self.num_layers * 2, x.size(0), self.hidden_size).to(x.device)
c0 = torch.zeros(self.num_layers * 2, x.size(0), self.hidden_size).to(x.device)
# Forward propagate
out, _ = self.lstm(x, (h0, c0))
# Decode
out = self.fc(out[:, -1, :])
return out
5. Real-world application: Text classification
Let’s apply our RNN knowledge to a practical text classification task. We’ll build a sentiment analysis model that classifies movie reviews as positive or negative.
Data preprocessing
# Example: Simple text preprocessing
class TextPreprocessor:
def __init__(self, max_vocab_size=10000):
self.max_vocab_size = max_vocab_size
self.word_to_idx = {}
self.idx_to_word = {}
def build_vocab(self, texts):
"""Build vocabulary from texts"""
word_freq = {}
for text in texts:
for word in text.lower().split():
word_freq[word] = word_freq.get(word, 0) + 1
# Sort by frequency and take top words
sorted_words = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)
vocab_words = [word for word, _ in sorted_words[:self.max_vocab_size - 2]]
# Build mappings (reserve 0 for padding, 1 for unknown)
self.word_to_idx = {word: idx + 2 for idx, word in enumerate(vocab_words)}
self.word_to_idx['<PAD>'] = 0
self.word_to_idx['<UNK>'] = 1
self.idx_to_word = {idx: word for word, idx in self.word_to_idx.items()}
def text_to_sequence(self, text, max_length=100):
"""Convert text to sequence of indices"""
words = text.lower().split()
sequence = [self.word_to_idx.get(word, 1) for word in words]
# Pad or truncate
if len(sequence) < max_length:
sequence += [0] * (max_length - len(sequence))
else:
sequence = sequence[:max_length]
return sequence
Text classification RNN model
class TextClassificationRNN(nn.Module):
def __init__(self, vocab_size, embedding_dim, hidden_size, num_classes, num_layers=2):
super(TextClassificationRNN, self).__init__()
# Embedding layer
self.embedding = nn.Embedding(vocab_size, embedding_dim, padding_idx=0)
# LSTM layer with dropout
self.lstm = nn.LSTM(embedding_dim, hidden_size, num_layers,
batch_first=True, dropout=0.3, bidirectional=True)
# Fully connected layers
self.fc1 = nn.Linear(hidden_size * 2, 64)
self.fc2 = nn.Linear(64, num_classes)
self.dropout = nn.Dropout(0.5)
self.relu = nn.ReLU()
def forward(self, x):
# Embedding
embedded = self.embedding(x)
# LSTM
lstm_out, _ = self.lstm(embedded)
# Global max pooling
pooled = torch.max(lstm_out, dim=1)[0]
# Fully connected layers
out = self.dropout(self.relu(self.fc1(pooled)))
out = self.fc2(out)
return out
# Example usage
vocab_size = 10000
embedding_dim = 128
hidden_size = 256
num_classes = 2
model = TextClassificationRNN(vocab_size, embedding_dim, hidden_size, num_classes).to(device)
# Example training setup
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)
6. Best practices and optimization tips
Handling vanishing and exploding gradients
Gradient problems are common in RNN deep learning. Here are techniques to address them:
# Gradient clipping
max_grad_norm = 5.0
torch.nn.utils.clip_grad_norm_(model.parameters(), max_grad_norm)
# Using LSTM or GRU instead of vanilla RNN
# Initialize with Xavier initialization
for name, param in model.named_parameters():
if 'weight' in name:
nn.init.xavier_uniform_(param)
elif 'bias' in name:
nn.init.constant_(param, 0.0)
Optimizing RNN performance
Several techniques can improve your RNN model’s performance:
Batch processing: Process multiple sequences simultaneously to leverage parallel computation.
# Create DataLoader for efficient batching
from torch.utils.data import TensorDataset, DataLoader
dataset = TensorDataset(X_train, y_train)
batch_size = 32
train_loader = DataLoader(dataset, batch_size=batch_size, shuffle=True)
# Training with batches
for epoch in range(num_epochs):
for batch_X, batch_y in train_loader:
optimizer.zero_grad()
outputs, _ = model(batch_X)
loss = criterion(outputs, batch_y)
loss.backward()
optimizer.step()
Learning rate scheduling: Adjust the learning rate during training for better convergence.
scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, mode='min',
factor=0.5, patience=5)
# In training loop
scheduler.step(validation_loss)
Dropout and regularization: Prevent overfitting by adding dropout layers.
self.lstm = nn.LSTM(input_size, hidden_size, num_layers,
batch_first=True, dropout=0.3)
Choosing the right RNN variant
Different tasks require different RNN architectures:
- Basic RNN: Simple sequences with short-term dependencies
- LSTM: Long sequences requiring long-term memory (text generation, machine translation)
- GRU: Similar to LSTM but faster training, good for most sequence tasks
- Bidirectional RNN: Tasks requiring both past and future context (named entity recognition, part-of-speech tagging)
7. Common pitfalls and debugging
Memory issues with long sequences
When working with very long sequences, you may encounter out-of-memory errors. Solutions include:
# Truncate sequences
max_sequence_length = 500
X = X[:, :max_sequence_length, :]
# Use gradient checkpointing for very deep networks
from torch.utils.checkpoint import checkpoint
# Process in smaller chunks
def process_long_sequence(model, sequence, chunk_size=100):
outputs = []
for i in range(0, len(sequence), chunk_size):
chunk = sequence[i:i+chunk_size]
output = model(chunk)
outputs.append(output)
return torch.cat(outputs, dim=0)
Debugging RNN training
Monitor these metrics to identify training issues:
def train_with_monitoring(model, train_loader, num_epochs):
train_losses = []
grad_norms = []
for epoch in range(num_epochs):
epoch_loss = 0
for batch_X, batch_y in train_loader:
optimizer.zero_grad()
outputs, _ = model(batch_X)
loss = criterion(outputs, batch_y)
loss.backward()
# Monitor gradient norms
total_norm = 0
for p in model.parameters():
if p.grad is not None:
param_norm = p.grad.data.norm(2)
total_norm += param_norm.item() ** 2
total_norm = total_norm ** 0.5
grad_norms.append(total_norm)
optimizer.step()
epoch_loss += loss.item()
avg_loss = epoch_loss / len(train_loader)
train_losses.append(avg_loss)
if (epoch + 1) % 10 == 0:
print(f'Epoch [{epoch+1}/{num_epochs}]')
print(f'Loss: {avg_loss:.4f}, Avg Gradient Norm: {np.mean(grad_norms[-len(train_loader):]):.4f}')
return train_losses, grad_norms
