Redes recurrentes

Las redes recurrentes son una arquitectura que nos permite lidiar con cadenas estocásticas y poder emitir cadenas como salidas. Existen diferentes tipos de redes recurrentes. Aquí implementamos una con la ayuda de torch que se enfoca en procesar datos textuales y etiquetar un texto de acuerdo a la función de las palabras en este.

import torch
import torch.nn as nn
import numpy as np
from collections import defaultdict, Counter
from itertools import chain
import matplotlib.pyplot as plt
from sklearn.decomposition import PCA
from sklearn.manifold import TSNE
from operator import itemgetter
from tqdm import tqdm

#Funcion que crea un vocabulario de palabras con un indice numerico
def vocab():
    vocab = defaultdict()
    vocab.default_factory = lambda: len(vocab)
    return vocab    

#Funcion que pasa la cadena de simbolos a una secuencia con indices numericos
def text2numba(corpus, vocab):
    for doc in corpus:
        yield [vocab[w] for w in doc.split()]

Preparación de los datos

Para trabajar con datos textuales, estos requieren de un tratamiento especial. En primer lugar se sustituirán las cadenas de texto por índices numéricos, esto permitirá generar embeddings a partir de una capa de embedding.

La indexación se hará tanto en el texto de entrada como el de salida, por tanto tendremos dos vocabularios. Los datos, como se observa, se traducen de una cadena de índices a otra cadena de índices.

#Entradas
inputs = ['el perro come un hueso', 'un muchacho jugaba', 'el muchacho saltaba la cuerda', 'el perro come mucho',
          'un perro come croquetas', 'el perro come', 'el gato come croquetas', 'un gato come', 'yo juego mucho', 
          'el juego', 'un juego', 'yo juego un juego', 'el gato come mucho']

#Etiquetas de salida
outputs = ['DA NC V DD NC', 'DD NC V', 'DA NC V DA NC', 'DD NC V NC', 'DA NC V Adv', 'DA NC V', 'DA NC V NC', 
           'DD NC V', 'DP V Adv', 'DA NC', 'DD NC', 'DP V DD NC', 'DA NC V Adv']

#Llamamos la funcion para crear el vocabulario
in_voc = vocab()
#Creamos el vocabulario y le asignamos un indice a cada simbolo segun su aparicion
x = list(text2numba(inputs,in_voc))

#Vocabulario de emisiones
out_voc = vocab()
#Se susituyen las emisiones por sus índices numéricos
y = list(text2numba(outputs,out_voc))

print(x)
print(y)

[[0, 1, 2, 3, 4], [3, 5, 6], [0, 5, 7, 8, 9], [0, 1, 2, 10], [3, 1, 2, 11], [0, 1, 2], [0, 12, 2, 11], [3, 12, 2], [13, 14, 10], [0, 14], [3, 14], [13, 14, 3, 14], [0, 12, 2, 10]]
[[0, 1, 2, 3, 1], [3, 1, 2], [0, 1, 2, 0, 1], [3, 1, 2, 1], [0, 1, 2, 4], [0, 1, 2], [0, 1, 2, 1], [3, 1, 2], [5, 2, 4], [0, 1], [3, 1], [5, 2, 3, 1], [0, 1, 2, 4]]

Red recurrente

Definimos la red recurrente de la siguiente forma:

• Capa de embedding: Esta priemra capa generará vectores de embeddings a partir de los índices numéricos de entrada.
• Capa recurrente: Es la capa donde se obtienen las recurrencias; esta capa puede ser bidireccional o sólo en la dirección de avance. Ocuparemos además una función de activación para la capa recurrente.
• Capa de salida: La capa con activación Sofmax para obtener los valores de salida.

class RecurrentNetwork(nn.Module):
    def __init__(self, dim_in, dim_out, dim=100, dim_h=200):
        super().__init__()
        #Capa de embedding
        self.emb = nn.Embedding(dim_in,dim)
        #Capa de RNN (se toma bidireccional)
        self.recurrence = nn.RNN(dim, dim_h, bidirectional=True)
        #Salida
        self.ffw = nn.Sequential(nn.Linear(2*dim_h,dim_out), nn.Softmax(dim=2))
        
    def forward(self, x):
        #Se pasa a formato torch
        x = torch.tensor(x)
        #Embedding
        x = self.emb(x)
        #Ajustes de tamaño
        x = x.unsqueeze(1)
        #Estados de recurrencia
        h, c = self.recurrence(x)
        #Activación
        h = h.tanh()
        #Salida
        y_pred = self.ffw(h)
        #Se acomoda la salida para que la tome el loss
        y_pred = y_pred.transpose(1, 2)
        
        return y_pred

Definiremos la rec recurrente con los parámetros que indiquemos.

rnn = RecurrentNetwork(len(in_voc), len(out_voc))

Entrenamiento de la red

Para entrenar la red se utiliza el BPT, Back Propagation Through Time, que es una variación del método de backpropagation. En este caso, el error se propaga a través del tiempo; como está retropropagación está determinada por la arquitectura recurrente de la red, no hace falta indicar en pytorch que se utiliza esta retropropagación, pues el método entiende que se debe considerar los diferentes estados recurrentes.

#Numero de iteraciones
epochs = 100
#La función de riesgo es la entropía cruzada
criterion = torch.nn.CrossEntropyLoss()
#Los parametros que se van a actualizar
optimizer = torch.optim.Adagrad(rnn.parameters(), lr=0.1)

#Se entrena el modelo
for epoch in tqdm(range(epochs)):
    for x_i, y_i in zip(x, y):
        #FORWARD
        y_pred = rnn(x_i)

        #BACKWARD
        #Resize de las variables esperadas (se agrega dimension de length_seq)
        y_i = (torch.tensor(y_i)).unsqueeze(1)
        #Se calcula el eror
        loss = criterion(y_pred, y_i)
        #zero grad
        optimizer.zero_grad()
        #Backprop
        loss.backward()
        #Se actualizan los parametros
        optimizer.step()

100%|██████████| 100/100 [00:02<00:00, 36.28it/s]

Resultados

Ya que la red toma como entrada índices numéricos y emite como salidas también índices numéricos, definiremos funciones que nos permitan aplicar la red a texto para hacer más amigable la interacción con la red:

#Vocanulario de indice a texto
tagger_voc = {i:w for w,i in out_voc.items()}
def tagger(text):
    """Función que toma el texto aplica la red y arroja las etiquetas textuales."""
    x_in = [in_voc[w] for w in text.split()]
    probs = rnn(x_in)
    values = probs.argmax(axis=1)
    tags = [tagger_voc[int(v[0])] for v in values]
    
    return ' '.join(tags)

tagger('yo juego mucho un juego')

'DP V Adv DD NC'