В этой статье я собираюсь рассмотреть знаменитый генеративный предварительно обученный преобразователь из статьи Улучшение понимания языка с помощью генеративного предварительного обучения. В этой статье предлагается изучить генеративную языковую модель с использованием неразмеченных данных, а затем отрегулировать модель, предоставив примеры конкретных последующих задач, таких как классификация, анализ тональности, вывод текста и т. д. Это первая GPT из серии GPT-n, и оказалось, очень перспективный подход. Следующие версии GPT сохраняют ту же архитектуру, но увеличивают количество параметров и объем обучающих данных.

Во-первых, я рассматриваю теорию архитектуры и ее компонентов. Во-вторых, я собираюсь реализовать GPT в PyTorch с нуля и протестировать его на задаче генерации имен.

Реализация не полностью оптимизирована для обеспечения высокой производительности, поскольку основная цель кода — образовательная. Полная реализация, готовая к запуску, доступна на Google Colab и GitHub.

Теория

Обзор архитектуры

В целом архитектура GPT основана на преобразователе. Архитектура преобразователя состоит из блоков энкодера и декодера:

GPT построен из декодеров без вывода кодировщика и, следовательно, без операции внимания кодировщик-декодер:

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

В целом архитектура GPT, представленная в статье Улучшение понимания языка с помощью генеративного предварительного обучения выглядит следующим образом:

Вложения

Для векторизации токенов мы используем линейное преобразование; для сохранения позиционной информации о токенах используются позиционные вложения, которые также являются результатом линейного преобразования. Обратите внимание, что в оригинальном преобразователе (статья «Внимание — это все, что вам нужно») авторы используют синусоидальное кодирование, тогда как авторы GPT используют обычное обучаемое линейное преобразование.

Замаскированное внимание к себе

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

Учитывая последовательность векторов токенов v_i, модель вычисляет вес w_i для каждого v_i так, что

Средневзвешенное значение

является результатом работы механизма внимания. Вес w_i получается из запросов и ключей:

Значения V являются векторами v_i.

Если запросы и ключи получены из кодировщика, это называется вниманием кодировщика-декодера. Если запросы и ключи получены из одних и тех же входных векторов v_i , это называется самостоятельным вниманием.

Для повышения производительности модели используется несколько головок механизма внутреннего внимания. Это означает, что изначально векторы Q, V, K разбиваются на n частей (где n — количество головок) и обрабатываются независимо друг от друга. После операции внутреннего внимания результаты объединяются и передаются на линейный уровень.

Подача вперед NN

Нейронная сеть с прямой связью является последней частью структуры декодера. Он состоит из двух линейных слоев, GELU и дропаута:

Первое линейное преобразование определяется с помощью (d_model, d_ff), а второе определяется с помощью (d_ff, d_model), где d_model – размер вложений токенов иd_ffявляетсяскрытым размером состояния для этого Feed Forward NN. GELU (линейная единица ошибки Гаусса) — это функция активации:

Выполнение

Мы собираемся внедрить архитектуру GPT с помощью PyTorch и протестировать задачу генерации фамилии.

Архитектура модели

Начнем с контура слоя декодера:

class DecoderLayer(nn.Module):
    def __init__(self, d_model: int, heads: int, dropout: float = 0.1, d_ff: int = 2048):
        super().__init__()
        self.norm_1 = Norm(d_model)
        self.norm_2 = Norm(d_model)
        self.dp = nn.Dropout(dropout)
        self.attn = MultiHeadAttention(heads, d_model, dropout, masked=True)
        self.ff = FeedForward(d_model, d_ff=d_ff)

    def forward(self, x):
        x = self.norm_1(x + self.dp(self.attn(x, x, x)))
        x = self.norm_2(x + self.dp(self.ff(x)))
        return x

Теперь давайте реализуем MultiHead Attention. Он состоит из механизма внимания и трех линейных преобразований, предназначенных для обработки входных запросов, ключей и значений; кроме того, он содержит конечный линейный слой.

