xbe.at

Deep learning / RNN LSTM and sequence

Implementing Recurrent Neural Networks in Python

12 min read, 15 sections

Contents
  1. Introduction to Recurrent Neural Networks (RNNs)
  2. RNN Architecture
  3. Implementing the Forward Pass
  4. Backpropagation Through Time (BPTT)
  5. Training the RNN
  6. Handling Vanishing and Exploding Gradients
  7. Real-Life Example: Text Generation
  8. Real-Life Example: Time Series Prediction
  9. Variations of RNNs: LSTM
  10. Variations of RNNs: GRU
  11. Bidirectional RNNs
  12. Attention Mechanism
  13. Practical Considerations and Best Practices
  14. Conclusion and Future Directions
  15. Additional Resources

Introduction to Recurrent Neural Networks (RNNs)

Recurrent Neural Networks are a class of neural networks designed to process sequential data. Unlike traditional feedforward networks, RNNs have loops that allow information to persist, making them ideal for tasks involving time series, natural language, or any data with temporal dependencies.

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))
        self.last_inputs = inputs
        self.last_hs = { 0: h }

        for i, x in enumerate(inputs):
            h = np.tanh(np.dot(self.Wxh, x) + np.dot(self.Whh, h) + self.bh)
            self.last_hs[i + 1] = h

        y = np.dot(self.Why, h) + self.by
        return y, h

RNN Architecture

The core idea behind RNNs is the ability to maintain a hidden state that acts as a memory of previous inputs. This hidden state is updated at each time step, allowing the network to learn and utilize patterns across long sequences.

import matplotlib.pyplot as plt
import networkx as nx

def draw_rnn():
    G = nx.DiGraph()
    G.add_edges_from([
        ('x_t', 'h_t'), ('h_t-1', 'h_t'), ('h_t', 'h_t+1'),
        ('h_t', 'y_t'), ('x_t-1', 'h_t-1'), ('h_t-1', 'y_t-1'),
        ('x_t+1', 'h_t+1'), ('h_t+1', 'y_t+1')
    ])
    pos = nx.spring_layout(G)
    nx.draw(G, pos, with_labels=True, node_color='lightblue', 
            node_size=3000, arrowsize=20)
    plt.title("RNN Architecture")
    plt.axis('off')
    plt.show()

draw_rnn()

Implementing the Forward Pass

The forward pass of an RNN involves processing each input in the sequence and updating the hidden state. We'll implement this using numpy for efficient matrix operations.

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

# Example usage
rnn = SimpleRNN(input_size=10, hidden_size=20, output_size=5)
inputs = [np.random.randn(10, 1) for _ in range(5)]  # 5 time steps
outputs, final_h = rnn.forward(inputs)

print(f"Number of outputs: {len(outputs)}")
print(f"Shape of final output: {outputs[-1].shape}")
print(f"Shape of final hidden state: {final_h.shape}")

Backpropagation Through Time (BPTT)

BPTT is the algorithm used to train RNNs. It involves unrolling the network through time and applying the chain rule to compute gradients. The process can be computationally expensive for long sequences.

def bptt(self, dh_next, cache, targets):
    x, h, y = cache
    dWxh, dWhh, dWhy = np.zeros_like(self.Wxh), np.zeros_like(self.Whh), np.zeros_like(self.Why)
    dbh, dby = np.zeros_like(self.bh), np.zeros_like(self.by)
    dh = np.zeros_like(h[0])

    for t in reversed(range(len(x))):
        dy = y[t] - targets[t]
        dWhy += np.dot(dy, h[t].T)
        dby += dy
        dh = np.dot(self.Why.T, dy) + dh_next
        dhraw = (1 - h[t] ** 2) * dh
        dbh += dhraw
        dWxh += np.dot(dhraw, x[t].T)
        dWhh += np.dot(dhraw, h[t-1].T)
        dh_next = np.dot(self.Whh.T, dhraw)

    return dWxh, dWhh, dWhy, dbh, dby, dh_next

# Note: This is a simplified version of BPTT. In practice, we often use truncated BPTT
# to limit the number of time steps we backpropagate through.

Training the RNN

Training an RNN involves iterating over the dataset, performing forward and backward passes, and updating the weights. We'll implement a simple training loop with gradient descent optimization.

def train(self, inputs, targets, learning_rate=0.01, epochs=100):
    for epoch in range(epochs):
        total_loss = 0
        h = np.zeros((self.hidden_size, 1))
        
        for i in range(len(inputs)):
            x, target = inputs[i], targets[i]
            y, h = self.forward(x, h)
            loss = np.sum((y - target) ** 2) / 2
            total_loss += loss
            
            # Backpropagation
            dy = y - target
            dWhy = np.dot(dy, h.T)
            dby = dy
            dh = np.dot(self.Why.T, dy)
            
            # Update weights
            self.Why -= learning_rate * dWhy
            self.by -= learning_rate * dby
            # ... (update other weights similarly)
        
        if epoch % 10 == 0:
            print(f"Epoch {epoch}, Loss: {total_loss}")

# Example usage
rnn = SimpleRNN(input_size=10, hidden_size=20, output_size=5)
inputs = [np.random.randn(10, 1) for _ in range(100)]
targets = [np.random.randn(5, 1) for _ in range(100)]
rnn.train(inputs, targets)

Handling Vanishing and Exploding Gradients

RNNs often struggle with long-term dependencies due to vanishing or exploding gradients. We can mitigate this using techniques like gradient clipping and careful weight initialization.

def clip_gradients(self, dWxh, dWhh, dWhy, dbh, dby, clip_value=5):
    for grad in [dWxh, dWhh, dWhy, dbh, dby]:
        np.clip(grad, -clip_value, clip_value, out=grad)

def initialize_weights(self):
    scale = 1.0 / np.sqrt(self.hidden_size)
    self.Wxh = np.random.randn(self.hidden_size, self.input_size) * scale
    self.Whh = np.random.randn(self.hidden_size, self.hidden_size) * scale
    self.Why = np.random.randn(self.output_size, self.hidden_size) * scale
    self.bh = np.zeros((self.hidden_size, 1))
    self.by = np.zeros((self.output_size, 1))

# Usage in training loop
dWxh, dWhh, dWhy, dbh, dby = self.bptt(dh_next, cache, targets)
self.clip_gradients(dWxh, dWhh, dWhy, dbh, dby)
# Then update weights...

Real-Life Example: Text Generation

RNNs are commonly used for text generation tasks. Let's implement a simple character-level language model that can generate text based on a given seed.

class CharRNN(SimpleRNN):
    def __init__(self, vocab_size, hidden_size):
        super().__init__(vocab_size, hidden_size, vocab_size)
        self.vocab_size = vocab_size

    def sample(self, seed, n_chars):
        h = np.zeros((self.hidden_size, 1))
        x = np.zeros((self.vocab_size, 1))
        x[seed] = 1
        generated = [seed]

        for _ in range(n_chars):
            h = np.tanh(np.dot(self.Wxh, x) + np.dot(self.Whh, h) + self.bh)
            y = np.dot(self.Why, h) + self.by
            p = np.exp(y) / np.sum(np.exp(y))
            ix = np.random.choice(range(self.vocab_size), p=p.ravel())
            x = np.zeros((self.vocab_size, 1))
            x[ix] = 1
            generated.append(ix)

        return generated

# Example usage
vocab = "abcdefghijklmnopqrstuvwxyz "
char_to_ix = {ch: i for i, ch in enumerate(vocab)}
ix_to_char = {i: ch for i, ch in enumerate(vocab)}

rnn = CharRNN(vocab_size=len(vocab), hidden_size=100)
# Assume the RNN has been trained on some text data

seed = char_to_ix['t']
generated_indices = rnn.sample(seed, n_chars=50)
generated_text = ''.join([ix_to_char[ix] for ix in generated_indices])
print("Generated text:", generated_text)

Real-Life Example: Time Series Prediction

RNNs are excellent for time series prediction tasks. Let's implement a simple RNN to predict future values in a time series.

import numpy as np
import matplotlib.pyplot as plt

