Red convolucional con PyTorch

En el artículo Introducción a PyTorch, primera red neuronal creamos y entrenamos nuestra primera red neuronal, concretamente se conoce a la topología que creamos como red feed-forward o perceptron multicapa. Ya que únicamente hace uso de capas densamente conectadas. En este tutorial vamos a implementar redes neuronales con PyTorch.

¿Qué es una red convolucional?

Una red convolucional es un tipo de red neuronal que contiene al menos una capa que utiliza la operación de convolución. Esta operación consiste en realizar una multiplicación de matrices entre una matriz K llamada kernel que será constante y una ventana M de la imagen del mismo tamaño que K. Esta operación se repite para cada pixel de la imagen donde se pueda extraer una ventana del mismo tamaño de K. En el caso de no querer reducir el tamaño de la imagen se añaden a la imagen una serie de píxeles conocidos como padding para que el centro de la ventana M pueda llegar hasta as posiciones de las esquinas, estos píxeles pueden tomar diferentes valores (0, 1 o incluso repetir los valores de sus píxeles vecinos). En la siguiente figura tenemos un ejemplo de como se realiza una convolución con un kernel de tamaño 3x3 sobre una imagen de tamaño 5x5 sin utilizar padding, por lo que la imagen resultante es de tamaño 3x3.

Operación de convolución (fuente)La operación convolución es llamada en ciertas ocasiones filtrado, ya que dependiendo de como configuremos el kernel podemos realizar un filtrado específico. Si añadimos estas operaciones a una red neuronal podremos calcular su aportación a la salida y con ello podemos aprender los valores de los kernels mediante el gradiente y el optimizador al igual que lo realizamos con las capas densamente conectadas. Al igual que el perceptron multicapa, la red convolucional podrían aprender de forma automática mediante el algoritmo backpropagation. Para calcular el tamaño de salida que tendrá una imagen tras pasar por una convolución se puede utilizar la siguiente formula:

Donde n_s hace referencia al tamaño de salida. N_e al tamaño de entrada. P al padding que se le aplica a la imagen, es decir, el número de píxeles que se le añaden a la imagen contandolo solo en uno de los lados. K es el tamaño del kernel. S hace referencia al stride, esto se conoce al número de pixeles que avanza la ventana en cada cálculo, en la primera imagen el stride era 1 ya que realiza el filtrado en cada uno de los píxeles.

Operación de pooling

En las redes convolucionales también se ha utilizado otra operación conocida como pooling. Al igual que la convolución se realiza mediante una aplicación de un kernel a una ventana. En vez de utilizar una multiplicación de matrices entre los valores del kernel y la ventana, el kernel contiene un función ya definida. En la literatura podemos encontrar que algunos autores utilizan el calculo de la media de los valores de la ventana, conocido como Average Pooling o la obtención del máximo valor de la ventana, conocido como Max Pooling. Esta operación al contrario que la convolución no cambia sus valores y por tanto ningún parámetro ha de aprender mediante el algoritmo backpropagation. En la siguiente foto podemos ver un ejemplo de la aplicación del Max Pooling.

Operación de Max Pooling (fuente)Como podemos ver la operación escoge el valor máximo en una ventana de 2x2 y esta ventana se mueve 2 casillas en cada filtrado, por tanto tiene un stride de 2. La aplicación de esta función se basa principalmente en la reducción de dimensionalidad lo que reduce el coste computacional junto con el número de parámetros que se han de aprender en las capas siguientes. Además proporciona una pequeña invarianza a la traslación en la representación interna de las imágenes.

Red convolucional con PyTorch

Una vez conocemos las operaciones, vamos a ver las capas que podemos utilizar de PyTorch para crear nuestra propia red convolucional. La primera de ellas es conv2d, esta capa se encarga de aplicar una convolución a un tensor. Tiene los siguientes parámetros obligatorios:

• Canales de entrada: número de canales que contiene el tensor que le llega a esta capa, en caso de ser la primera, si la imagen es RGB tendrá 3 canales, en caso de ser escala de grises será 1 solo canal.
• Canales de salida: número de kernels que se van a utilizar en esta capa.
• tamaño de kernel: número de filas y columnas que tendrá de tamaño cada uno de los kernels utilizados.

Podemos modificar el stride que está por defecto a 1 o añadir padding con valor 0, además de otros parámetros en los que no vamos a entrar. Tanto el tamaño de kernel como el stride como el padding pueden recibir un número entero que se usará tanto para la altura como para la anchura o en una tupla indicando los tamaños.

Red neuronal convolucional

Para crear la primera red convolucional en PyTorch vamos a reutilizar el código creado en Introducción a PyTorch, primera red neuronal, ya que modificando únicamente el forward de la red neuronal podemos entrenar una nueva topología. Al final de este tutorial tendremos el código completo con la nueva red.

