5 июня 2021 г. 5 min read Продвинуто

Рекуррентная нейросеть (RNN)

Рекуррентная нейронная сеть – Модель (Model) Машинного обучения (ML), работающая по принципу сохранения слоя и его переподачи.

Нейронные сети

Нейронная сеть (Neural Network) состоит из разных слоев, связанных друг с другом и работающих над функциями человеческого мозга. Последний учится на огромных объемах данных и использует сложные алгоритмы для обучения нейронной сети.

Вот пример того, как нейронные сети могут определять породу собаки по ее характеристикам:

Пиксели изображения двух разных пород собак поступают на входной слой нейронной сети.
Пиксели изображения обрабатываются в скрытых слоях для выделения признаков.
Выходной слой дает результат, позволяющий определить, является ли это немецкой овчаркой или лабрадором.

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

Почему RNN?

Рекуррентные нейронные сети были созданы, потому что в нейронной сети с прямой связью было несколько проблем:

Не могут обрабатывать последовательные данные
Учитывают только текущую вводную информацию
Невозможно запомнить предыдущий ввод

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

Приложения рекуррентных нейронных сетей

RNN используются для:

Распознавания предметов на изображениях:

RNN узнает собаку с мячом. Фото: Anna Dudkova / Unsplash

Прогнозирования временных рядов: любую задачу временных рядов, такую как прогнозирование цен акций в конкретном месяце, можно решить с помощью RNN.
Обработка естественного языка (NLP): эмоциональная оценка, анализ тональности.
Машинный перевод: текст с одного языка переводится на несколько из них.

RNN: Torch

Рекуррентная нейросеть вполне доступно реализована на Keras. Для начала импортируем все необходимые библиотеки:

import torch
from torch import nn

import numpy as np

Объединим все предложения и вычленим уникальные символы, затем создадим словарь из символов и их порядковых номеров и его инверсированную версию:

text = ['hey how are you','good i am fine','have a nice day']
chars = set(''.join(text))

int2char = dict(enumerate(chars))

char2int = {char: ind for ind, char in int2char.items()}

Найдем наиболее длинное слово. Создадим цикл, который пройдет через список предложений и добавит столько пробелов, чтобы длина у всех предложений стала одинаковая:

maxlen = len(max(text, key = len))

for i in range(len(text)):
  while len(text[i]) < maxlen:
      text[i] += ' '

Создадим списки для входных и целевых последовательностей. В цикле удалим первые символ и букву первой последовательности

input_seq = []
target_seq = []

for i in range(len(text)):
  input_seq.append(text[i][:-1])
    
  target_seq.append(text[i][1:])
  print("Входная последовательность: {}\nЦелевая последовательность: {}".format(input_seq[i], target_seq[i]))

Вот так странновато выглядят входная и целевая последовательности символов:

Входная последовательность: hey how are yo
Целевая последовательность: ey how are you
Входная последовательность: good i am fine
Целевая последовательность: ood i am fine 
Входная последовательность: have a nice da
Целевая последовательность: ave a nice day

Преобразуем с помощью цикла символы в целочисленные значения-псевдонимы:

for i in range(len(text)):
    input_seq[i] = [char2int[character] for character in input_seq[i]]
    target_seq[i] = [char2int[character] for character in target_seq[i]]

Инициируем функцию one_hot_encode и создадим многомерный ряд нулей с желаемым разрешением. Создадим для каждой буквы по объекту, где заменим единицами вхождения этой буквы в предложении и нолями – другие буквы

dict_size = len(char2int)
seq_len = maxlen - 1
batch_size = len(text)

def one_hot_encode(sequence, dict_size, seq_len, batch_size):
    
    features = np.zeros((batch_size, seq_len, dict_size), dtype = np.float32)
    
    for i in range(batch_size):
        for u in range(seq_len):
            features[i, u, sequence[i][u]] = 1
    return features

Входноsе разрешение -> (Размер пакета, длина последовательности, масштаб Горячего кодирования (One-Hot Encoding)):

input_seq = one_hot_encode(input_seq, dict_size, seq_len, batch_size)

Инициируем объекты – входную и целевую последовательности:

input_seq = torch.from_numpy(input_seq)
target_seq = torch.Tensor(target_seq)