class TimeSeriesRNN(SimpleRNN):
    def __init__(self, input_size, hidden_size, output_size, sequence_length):
        super().__init__(input_size, hidden_size, output_size)
        self.sequence_length = sequence_length

    def prepare_data(self, data):
        X, y = [], []
        for i in range(len(data) - self.sequence_length):
            X.append(data[i:i+self.sequence_length])
            y.append(data[i+self.sequence_length])
        return np.array(X), np.array(y)

    def predict(self, X):
        outputs, _ = self.forward(X)
        return outputs[-1]

# Generate sample time series data
t = np.linspace(0, 100, 1000)
data = np.sin(0.1 * t) + np.random.normal(0, 0.1, 1000)

rnn = TimeSeriesRNN(input_size=1, hidden_size=50, output_size=1, sequence_length=20)
X, y = rnn.prepare_data(data)

# Train the RNN (simplified)
for epoch in range(100):
    total_loss = 0
    for i in range(len(X)):
        outputs, _ = rnn.forward(X[i])
        loss = np.sum((outputs[-1] - y[i]) ** 2)
        total_loss += loss
        # Perform backpropagation and update weights (not shown for brevity)
    if epoch % 10 == 0:
        print(f"Epoch {epoch}, Loss: {total_loss}")

# Make predictions
predictions = [rnn.predict(X[i]) for i in range(len(X))]

plt.figure(figsize=(12, 6))
plt.plot(t[20:], data[20:], label='Actual')
plt.plot(t[20:], predictions, label='Predicted')
plt.legend()
plt.title('Time Series Prediction using RNN')
plt.xlabel('Time')
plt.ylabel('Value')
plt.show()

Variations of RNNs: LSTM

Long Short-Term Memory (LSTM) networks are a popular variant of RNNs designed to better capture long-term dependencies. They introduce gates to control the flow of information.

class LSTM:
    def __init__(self, input_size, hidden_size, output_size):
        self.hidden_size = hidden_size
        self.Wf = np.random.randn(hidden_size, input_size + hidden_size) * 0.01
        self.Wi = np.random.randn(hidden_size, input_size + hidden_size) * 0.01
        self.Wc = np.random.randn(hidden_size, input_size + hidden_size) * 0.01
        self.Wo = np.random.randn(hidden_size, input_size + hidden_size) * 0.01
        self.Wy = np.random.randn(output_size, hidden_size) * 0.01
        self.bf = np.zeros((hidden_size, 1))
        self.bi = np.zeros((hidden_size, 1))
        self.bc = np.zeros((hidden_size, 1))
        self.bo = np.zeros((hidden_size, 1))
        self.by = np.zeros((output_size, 1))

    def forward(self, x, h_prev, c_prev):
        z = np.row_stack((h_prev, x))
        f = sigmoid(np.dot(self.Wf, z) + self.bf)
        i = sigmoid(np.dot(self.Wi, z) + self.bi)
        c_bar = np.tanh(np.dot(self.Wc, z) + self.bc)
        c = f * c_prev + i * c_bar
        o = sigmoid(np.dot(self.Wo, z) + self.bo)
        h = o * np.tanh(c)
        y = np.dot(self.Wy, h) + self.by
        return y, h, c

def sigmoid(x):
    return 1 / (1 + np.exp(-x))

# Example usage
lstm = LSTM(input_size=10, hidden_size=20, output_size=5)
x = np.random.randn(10, 1)
h_prev = np.zeros((20, 1))
c_prev = np.zeros((20, 1))
y, h, c = lstm.forward(x, h_prev, c_prev)
print(f"Output shape: {y.shape}")
print(f"Hidden state shape: {h.shape}")
print(f"Cell state shape: {c.shape}")

Variations of RNNs: GRU

Gated Recurrent Units (GRUs) are another popular variant of RNNs. They use a simplified gating mechanism compared to LSTMs, making them computationally more efficient while still addressing the vanishing gradient problem.

