Come creare una RNN da zero con PyTorch

Obiettivo

Proviamo a creare una Recurrent Neural Network (RNN) che, dato un carattere tra a,b,c,d, lo continui secondo il pattern: 'aaaaabbbbbccccacdddddaaaa'

Importiamo le librerie:

import torch 
import torch.nn as nn 
import torch.optim as optim

sequence = 'aaaaabbbbbccccacdddddaaaa'

• torch: il core di PyTorch, per operazioni tensoriali e autograd.
• torch.nn: modulo per costruire reti neurali in PyTorch.
• torch.optim: modulo per gli ottimizzatori, usato per aggiornare i pesi della rete durante l'addestramento.

Mappatura Caratteri-indici:

# Funzioni di utilità per convertire da carattere a intero e viceversa

chars = list(set(sequence))
char_to_ix = {ch: i for i, ch in enumerate(chars)} # {'b': 0, 's': 1, 'c': 2, 'd': 3, 'a': 4}
ix_to_char = {i: ch for i, ch in enumerate(chars)} # {0: 'b', 1: 's', 2: 'c', 3: 'd', 4: 'a'}

Creazione di due dizionari per convertire i caratteri in indici e viceversa.

Questo è utile per manipolare i caratteri in forma numerica, facilitando le operazioni matematiche sui dati. char_to_ix mappa ogni carattere unico a un indice intero univoco, e ix_to_char fa l'opposto.

Funzione di Conversione in One-Hot

Creiamo dei Tensori per codificare i caratteri sia di input che di addestramento in one-hot encodings:

def char_to_onehot(char):
   tensor = torch.zeros(len(chars))
   tensor[char_to_ix[char]] = 1
   return tensor

