Transformeri

Preporučujem da pogledate sledeće materijale:
Prvi deo
Drugi deo
https://huggingface.co/blog/designing-positional-encoding
https://huggingface.co/learn/llm-course/chapter1/4

Originalni rad: Attention is all you need

Nakon potpuno povezanih i konvolutivnih neuronskih mreža, prirodno je postaviti pitanje: da li postoji arhitektura koja može efikasno da obradi sekvence, a da pri tome lakše uočava odnose između i vrlo udaljenih delova ulaza? Upravo iz te potrebe razvijen je attention mehanizam, a zatim i čitava porodica modela poznata pod nazivom transformeri.

Transformeri su danas veoma važni u obradi prirodnog jezika, ali se ideja attention mehanizma kasnije proširila i na druge oblasti. Važno je ipak naglasiti da transformeri nisu "izbrisali" ranije arhitekture: u praksi se i dalje koriste i potpuno povezane mreže i konvolutivne mreže.

Kod sekvencijalnih podataka, kao što su rečenice, vremenske serije ili nizovi simbola, često nije dovoljno posmatrati samo lokalne obrasce. Nekada je za donošenje odluke presudna veza između elemenata koji su daleko jedan od drugog. Klasični rekurentni modeli pokušavali su da tu informaciju "prenose" kroz vreme, ali je to u praksi često bilo teško, naročito kada sekvence postanu duže.

Osnovna ideja attention mehanizma jeste da model, pri obradi svakog elementa sekvence, može da proceni koji su mu drugi elementi najvažniji i da na njih "obrati pažnju". Time se informacija ne prenosi isključivo korak po korak, već se relevantne veze mogu uspostaviti direktnije.

Attention mehanizam

Za svaki element sekvence formiraju se tri reprezentacije:

• query (Q),
• key (K),
• value (V).

Intuitivno, query možemo posmatrati kao pitanje koje trenutno postavljamo, key kao opis onoga što svaki element nudi, a value kao informaciju koju zapravo preuzimamo ukoliko procenimo da je taj element važan.

Osnovna formula za tzv. scaled dot-product attention glasi

$$ \mathrm{Attention}(Q, K, V) = \mathrm{softmax}\left(\frac{QK^T}{\sqrt{d_k}}\right)V. $$

Matrica $QK^T$ meri sličnost između upita i ključeva, faktor $\sqrt{d_k}$ služi za stabilizaciju vrednosti pre ulaska u funkciju softmax, dok softmax pretvara dobijene skorove u težine koje možemo tumačiti kao relativnu pažnju usmerenu ka pojedinim elementima.

Kada se isti ulaz koristi za formiranje i upita i ključeva i vrednosti, govorimo o self-attention mehanizmu. To znači da svaki element sekvence može da "pogleda" sve ostale elemente iste sekvence i da u odnosu na njih formira novu reprezentaciju. Upravo taj mehanizam predstavlja centralni deo transformera.

U praksi se često koristi i multi-head attention. Umesto jedne jedine matrice pažnje, model formira više „glava“, pri čemu svaka glava može da nauči drugačiji tip odnosa između tokena:

$$ \mathrm{MultiHead}(Q, K, V) = \mathrm{Concat}(\mathrm{head}_1, \dots, \mathrm{head}_h) W^O. $$

Na taj način model ne traži samo jednu vrstu zavisnosti, već istovremeno može da uočava više različitih obrazaca.

Pošto transformer nema rekurenciju, on sam po sebi ne zna kojim redosledom elementi dolaze u sekvenci. Zbog toga je potrebno da se informacija o poziciji eksplicitno doda. Jedan od klasičnih pristupa jeste sinusoidno poziciono enkodovanje:

$$ PE(\mathrm{pos}, 2i) = \sin\left(\frac{\mathrm{pos}}{10000^{2i/d}}\right), \qquad PE(\mathrm{pos}, 2i+1) = \cos\left(\frac{\mathrm{pos}}{10000^{2i/d}}\right). $$

Na ovaj način svakoj poziciji u sekvenci pridružujemo vektor iste dimenzije kao i vektor ugnežđenja reči, pa model dobija i informaciju o sadržaju i informaciju o redosledu.

