Preprocesamiento de texto para NLP (parte 2)
=^._.^= ∫ PyTorch, Dataset y DataLoader
- PyTorch
- Dataset
- DataLoader
- Tensores
- Tókenes especiales
- La función que rellena
- Tamaño del documento
- Bonus: AtributoDiccionario
- Función collate
- Avanzado: Memory pinning
- El pre-procesamiento hasta ahora
En la primera parte llegamos a convertir esto
[
'que se requiere para un prestamo personal',
'me piden mi numero de cuenta es mi cbu',
]
en esto
[
[4160, 4683, 4484, 3703, 5294, 4011, 3825],
[3275, 3854, 3319, 3554, 1532, 1462, 2151, 3319, 950],
]
donde dijimos que las partes fundamentales son la tokenización —separar a los documentos en unidades de información— y la numericalización —el asignarle a cada uno de los tókenes un número, más que nada para que la computadora, que gusta mucho de los números, sea feliz—.
También habíamos dicho que un tóken es un atributo pero no dijimos mucho más al respecto. Veamos cómo puede ser esto. La tarea de ejemplo es clasificar documentos. Estamos acustumbrados a tener muestras y etiquetas como X e y en las que la primera es una matriz de muestras (filas) y atributos (columnas), y la segunda suele ser una columna. Cuando el dataset está sin pre-procesar tenemos las muestras (filas) pero no los atributos (columnas), por lo general tenemos una única columna con los documentos en forma de strings, lo que mucha forma de atributos no tiene.
Ahora que hemos pre-procesado el texto estamos a un paso de obtener los atributos. La función de los atributos es describir o caracterizar a las muestras. El modelo lee estos atributos para realizar inferencias. Hay distintas maneras de describir a los documentos, algunas más sofisticadas que otras, una intuitiva es aprovechar que los tókenes están numerados desde 0 hasta L (len(vocabulario)) y otorgarle una columna a cada uno en la matriz de atributos de tamaño N x L (donde N es la cantidad de muestras).
Hecho esto, solo resta contar cuántas veces aparece cada tóken en cada documento y asentarlo en la matriz.
| bien hola si todo
-------------------------------------------
'hola todo bien' | 1 1 0 1
'si bien bien' | 2 0 1 0Como comentario, esta forma de describir los documentos ignora enteramente el órden de los tókenes, sabemos que el sentido de una oración puede cambir completamente si cambiamos algunas palabras de lugar. Para el problema en cuestión, no parece ser tan grave ya que para clasificar una pregunta podría bastar con reconocer algunas palabras claves como cambio y clave o requisito y préstamo.
Ver tf-idf.
PyTorch
El típico bucle de entrenamiento de PyTorch tiene esta pinta.
for época in range(N_ÉPOCAS):
for lote in datos_entrenamiento:
# reseteamos los gradientes
optimizador.zero_grad()
predicciones = red_neuronal(lote.X)
pérdida = criterio(predicciones, lote.y)
# calculamos los gradientes
pérdida.backward()
# aplicamos los gradientes
optimizador.step()
Recordemos que a diferencia de otros modelos las redes neuronales revisitan varias veces el dataset, en lo que se llaman épocas, cada época es un recorrido por todas las muestras de entrenamiento.
En una época el dataset se puede mostrar entero, de a una muestra, o como es común hoy en día de a grupos o lotes (batches). La experiencia mostró que es útil variar el orden de las muestras en cada época.
PyTorch provee ciertas facilidades para el manejo de los datos con las clases definidas en torch.utils.data a ser:
-
Dataset. Organiza los datos. Le pasamos un número o índice de muestra y nos devuelve la muestra usualmente como una tupla(atributos, etiqueta). -
Sampler. Salvo que lo queramos de otra manera, se encarga de brindar un orden aleatoreo de los índices del dataset; uno diferente cada vez que le preguntamos. -
BatchSampler. Por defecto, se inicializa con unSamplery el tamaño de lote. Se encarga de armar grupos de índices; diferentes cada vez que le preguntamos. -
DataLoader. Valiéndose de los grupos de índices deBatchSampler, obtiene muestras deDataset. De esta manera para cada época devuelve lotes de muestras al azar.
Por Sampler y BatchSampler no nos detendremos ya el comportamiento por defecto, que es barajar el dataset en cada época y armar lotes del mismo tamaño es todo lo que necesitamos.
from torch.utils.data import Dataset
class Textset(Dataset):
def __init__(self, documentos, etiquetas=None):
self.documentos = documentos
self.etiquetas = etiquetas or np.full(len(documentos), np.nan)
def __len__(self):
return len(self.documentos)
def __getitem__(self, item):
return self.documentos[item], self.etiquetas[item]
Es una clase que necesita implementar __len__ y __getitem__. Podría encargarse de levantar y pre-procesar el dataset, que por comodidad lo hemos cargado con Pandas y pre-procesado por fuera: el constructor (__init__) podría recibir el nombre del archivo, leerlo y aplicarle las funciones pertinentes. No lo hemos hecho internamente porque el vocabulario debe nutrirse del dataset de entrenamiento ya pre-procesado [falta].
También necesitaremos crear un Textset para el dataset de inferencia, para el cual no contamos con las etiquetas. En el caso de no pasar etiquetas generamos una lista llena de NaNs del mismo largo que la lista de documentos.
train_ds = Textset(train_índices, etiquetas_train_índices)
len(train_ds)
train_ds[10_000]
Es bastante similar a lo que una lista de tuplas podría lograr, aunque fue una buena oportunidad para juntar los documentos y las etiquetas que luego de cargar el DataFrame y hasta ahora recorrieron caminos separados. Lo realmente importante es el DataLoader, no podemos usar una lista como dataset porque requiere que sea una instancia de Dataset.
DataLoader
DataLoader es un iterable. Los iterables son colecciones de elementos que se pueden recorrer; implementan el método __iter__, del que se espera que devuelva un objeto iterador (iterador = iter(iterable)). A su vez el iterador implementa el método __next__ que se encarga devolver secuencialmente los elementos de la colección hasta que se agota; una vez que esto sucede el iterador debe ser descartado y en todo caso le pedimos al iterable que nos arme un nuevo iterador. Cuando usamos la construcción for ítem in iterable, el intérprete de Python implícitamente obtiene un iterador.
Ver la sección de interables en el tutorial de Python.
lista = iter(['uno','dos'])
next(lista)
next(lista)
next(lista)
No hay próximo elemento. Cuando se llega al fin del iterador se levanta la excepción StopIteration.
Suficientes detalles por ahora. Todo esto para decir que DataLoader es un iterable que particularmente devuelve un iterador distinto cada vez, a diferencia de una lista en la que los elementos siempre se recorren en el mismo orden. Es decir, se trata de una colección de lotes pero cada iterador agrupa lotes según como dicte BatchSampler, que suele ser aleatorio.
En cada época le pedimos un iterador a DataLoader, por lo que recorremos todo el Dataset agrupado en lotes de manera diferente cada vez.
from torch.utils.data import DataLoader
train_dl = DataLoader(train_ds, batch_size=32, shuffle=True)
Le estamos diciendo a DataLoader que queremos lotes de 32 muestras (batch_size) y que el armado de los lotes sea aleatorio (shuffle).
un_lote = next(iter(train_dl))
un_lote
Está bueno que ya veamos tensores de PyTorch porque vamos a necesitar los datos en forma de tensor para alimentar a la red neuronal. Sin embargo, algo no parece andar bien con el lote que acabamos de obtener.
len(un_lote)
Tenemos dos elementos adentro del lote, podríamos pensar que el primero agrupa documentos y el segundo, etiquetas.
type(un_lote[1]), len(un_lote[1])
Las etiquetas del lote están perfecto, son un tensor de una dimensión con largo 32.
type(un_lote[0]), len(un_lote[0])
En cambio la agrupación de documentos no tiene sentido. Es otra lista de tamaño 4 con tensores adentro. ¿Qué está pasando?
Tensores
El problema parece radicar en los tensores. Son estructuras que las podemos imaginar como una columna cuando tienen una dimensión, una tabla cuando son dos, un cubo cuando tres...
Los tensores son similares a los ndarrays de NumPy, con el aditivo que también pueden ser usados en la GPU para acelerar los cómputos. Ver más de tensores en el tutorial de PyTorch.
En el caso de los documentos que a la altura del Dataset son listas de listas de índices, son dos dimensiones, y al llevarlos a una tabla vemos que tendríamos tantas filas como documentos y tantas columnas como índices tenga el documento más largo de la colección pero que no todos los documentos tienen tantos índices como columnas la tabla.
índices = [
[2,2],
[4,4,4,4],
[7,7,7,7,7,7,7],
]
índices
import torch
torch.tensor(índices)
Como anticipamos, no le gustó nada.
Tókenes especiales
Lo mencionamos al pasar, a veces se utilizan tókenes especiales como <separador de palabra>, <separador de oración>, <inicio del texto>, <fin del texto>. Hay de todo tipo, según la tarea a realizar. Uno que está presente generalmente en los proyectos es el tóken de relleno <relleno> (en inglés padding).
El tóken de relleno nos va a servir para hacer que todos los documentos tengan el mismo largo y finalmente podamos convertirlos en un tensor. No lo vamos a hacer inmediatamente ya que no nos interesa que tengan el mismo largo en todo el dataset sino en todo el lote. Como los lotes son generados en el DataLoader, este último tendrá que encargarse de rellenar los documentos.
Vamos a modificar Vocab quien se encarga de la lista de tókenes para que incluya a <relleno>.
import numpy as np
from itertools import chain
from collections import Counter
class Vocab():
@property
def índice_relleno(self):
return self.mapeo.get(self.tóken_relleno)
def __init__(self, tóken_desconocido='<unk>', tóken_relleno='<pad>', frecuencia_mínima=0.0, frecuencia_máxima=1.0,
longitud_mínima=1, longitud_máxima=np.inf, stop_words=[], límite_vocabulario=None):
self.tóken_desconocido = tóken_desconocido
self.tóken_relleno = tóken_relleno
self.frecuencia_mínima = frecuencia_mínima
self.frecuencia_máxima = frecuencia_máxima
self.longitud_mínima = longitud_mínima
self.longitud_máxima = longitud_máxima
self.stop_words = stop_words
self.límite_vocabulario = límite_vocabulario
# ningún cambio aquí
def reducir_vocabulario(self, lote):
contador_absoluto = Counter(chain(*lote))
contador_documentos = Counter()
for doc in lote:
contador_documentos.update(set(doc))
# frecuencia mínima
if isinstance(self.frecuencia_mínima, int): # frecuencia de tóken
vocabulario_mín = [tóken for tóken, frecuencia in contador_absoluto.most_common() if frecuencia >= self.frecuencia_mínima]
else: # frecuencia de documento
vocabulario_mín = [tóken for tóken, frecuencia in contador_documentos.most_common() if frecuencia/len(lote) >= self.frecuencia_mínima]
# frecuencia máxima
if isinstance(self.frecuencia_máxima, int): # frecuencia de tóken
vocabulario_máx = [tóken for tóken, frecuencia in contador_absoluto.most_common() if self.frecuencia_máxima >= frecuencia]
else: # frecuencia de documento
vocabulario_máx = [tóken for tóken, frecuencia in contador_documentos.most_common() if self.frecuencia_máxima >= frecuencia/len(lote)]
# intersección de vocabulario_mín y vocabulario_máx preservando el órden
vocabulario = [tóken for tóken in vocabulario_mín if tóken in vocabulario_máx]
# longitud
vocabulario = [tóken for tóken in vocabulario if self.longitud_máxima >= len(tóken) >= self.longitud_mínima]
# stop words
vocabulario = [tóken for tóken in vocabulario if tóken not in self.stop_words]
# límite
vocabulario = vocabulario[:self.límite_vocabulario]
return vocabulario
def fit(self, lote):
vocabulario = self.reducir_vocabulario(lote)
if self.tóken_desconocido:
vocabulario.append(self.tóken_desconocido)
if self.tóken_relleno:
vocabulario.insert(0, self.tóken_relleno)
self.mapeo = {tóken: índice for índice, tóken in enumerate(vocabulario)}
return self
# ningún cambio aquí
def transform(self, lote):
if self.tóken_desconocido: # reemplazar
return [[tóken if tóken in self.mapeo else self.tóken_desconocido for tóken in doc] for doc in lote]
else: # ignorar
return [[tóken for tóken in doc if tóken in self.mapeo] for doc in lote]
# ningún cambio aquí
def tókenes_a_índices(self, lote):
lote = self.transform(lote)
return [[self.mapeo[tóken] for tóken in doc] for doc in lote]
# ningún cambio aquí
def índices_a_tókenes(self, lote):
mapeo_inverso = list(self.mapeo.keys())
return [[mapeo_inverso[índice] for índice in doc] for doc in lote]
def __len__(self):
return len(self.mapeo)
El índice del tóken de relleno suele ser 0 y para continuar con esta tradición en vez de hacerle append al vocabulario le hicimos un prepend para que el tóken encabece el listado. Además usamos el decorador @property para tener un atributo índice_relleno (en vez de un método) que nos devuelva el índice del tóken.
v = Vocab().fit(train_docs)
v.índice_relleno
def rellenar_documentos(lote, largos, índice_relleno):
máximo_largo = max(largos)
return [doc + [índice_relleno] * (máximo_largo - largos[i]) for i, doc in enumerate(lote)]
Le tenemos que pasar el lote, el largo o tamaño de cada documento del lote y el índice de relleno.
índices = [
[2,2],
[4,4,4,4],
[7,7,7,7,7,7,7],
]
largos = [2,4,7]
rellenos = rellenar_documentos(índices, largos, v.índice_relleno)
rellenos
torch.tensor(rellenos)
¡Ahora sí funcionó!
from torch.utils.data import Dataset
class Textset(Dataset):
def __init__(self, documentos, etiquetas=None):
self.documentos = documentos
self.etiquetas = etiquetas or np.full(len(documentos), np.nan)
def __len__(self):
return len(self.documentos)
def __getitem__(self, item):
return self.documentos[item], len(self.documentos[item]), self.etiquetas[item]
train_ds = Textset(train_índices, etiquetas_train_índices)
train_ds[10_000]
class AtriDicc():
def __init__(self, *args, **kwargs):
self.__dict__ = dict(*args, **kwargs)
def __repr__(self):
return repr(self.__dict__)
AtriDicc(uno=1, dos=2, tres=3).uno
Vamos a pimpiar la clase Textset con esto para que en vez de devolver elementos del dataset como tuplas (documento, largo, etiqueta) en el que debemos acordarnos que el orden de los elementos, devolvemos un AtriDicc en el que accedemos las cosas por su nombre y es más cómodo que un diccionario.
from torch.utils.data import Dataset
class Textset(Dataset):
def __init__(self, documentos, etiquetas=None):
self.documentos = documentos
self.etiquetas = etiquetas or np.full(len(documentos), np.nan)
def __len__(self):
return len(self.documentos)
def __getitem__(self, item):
return AtriDicc(
documento = self.documentos[item],
largo = len(self.documentos[item]),
etiqueta = self.etiquetas[item],
)
train_ds = Textset(train_índices, etiquetas_train_índices)
train_ds[10_000].documento
Función collate
Collate significa juntar diferentes piezas de información para ver sus similaridades y diferencias, también puede ser colectar y organizar las hojas de un reporte, un libro. En el contexto de DataLoader quiere decir arreglar el lote. Entonces esta función recibe una lista de elementos del Dataset, en nuestro caso una lista de de AtriDiccs, y debe devolver el lote en una forma útil y en lo posible realizar conversiones a tensores.
DataLoader posee una collate function por defecto que utiliza internamente y que en muchos casos funciona correctamente, pero otros como ahora que tenemos documentos de distinto largo nos toca definir una función propia.
def rellenar_lote(lote):
"""Prepara lotes para ingresar a nn.Embedding"""
documentos = [elemento.documento for elemento in lote]
largos = [elemento.largo for elemento in lote]
etiquetas = [elemento.etiqueta for elemento in lote]
rellenos = rellenar_documentos(documentos, largos, v.índice_relleno)
return AtriDicc(
documentos = torch.tensor(rellenos),
etiquetas = torch.tensor(etiquetas),
)
Cuando instanciamos un DataLoader le pasamos la función que acabamos de definir.
train_dl = DataLoader(train_ds, collate_fn=rellenar_lote, batch_size=3, shuffle=True)
un_lote = next(iter(train_dl))
un_lote.documentos
un_lote.etiquetas
Funciona de maravillas.
La función anterior es compatible con el módulo de PyTorch nn.Embedding que suele se la puerta de entrada en los modelos de procesamiento de texto. Todavía no hemos hablado nada de los embeddings. Quizás sea un momento para mencionar a nn.EmbeddingBag, que tiene requerimientos completamente diferentes al primer módulo.
def offsetear_lote(lote):
"""Prepara lotes para ingresar a nn.EmbeddingBag"""
documentos = [torch.tensor(elemento.documento) for elemento in lote]
offsets = [0] + [elemento.largo for elemento in lote][:-1]
etiquetas = [elemento.etiqueta for elemento in lote]
return AtriDicc(
documentos = torch.cat(documentos),
offsets = torch.tensor(offsets).cumsum(dim=0),
etiquetas = torch.tensor(etiquetas),
)
Esta función yuxtapone los documentos por un lado, y por otro (offsets) indica cuándo comienza cada documento en ese continuo.
train_dl = DataLoader(train_ds, collate_fn=offsetear_lote, batch_size=3, shuffle=True)
un_lote = next(iter(train_dl))
un_lote.documentos
un_lote.offsets
Avanzado: Memory pinning
https://pytorch.org/docs/stable/data.html#memory-pinning
Esta técnica consiste en pre-disponibilizar los tensores en el GPU. Llevar un lote del disco o de la memoria a la GPU insume tiempo y puede causar un cuello de botella durante el entrenamiento.
Pasar la opción pin_memory=True al DataLoader pondrá automáticamente a los tensores en la pinned memory. Por defecto funciona con tensores y colecciones de tensores. Cuando los lotes son de un tipo personalizado (por ejemplo AtriDicc), normal cuando se utiliza una collate function propia, es necesario que el tipo defina el método pin_memory.
class AttrDict():
def __init__(self, *args, **kwargs):
self.__dict__ = dict(*args, **kwargs)
def __repr__(self):
return repr(self.__dict__)
def pin_memory(self):
for atributo, valor in self.__dict__.items():
self.__dict__[atributo] = valor.pin_memory() if hasattr(valor, 'pin_memory') else valor
return self
vocabulario_documentos = Vocab().fit(train_docs)
train_índices = vocabulario_documentos.tókenes_a_índices(train_docs)
valid_índices = vocabulario_documentos.tókenes_a_índices(valid_docs)
infer_índices = vocabulario_documentos.tókenes_a_índices(infer_docs)
train_ds = Textset(train_índices)
valid_ds = Textset(valid_índices)
infer_ds = Textset(infer_índices)
Solo definimos el modelo, no lo entrenamos. Elegimos la función offsetear_lote ya que el modelo usa nn.EmbeddingBag.
from torch.utils.data import DataLoader
train_dl = DataLoader(train_ds, batch_size=32, shuffle=True, collate_fn=offsetear_lote, pin_memory=True)
# validación e inferencia no requieren `shuffle`
valid_dl = DataLoader(valid_ds, batch_size=32, shuffle=False, collate_fn=offsetear_lote, pin_memory=True)
infer_dl = DataLoader(infer_ds, batch_size=32, shuffle=False, collate_fn=offsetear_lote, pin_memory=True)
Hacemos unas definiciones necesarias. No es el punto de lo que queremos mostrar, lo podés pasar por alto. Para mayores detalles recomendamos ver Text Classification with TorchText.
import torch.nn as nn
import torch.nn.functional as F
DIM_EMBEDDINGS = 8
class ClasificadorBolsa(nn.Module):
def __init__(self, vocab_size, embed_dim, num_class):
super().__init__()
self.embedding = nn.EmbeddingBag(vocab_size, embed_dim, sparse=False, mode='max')
self.fc = nn.Linear(embed_dim, num_class)
def forward(self, text, offsets):
embedded = self.embedding(text, offsets)
return self.fc(embedded)
modelo = ClasificadorBolsa( len(vocabulario_documentos), DIM_EMBEDDINGS, len(vocabulario_etiquetas) ).to(device)
Ídem.
if torch.cuda.is_available():
device = torch.device('cuda')
else:
device = torch.device('cpu')
Nuevamente.
optimizador = torch.optim.Adam(modelo.parameters(), lr=1e-3, weight_decay=1e-5)
criterio = nn.CrossEntropyLoss()
Alto aquí. Así es como se usa un DataLoader.
ÉPOCAS = 10
for época in range(ÉPOCAS):
for lote in train_dl:
optimizador.zero_grad()
predicciones = modelo(lote.documentos.to(device), lote.offsets.to(device))
pérdida = criterio(predicciones, lote.etiquetas.to(device))
pérdida.backward()
optimizador.step()
Con esto concluye la segunda parte. Quedaron los embeddings para la tercera. Ahora deberíamos tener más control sobre la carga de datos en PyTorch. Muchos ejemplos de uso y tutoriales dan por sentada esta parte al utilizar datasets de ejemplos, que ya vienen pre-procesados y/o que la carga por defecto de PyTorch maneja sin inconvenientes.