tensor([[0., 0., 0., 0., 1.],

       [0., 0., 0., 0., 1.],

       [0., 0., 0., 0., 1.],

Quando si usa CrossEntropyLoss di PyTorch, l'output atteso è l'indice della classe target, non la rappresentazione one-hot.

CrossEntropyLoss combina LogSoftmax e NLLLoss (Negative Log Likelihood Loss) in un'unica classe. Questa funzione si aspetta logits non normalizzati come input (cioè, l'output grezzo del modello prima dell'applicazione di una funzione di attivazione come Softmax e l'indice della classe come target. PyTorch gestisce internamente la conversione degli indici in una rappresentazione one-hot e l'applicazione della Softmax per calcolare la perdita, il che rende l'uso di CrossEntropyLoss più efficiente dal punto di vista computazionale rispetto al calcolo manuale di Softmax seguito da una funzione di perdita come NLLLoss su output one-hot.

   # INIZIALIZZAZIONE DELL'ARCHITETTURA
   def __init__(self, input_size, hidden_size, output_size):
       super(RNN, self).__init__()
       self.hidden_size = hidden_size

       # La funzione è:
       # torch.nn.RNN(self, input_size, hidden_size, num_layers=1, 
       # nonlinearity='tanh', bias=True, batch_first=False, dropout=0.0,
       # bidirectional=False, device=None, dtype=None)
       #
       # input size è la dimensione dell'input
       # hidden size la dimensione del layer "nascosto". Piu grande è più c'è 
       # spazio di apprendimento.
       #
       # batch first true se diamo l'input e outpur nella forma: (seq_len, batch_size, features)
       # batch first false se abbiamo l'input e output come (batch_size, seq_len, features)
       
       self.rnn = nn.RNN(input_size, hidden_size, batch_first=True)

       # Layer totalmente connesso
       self.fc = nn.Linear(hidden_size, output_size)

   def forward(self, x, hidden):
       # hidden lo aspettiamo sempre come (batch_size, seq_len, features) che in questo caso "unario"
       # è [1,1,hidden_size]
       #
       # X ce lo aspettiamo egualmente come (batch_size, seq_len, input_size) che diventa [1,1,one-hot-size]
       # nel caso di batch 1 e seq 1
       out, hidden = self.rnn(x, hidden)

       # passiamo direttamente a un layer totalmente connesso che darà in output i LOGIT
       # siccome per ogni sequenza (che qui è da "uno") vogliamo sempre prendere l'ultimo step 
       # facciamo quest'operazione di slicing.
       # ricordando che out ritorna il formato (batch_size, seq_len, output_size (onr-hot size)
       # out[:, -1, :] significa:
       # - : per ogni esempio nel batch (qui è da uno)
       # - prendi l'ultimo output emesso dalla RNN (-1)
       # - in tutte le sue componenti/faetures (: output vector size - one hot size)
       
       out = self.fc(out[:, -1, :])  

       # Out contiene ora i LOGIT in formato: (batch_size, output_size) 
       # ad esempio -0.3382,  0.2435, -0.4452,  0.4239,  0.1288
       return out, hidden

   def init_hidden(self):
       # hidden ha formato (batch_size, seq_len, dimensione hidden), qui (1,1,hidden)
       return torch.zeros(1, 1, self.hidden_size)

1. Model instantiation

n_hidden = 128 model = RNN(len(chars), n_hidden, len(chars))

1. Loss and optimizer

criterion = nn.CrossEntropyLoss() optimizer = optim.Adam(model.parameters(), lr=0.005)

1. Addestramento

In PyTorch, per ogni epoch (o mini batch) durante la fase di addestramento, vogliamo azzerare esplicitamente i gradienti prima di iniziare a fare la backpropagation (cioè, aggiornare i pesi e i bias) perché PyTorch accumula i gradienti su passaggi backward successivi.

__l'azione predefinita è stata impostata per accumulare (cioè sommare) i gradienti ad ogni chiamata di loss.backward()__.

Per questa ragione, quando inizi il tuo ciclo di addestramento, dovresti __azzerare i gradienti__ in modo che si possa fare aggiornare dei parametri correttamente. Altrimenti il gradiente sarebbe una combinazione del vecchio gradiente, che hai già utilizzato per aggiornare i parametri del tuo modello, e del nuovo gradiente calcolato. Di conseguenza, questo punterebbe in una direzione diversa da quella intesa verso il minimo (o massimo, in caso di obiettivi di massimizzazione).

1. Training the model

for epoch in range(1000):

   # torch.Size([1, 1, 128])
   hidden = model.init_hidden() 

   # 
   model.zero_grad()

   # inizializza la loss per questa epoch
   loss = 0

   # inputs.size(0) è il numero di caratteri nella stringa di input: 20
   for i in range(inputs.size(0)):
       # inputs[i] è un tensore tensor([0., 0., 0., 0., 1.]), quindi torch.Size([5])
       # Siccome la rete si aspetta (1,1,5)
       # Dobbiamo usare unsqueeze due volte che aggiunge le dimensioni "batch size" e "seq length"
       # da [input_size] a [1, 1, input_size]
       input_tensor = inputs[i].unsqueeze(0).unsqueeze(0)
       output, hidden = model(input_tensor, hidden)

       # Anche per il calcolo della loss dobbiamo usare unsqueeze una volte per aggiungere la dimensione "batch size" 
       # da [1, one-hot size] a [1, 1, one-hot size]
       loss += criterion(output, targets[i].unsqueeze(0))

   loss.backward()
   optimizer.step()

   if epoch % 100 == 0:
       print(f'Epoch: {epoch}, Loss: {loss.item()/inputs.size(0)}')

1. Funzione per generare una sequenza in maniera autoregressiva

def generate_seq(start_char='a', length=20):

   hidden = model.init_hidden()
   input = char_to_onehot(start_char)

   # questa è la lista di caratteri in output
   output_seq = start_char

   for i in range(length-1):
       # anche qui unsqueeze
       input_tensor = input.unsqueeze(0).unsqueeze(0)
       output, hidden = model(input_tensor, hidden)

       # prendiamo direttamente la posizione col valore massimo dei logit
       predicted_char_idx = torch.argmax(output).item()
       predicted_char = ix_to_char[predicted_char_idx]

       # accodiamo il carattere alla sequenza
       output_seq += predicted_char

       # usiamo l'ultimo carattere predetto come input al prossimo giro
       input = char_to_onehot(predicted_char)

   return output_seq

1. Generating a sequence

print(generate_seq('a', 25))