class MultiHeadAttention(nn.Module):
    def __init__(self, heads: int, d_model: int, dropout: float = 0.1, masked: bool = False):
        """
        params
        ---
        heads : number of heads
        d_model : model dimension (embeddings size)
        dropout :
        """
        super().__init__()

        self.d_model = d_model
        self.d_k = d_model // heads
        self.h = heads

        self.q_linear = nn.Linear(d_model, d_model)
        self.v_linear = nn.Linear(d_model, d_model)
        self.k_linear = nn.Linear(d_model, d_model)

        self.attention = ScaledDotProductAttention(masked=masked)
        self.dropout = nn.Dropout(dropout)
        self.out = nn.Linear(d_model, d_model)

Как это работает?

def forward(self, q, k, v):
        """
        params
        ---
        q : tensor [batch_size, length, emb_dim]
        k : tensor [batch_size, length, emb_dim]
        v : tensor [batch_size, length, emb_dim]
        returns
        ---
        output : [batch_size, length, emb_dim]
        """
        bs = q.size(0)
        # perform linear operation, split into h heads and get [bs, h, length, d_k]
        k = self.k_linear(k).view(bs, -1, self.h, self.d_k).transpose(1, 2)
        q = self.q_linear(q).view(bs, -1, self.h, self.d_k).transpose(1, 2)
        v = self.v_linear(v).view(bs, -1, self.h, self.d_k).transpose(1, 2)

        scores = self.attention(q, k, v)

        # concatenate heads and put through final linear layer
        concat = scores.transpose(1, 2).contiguous().view(bs, -1, self.d_model)
        output = self.out(concat)
        return output

Требуется Q, Kи V, которые в случае GPT являются просто входными векторамиX: Q = K = V = X.После линейного преобразования нам нужно разбить векторы результатов на h частей: по одной части каждой головы. Мы можем думать об этом как об ортогональной проекции векторов в d_k = d_model // hпространства. После этого применяется маскированное внимание:

class ScaledDotProductAttention(nn.Module):
    def __init__(self, masked: bool = False):
        super().__init__()
        self.masked = masked
        if masked:
            self.register_buffer("mask", torch.tril(torch.ones(512, 512))
                                 .view(1, 1, 512, 512))

    def forward(self, q, k, v):
        """
        q : tensor [batch_size, heads, length, d_model//heads]
        k : tensor [batch_size, heads, length, d_model//heads]
        v : tensor [batch_size, heads, length, d_model//heads]
        """
        d_k = k.shape[-1]
        scores = (q @ k.transpose(-2, -1)) / math.sqrt(d_k)
        if self.masked:
            scores = scores.masked_fill(self.mask[:, :, :q.shape[2], :q.shape[2]] == 0, float('-inf'))
        scores = torch.softmax(scores, dim=-1)
        output = scores @ v
        return output

После операции внимания блок MultiHead Attention объединяет все векторы h и применяет окончательное линейное преобразование.

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

class Norm(nn.Module):
    def __init__(self, d_model: int, eps: float = 1e-6):
        super().__init__()
        self.size = d_model
        # create two learnable parameters to calibrate normalisation
        self.weight = nn.Parameter(torch.ones(self.size))
        self.bias = nn.Parameter(torch.zeros(self.size))
        self.eps = eps

    def forward(self, x):
        norm = self.weight * (x - x.mean(dim=-1, keepdim=True)) \
               / (x.std(dim=-1, keepdim=True) + self.eps) + self.bias
        return 

Наконец, соберите все вместе:

class GPT(nn.Module):
    def __init__(self,
                 vocab_size: int,
                 d_model: int,
                 n_layers: int,
                 heads: int,
                 d_ff: int = 2048,
                 dropout: float = 0.1,
                 max_length: int = 128,
                 device: str = 'cpu'):
        """
        params
        ---
        vocab_size : now many unique tokens
        d_model : tokens embeddings size
        n_layers : how many decoder layers
        heads : how many heads in attention block
        d_ff : embeddings size in decoder's feed forward NN
        dropout : probability of dropout
        max_length : maximum length of token sequence
        device : cpu or cuda
        """
        super().__init__()
        self.device = device
        self.max_length = max_length
        self.embedder = nn.Embedding(vocab_size, d_model)
        self.pe = nn.Embedding(max_length, d_model)
        self.layers = nn.ModuleList([DecoderLayer(d_model, heads, dropout, d_ff) for _ in range(n_layers)])
        self.ff = nn.Linear(d_model, vocab_size)
        self.sm = nn.Softmax(dim=1)
        self.to(device)

    def count_parameters(self):
        return sum(p.numel() for p in self.parameters() if p.requires_grad)

    def forward(self, tokens):
        """
        tokens : [bs, length]
        """
        pos = torch.arange(0, tokens.shape[1], dtype=torch.long).unsqueeze(0).to(self.device)
        x = self.embedder(tokens) + self.pe(pos)
        for layer in self.layers:
            x = layer(x)
        logits = self.ff(x)
        return logits

    @torch.no_grad()
    def generate(self, idx, idx2token, k=3, temperature=1.0):
      self.eval()
      for _ in range(self.max_length):
          logits = self(idx)
          # pluck the probs at the final step and scale by desired temperature
          logits = logits[:, -1, :] / temperature
          probs = self.sm(logits)
          probs, indicies = torch.topk(probs, k=k)
          idx_next = indicies[0][torch.multinomial(probs, 1)[0][0]]
          if idx2token[idx_next.item()] == '[STOP]':
            break
          idx = torch.cat((idx, idx_next[None][None]), dim=1)
      result = [idx2token[i.item()] for i in idx[0] if i.item() not in {0, 1}]
      return ''.join(result).strip()

