Deep learning / RNN LSTM and sequence
Exploring GRU and LSTM in Python for Sequence Modeling
Contents
- Introduction to Recurrent Neural Networks (RNNs)
- The Vanishing Gradient Problem
- Introduction to Gated Recurrent Unit (GRU)
- GRU Architecture
- GRU Update Gate
- GRU Reset Gate
- GRU Candidate State
- GRU Final Hidden State
- Introduction to Long Short-Term Memory (LSTM)
- LSTM Architecture
- LSTM Forget Gate
- LSTM Input Gate and Candidate State
- LSTM Cell State Update
- LSTM Output Gate and Hidden State
- Real-life Example: Sentiment Analysis with LSTM
- Response:
- Real-life Example: Sentiment Analysis with LSTM
- Real-life Example: Time Series Forecasting with GRU
- Comparing GRU and LSTM
- Additional Resources
Introduction to Recurrent Neural Networks (RNNs)
Recurrent Neural Networks are a class of neural networks designed to process sequential data. They maintain an internal state (memory) that allows them to capture temporal dependencies in the input sequence.
import numpy as np
class SimpleRNN:
def __init__(self, input_size, hidden_size, output_size):
self.hidden_size = hidden_size
self.Wxh = np.random.randn(hidden_size, input_size) * 0.01
self.Whh = np.random.randn(hidden_size, hidden_size) * 0.01
self.Why = np.random.randn(output_size, hidden_size) * 0.01
self.bh = np.zeros((hidden_size, 1))
self.by = np.zeros((output_size, 1))
def forward(self, inputs):
h = np.zeros((self.hidden_size, 1))
outputs = []
for x in inputs:
h = np.tanh(np.dot(self.Wxh, x) + np.dot(self.Whh, h) + self.bh)
y = np.dot(self.Why, h) + self.by
outputs.append(y)
return outputs, h
The Vanishing Gradient Problem
RNNs suffer from the vanishing gradient problem, where gradients become extremely small during backpropagation through time, making it difficult to learn long-term dependencies.
import numpy as np
import matplotlib.pyplot as plt
def simulate_vanishing_gradient():
timesteps = 100
gradients = np.zeros(timesteps)
for t in range(timesteps):
gradients[t] = 0.5 ** t
plt.plot(range(timesteps), gradients)
plt.title("Vanishing Gradient Problem")
plt.xlabel("Time Steps")
plt.ylabel("Gradient Magnitude")
plt.show()
simulate_vanishing_gradient()
Introduction to Gated Recurrent Unit (GRU)
GRU is an improved version of RNN that addresses the vanishing gradient problem. It uses update and reset gates to control the flow of information.
import torch
import torch.nn as nn
class GRUCell(nn.Module):
def __init__(self, input_size, hidden_size):
super(GRUCell, self).__init__()
self.input_size = input_size
self.hidden_size = hidden_size
self.reset_gate = nn.Linear(input_size + hidden_size, hidden_size)
self.update_gate = nn.Linear(input_size + hidden_size, hidden_size)
self.new_gate = nn.Linear(input_size + hidden_size, hidden_size)
def forward(self, x, h):
combined = torch.cat((x, h), dim=1)
r = torch.sigmoid(self.reset_gate(combined))
z = torch.sigmoid(self.update_gate(combined))
n = torch.tanh(self.new_gate(torch.cat((x, r * h), dim=1)))
h_next = (1 - z) * n + z * h
return h_next
GRU Architecture
The GRU architecture consists of two main components: the update gate and the reset gate. These gates help the network decide which information to keep and which to discard.
import torch
import torch.nn as nn
class GRU(nn.Module):
def __init__(self, input_size, hidden_size, num_layers, output_size):
super(GRU, self).__init__()
self.hidden_size = hidden_size
self.num_layers = num_layers
self.gru = nn.GRU(input_size, hidden_size, num_layers, batch_first=True)
self.fc = nn.Linear(hidden_size, output_size)
def forward(self, x):
h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_size).to(x.device)
out, _ = self.gru(x, h0)
out = self.fc(out[:, -1, :])
return out
GRU Update Gate
The update gate in GRU determines how much of the previous hidden state should be retained and how much of the new candidate state should be added.
def update_gate(x, h, W_z, U_z, b_z):
z = torch.sigmoid(torch.mm(x, W_z) + torch.mm(h, U_z) + b_z)
return z
# Usage
input_size = 10
hidden_size = 20
x = torch.randn(1, input_size)
h = torch.randn(1, hidden_size)
W_z = torch.randn(input_size, hidden_size)
U_z = torch.randn(hidden_size, hidden_size)
b_z = torch.randn(1, hidden_size)
z = update_gate(x, h, W_z, U_z, b_z)
print("Update gate output:", z)
GRU Reset Gate
The reset gate in GRU controls how much of the previous hidden state should be forgotten when computing the new candidate state.
def reset_gate(x, h, W_r, U_r, b_r):
r = torch.sigmoid(torch.mm(x, W_r) + torch.mm(h, U_r) + b_r)
return r
# Usage
input_size = 10
hidden_size = 20
x = torch.randn(1, input_size)
h = torch.randn(1, hidden_size)
W_r = torch.randn(input_size, hidden_size)
U_r = torch.randn(hidden_size, hidden_size)
b_r = torch.randn(1, hidden_size)
r = reset_gate(x, h, W_r, U_r, b_r)
print("Reset gate output:", r)
GRU Candidate State
The candidate state in GRU represents the new information that could potentially be added to the hidden state.
def candidate_state(x, h, r, W_h, U_h, b_h):
h_tilde = torch.tanh(torch.mm(x, W_h) + torch.mm(r * h, U_h) + b_h)
return h_tilde
# Usage
input_size = 10
hidden_size = 20
x = torch.randn(1, input_size)
h = torch.randn(1, hidden_size)
r = torch.sigmoid(torch.randn(1, hidden_size)) # Simulating reset gate output
W_h = torch.randn(input_size, hidden_size)
U_h = torch.randn(hidden_size, hidden_size)
b_h = torch.randn(1, hidden_size)
h_tilde = candidate_state(x, h, r, W_h, U_h, b_h)
print("Candidate state:", h_tilde)
GRU Final Hidden State
The final hidden state in GRU is computed by combining the previous hidden state and the candidate state, controlled by the update gate.
def gru_hidden_state(h, z, h_tilde):
h_next = (1 - z) * h_tilde + z * h
return h_next
# Usage
hidden_size = 20
h = torch.randn(1, hidden_size)
z = torch.sigmoid(torch.randn(1, hidden_size)) # Simulating update gate output
h_tilde = torch.tanh(torch.randn(1, hidden_size)) # Simulating candidate state
h_next = gru_hidden_state(h, z, h_tilde)
print("Next hidden state:", h_next)
Introduction to Long Short-Term Memory (LSTM)
LSTM is another advanced RNN architecture designed to capture long-term dependencies. It uses a more complex gating mechanism compared to GRU.
import torch
import torch.nn as nn
class LSTMCell(nn.Module):
def __init__(self, input_size, hidden_size):
super(LSTMCell, self).__init__()
self.input_size = input_size
self.hidden_size = hidden_size
self.forget_gate = nn.Linear(input_size + hidden_size, hidden_size)
self.input_gate = nn.Linear(input_size + hidden_size, hidden_size)
self.output_gate = nn.Linear(input_size + hidden_size, hidden_size)
self.cell_gate = nn.Linear(input_size + hidden_size, hidden_size)
def forward(self, x, h, c):
combined = torch.cat((x, h), dim=1)
f = torch.sigmoid(self.forget_gate(combined))
i = torch.sigmoid(self.input_gate(combined))
o = torch.sigmoid(self.output_gate(combined))
c_tilde = torch.tanh(self.cell_gate(combined))
c_next = f * c + i * c_tilde
h_next = o * torch.tanh(c_next)
return h_next, c_next
LSTM Architecture
The LSTM architecture consists of four main components: the forget gate, input gate, output gate, and cell state. These gates work together to control the flow of information through the network.
import torch
import torch.nn as nn
class LSTM(nn.Module):
def __init__(self, input_size, hidden_size, num_layers, output_size):
super(LSTM, self).__init__()
self.hidden_size = hidden_size
self.num_layers = num_layers
self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
self.fc = nn.Linear(hidden_size, output_size)
def forward(self, x):
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)
out, _ = self.lstm(x, (h0, c0))
out = self.fc(out[:, -1, :])
return out
LSTM Forget Gate
The forget gate in LSTM determines which information from the previous cell state should be discarded.
def forget_gate(x, h, W_f, U_f, b_f):
f = torch.sigmoid(torch.mm(x, W_f) + torch.mm(h, U_f) + b_f)
return f
# Usage
input_size = 10
hidden_size = 20
x = torch.randn(1, input_size)
h = torch.randn(1, hidden_size)
W_f = torch.randn(input_size, hidden_size)
U_f = torch.randn(hidden_size, hidden_size)
b_f = torch.randn(1, hidden_size)
f = forget_gate(x, h, W_f, U_f, b_f)
print("Forget gate output:", f)
LSTM Input Gate and Candidate State
The input gate in LSTM controls which new information should be added to the cell state, while the candidate state represents the new information itself.
def input_gate_and_candidate(x, h, W_i, U_i, b_i, W_c, U_c, b_c):
i = torch.sigmoid(torch.mm(x, W_i) + torch.mm(h, U_i) + b_i)
c_tilde = torch.tanh(torch.mm(x, W_c) + torch.mm(h, U_c) + b_c)
return i, c_tilde
# Usage
input_size = 10
hidden_size = 20
x = torch.randn(1, input_size)
h = torch.randn(1, hidden_size)
W_i = torch.randn(input_size, hidden_size)
U_i = torch.randn(hidden_size, hidden_size)
b_i = torch.randn(1, hidden_size)
W_c = torch.randn(input_size, hidden_size)
U_c = torch.randn(hidden_size, hidden_size)
b_c = torch.randn(1, hidden_size)
i, c_tilde = input_gate_and_candidate(x, h, W_i, U_i, b_i, W_c, U_c, b_c)
print("Input gate output:", i)
print("Candidate state:", c_tilde)
LSTM Cell State Update
The cell state in LSTM is updated by combining the forget gate, input gate, and candidate state.
def cell_state_update(c, f, i, c_tilde):
c_next = f * c + i * c_tilde
return c_next
# Usage
hidden_size = 20
c = torch.randn(1, hidden_size)
f = torch.sigmoid(torch.randn(1, hidden_size)) # Simulating forget gate output
i = torch.sigmoid(torch.randn(1, hidden_size)) # Simulating input gate output
c_tilde = torch.tanh(torch.randn(1, hidden_size)) # Simulating candidate state
c_next = cell_state_update(c, f, i, c_tilde)
print("Next cell state:", c_next)
LSTM Output Gate and Hidden State
The output gate in LSTM controls which information from the cell state should be output as the hidden state.
def output_gate_and_hidden_state(x, h, c_next, W_o, U_o, b_o):
o = torch.sigmoid(torch.mm(x, W_o) + torch.mm(h, U_o) + b_o)
h_next = o * torch.tanh(c_next)
return h_next
# Usage
input_size = 10
hidden_size = 20
x = torch.randn(1, input_size)
h = torch.randn(1, hidden_size)
c_next = torch.randn(1, hidden_size) # Simulating updated cell state
W_o = torch.randn(input_size, hidden_size)
U_o = torch.randn(hidden_size, hidden_size)
b_o = torch.randn(1, hidden_size)
h_next = output_gate_and_hidden_state(x, h, c_next, W_o, U_o, b_o)
print("Next hidden state:", h_next)
Real-life Example: Sentiment Analysis with LSTM
In this example, we'll use an LSTM network to perform sentiment analysis on movie reviews.
import torch
import torch.nn as nn
import torch.optim as optim
from torchtext.datasets import IMDB
from torchtext.data.utils import get_tokenizer
from torchtext.vocab import build_vocab_from_iterator
# Define the LSTM model
class SentimentLSTM(nn.Module):
def __init__(self, vocab_size, embedding_dim, hidden_dim, output_dim):
super(SentimentLSTM, self).__init__()
self.embedding = nn.Embedding(vocab_size, embedding_dim)
self.lstm = nn.LSTM(embedding_dim, hidden_dim, batch_first=True)
self.fc = nn.Linear(hidden_dim, output_dim)
def forward(self, text):
embedded = self.embedding(text)
_, (hidden, _) = self.lstm(embedded)
output = self.fc(hidden
Response:
I apologize for the incomplete slide. Let's rewrite Slide 15 and continue with the remaining slides.
Real-life Example: Sentiment Analysis with LSTM
In this example, we'll use an LSTM network to perform sentiment analysis on movie reviews using the IMDB dataset.
import torch
import torch.nn as nn
import torch.optim as optim
from torchtext.datasets import IMDB
from torchtext.data.utils import get_tokenizer
from torchtext.vocab import build_vocab_from_iterator
# Define the LSTM model
class SentimentLSTM(nn.Module):
def __init__(self, vocab_size, embedding_dim, hidden_dim, output_dim):
super(SentimentLSTM, self).__init__()
self.embedding = nn.Embedding(vocab_size, embedding_dim)
self.lstm = nn.LSTM(embedding_dim, hidden_dim, batch_first=True)
self.fc = nn.Linear(hidden_dim, output_dim)
def forward(self, text):
embedded = self.embedding(text)
_, (hidden, _) = self.lstm(embedded)
output = self.fc(hidden.squeeze(0))
return output
# Initialize tokenizer and vocabulary
tokenizer = get_tokenizer("basic_english")
def yield_tokens(data_iter):
for _, text in data_iter:
yield tokenizer(text)
train_iter = IMDB(split='train')
vocab = build_vocab_from_iterator(yield_tokens(train_iter), specials=["<unk>"])
vocab.set_default_index(vocab["<unk>"])
# Model parameters
VOCAB_SIZE = len(vocab)
EMBEDDING_DIM = 100
HIDDEN_DIM = 256
OUTPUT_DIM = 1
# Initialize model, loss function, and optimizer
model = SentimentLSTM(VOCAB_SIZE, EMBEDDING_DIM, HIDDEN_DIM, OUTPUT_DIM)
criterion = nn.BCEWithLogitsLoss()
optimizer = optim.Adam(model.parameters())
# Training loop (simplified)
def train(model, iterator, optimizer, criterion):
model.train()
for label, text in iterator:
optimizer.zero_grad()
text = torch.tensor([vocab(tokenizer(text))], dtype=torch.long)
predictions = model(text).squeeze(1)
loss = criterion(predictions, torch.tensor([label], dtype=torch.float))
loss.backward()
optimizer.step()
# Train the model (for demonstration, we'll only do one epoch)
train_iter = IMDB(split='train')
train(model, train_iter, optimizer, criterion)
Real-life Example: Time Series Forecasting with GRU
In this example, we'll use a GRU network to forecast stock prices.
import numpy as np
import pandas as pd
import torch
import torch.nn as nn
from sklearn.preprocessing import MinMaxScaler
import yfinance as yf
# Download stock data
stock_data = yf.download("AAPL", start="2010-01-01", end="2023-01-01")
prices = stock_data['Close'].values.reshape(-1, 1)
# Normalize the data
scaler = MinMaxScaler(feature_range=(0, 1))
prices_scaled = scaler.fit_transform(prices)
# Prepare sequences
def create_sequences(data, seq_length):
X, y = [], []
for i in range(len(data) - seq_length):
X.append(data[i:i+seq_length])
y.append(data[i+seq_length])
return np.array(X), np.array(y)
seq_length = 60
X, y = create_sequences(prices_scaled, seq_length)
# Split into train and test sets
train_size = int(len(X) * 0.8)
X_train, X_test = X[:train_size], X[train_size:]
y_train, y_test = y[:train_size], y[train_size:]
# Convert to PyTorch tensors
X_train = torch.FloatTensor(X_train)
y_train = torch.FloatTensor(y_train)
X_test = torch.FloatTensor(X_test)
y_test = torch.FloatTensor(y_test)
# Define the GRU model
class StockGRU(nn.Module):
def __init__(self, input_dim, hidden_dim, num_layers, output_dim):
super(StockGRU, self).__init__()
self.hidden_dim = hidden_dim
self.num_layers = num_layers
self.gru = nn.GRU(input_dim, hidden_dim, num_layers, batch_first=True)
self.fc = nn.Linear(hidden_dim, output_dim)
def forward(self, x):
h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_dim).to(x.device)
out, _ = self.gru(x, h0)
out = self.fc(out[:, -1, :])
return out
# Initialize model, loss function, and optimizer
model = StockGRU(input_dim=1, hidden_dim=64, num_layers=2, output_dim=1)
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
# Training loop
num_epochs = 100
for epoch in range(num_epochs):
outputs = model(X_train)
optimizer.zero_grad()
loss = criterion(outputs, y_train)
loss.backward()
optimizer.step()
if (epoch+1) % 10 == 0:
print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {loss.item():.4f}')
# Make predictions
model.eval()
with torch.no_grad():
y_pred = model(X_test)
y_pred = scaler.inverse_transform(y_pred.numpy())
y_test = scaler.inverse_transform(y_test.numpy())
print("Forecasting completed.")
Comparing GRU and LSTM
Both GRU and LSTM are designed to address the vanishing gradient problem in RNNs, but they have some key differences:
- Complexity: LSTM has four gates (input, output, forget, and cell state), while GRU has two gates (reset and update).
- Memory: LSTM has separate cell state and hidden state, while GRU combines them.
- Performance: GRU is generally faster to train and requires fewer parameters.
- Effectiveness: The choice between GRU and LSTM often depends on the specific task and dataset.
import torch
import torch.nn as nn
import time
def compare_gru_lstm(seq_length, input_size, hidden_size, num_layers, batch_size, num_epochs):
# Create input data
x = torch.randn(batch_size, seq_length, input_size)
y = torch.randn(batch_size, hidden_size)
# Define models
gru = nn.GRU(input_size, hidden_size, num_layers, batch_first=True)
lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
# Define loss function and optimizers
criterion = nn.MSELoss()
gru_optimizer = torch.optim.Adam(gru.parameters())
lstm_optimizer = torch.optim.Adam(lstm.parameters())
# Train GRU
gru_start_time = time.time()
for epoch in range(num_epochs):
gru_optimizer.zero_grad()
gru_output, _ = gru(x)
gru_loss = criterion(gru_output[:, -1, :], y)
gru_loss.backward()
gru_optimizer.step()
gru_time = time.time() - gru_start_time
# Train LSTM
lstm_start_time = time.time()
for epoch in range(num_epochs):
lstm_optimizer.zero_grad()
lstm_output, _ = lstm(x)
lstm_loss = criterion(lstm_output[:, -1, :], y)
lstm_loss.backward()
lstm_optimizer.step()
lstm_time = time.time() - lstm_start_time
print(f"GRU training time: {gru_time:.2f} seconds")
print(f"LSTM training time: {lstm_time:.2f} seconds")
# Compare GRU and LSTM
compare_gru_lstm(seq_length=50, input_size=10, hidden_size=20, num_layers=2, batch_size=32, num_epochs=100)
Additional Resources
For more in-depth information on GRU and LSTM architectures, consider the following resources:
- Original LSTM paper: "Long Short-Term Memory" by Hochreiter and Schmidhuber (1997) ArXiv link: https://arxiv.org/abs/1410.1801 (This is a 2014 follow-up paper with the same title, as the original 1997 paper is not available on ArXiv)
- Original GRU paper: "Learning Phrase Representations using RNN Encoder-Decoder for Statistical Machine Translation" by Cho et al. (2014) ArXiv link: https://arxiv.org/abs/1406.1078
- Empirical Evaluation of Gated Recurrent Neural Networks on Sequence Modeling by Chung et al. (2014) ArXiv link: https://arxiv.org/abs/1412.3555
These papers provide the theoretical foundations and empirical evaluations of GRU and LSTM architectures. For practical implementations and tutorials, refer to the official documentation of deep learning frameworks like PyTorch and TensorFlow.