xbe.at

Deep learning / RNN LSTM and sequence

Exploring GRU and LSTM in Python for Sequence Modeling

12 min read, 20 sections

Contents
  1. Introduction to Recurrent Neural Networks (RNNs)
  2. The Vanishing Gradient Problem
  3. Introduction to Gated Recurrent Unit (GRU)
  4. GRU Architecture
  5. GRU Update Gate
  6. GRU Reset Gate
  7. GRU Candidate State
  8. GRU Final Hidden State
  9. Introduction to Long Short-Term Memory (LSTM)
  10. LSTM Architecture
  11. LSTM Forget Gate
  12. LSTM Input Gate and Candidate State
  13. LSTM Cell State Update
  14. LSTM Output Gate and Hidden State
  15. Real-life Example: Sentiment Analysis with LSTM
  16. Response:
  17. Real-life Example: Sentiment Analysis with LSTM
  18. Real-life Example: Time Series Forecasting with GRU
  19. Comparing GRU and LSTM
  20. 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:

  1. Complexity: LSTM has four gates (input, output, forget, and cell state), while GRU has two gates (reset and update).
  2. Memory: LSTM has separate cell state and hidden state, while GRU combines them.
  3. Performance: GRU is generally faster to train and requires fewer parameters.
  4. 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:

  1. 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)
  2. 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
  3. 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.

More in RNN LSTM and sequence

  1. Exploring the Input Gate in LSTM Networks with Python10 min

    Long Short-Term Memory (LSTM) networks are a type of recurrent neural network designed to handle long-term dependencies in sequential data. The input gate is...

  2. Explaining LSTM Networks in Deep Learning with Python12 min

    Long Short-Term Memory (LSTM) networks are a type of recurrent neural network (RNN) designed to address the vanishing gradient problem in traditional RNNs....

  3. Forget Gate in LSTM Networks! A Deep Dive11 min

    Long Short-Term Memory (LSTM) networks are a type of recurrent neural network designed to handle long-term dependencies in sequential data. They are...

  4. Deep Recurrent Neural Networks with Keras and Python12 min

    Deep Recurrent Neural Networks (RNNs) are a class of neural networks designed to process sequential data. They have loops that allow information to persist...

  5. Gated Recurrent Units (GRUs) in Deep Learning with Python12 min

    Gated Recurrent Units (GRUs) are a type of recurrent neural network architecture designed to solve the vanishing gradient problem in traditional RNNs....

  6. Deep Recurrent Neural Networks Mastering Sequential Data12 min

    Deep Recurrent Neural Networks (RNNs) are advanced neural architectures designed to process sequential data by maintaining an internal state or 'memory'....