PyTorch — среда глубокого обучения, которая была принята такими технологическими гигантами, как Tesla, OpenAI и Microsoft для ключевых исследовательских и производственных рабочих нагрузок.

PyTorch-Ignite — это библиотека высокого уровня, помогающая гибко и прозрачно обучать и оценивать нейронные сети в PyTorch. Основная проблема с реализацией глубокого обучения заключается в том, что коды могут быстро расти, становиться повторяющимися и слишком длинными. Рассматривать данную библиотеку буду, решая задачу оценки вероятности отнесения изображения к определенному классу на примере датасета CIFAR10. Чуть позже расскажу о нем подробнее. А сейчас начнем подготовку с установки и импорта необходимых библиотек.

Установка и импорт необходимых библиотек

Советую работать в сервисе GoogleColab

!pip install pytorch-ignite
import numpy as np
import matplotlib.pyplot as plt
import torch
from torch import nn, optim
import torch.nn.functional as F
from torch.utils.data import DataLoader
from torchvision import datasets, transforms
import torchvision
from torch.utils.data import sampler
from torch.utils.data.sampler import SubsetRandomSampler
from ignite.engine import Events, create_supervised_trainer, create_supervised_evaluator
from ignite.metrics import Accuracy, Loss, RunningAverage, ConfusionMatrix
from ignite.handlers import ModelCheckpoint, EarlyStopping

Теперь подробнее рассмотрим датасет CIFAR10. Он содержит 60 000 изображений в 10 классах. Все изображения размером 32х32. С помощью следующего блока кода разобью данные на тестовые, тренировочные и валидационные. 

transform = transforms.Compose( [transforms.ToTensor(),transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))])

trainset = datasets.CIFAR10('./data', download=True, train=True, transform=transform)
train_loader = DataLoader(trainset, batch_size=4, shuffle=True)

validationset = datasets.CIFAR10('./data', download=True, train=False, transform=transform)

classes = ('plane', 'car', 'bird', 'cat','deer', 'dog', 'frog', 'horse', 'ship', 'truck')

validation_split = .2
shuffle_dataset = True
random_seed= 42
dataset_size = len(validationset)
indices = list(range(dataset_size))
split = int(np.floor(validation_split * dataset_size))
if shuffle_dataset :
    np.random.seed(random_seed)
    np.random.shuffle(indices)
val_indices, test_indices = indices[split:], indices[:split]

val_sampler = SubsetRandomSampler(val_indices)
test_sampler = SubsetRandomSampler(test_indices)

val_loader = torch.utils.data.DataLoader(validationset, batch_size=4, 
                                           sampler=val_sampler)
test_loader = torch.utils.data.DataLoader(validationset, batch_size=4,
                                                sampler=test_sampler)
print(f'Кол-во валидационных данных {len(val_loader)}')
print(f'Кол-во тестовых данных {len(test_loader)}')
print(f'Кол-во обучающих данных {len(train_loader)}')

Предлагаю посмотреть, как выглядят наши изображения. Код ниже выводит изображение из датасета и подписывает к какому классу оно принадлежит. 

def imshow(img):
    img = img / 2 + 0.5 
    npimg = img.numpy()
    print(f'Размер изображения  {npimg.shape}')
    plt.imshow(np.transpose(npimg, (1, 2, 0)))
    plt.show()

train_loader = DataLoader(trainset, batch_size=1, shuffle=True)
dataiter = iter(train_loader)
images, labels = dataiter.next()
imshow(torchvision.utils.make_grid(images))
# print(' '.join('%5s' % classes[labels[j]] for j in range(1)))
print(f'Класс изображения {" ".join("%5s" % classes[labels[j]] for j in range(1))}')

train_loader = DataLoader(trainset, batch_size=64, shuffle=True)

Теперь разберу архитектуру модели. На изображении ниже схематично показано, как работают сверточные нейронные сети. У нас есть 2 основных слоя: Conv_1 и Conv_2. Между ними идет увеличение найденных уникальных признаков Max-polling. После прохождения по всем слоям происходит подключение функции активации Relu. Перед тем как сеть определит, к какому классу принадлежит изображение, подключается dropout, чтобы избежать переобучения модели. И после этого на выходе получается предполагаемый класс изображения.

Далее разберу основные компоненты сети. 

class Mod(nn.Module):
    def __init__(self):
        super(Mod, self).__init__()

        self.conv1 = nn.Conv2d(3, 16, 5) 
        self.pool = nn.MaxPool2d(2, 2)
        self.bn1 = nn.BatchNorm2d(16)

        self.conv2 = nn.Conv2d(16, 32, 5)
        self.bn2 = nn.BatchNorm2d(32)

        self.fc1 = nn.Linear(32 * 5 * 5, 120)
        self.fc2 = nn.Linear(120, 84)
        self.fc3 = nn.Linear(84, 10)

    def forward(self, x):
        x = self.pool(F.relu(self.conv1(x)))
        x = self.bn1(x)

        x = self.pool(F.relu(self.conv2(x)))
        x = self.bn2(x)

        x = torch.flatten(x, 1)
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        x = self.fc3(x)
        # return x
        
        return F.log_softmax(x,dim=1)