class GRU:
    def __init__(self, input_size, hidden_size, output_size):
        self.hidden_size = hidden_size
        self.Wz = np.random.randn(hidden_size, input_size + hidden_size) * 0.01
        self.Wr = np.random.randn(hidden_size, input_size + hidden_size) * 0.01
        self.Wh = np.random.randn(hidden_size, input_size + hidden_size) * 0.01
        self.Wy = np.random.randn(output_size, hidden_size) * 0.01
        self.bz = np.zeros((hidden_size, 1))
        self.br = np.zeros((hidden_size, 1))
        self.bh = np.zeros((hidden_size, 1))
        self.by = np.zeros((output_size, 1))

    def forward(self, x, h_prev):
        z = np.row_stack((h_prev, x))
        update = sigmoid(np.dot(self.Wz, z) + self.bz)
        reset = sigmoid(np.dot(self.Wr, z) + self.br)
        h_candidate = np.tanh(np.dot(self.Wh, np.row_stack((reset * h_prev, x))) + self.bh)
        h = (1 - update) * h_prev + update * h_candidate
        y = np.dot(self.Wy, h) + self.by
        return y, h

# Helper function
def sigmoid(x):
    return 1 / (1 + np.exp(-x))

# Example usage
gru = GRU(input_size=10, hidden_size=20, output_size=5)
x = np.random.randn(10, 1)
h_prev = np.zeros((20, 1))
y, h = gru.forward(x, h_prev)
print(f"Output shape: {y.shape}")
print(f"Hidden state shape: {h.shape}")

Bidirectional RNNs

Bidirectional RNNs process input sequences in both forward and backward directions, allowing the network to capture context from both past and future states.

class BidirectionalRNN:
    def __init__(self, input_size, hidden_size, output_size):
        self.forward_rnn = SimpleRNN(input_size, hidden_size, output_size)
        self.backward_rnn = SimpleRNN(input_size, hidden_size, output_size)
        self.Wy = np.random.randn(output_size, 2 * hidden_size) * 0.01
        self.by = np.zeros((output_size, 1))

    def forward(self, inputs):
        forward_outputs, _ = self.forward_rnn.forward(inputs)
        backward_outputs, _ = self.backward_rnn.forward(inputs[::-1])
        
        combined_outputs = []
        for f, b in zip(forward_outputs, backward_outputs[::-1]):
            h = np.concatenate((f, b), axis=0)
            y = np.dot(self.Wy, h) + self.by
            combined_outputs.append(y)
        
        return combined_outputs

# Example usage
bi_rnn = BidirectionalRNN(input_size=10, hidden_size=20, output_size=5)
inputs = [np.random.randn(10, 1) for _ in range(5)]  # 5 time steps
outputs = bi_rnn.forward(inputs)
print(f"Number of outputs: {len(outputs)}")
print(f"Shape of each output: {outputs[0].shape}")

Attention Mechanism

Attention mechanisms allow RNNs to focus on different parts of the input sequence when producing each output, greatly improving performance on tasks like machine translation.

def attention(query, keys, values):
    # Simplified dot-product attention
    scores = np.dot(query.T, keys)
    attention_weights = softmax(scores)
    context = np.dot(values, attention_weights.T)
    return context, attention_weights

def softmax(x):
    exp_x = np.exp(x - np.max(x, axis=-1, keepdims=True))
    return exp_x / np.sum(exp_x, axis=-1, keepdims=True)

class AttentionRNN:
    def __init__(self, input_size, hidden_size, output_size):
        self.rnn = SimpleRNN(input_size, hidden_size, hidden_size)
        self.Wq = np.random.randn(hidden_size, hidden_size) * 0.01
        self.Wk = np.random.randn(hidden_size, hidden_size) * 0.01
        self.Wv = np.random.randn(hidden_size, hidden_size) * 0.01
        self.Wy = np.random.randn(output_size, hidden_size) * 0.01
        self.by = np.zeros((output_size, 1))

    def forward(self, inputs):
        hidden_states, _ = self.rnn.forward(inputs)
        outputs = []
        
        for t in range(len(inputs)):
            query = np.dot(self.Wq, hidden_states[t])
            keys = np.dot(self.Wk, np.hstack(hidden_states))
            values = np.dot(self.Wv, np.hstack(hidden_states))
            
            context, _ = attention(query, keys, values)
            output = np.dot(self.Wy, context) + self.by
            outputs.append(output)
        
        return outputs

