Bir Bakışta Yapay Sinir Ağları 11: Bir transformatörün uygulanması

Nöronal ağlar birçok AI ve Genai uygulamasının motorudur. Bu makale dizisi bireysel unsurlar hakkında bir fikir sunar. Onbirinci ve son kısım bir transformatör uygular.

Dr. Michael Stal, 1991 yılından bu yana Siemens teknolojisi üzerinde çalışmaktadır. Yazılım mimarisi sorunları hakkında iş alanları önermektedir ve Siemens'in kıdemli yazılım mimarlarının mimari oluşumundan sorumludur.

Transformatör, tamamen dikkat mekanizmalarına dayanan ve paralel dizileri işlemek için özyineleme ve katlama gerektirmeyen bir nöral mimaridir. En önemli yeniliği, aynı zamanda girdideki tüm pozisyonlar arasındaki ilişkileri hesaplayan ölçekli ürün üzerindeki görevdir. Bir transformatör kodu, her biri artık bağlantılara dahil edilen konumla ilgili başlık ve ileri besleme ağlarına birkaç düzey kendi kendine postlama istifler. Kod çözücü, bir nesil özdenetim sağlamak için maskeli bir benlik yetkisi ve kodlayıcı kod çözücünün dikkatini ekler.

Tırmanma ürününün dikkatini ölçeklendirmeye başlıyoruz. Sorguların bildirimleri durumunda, anahtar ve q, k ve v değeri formlarla (batch_size,, num_heads,, seq_len,, d_k) Q noktasının noktasını oluşturan ham değerleri K'nin geçişiyle hesaplıyoruz. Bu nedenle, aşırı değerlerin kaybolma gradyanlarına yol açmasını önlemek için √d_k ile bu değerleri azaltıyoruz, ABD Softmax dikkat ağırlıklarını korumak ve saygın baskıyı elde etmek için çoğalıyoruz:

Dikkat (q, k, v) = softmax ((q · kᵀ) / √d_k) · v

Pytorch'ta aşağıdaki gibi uygulanabilir:


import torch
import torch.nn.functional as F

def scaled_dot_product_attention(Q, K, V, mask=None):
    """
    Berechnet die skalierte Skalarprodukt-Aufmerksamkeit.
    
    Q, K, V haben die Form (batch_size, num_heads, seq_len, d_k).
    Mask, falls angegeben, wird zu den Bewertungen hinzugefügt, um die Aufmerksamkeit auf bestimmte Positionen zu verhindern.
    """
    d_k = Q.size(-1)
    # Berechne die rohen Aufmerksamkeitswerte.
    scores = torch.matmul(Q, K.transpose(-2, -1)) / torch.sqrt(torch.tensor(d_k, dtype=torch.float32))
    # Wende die Maske an (z. B. um zu verhindern, dass nachfolgende Token im Decoder Beachtung finden).
    if mask is not None:
        scores = scores + mask
    # Normalisieren, um Aufmerksamkeitsgewichte zu erhalten.
    attn_weights = F.softmax(scores, dim=-1)
    # Berechne die gewichtete Summe der Werte.
    output = torch.matmul(attn_weights, V)
    return output, attn_weights

Bu işlevde boyutu çıkarıyoruz d_k Q'dan skaler ürünler, Softmax'ın önüne bir maske hesaplar, tırmanır ve isteğe bağlı olarak ekler. Maske, kabul edilemez pozisyonlarda büyük negatif değerler (−∞) içerir, böylece bu konumlar Softmax'tan sonra sıfır ağırlık alır.

Birkaç kafalı yayın, modelin farklı sunum boşluklarından bilgi almasına izin vererek bu yaklaşımı genişletiyor. Her şeyden önce, formun giriş x sensörünü yansıtmak (batch_size, seq_len, d_model) Sorgular, anahtarlar ve değerler üzerinde eğitilmiş doğrusal katmanların yardımıyla. Bu nedenle, bu projeksiyonların her birini karakteristik boyut boyunca paylaşıyoruz num_heads Ayrı kafaları, tırmanma ürününün dikkatini her kafaya paralel olarak uygulayın, sonuçları bağlayın ve orijinaline tekrar projelendirin d_model:


import torch
import torch.nn as nn

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, num_heads):
        super(MultiHeadAttention, self).__init__()
        assert d_model % num_heads == 0, "d_model muss durch num_heads teilbar sein"
        self.num_heads = num_heads
        self.d_k = d_model // num_heads
        # Lineare Projektionen für Abfragen, Schlüssel, Werte und die endgültige Ausgabe.
        self.W_q = nn.Linear(d_model, d_model)
        self.W_k = nn.Linear(d_model, d_model)
        self.W_v = nn.Linear(d_model, d_model)
        self.W_o = nn.Linear(d_model, d_model)

    def forward(self, X, mask=None):
        batch_size, seq_len, _ = X.size()
        # Projektionen der Eingaben auf Q, K, V.
        Q = self.W_q(X)
        K = self.W_k(X)
        V = self.W_v(X)
        # Umformen und transponieren, um Köpfe zu trennen.
        Q = Q.view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
        K = K.view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
        V = V.view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
        # Skalarprodukt-Aufmerksamkeit anwenden.
        attn_output, _ = scaled_dot_product_attention(Q, K, V, mask)
        # Köpfe verknüpfen und zurück auf d_model projizieren.
        concat = attn_output.transpose(1, 2).contiguous().view(batch_size, seq_len, -1)
        output = self.W_o(concat)
        return output

Transformatörün sipariş hakkında entegre bir fikri olmadığından, sıradaki her bir öğenin konumunda model hakkında bilgi sağlamak için katlanır jetona konum kodlaması eklenir. Orijinal transformatör, aşağıdaki gibi tanımlanan göğüs şeklinde bir kodlama kullanır:

P[pos, 2i ] = Sin (POS / (10000^(2i / d_model)))

P[pos, 2i+1 ] = Cos (pos / (10000^(2i / d_model))))

POS için [0, L−1] Ve ben içeri [0, d_model/2−1].

Bunu aşağıdaki gibi uyguluyoruz:


import torch
import math

def get_sinusoidal_positional_encoding(L, d_model):
    # Erstelle einen Tensor der Form (L, d_model).
    P = torch.zeros(L, d_model)
    position = torch.arange(0, L).unsqueeze(1).float()
    div_term = torch.exp(torch.arange(0, d_model, 2).float() * -(math.log(10000.0) / d_model))
    # Sinus auf gerade Indizes anwenden
    P[:, 0::2] = torch.sin(position * div_term)
    # Cosinus auf ungerade Indizes anwenden.
    P[:, 1::2] = torch.cos(position * div_term)
    return P

Her kodlama seviyesi bir otoustodian ünitesinden oluşur ve ardından konuma bağlı bir besleme ağından oluşur. Her iki alt sınıf da kalan bağlantılara dahildir. Bunu turun normalleştirilmesi ve bir terk etme izler. BeedForward ağı formuna sahiptir:

Ffn (x) = relu (x · w₁ + b₁) · w₂ + b₂

ve her pozisyondan bağımsız olarak uygulanır. Pytorch'ta bir kodlayıcı katmanı oluşturalım:


class TransformerEncoderLayer(nn.Module):
    def __init__(self, d_model, num_heads, d_ff, dropout=0.1):
        super(TransformerEncoderLayer, self).__init__()
        self.self_attn = MultiHeadAttention(d_model, num_heads)
        self.ffn = nn.Sequential(
            nn.Linear(d_model, d_ff),
            nn.ReLU(),
            nn.Linear(d_ff, d_model),
        )
        self.norm1 = nn.LayerNorm(d_model)
        self.norm2 = nn.LayerNorm(d_model)
        self.dropout1 = nn.Dropout(dropout)
        self.dropout2 = nn.Dropout(dropout)

    def forward(self, x, mask=None):
        # Selbstaufmerksamkeit mit Restverbindung und Normalisierung.
        attn_out = self.self_attn(x, mask)
        x = x + self.dropout1(attn_out)
        x = self.norm1(x)
        # Feed-Forward mit Restverbindung und Normalisierung.
        ffn_out = self.ffn(x)
        x = x + self.dropout2(ffn_out)
        x = self.norm2(x)
        return x

Tam kodlamayı oluşturmak için bu seviyeleri ayarladık ve girişteki konum kodlarını açtık:


class TransformerEncoder(nn.Module):
    def __init__(self, num_layers, d_model, num_heads, d_ff, dropout):
        super(TransformerEncoder, self).__init__()
        self.pos_encoder = get_sinusoidal_positional_encoding
        self.layers = nn.ModuleList([
            TransformerEncoderLayer(d_model, num_heads, d_ff, dropout)
            for _ in range(num_layers)
        ])
        self.norm = nn.LayerNorm(d_model)

    def forward(self, src, src_mask=None):
        # src: (batch_size, seq_len, d_model)
        seq_len = src.size(1)
        # Positionskodierung hinzufügen.
        pos_enc = self.pos_encoder(seq_len, src.size(2)).to(src.device)
        x = src + pos_enc.unsqueeze(0)
        # Durchlaufen jeder Encoder-Schicht.
        for layer in self.layers:
            x = layer(x, src_mask)
        return self.norm(x)

Bir kod çözücünün uygulanması aynı modeli takip eder, ancak sonraki konumların dikkate alınmasını önlemek için maskelenmiş bir araba-uygunluk ve ayrıca kodlayıcının çıktısını dikkate alan NotCoder-Decoder Note-Decoders altyazı içerir. Nihai doğrusal bir softmax seviyesi, kod çözücünün çıkışını hedef kelime bilgisindeki olasılığa atar.

Her bir bileşenin en çok kafaya dikkate, konum kodlamasına, besleme ağlarına kodlama katmanlarına kadar dikkat çeken üründen kodlanarak, transformatörden bilgi akışı hakkında bir vizyon elde edersiniz. Bu tabanda, model, makinelerin çevirisi, metin kaydı veya hatta görüntü üretimi gibi görevlere kolayca uyarlanabilir veya genişletilebilir.

Makinenin çevirisi veya özeti gibi sekansdan sekansa diziyi yapmak için, kodlayıcıyı birleştirilmiş bir kod çözücüye genişletmeliyiz, her biri bir jeton oluşturur ve kodlayıcının çıkışını dikkate alır. Bu nedenle tam bir transformatör, jeton, konumlandırma kodları, bir kodlama seviyesi yığını, bir yığın kod çözme seviyeleri ve hedef kelime dağarcığında son doğrusal projeksiyonu içerir.

Aşağıda, her satır için bir açıklama ile Pytorch için kademeli bir uygulama bulunmaktadır.


import math
import torch
import torch.nn as nn
import torch.nn.functional as F

def scaled_dot_product_attention(Q, K, V, mask=None):
    """
    Berechnet die skalierte Skalarprodukt-Aufmerksamkeit.
    
    Q, K, V sind Form (batch_size, num_heads, seq_len, d_k).
    Mask, falls angegeben, enthält -inf unzulässige Positionen.
    """
    d_k = Q.size(-1)
    
    # Berechne die rohen Aufmerksamkeitswerte durch Matrixmultiplikation der Abfragen mit den Schlüsseln.
    scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(d_k)
    # Wenn eine Maske angegeben ist, füge sie hinzu (Positionen mit -inf bleiben nach Softmax Null).
    if mask is not None:
        scores = scores + mask
    # Normalisiere die Werte zu Wahrscheinlichkeiten.
    attn_weights = F.softmax(scores, dim=-1)
    # Multipliziere die Wahrscheinlichkeiten mit den Werten, um die beachteten Ausgaben zu erhalten.
    output = torch.matmul(attn_weights, V)
    return output, attn_weights

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, num_heads):
        super(MultiHeadAttention, self).__init__()
        # Sicherstellen, dass d_model gleichmäßig durch die Anzahl der Köpfe teilbar ist.
        assert d_model % num_heads == 0
        self.num_heads = num_heads
        self.d_k = d_model // num_heads
        # Lineare Projektionen für Abfragen, Schlüssel, Werte.
        self.W_q = nn.Linear(d_model, d_model)
        self.W_k = nn.Linear(d_model, d_model)
        self.W_v = nn.Linear(d_model, d_model)
        # Endgültige lineare Projektion nach Verkettung aller Köpfe.
        self.W_o = nn.Linear(d_model, d_model)

    def forward(self, query, key, value, mask=None):
        batch_size = query.size(0)
        # Projektion der Eingabetensoren in Q, K, V.
        Q = self.W_q(query)
        K = self.W_k(key)
        V = self.W_v(value)
        # In (batch, heads, seq_len, d_k) umformen und transponieren.
        Q = Q.view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2)
        K = K.view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2)
        V = V.view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2)
        # Skalierte Skalarprodukt-Aufmerksamkeit pro Kopf anwenden.
        attn_output, _ = scaled_dot_product_attention(Q, K, V, mask)
        # Köpfe verknüpfen: zurück transponieren und Kopfdimension zusammenführen.
        concat = attn_output.transpose(1, 2).contiguous().view(batch_size, -1, self.num_heads * self.d_k)
        # Endgültige lineare Projektion.
        output = self.W_o(concat)
        return output

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=5000):
        super(PositionalEncoding, self).__init__()
        # Erstelle einmal sinusförmige Positionskodierungen.
        pe = torch.zeros(max_len, d_model)
        position = torch.arange(0, max_len).unsqueeze(1).float()
        
        div_term = torch.exp(torch.arange(0, d_model, 2).float() * -(math.log(10000.0) / d_model))
        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)
        
        # Batch-Dimension hinzufügen und als Puffer registrieren, damit sie sich mit dem Modell mitbewegt.
        self.register_buffer('pe', pe.unsqueeze(0))

    def forward(self, x):
        # x hat die Form (batch_size, seq_len, d_model).
        # Die Positionskodierungen bis zur Eingabelänge hinzufügen.
        x = x + self.pe[:, :x.size(1)]
        return x

class TransformerEncoderLayer(nn.Module):
    def __init__(self, d_model, num_heads, d_ff, dropout):
        super(TransformerEncoderLayer, self).__init__()
        # Selbstaufmerksamkeits-Unterschicht.
        self.self_attn = MultiHeadAttention(d_model, num_heads)
        # Positionsbezogenes Feedforward-Netzwerk.
        self.ffn = nn.Sequential(
            nn.Linear(d_model, d_ff),
            nn.ReLU(),
            nn.Linear(d_ff, d_model),
        )
        # Layer-Normalisierungsmodule.
        self.norm1 = nn.LayerNorm(d_model)
        self.norm2 = nn.LayerNorm(d_model)
        # Dropout zur Regularisierung.
        self.dropout1 = nn.Dropout(dropout)
        self.dropout2 = nn.Dropout(dropout)

    def forward(self, x, src_mask=None):
        # Selbstaufmerksamkeit anwenden, dann addieren und normieren.
        attn_out = self.self_attn(x, x, x, src_mask)
        x = self.norm1(x + self.dropout1(attn_out))
        # Feedforward-Netzwerk anwenden, dann addieren und normieren.
        ffn_out = self.ffn(x)
        x = self.norm2(x + self.dropout2(ffn_out))
        return x

class TransformerDecoderLayer(nn.Module):
    def __init__(self, d_model, num_heads, d_ff, dropout):
        super(TransformerDecoderLayer, self).__init__()
        # Maskierte Selbstaufmerksamkeit für Zielsequenz.
        self.self_attn = MultiHeadAttention(d_model, num_heads)
        # Encoder-Decoder-Aufmerksamkeit für die Quelle.
        self.src_attn = MultiHeadAttention(d_model, num_heads)
        # Feed-Forward-Netzwerk.
        self.ffn = nn.Sequential(
            nn.Linear(d_model, d_ff),
            nn.ReLU(),
            nn.Linear(d_ff, d_model),
        )
        
        # Layer-Normen und Dropouts für jede Unterschicht.
        self.norm1 = nn.LayerNorm(d_model)
        self.norm2 = nn.LayerNorm(d_model)
        self.norm3 = nn.LayerNorm(d_model)
        
        self.dropout1 = nn.Dropout(dropout)
        self.dropout2 = nn.Dropout(dropout)
        self.dropout3 = nn.Dropout(dropout)

    def forward(self, x, memory, src_mask=None, tgt_mask=None):
        # Maskierte Selbstaufmerksamkeit auf dem Decodereingang.
        self_attn_out = self.self_attn(x, x, x, tgt_mask)
        x = self.norm1(x + self.dropout1(self_attn_out))
        # Encoder-Decoder-Aufmerksamkeit über Encoder-Ausgaben.
        src_attn_out = self.src_attn(x, memory, memory, src_mask)
        x = self.norm2(x + self.dropout2(src_attn_out))
        # Feed-forward und add & norm.
        ffn_out = self.ffn(x)
        x = self.norm3(x + self.dropout3(ffn_out))
        return x

class Transformer(nn.Module):
    def __init__(self,
                 src_vocab_size,
                 tgt_vocab_size,
                 d_model=512,
                 num_heads=8,
                 d_ff=2048,
                 num_encoder_layers=6,
                 num_decoder_layers=6,
                 dropout=0.1):
        super(Transformer, self).__init__()
        # Token-Einbettung für Quelle und Ziel.
        self.src_embed = nn.Sequential(
            nn.Embedding(src_vocab_size, d_model),
            PositionalEncoding(d_model)
        )
        
        self.tgt_embed = nn.Sequential(
            nn.Embedding(tgt_vocab_size, d_model),
            PositionalEncoding(d_model)
        )
        # Gestapelte Encoder- und Decoder-Schichten.
        self.encoder_layers = nn.ModuleList([
            TransformerEncoderLayer(d_model, num_heads, d_ff, dropout)
            for _ in range(num_encoder_layers)
        ])
        self.decoder_layers = nn.ModuleList([
            TransformerDecoderLayer(d_model, num_heads, d_ff, dropout)
            for _ in range(num_decoder_layers)
        ])
        # Endgültige lineare Projektion auf die Vokabulargröße.
        self.generator = nn.Linear(d_model, tgt_vocab_size)
        self.d_model = d_model

    def encode(self, src, src_mask=None):
        # Positionskodierung einbetten und hinzufügen.
        x = self.src_embed(src) * math.sqrt(self.d_model)
        # Durch jede Encoder-Schicht hindurchlaufen.
        for layer in self.encoder_layers:
            x = layer(x, src_mask)
        return x

    def decode(self, tgt, memory, src_mask=None, tgt_mask=None):
        # Ziel einbetten und Positionskodierung hinzufügen.
        x = self.tgt_embed(tgt) * math.sqrt(self.d_model)
        # Durchlaufen jeder Decoderschicht.
        for layer in self.decoder_layers:
            x = layer(x, memory, src_mask, tgt_mask)
        return x

    def forward(self, src, tgt, src_mask=None, tgt_mask=None):
        # Encoder-Ausgabe berechnen.
        memory = self.encode(src, src_mask)
        # Decoder-Ausgabe unter Berücksichtigung des Encoder-Speichers berechnen.
        output = self.decode(tgt, memory, src_mask, tgt_mask)
        # Auf Vokabular-Logits projizieren.
        return self.generator(output)

def generate_square_subsequent_mask(sz):
    """
    Erstellt eine Maske für kausale Aufmerksamkeit, sodass Position i nur
    auf Positionen ≤ i achten kann. Masken-Einträge sind 0, wo erlaubt, und
    -inf, wo nicht erlaubt.
    """
    mask = torch.triu(torch.full((sz, sz), float('-inf')), diagonal=1)
    return mask

# Anwendungsbeispiel mit Dummy-Daten:
# Vokabulargrößen und Sequenzlängen definieren.
src_vocab_size, tgt_vocab_size = 10000, 10000
batch_size, src_len, tgt_len = 2, 20, 22

# Transformer instanziieren.
model = Transformer(src_vocab_size, tgt_vocab_size)

# Beispiel für Quell- und Zieltoken-Indizes.
src = torch.randint(0, src_vocab_size, (batch_size, src_len))
tgt = torch.randint(0, tgt_vocab_size, (batch_size, tgt_len))

# Keine Füllmaske für dieses Beispiel.
src_mask = None
# Kausale Maske für den Decoder.
tgt_mask = generate_square_subsequent_mask(tgt_len)

# Der Vorwärtsdurchlauf liefert Logits der Form (batch_size, tgt_len, tgt_vocab_size).
logits = model(src, tgt, src_mask, tgt_mask)

Bu uygulamada, her kodlama seviyesi, her biri artık bağlantıları ve şanzımanın normalleştirilmesine sahip, konumla ilgili en çok tam kamyon ve bir besleme ağını kullanır. Her kod çözme seviyesi, gelecekteki jetonların görünümünü önlemek için bir maskelenmiş öz-tedirginlik aşamasının yanı sıra, kod çözücünün menşe dizisinin ilgili kısımlarına konsantre olmasını sağlayan ek kodlama kod çözücünün dikkatini çeker. Sinüs eğrileri ile oluşturulan konum kodları modeli modele ekler ve son doğrusal seviye, kod çözücünün sürümlerini ham jeton skorlarına yansıtır.

Bu temelde, modeli yeterli bir kayıp (örneğin, beklenen logitit ve gerçek jeton indeksleri arasındaki haç) ve yukarıda açıklanan optimize edicilerden biri tanımlayarak etiketlenmiş metin verilerine (birleşmiş) dayanarak eğitmek mümkündür.

Bu blog dizisi, yapay nöronal ağların (KNN'ler) büyük zeminde bir turistik uçuş sundu. KNN'nin rol oynadığı tüm uygulamaları henüz dikkate almadık. DQN (Deep-Q öğrenimi) gibi takviye öğrenimi örneği burada belirtilmiştir.

Yapısı nedeniyle, KNN, istatistiksel modelin yazışmalarını gerçekleştirmek veya varış verilerindeki bazı kalıpları izlemek için yeterli bir araç olduğunu kanıtlamaktadır. Bu yetenek, özellikle onlardan geri bildirim sağlamak için ayrıntılı olarak ayrılan transformatörün mimarisiyle devreye giriyor. Buna ek olarak, takviye öğrenimi genellikle istenen bazı süreçlerde, örneğin sözde akıl yürütme için büyük dilsel modelleri “ikna etmek” için gerçekleşir.

Bugünün KNN'leri biyolojik sinir ağlarına (BNN) dayanmaktadır, ancak bunlarla çok sınırlıdır. Gelecekte, bilim adamları biyolojik rol modelleriyle KNN ile daha net yüzleşmeye çalışabilirler. Genel olarak, gelecekte yapay sinir ağları üretken için temel ve merkezi bir rol oynamalıdır.


(RME)


Yayımlandı

kategorisi

yazarı:

Etiketler:

Yorumlar

Bir yanıt yazın

E-posta adresiniz yayınlanmayacak. Gerekli alanlar * ile işaretlenmişlerdir