Метод Conv2d создает набор сверточных фильтров, их задача состоит в обработке изображений небольшим фильтром, который переходит по изображению небольшими шагами. Это основа сети.

BatchNorm2d выполняет функцию нормализации.

Maxpooling сохраняет ��аиболее активные пиксели из предыдущего слоя, укрупняет масштаб полученных признаков. Другими словами, ключевые области, которые определяют изображение к определённому классу, делает более крупными.

Слои Linear и Dropout, нужны для того чтобы избежать переобучения.

В качестве функции активации используем Relu. Relu все отрицательные числа делает равным 0, а положительные остаются без изменения. Блок кода ниже создает экземпляр созданной ранее сверточной сети, а также задает функцию потерь и скорость обучения. Теперь можно провести небольшое исследование с переменной lr(Learning rate), чтобы посмотреть, как меняется результат в зависимости от скорости обучения. На графике оранжевая линия означает lr=0.001, а синяя lr=0.1. Как видим, данная переменная играет важную роль в настройке модели.

model_classification = Mod()
criterion = nn.CrossEntropyLoss()
def opt(rate):
  optimizer = optim.SGD(model_classification.parameters(), lr=rate, momentum=0.9)
  return optimizer
optimizer = opt(0.001)

Воспользуемся метриками из ignite.metrics, которые хотим рассчитать для модели: Accuracy, ConfusionMatrix и Loss. Далее передам их механизмам оценки, которые будут вычислять эти показатели для каждой итерации. 

epochs = 12
trainer = create_supervised_trainer(model_classification, optimizer, criterion)
metrics = {'accuracy':Accuracy(),'nll':Loss(criterion),'cm':ConfusionMatrix(num_classes=len(classes))}
train_evaluator = create_supervised_evaluator(model_classification, metrics=metrics)
val_evaluator = create_supervised_evaluator(model_classification, metrics=metrics)
training_history = {'accuracy':[],'loss':[]}
validation_history = {'accuracy':[],'loss':[]}
last_epoch = []

Отследить потери для каждого шага можно, запустив данный блок. 

RunningAverage(output_transform=lambda x: x).attach(trainer, 'loss')

Теперь настрою обработчик потерь EarlyStopping. 

def score_function(engine):
    val_loss = engine.state.metrics['nll']
    return val_loss

handler = EarlyStopping(patience=20, score_function=score_function, trainer=trainer)
val_evaluator.add_event_handler(Events.COMPLETED, handler)

Данный обработчик может приостановить процесс обучения. В случае, если потери проверочного множества не уменьшатся, процесс обучения остановится досрочно.

Далее просмотрим 2 функции, которые печатают результаты работы с набором данных. Одна функция делает это с обучающими данными, а другая с данными для проверки (валидационными).

Первая функция использует декоратор “trainer.on()”. Это означает, что декорируемая функция будет прикреплена к тренировочной функции и будет вызываться в конце каждого прогона обучения.

@trainer.on(Events.EPOCH_COMPLETED)
def log_training_results(trainer):
    train_evaluator.run(train_loader)
    metrics_trian = train_evaluator.state.metrics
    accuracy = metrics_trian['accuracy']*100
    loss = metrics_trian['nll']
    last_epoch.append(0)
    training_history['accuracy'].append(accuracy)
    training_history['loss'].append(loss)
    print(f"Training Results - Epoch: {trainer.state.epoch}  TRAIN_accuracy: {round(accuracy,2)} TRAIN_loss: {round(loss,2)}".format(trainer.state.epoch, accuracy, loss))

Вторая функция предполагает использование метода add_event_handler. При этом достигается тот же результат, что и выше. 

def log_validation_results(trainer):
    val_evaluator.run(val_loader)
    metrics_val = val_evaluator.state.metrics
    accuracy = metrics_val['accuracy']*100
    loss = metrics_val['nll']
    validation_history['accuracy'].append(accuracy)
    validation_history['loss'].append(loss)
    print(f"VAL_Training Results - Epoch: {trainer.state.epoch}  VAL_accuracy: {round(accuracy,2)} VAL_loss: {round(loss,2)}".format(trainer.state.epoch, accuracy, loss))
    
trainer.add_event_handler(Events.EPOCH_COMPLETED, log_validation_results)

Теперь проверю модель. Это важный шаг, так как процессы обучения могут занимать много времени и, если по какой-то причине во время обучения что-то пойдет не так, контрольная точка модели может помочь перезапустить обучение с точки отказа.