Обучение

Мы собираемся протестировать реализацию генерации имен и фамилий (например, Тед Смит). Набор данных выглядит следующим образом:

Renee Deleon
Laurie Sutton
Janice Hodge
Sandy Alvarado
Patrice Fields
Morgan Sullivan

Каждая строка представляет собой одну часть входных данных. Модель будет отслеживать каждую входную букву за буквой и предсказывать следующую букву. Подготовим данные:

with open("names.txt", 'r') as f:
  names = [line.lower() for line in f.read().split('\n')]

# create vocabulary of tokens (letters) with stop and padding tokens 
vocab = ['[PAD]', '[STOP]'] + sorted(list(set(''.join(names))))
idx2token = {k:v for k,v in enumerate(vocab)}
token2idx = {v:k for k,v in idx2token.items()}
print(len(vocab))

# create input and target by shifting the text by one letter
# e.g. "river" -> "river", "iver[PAD]" 
X, Y = [], []
for name in names:
  tokens = list(map(token2idx.get, name))[:MAX_LENGTH-1] + [token2idx['[STOP]']]
  while len(tokens) < MAX_LENGTH:
    tokens.append(token2idx['[PAD]'])
  x, y = tokens, tokens[1:] + [token2idx['[PAD]']]
  X.append(torch.tensor(x))
  Y.append(torch.tensor(y))

Затем мы загружаем эти данные в даталоадер и переходим к обучающей части, которая вполне стандартна:

model = GPT(VOCAB_SIZE, EMB_DIM, N_LAYERS, N_HEADS, 512, dropout=DROPOUT, max_length=MAX_LENGTH, device=DEVICE)
print(model.count_parameters())
optimizer = torch.optim.Adam(model.parameters(), lr=LR)

for epoch in range(N_EPOCHS):
  epoch_loss_value = 0.
  for x, y in dataloader:
    x, y = x.to(DEVICE), y.to(DEVICE)
    probs = model(x)
    loss_value = F.cross_entropy(probs.view(-1, probs.size(-1)), y.view(-1), ignore_index=0)
    loss_value.backward()
    optimizer.step()
    optimizer.zero_grad()
    epoch_loss_value += loss_value.item()
  
  if epoch % LOG_EVERY_N == 0:
    print(epoch_loss_value/len(dataloader))

Заключение

Мы рассмотрели одну из самых популярных архитектур для языкового моделирования. Основная часть этой архитектуры — маскированный механизм самоконтроля, который 1) делает модель авторегрессивной, маскируя токены прямо к текущему 2) заставляет модель изучать наиболее важные части данных. Решая задачу генерации языка, GPT приобретает надежные «знания» языковой семантики и контекста, которые широко используются в последующих задачах, таких как классификация текста, ответы на вопросы, обобщение текста и другие.

Ссылки

  1. Рэдфорд, А., Нарасимхан, К., Салиманс, Т. и Суцкевер, И., 2018. Улучшение понимания языка с помощью генеративного предварительного обучения.
  2. Васвани А., Шазир Н., Пармар Н., Ушкорейт Дж., Джонс Л., Гомес А.Н., Кайзер Л. и Полосухин И., 2017. Внимание — это все, что вам нужно. В NIPS.