Para la nueva topología hemos decidido utilizar dos capas convolucionales con 32 y 64 kernels respectivamente y un tamaño de kernel común de 3x3. Seguidas por una capa que realiza la función Max Pooling con stride=2 y una capa densamente conectada de tamaño 128. El código de la clase NET quedaría así:

class NET(nn.Module):
    def __init__(self):
        super(NET, self).__init__()
        self.conv1 = nn.Conv2d(1, 32, kernel_size=3)
        self.conv2 = nn.Conv2d(32, 64, kernel_size=3)
        self.fc1 = nn.Linear(64*12*12, 128)
        self.fc2 = nn.Linear(128, 10)#capa de salida
        self.loss_criterion = nn.CrossEntropyLoss()#Función de pérdida
        
    def forward(self, x, target):
        x = x.view(-1, 1, 28, 28)#transforma las imágenes a tamaño (n, 1, 28, 28)
        x = F.relu(self.conv1(x))
        # la salida de conv1 es 32x26x26
        x = F.max_pool2d(F.relu(self.conv2(x)), 2)#El 2 es el stride
        # la salida de conv2 es 64x24x24, la salida de max_pool es 64x12x12
        x = x.view(-1, 64*12*12)#transformamos la salida en un tensor de tamaño (n, 9216)
        x = F.relu(self.fc1(x))#Función de activación relu en la salida de la capa oculta
        x = F.softmax(self.fc2(x), dim=1)#Función de activación softmax en la salida de la capa oculta
        return x

PyTorch necesita que en las operaciones convolucionales las imagen vengan en formato (tamaño_batch, n_canales, altura, anchura), la primera operación del forward se encarga de cambiar la forma del tensor a ese formato pero no intercambia los valores, en el caso de tener una imgen con formato (tamaño_batch, altura, anchura, n_canales) debemos usar una función como numpy.swapaxes para poner los canales de la imagen primero. La segunda llamada a la función view transforma el tensor de la salida de la capa Max Pooling de tamaño (tamaño_batch, 64, 12, 12) a tamaño (tamaño_batch, 9216) dado que las capas lineales deben recibir un tensor con 2 dimensiones. Este cálculo se debe realizar utilizando la formula dada anteriormente para calcular el tamaño de salida de cada capa convolucional.

Esta nueva topología tarda bastante más que la anterior en entrenar, no solo por el aumento de tamaño sino también por incluir nuevas capas. En el caso de disponer de una GPU capaz de realizar el entrenamiento, es recomendable utilizarla ya que veremos muy reducido el tiempo.Entrenando este modelo durante solo 10 epochs conseguimos un 96.50% de precisión. No es para nada mal resultado aunque esta muy lejos del estado del arte de MNIST. Para conseguir mejorar la precisión debemos entrenar la red durante más epochs sin caer en el sobre-entrenamiento (overfitting), para ello podemos utilizar capas especiales que intentar reducir la probabilidad de sobre-entrenar, estas capas las veremos en el siguiente tutorial.

Como he comentado antes, aquí dejo el código completo de la red convolucional con PyTorch:

import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.autograd import Variable
import torchvision.datasets as dset
import torchvision.transforms as transforms
import torch.optim as optim
import time

CUDA = torch.cuda.is_available()

class NET(nn.Module):
    def __init__(self):
        super(NET, self).__init__()
        self.conv1 = nn.Conv2d(1, 32, kernel_size=3)
        self.conv2 = nn.Conv2d(32, 64, kernel_size=3)
        self.fc1 = nn.Linear(64*12*12, 128)
        self.fc2 = nn.Linear(128, 10)#capa de salida
        self.loss_criterion = nn.CrossEntropyLoss()#Función de pérdida
        
    def forward(self, x, target):
        x = x.view(-1, 1, 28, 28)#transforma las imágenes a tamaño (n, 1, 28, 28)
        x = F.relu(self.conv1(x))
        # la salida de conv1 es 32x26x26
        x = F.max_pool2d(F.relu(self.conv2(x)), 2)#El 2 es el stride
        # la salida de conv2 es 64x24x24, la salida de max_pool es 64x12x12
        x = x.view(-1, 64*12*12)#transformamos la salida en un tensor de tamaño (n, 9216)
        x = F.relu(self.fc1(x))#Función de activación relu en la salida de la capa oculta
        x = F.softmax(self.fc2(x), dim=1)#Función de activación softmax en la salida de la capa oculta
        return x
        