# Example usage
attn_rnn = AttentionRNN(input_size=10, hidden_size=20, output_size=5)
inputs = [np.random.randn(10, 1) for _ in range(5)]  # 5 time steps
outputs = attn_rnn.forward(inputs)
print(f"Number of outputs: {len(outputs)}")
print(f"Shape of each output: {outputs[0].shape}")

Practical Considerations and Best Practices

When implementing RNNs from scratch, consider these best practices:

  1. Use gradient clipping to prevent exploding gradients.
  2. Initialize weights carefully to avoid vanishing gradients.
  3. Consider using more advanced architectures like LSTMs or GRUs for long sequences.
  4. Implement truncated backpropagation through time for long sequences.
  5. Use regularization techniques like dropout to prevent overfitting.
  6. Experiment with different optimizers, such as Adam or RMSprop.
  7. Monitor validation loss to avoid overfitting and implement early stopping.
  8. For large datasets, implement mini-batch training for efficiency.
def train_with_best_practices(rnn, inputs, targets, learning_rate=0.01, epochs=100, clip_value=5):
    for epoch in range(epochs):
        total_loss = 0
        h = np.zeros((rnn.hidden_size, 1))
        
        for i in range(len(inputs)):
            x, target = inputs[i], targets[i]
            y, h = rnn.forward(x, h)
            loss = np.sum((y - target) ** 2) / 2
            total_loss += loss
            
            # Backpropagation (simplified)
            grads = rnn.backward(x, h, y, target)
            
            # Gradient clipping
            for grad in grads:
                np.clip(grad, -clip_value, clip_value, out=grad)
            
            # Update weights (simplified)
            for param, grad in zip(rnn.params, grads):
                param -= learning_rate * grad
        
        if epoch % 10 == 0:
            print(f"Epoch {epoch}, Loss: {total_loss}")
            
        # Implement early stopping and learning rate decay here

Conclusion and Future Directions

Recurrent Neural Networks have revolutionized sequence modeling tasks, but they continue to evolve. Some future directions and advanced topics include:

  1. Transformer architectures, which have largely replaced traditional RNNs in many NLP tasks.
  2. Neural Turing Machines and Differentiable Neural Computers for more complex reasoning tasks.
  3. Combining RNNs with reinforcement learning for sequence decision-making problems.
  4. Exploring sparse attention mechanisms for improved efficiency and interpretability.
  5. Investigating continual learning techniques to allow RNNs to adapt to changing data distributions over time.

As the field of deep learning continues to advance, the principles behind RNNs remain fundamental to understanding and implementing sequence models.

Additional Resources

For those interested in diving deeper into RNNs and their implementations, here are some valuable resources:

  1. "Sequence Models" course by Andrew Ng on Coursera
  2. "The Unreasonable Effectiveness of Recurrent Neural Networks" by Andrej Karpathy
  3. "Understanding LSTM Networks" by Christopher Olah
  4. "Attention Is All You Need" paper (Vaswani et al., 2017) - ArXiv:1706.03762
  5. "On the difficulty of training Recurrent Neural Networks" paper (Pascanu et al., 2013) - ArXiv:1211.5063
  6. "Learning long-term dependencies with gradient descent is difficult" paper (Bengio et al., 1994) - IEEE Transactions on Neural Networks

Remember to verify these resources and their current availability, as the field of machine learning is rapidly evolving.

More in RNN LSTM and sequence

  1. LSTM and GRU Basics13 min

    Long Short-Term Memory (LSTM) and Gated Recurrent Unit (GRU) are advanced recurrent neural network architectures designed to handle sequential data and...

  2. 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...

  3. LSTM and GRU Deep Learning Architectures in Python13 min

    Long Short-Term Memory (LSTM) and Gated Recurrent Unit (GRU) are advanced recurrent neural network architectures designed to address the vanishing gradient...

  4. 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...

  5. LSTM for Time Series Forecasting in Python10 min

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

  6. Exploring Recurrent Neural Networks for Sequence Learning in Python15 min

    Recurrent Neural Networks (RNNs) are a class of neural networks designed to process sequential data. Unlike traditional feedforward networks, RNNs have...