Enkoder i dekoder

Originalna transformer arhitektura sastoji se iz dva dela: enkodera i dekodera. Enkoder prima ulaznu sekvencu i formira njenu bogatiju reprezentaciju, dok dekoder koristi tu reprezentaciju kako bi generisao izlaznu sekvencu, najčešće korak po korak.

Jedan enkoderski blok tipično sadrži:

1. multi-head self-attention,
2. rezidualnu vezu i normalizaciju,
3. potpuno povezanu mrežu primenjenu nezavisno po pozicijama,
4. još jednu rezidualnu vezu i normalizaciju.

U zadacima klasifikacije često je dovoljan samo enkoderski deo modela, dok su za generisanje teksta važni i dekoderski mehanizmi i maskiranje budućih tokena.

Kvaliteti i mane

Transformeri imaju nekoliko veoma važnih prednosti. Pre svega, omogućavaju da se odnosi između udaljenih delova sekvence modeluju direktno. Pored toga, obrada svih tokena može se u velikoj meri paralelizovati, što je posebno značajno pri radu sa velikim skupovima podataka i snažnim hardverom.

Sa druge strane, cena takve fleksibilnosti može biti velika računska složenost, naročito kada sekvence postanu veoma duge. Takođe, transformeri obično zahtevaju više podataka. Zbog toga u praksi i dalje biramo arhitekturu prema zadatku, a ne prema tome šta je trenutno "najmodernije".

U obradi prirodnog jezika transformeri su postali centralna arhitektura. U obradi slika njihov uticaj je takođe veliki. Danas se u praksi sreću i čisti konvolutivni modeli, i modeli zasnovani na transformerima, i razne kombinacije ova dva pristupa.

Sada ćemo napraviti mali model za klasifikaciju sentimenta tekstova na pozitivne i negativne. Prvo ćemo učitati biblioteke koje će nam biti potrebne u nastavku.

import math
import random
import re
from collections import Counter

import matplotlib.pyplot as plt
import numpy as np
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset
from sklearn.model_selection import train_test_split

torch.set_num_threads(1)
device = torch.device('cpu')

Formiranje malog korpusa

Kako bismo zadržali fokus na samoj arhitekturi, napravićemo mali skup kratkih rečenica. Rečenice će imati pozitivnu ili negativnu ocenu. Na taj način možemo lako da pripremimo podatke, a da se i dalje jasno vidi kako se transformer koristi u klasifikaciji.

subjects = [
    "film", "ovaj film", "serija", "ova serija", "dokumentarac", "predavanje",
    "kurs", "aplikacija", "knjiga", "album", "nastup", "projekat"]

positive_phrases = [
    "je veoma zanimljiv", "je odlično urađen",
    "mi se mnogo dopao", "je bio vrlo koristan",
    "ostavlja veoma dobar utisak", "je kvalitetan i jasan",
    "je prijatno iznenađenje", "vredi preporučiti drugima"]

negative_phrases = [
    "je veoma dosadan", "je loše urađen",
    "mi se nimalo nije dopao", "je bio veoma naporan",
    "ostavlja loš utisak", "je konfuzan i slab",
    "je veliko razočaranje", "ne vredi preporučiti drugima"]

texts = []
labels = []

for s in subjects:
    for p in positive_phrases:
        texts.append(f"{s} {p}")
        labels.append(1)
    for p in negative_phrases:
        texts.append(f"{s} {p}")
        labels.append(0)

combined = list(zip(texts, labels))
random.shuffle(combined)
texts, labels = zip(*combined)
texts, labels = list(texts), list(labels)

print("Ukupan broj instanci:", len(texts))
print("Broj pozitivnih:", sum(labels))
print("Broj negativnih:", len(labels) - sum(labels))
print()
for i in range(5):
    print(f"{labels[i]} -> {texts[i]}")

Ukupan broj instanci: 192
Broj pozitivnih: 96
Broj negativnih: 96

1 -> knjiga vredi preporučiti drugima
0 -> predavanje je veliko razočaranje
0 -> aplikacija je veliko razočaranje
1 -> ovaj film je prijatno iznenađenje
0 -> nastup je veliko razočaranje