class NN():
    """docstring for NN"""
    def __init__(self):
        self.model = NET()
        if CUDA:
            self.model.cuda()
        self.optimizer = optim.SGD(self.model.parameters(), lr=0.01, momentum=0.9)

    def epoch_step(self, dataset_loader, train=False, verbose=2):
        correct_cnt, ave_loss = 0, 0#Contador de aciertos y acumulador de la función de pérdida
        count = 0#Contador de muestras
        for batch_idx, (x, target) in enumerate(dataset_loader):
            start_time_epch = time.time()
            count += len(x)#sumamos el tamaño de batch, esto es porque n_batches*tamaño_batch != n_muestras
            if train:
                self.optimizer.zero_grad()#iniciamos a 0 los valores de los gradiente
            x, target = Variable(x), Variable(target)#Convertimos el tensor a variable del modulo autograd
            if CUDA:
                x = x.cuda()
                target = target.cuda()
            score = self.model(x, target)#realizamos el forward
            loss = self.model.loss_criterion(score, target)
            _, pred_label = torch.max(score.data, 1)#pasamos de one hot a número
            correct_cnt_epch = (pred_label == target.data).sum()#calculamos el número de etiquetas correctas
            correct_cnt += correct_cnt_epch
            ave_loss += loss.data[0]#sumamos el resultado de la función de pérdida para mostrar después
            if train:
                loss.backward()#Calcula los gradientes y los propaga 
                self.optimizer.step()#adaptamos los pesos con los gradientes propagados
            elapsed_time_epch = time.time() - start_time_epch

            if verbose == 1:
                elapsed_time_epch = time.strftime("%Hh,%Mm,%Ss", time.gmtime(elapsed_time_epch))
                print ('\t\tbatch: {} loss: {:.6f}, accuracy: {:.4f}, time: {}'.format(
                    batch_idx, loss.data[0], correct_cnt_epch/len(x), elapsed_time_epch))

        accuracy = correct_cnt/count#Calculamos la precisión total
        ave_loss /= count#Calculamos la pérdida media

        return ave_loss, accuracy

    def train(self, epoch, train_loader, test_loader=None, verbose=2):
        for epoch in range(epoch):
            if verbose > 0:
                print("\n***Epoch {}***\n".format(epoch))
                if verbose == 1:
                    print("\tTraining:")
            start_time = time.time()
            ave_loss, accuracy = self.epoch_step(train_loader, train=True, verbose=verbose)
            elapsed_time = time.time() - start_time
            if verbose == 2:
                elapsed_time = time.strftime("%Hh,%Mm,%Ss", time.gmtime(elapsed_time))
                print ('\tTraining loss: {:.6f}, accuracy: {:.4f}, time: {}'.format(
                    ave_loss, accuracy, elapsed_time))

            if test_loader != None:
                start_time = time.time()
                ave_loss, accuracy = self.epoch_step(test_loader, train=False, verbose=verbose)
                elapsed_time = time.time() - start_time
                if verbose == 2:
                    elapsed_time = time.strftime("%Hh,%Mm,%Ss", time.gmtime(elapsed_time))
                    print ('\tTest loss: {:.6f}, accuracy: {:.4f}, time: {}'.format(
                        ave_loss, accuracy, elapsed_time))

    def evaluate(self, test_loader, verbose=2):
        print("\n***Evaluate***\n")
        start_time = time.time()
        ave_loss, accuracy = self.epoch_step(test_loader, train=False, verbose=verbose)
        elapsed_time = time.time() - start_time
        if verbose == 2:
            elapsed_time = time.strftime("%Hh,%Mm,%Ss", time.gmtime(elapsed_time))
            print ('\tloss: {:.6f}, accuracy: {:.4f}, time: {}'.format(
                ave_loss, accuracy, elapsed_time))

    def save_weights(self, path):
        torch.save(self.model.state_dict(), path)
    
    def load_weights(self, path):
        self.model.load_state_dict(torch.load(path))


if __name__ == "__main__":
    torch.manual_seed(123) #fijamos la semilla
    epochs = 10

    trans = transforms.Compose([transforms.ToTensor()]) #Transformador para el dataset
    root="../data"
    train_set = dset.MNIST(root=root, train=True, transform=trans, download=True)
    test_set = dset.MNIST(root=root, train=False, transform=trans)

    batch_size = 128
    train_loader = torch.utils.data.DataLoader(
                     dataset=train_set,
                     batch_size=batch_size,
                     shuffle=True)
    test_loader = torch.utils.data.DataLoader(
                    dataset=test_set,
                    batch_size=batch_size,
                    shuffle=False)

    print ('Trainning batch number: {}'.format(len(train_loader)))
    print ('Testing batch number: {}'.format(len(test_loader)))

    net = NN()
    net.train(epochs, train_loader, test_loader=test_loader, verbose=2)
    net.evaluate(test_loader, verbose=2)
    net.save_weights("./weights")