Буду использовать обработчик Ignite ModelCheckpoint для проверки моделей в конце каждого прогона обучения.

checkpointer = ModelCheckpoint('./saved_models', 'CIFAR10', n_saved=2, create_dir=True, save_as_state_dict=True, require_empty=False)
trainer.add_event_handler(Events.EPOCH_COMPLETED, checkpointer, {'CIFAR10': model_classification})

Далее запущу обучение нашей модели на 24 epochs. Другими словами, модель пройдет обучение 24 раза. И после каждого раза будет виден результат обучения. 

trainer.run(train_loader, max_epochs=24)

Результат обучения модели

Если при тесте данной модели вы увидите accuracy 100, это может означать два варианта:

  1. Модель начала переобучение. Она, как студент, который просто заучивает правильные ответы и даже не понимает, что учит.

  2. Ваши тестовые данные такие же, как тренировочные. Важно следить за качеством данных и не забывать их менять.

Визуализирую данные об обучении модели на обучающих и валидационных данных.

Вывод графика истории на обучающих данных.

plt.plot(training_history['accuracy'],label="Training Accuracy")
plt.plot(validation_history['accuracy'],label="Validation Accuracy")
plt.xlabel('Epochs')
plt.ylabel('Accuracy')
plt.legend(frameon=False)
plt.show()

Вывод графика истории на валидационных данных. 

plt.plot(training_history['loss'],label="Training Loss")
plt.plot(validation_history['loss'],label="Validation Loss")
plt.xlabel('Epochs')
plt.ylabel('Loss')
plt.legend(frameon=False)
plt.show()

Код ниже будет использоваться для вывода данных из модели и визуализации результатов. 

classes = ['plane', 'car', 'bird', 'cat','deer', 'dog', 'frog', 'horse', 'ship', 'truck']

def predict_class(batch_size):
  test_loader = torch.utils.data.DataLoader(validationset, batch_size=batch_size,sampler=test_sampler)
  dataiter = iter(test_loader)
  images, labels = dataiter.next()
  imshow(torchvision.utils.make_grid(images))
  outputs =  model_classification(images)
  _, predicted = torch.max(outputs, 1)
  print(labels)
  print('DATA: '+' '.join(f'{classes[labels[j]]}' for j in range(batch_size)))
  print('PRED: '+' '.join(f'{classes[predicted[j]]}'for j in range(batch_size)))

  label_dataset = [classes[predicted[j]] for j in range(batch_size)]
  label_predict = [classes[labels[j]] for j in range(batch_size)]
  kol = 0
  for i in range(batch_size):
    if label_predict[i] == label_dataset[i]:
      kol+=1
  print(f'Правильно классифицировано: {(kol/batch_size)*100}%')
    
predict_class(40)

Результат:

В заключении я сравню реализацию нескольких задач с данной библиотекой и без нее.

Первая задача вывод значений accuracy и loss.

PyTorch-Ignite

train_evaluator.run(train_loader)
metrics_trian = train_evaluator.state.metrics
accuracy = metrics_trian['accuracy']*100
loss = metrics_trian['nll']
print(f"Training Results - Epoch: {trainer.state.epoch}  TRAIN_accuracy: {round(accuracy,2)} TRAIN_loss: {round(loss,2)}".format(trainer.state.epoch, accuracy, loss))

PyTorch 

train_epoch_loss=0
train_epoch_accuracy=0
def metrics_accuracy(pred, train):
    pred = torch.log_softmax(pred, dim = 1)
    _, pred = torch.max(pred, dim = 1)  
    res = (pred == train).sum().float()
    accuracy =res/train.shape[0]
    accuracy = torch.round(accuracy * 100)
    return accuracy
for train_img, train_label in train_loader:
        optimizer.zero_grad()
        label_train_pred = model_classification(train_img).squeeze()
        train_loss = criterion(label_train_pred, train_label)
        train_accuracy = metrics_accuracy(label_train_pred, train_label)
        train_loss.backward()
        optimizer.step()
        train_epoch_loss += train_loss.item()
        train_epoch_accuracy += train_accuracy.item()

Вторая задача преждевременная остановка обучения.

PyTorch-Ignite

handler = EarlyStopping(patience=10, score_function=score_function, trainer=trainer)
val_evaluator.add_event_handler(Events.COMPLETED, handler)

PyTorch 

val_loss_list.append(val_epoch_loss/len(val_loader))
if len(val_loss_list)>1:
  if val_loss_list[-2] < val_loss_list[-1]:
        trigger_times += 1
        if trigger_times > patience:
            print('Early stopping!')
            break
  else:
    trigger_times = 0

Даже по двум задачам видно, на сколько данная библиотека сокращает объем кода.

Я рассмотрел основные функции данной библиотеки, а также сравнил выполнение некоторых задач без использования PyTorch-Ignite и с использованием.