Tokenizacija i rečnik

Transformeri ne rade direktno nad sirovim tekstom. Najpre je potrebno tekst rastaviti na tokene, a zatim svakom tokenu pridružiti broj. Za potrebe ovog primera koristićemo veoma jednostavnu tokenizaciju zasnovanu na malim slovima i razdvajanju po rečima.

def tokenize(text):
    return re.findall(r"[a-zA-ZčćžšđČĆŽŠĐ]+", text.lower())

special_tokens = ["<pad>", "<unk>", "<cls>"]
counter = Counter()

for text in texts:
    counter.update(tokenize(text))

vocab = special_tokens + sorted(counter.keys())
stoi = {token: idx for idx, token in enumerate(vocab)}
itos = {idx: token for token, idx in stoi.items()}

PAD_IDX = stoi["<pad>"]
UNK_IDX = stoi["<unk>"]
CLS_IDX = stoi["<cls>"]

max_len = max(len(tokenize(text)) for text in texts) + 1  # +1 zbog <cls> tokena
print("Veličina rečnika:", len(vocab))
print("Maksimalna dužina sekvence:", max_len)

def encode(text):
    tokens = ["<cls>"] + tokenize(text)
    ids = [stoi.get(token, UNK_IDX) for token in tokens]
    if len(ids) < max_len:
        ids = ids + [PAD_IDX] * (max_len - len(ids))
    else:
        ids = ids[:max_len]
    return ids

encoded_texts = torch.tensor([encode(text) for text in texts], dtype=torch.long)
labels_tensor = torch.tensor(labels, dtype=torch.long)

print()
print("Primer originalne rečenice:", texts[0])
print("Primer tokena:", ["<cls>"] + tokenize(texts[0]))
print("Primer kodirane sekvence:", encoded_texts[0].tolist())

Veličina rečnika: 49
Maksimalna dužina sekvence: 8

Primer originalne rečenice: knjiga vredi preporučiti drugima
Primer tokena: ['<cls>', 'knjiga', 'vredi', 'preporučiti', 'drugima']
Primer kodirane sekvence: [2, 16, 46, 35, 10, 0, 0, 0]

Podela na trening, validacioni i test skup

Izdvojićemo validacioni skup kako bismo tokom treniranja pratili da li model zaista napreduje. Pošto je naš skup mali i uravnotežen, podelu ćemo napraviti tako da obe klase ostanu zastupljene u svim delovima skupa.

X_train, X_temp, y_train, y_temp = train_test_split(
    encoded_texts,
    labels_tensor,
    test_size=0.3,
    stratify=labels_tensor,
    random_state=42)

X_val, X_test, y_val, y_test = train_test_split(
    X_temp,
    y_temp,
    test_size=0.5,
    stratify=y_temp,
    random_state=42)

# TensorDataset
train_dataset = TensorDataset(X_train, y_train)
val_dataset = TensorDataset(X_val, y_val)
test_dataset = TensorDataset(X_test, y_test)

# DataLoader
batch_size = 16

train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False)
test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False)

print("Broj instanci u trening skupu:", len(train_dataset))
print("Broj instanci u validacionom skupu:", len(val_dataset))
print("Broj instanci u test skupu:", len(test_dataset))

Broj instanci u trening skupu: 134
Broj instanci u validacionom skupu: 29
Broj instanci u test skupu: 29

Pogledajmo kako izgleda jedna mini-serija podataka.

batch_inputs, batch_labels = next(iter(train_loader))
print("Oblik ulaza:", batch_inputs.shape)
print("Oblik ciljeva:", batch_labels.shape)
print()
print("Prva instanca iz mini-serije (indeksi tokena):")
print(batch_inputs[0])
print("Prva oznaka:", batch_labels[0].item())

Oblik ulaza: torch.Size([16, 8])
Oblik ciljeva: torch.Size([16])

Prva instanca iz mini-serije (indeksi tokena):
tensor([ 2,  7, 15, 45, 48,  0,  0,  0])
Prva oznaka: 1

Poziciono enkodovanje

Najpre definišemo modul za sinusoidno poziciono enkodovanje. Ono se sabira sa ugnežđenjima tokena, tako da model pri svakoj poziciji dobija i informaciju o samom tokenu i informaciju o mestu tog tokena u sekvenci.

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len):
        super().__init__()
        pe = torch.zeros(max_len, d_model)
        position = torch.arange(0, max_len, dtype=torch.float32).unsqueeze(1)
        div_term = torch.exp(
            torch.arange(0, d_model, 2, dtype=torch.float32) * (-math.log(10000.0) / d_model))

        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)

        pe = pe.unsqueeze(0)  # (1, max_len, d_model)
        self.register_buffer("pe", pe)

    def forward(self, x):
        return x + self.pe[:, :x.size(1), :]

Pojednostavljeni transformer enkoder

U nastavku definišemo pojednostavljeni model sa jednim enkoderskim blokom. Model sadrži:

• sloj ugnežđenja tokena,
• poziciono enkodovanje,
• multi-head self-attention,
• rezidualne veze i normalizaciju,
• mali potpuno povezani podsistem,
• klasifikacioni sloj koji koristi reprezentaciju <cls> tokena.

class TransformerClassifier(nn.Module):
    def __init__(self, vocab_size, d_model=64, num_heads=4, ff_dim=128, num_classes=2, max_len=32, dropout=0.1):
        super().__init__()
        self.d_model = d_model
        self.embedding = nn.Embedding(vocab_size, d_model, padding_idx=PAD_IDX)
        self.positional_encoding = PositionalEncoding(d_model, max_len)

        self.attention = nn.MultiheadAttention(
            embed_dim=d_model,
            num_heads=num_heads,
            dropout=dropout,
            batch_first=True)

        self.norm1 = nn.LayerNorm(d_model)
        self.feed_forward = nn.Sequential(
            nn.Linear(d_model, ff_dim),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(ff_dim, d_model))
        self.norm2 = nn.LayerNorm(d_model)
        self.dropout = nn.Dropout(dropout)
        self.classifier = nn.Linear(d_model, num_classes)

    def forward(self, x, return_attention=False):
        mask = (x == PAD_IDX)
        x = self.embedding(x) * math.sqrt(self.d_model)
        x = self.positional_encoding(x)

        attn_output, attn_weights = self.attention(
            x, x, x,
            key_padding_mask=mask,
            need_weights=True,
            average_attn_weights=False)

        x = self.norm1(x + self.dropout(attn_output))
        ff_output = self.feed_forward(x)
        x = self.norm2(x + self.dropout(ff_output))

        cls_representation = x[:, 0, :]
        logits = self.classifier(cls_representation)

        if return_attention:
            return logits, attn_weights
        return logits

Kao i ranije, možemo pogledati osnovnu strukturu modela i broj parametara koje mreža treba da nauči.

model = TransformerClassifier(
    vocab_size=len(vocab), d_model=32, num_heads=4,
    ff_dim=64, num_classes=2, max_len=max_len, dropout=0.1).to(device)

print(model)

num_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
print()
print("Broj parametara:", num_params)

TransformerClassifier(
  (embedding): Embedding(49, 32, padding_idx=0)
  (positional_encoding): PositionalEncoding()
  (attention): MultiheadAttention(
    (out_proj): NonDynamicallyQuantizableLinear(in_features=32, out_features=32, bias=True)
  )
  (norm1): LayerNorm((32,), eps=1e-05, elementwise_affine=True)
  (feed_forward): Sequential(
    (0): Linear(in_features=32, out_features=64, bias=True)
    (1): ReLU()
    (2): Dropout(p=0.1, inplace=False)
    (3): Linear(in_features=64, out_features=32, bias=True)
  )
  (norm2): LayerNorm((32,), eps=1e-05, elementwise_affine=True)
  (dropout): Dropout(p=0.1, inplace=False)
  (classifier): Linear(in_features=32, out_features=2, bias=True)
)

Broj parametara: 10178

Za funkciju gubitaka koristićemo unakrsnu entropiju, a za optimizaciju Adamov gradijentni spust.

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

Definisaćemo pomoćne funkcije za jednu epohu treniranja i za evaluaciju modela.

def run_epoch(model, dataloader, loss_fn, optimizer=None, device="cpu"):
    if optimizer is None:
        model.eval()
    else:
        model.train()

    total_loss = 0.0
    total_correct = 0
    total_examples = 0

    for X_batch, y_batch in dataloader:
        X_batch = X_batch.to(device)
        y_batch = y_batch.to(device)

        if optimizer is not None:
            optimizer.zero_grad()

        logits = model(X_batch)
        loss = loss_fn(logits, y_batch)

        if optimizer is not None:
            loss.backward()
            optimizer.step()

        total_loss += loss.item() * X_batch.size(0)
        predictions = logits.argmax(dim=1)
        total_correct += (predictions == y_batch).sum().item()
        total_examples += X_batch.size(0)

    avg_loss = total_loss / total_examples
    avg_accuracy = total_correct / total_examples
    return avg_loss, avg_accuracy

Model ćemo trenirati u 20 epoha i pratiti vrednost funkcije gubitaka i tačnost na trening i validacionom skupu.

epochs = 20

history = {
    "train_loss": [],
    "train_accuracy": [],
    "val_loss": [],
    "val_accuracy": []}

for epoch in range(epochs):
    train_loss, train_acc = run_epoch(model, train_loader, loss_fn, optimizer=optimizer, device=device)
    val_loss, val_acc = run_epoch(model, val_loader, loss_fn, optimizer=None, device=device)

    history["train_loss"].append(train_loss)
    history["train_accuracy"].append(train_acc)
    history["val_loss"].append(val_loss)
    history["val_accuracy"].append(val_acc)

    print(
        f"Epoha {epoch + 1:02d}/{epochs} | "
        f"train loss = {train_loss:.4f}, train acc = {train_acc:.4f} | "
        f"val loss = {val_loss:.4f}, val acc = {val_acc:.4f}")

Epoha 01/20 | train loss = 0.7246, train acc = 0.4851 | val loss = 0.6319, val acc = 0.7586
Epoha 02/20 | train loss = 0.6508, train acc = 0.6493 | val loss = 0.6185, val acc = 0.6552
Epoha 03/20 | train loss = 0.6194, train acc = 0.6418 | val loss = 0.5882, val acc = 0.7931
Epoha 04/20 | train loss = 0.5649, train acc = 0.8358 | val loss = 0.5509, val acc = 0.7931
Epoha 05/20 | train loss = 0.4953, train acc = 0.8806 | val loss = 0.5135, val acc = 0.7586
Epoha 06/20 | train loss = 0.4165, train acc = 0.9030 | val loss = 0.4327, val acc = 0.7931
Epoha 07/20 | train loss = 0.3301, train acc = 0.9104 | val loss = 0.3625, val acc = 0.7931
Epoha 08/20 | train loss = 0.2358, train acc = 0.9104 | val loss = 0.3101, val acc = 0.8276
Epoha 09/20 | train loss = 0.1711, train acc = 0.9403 | val loss = 0.1905, val acc = 0.9655
Epoha 10/20 | train loss = 0.1076, train acc = 0.9701 | val loss = 0.1455, val acc = 0.9655
Epoha 11/20 | train loss = 0.1338, train acc = 0.9403 | val loss = 0.0699, val acc = 1.0000
Epoha 12/20 | train loss = 0.0703, train acc = 0.9776 | val loss = 0.0389, val acc = 1.0000
Epoha 13/20 | train loss = 0.0620, train acc = 0.9851 | val loss = 0.0316, val acc = 1.0000
Epoha 14/20 | train loss = 0.0604, train acc = 0.9701 | val loss = 0.0340, val acc = 1.0000
Epoha 15/20 | train loss = 0.1118, train acc = 0.9701 | val loss = 0.0583, val acc = 0.9655
Epoha 16/20 | train loss = 0.0713, train acc = 0.9776 | val loss = 0.0658, val acc = 0.9655
Epoha 17/20 | train loss = 0.0702, train acc = 0.9776 | val loss = 0.0362, val acc = 1.0000
Epoha 18/20 | train loss = 0.0618, train acc = 0.9776 | val loss = 0.0174, val acc = 1.0000
Epoha 19/20 | train loss = 0.0488, train acc = 0.9851 | val loss = 0.0120, val acc = 1.0000
Epoha 20/20 | train loss = 0.0249, train acc = 0.9925 | val loss = 0.0133, val acc = 1.0000

Možemo nacrtati i grafike funkcije gubitaka i tačnosti u toku treniranja.

plt.figure(figsize=(12, 4))

plt.subplot(1, 2, 1)
plt.title("Funkcija gubitaka")
plt.plot(np.arange(1, epochs + 1), history["train_loss"], label="train")
plt.plot(np.arange(1, epochs + 1), history["val_loss"], label="validation")
plt.xlabel("Epoha")
plt.legend()

plt.subplot(1, 2, 2)
plt.title("Tačnost")
plt.plot(np.arange(1, epochs + 1), history["train_accuracy"], label="train")
plt.plot(np.arange(1, epochs + 1), history["val_accuracy"], label="validation")
plt.xlabel("Epoha")
plt.legend()

plt.tight_layout()
plt.show()

Model ćemo oceniti na test skupu.

test_loss, test_accuracy = run_epoch(model, test_loader, loss_fn, optimizer=None, device=device)

print("Test loss:", round(test_loss, 4))
print("Test accuracy:", round(test_accuracy, 4))

Test loss: 0.0735
Test accuracy: 0.9655

Pogledajmo nekoliko pojedinačnih predikcija modela.

def predict_sentiment(model, text):
    model.eval()
    encoded = torch.tensor([encode(text)], dtype=torch.long).to(device)

    with torch.no_grad():
        logits = model(encoded)
        probabilities = torch.softmax(logits, dim=1).cpu().numpy()[0]

    predicted_label = int(np.argmax(probabilities))
    return predicted_label, probabilities

label_names = {0: "negativno", 1: "pozitivno"}

examples = [
    "ovaj film je veoma zanimljiv",
    "kurs je bio veoma naporan",
    "knjiga je kvalitetna i jasna",
    "projekat je veliko razočaranje"
]

for text in examples:
    pred, probs = predict_sentiment(model, text)
    print(f"Rečenica: {text}")
    print(f"Predikcija: {label_names[pred]} | verovatnoće = {np.round(probs, 3)}")
    print()

Rečenica: ovaj film je veoma zanimljiv
Predikcija: pozitivno | verovatnoće = [0.015 0.985]

Rečenica: kurs je bio veoma naporan
Predikcija: negativno | verovatnoće = [0.997 0.003]

Rečenica: knjiga je kvalitetna i jasna
Predikcija: negativno | verovatnoće = [0.992 0.008]

Rečenica: projekat je veliko razočaranje
Predikcija: negativno | verovatnoće = [0.998 0.002]

Vizuelizacija pažnje

Jedna od najzanimljivijih osobina attention mehanizma jeste to što možemo da pogledamo matricu pažnje. Na taj način stičemo grubu predstavu o tome na koje pozicije u sekvenci model obraća više pažnje pri formiranju novih reprezentacija.

Treba imati u vidu da ova interpretacija nije savršena i da matrica pažnje ne daje celokupno objašnjenje rada modela, ali je ipak veoma korisna za sticanje intuicije.

def attention_for_sentence(model, text):
    model.eval()
    encoded = torch.tensor([encode(text)], dtype=torch.long).to(device)

    with torch.no_grad():
        logits, attn_weights = model(encoded, return_attention=True)

    # attn_weights: (batch, num_heads, target_len, source_len)
    attn = attn_weights[0].mean(dim=0).cpu().numpy()

    tokens = ["<cls>"] + tokenize(text)
    length = len(tokens)
    attn = attn[:length, :length]

    return tokens, attn

text = "ovaj film mi se mnogo dopao"
tokens, attn = attention_for_sentence(model, text)

plt.figure(figsize=(6, 5))
plt.imshow(attn)
plt.xticks(np.arange(len(tokens)), tokens, rotation=45)
plt.yticks(np.arange(len(tokens)), tokens)
plt.title("Prosečna attention matrica")
plt.colorbar()
plt.tight_layout()
plt.show()