torch.cuda.is_available() возвращает булевое True, если графический процессор доступен. Создадим переменную, обозначающую доступность графического процессора, чтобы в дальнейшем использовать ее.

is_cuda = torch.cuda.is_available()

if is_cuda:
    device = torch.device("cuda")
    print('Графический процессор доступен')
else:
    device = torch.device("cpu")
    print("Графический процессор доступен, используем центральный")

Cuda недоступен:

Графический процессор недоступен, используем центральный

Создадим класс модели. Определим некоторые параметры и определим слои RNN. Инициализируем скрытое состояние для первой порции входных данных используя метод ниже. Передаем входные данные и скрытое состояние в модель и получим вывод. Изменим разрешение выходных данных так, чтобы их можно было передать в полносвязный слой. Метод init_hidden генерирует первое скрытое состояние из нолей. Отправим тензор со скрытым состоянием в процессор:

class Model(nn.Module):
    def __init__(self, input_size, output_size, hidden_dim, n_layers):
        super(Model, self).__init__()

        self.hidden_dim = hidden_dim
        self.n_layers = n_layers

        self.rnn = nn.RNN(input_size, hidden_dim, n_layers, batch_first=True)   
        # Полносвязные слои
        self.fc = nn.Linear(hidden_dim, output_size)
    
    def forward(self, x):
        
        batch_size = x.size(0)
        hidden = self.init_hidden(batch_size)

        out, hidden = self.rnn(x, hidden)
        out = out.contiguous().view(-1, self.hidden_dim)
        out = self.fc(out)
        
        return out, hidden
    
    def init_hidden(self, batch_size):
        hidden = torch.zeros(self.n_layers, batch_size, self.hidden_dim)
        return hidden

Создадим экземпляр модели с заданными гиперпараметрами. Подадим модель процессору (центральный по умолчанию). Определим гиперпараметры. Определим тип потерь, функцию оптимизации

model = Model(input_size = dict_size, output_size = dict_size, hidden_dim = 12, n_layers = 1)
model.to(device)

n_epochs = 100
lr = 0.01

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

Запустим тренировку:

for epoch in range(1, n_epochs + 1):
    optimizer.zero_grad() # Очищает существующие градиенты от данных предыдущей эпохи
    input_seq.to(device)
    output, hidden = model(input_seq)
    loss = criterion(output, target_seq.view(-1).long())
    loss.backward() # Осуществляет обратное распространение ошибки и вычисляет градиенты
    optimizer.step() # Обновляет веса последовательно
    
    if epoch%10 == 0:
        print('Epoch: {}/{}.............'.format(epoch, n_epochs), end = ' ')
        print("Loss: {:.4f}".format(loss.item()))

Потери становятся очень показательными, когда рассматриваются в сравнении. И вот теперь отчетливо заметна динамика снижения потерь:

Epoch: 10/100............. Loss: 2.4213
Epoch: 20/100............. Loss: 2.1452
Epoch: 30/100............. Loss: 1.7736
Epoch: 40/100............. Loss: 1.3867
Epoch: 50/100............. Loss: 1.0286
Epoch: 60/100............. Loss: 0.7298
Epoch: 70/100............. Loss: 0.5091
Epoch: 80/100............. Loss: 0.3600
Epoch: 90/100............. Loss: 0.2619
Epoch: 100/100............. Loss: 0.1983

Создадим функцию, которая принимает символы как аргументы и возвращает предсказание – следующий символ. Применим горячее кодирование ко входным данным для передачи в модель. Выбираем класс с наибольшей вероятностью:

def predict(model, character):
    character = np.array([[char2int[c] for c in character]])
    character = one_hot_encode(character, dict_size, character.shape[1], 1)
    character = torch.from_numpy(character)
    character.to(device)
    
    out, hidden = model(character)

    prob = nn.functional.softmax(out[-1], dim = 0).data
    char_ind = torch.max(prob, dim = 0)[1].item()

    return int2char[char_ind], hidden

Эта функция принимает желаемую длину выходные данных и входные символы как аргументы, возвращая сгенерированное предложение:

Теперь посмотрим, какие символы модель ожидает увидеть после слова good: