""" Corso Machine Learning Engineer - Modulo 10, Lezione 03 "L'attenzione, derivata per intero: query, chiave, valore e il blocco Transformer scritto da zero" OBIETTIVO DI QUESTO FILE ------------------------ Quattro esercizi sui quattro punti in cui si sbaglia davvero quando si implementa l'attenzione a mano: 1. la divisione per sqrt(d_k): da dove viene, e cosa succede al softmax quando la si dimentica; 2. la maschera causale: -inf PRIMA del softmax, non azzeramento DOPO; 3. il conto dei parametri di un blocco Transformer, in forma chiusa, verificato contro l'inventario esplicito delle matrici; 4. la memoria della cache chiave-valore, che in generazione lunga supera il peso del modello. COME SI USA ----------- python3 esercizi.py Il file gira anche così com'è: stampa quali esercizi mancano ancora. Completa i TODO uno alla volta e rilancia. Le soluzioni di riferimento sono commentate in fondo al file: guardale solo dopo aver provato. AMBIENTE DICHIARATO ------------------- Python 3.12+ e NumPy 2.3. Nessun'altra dipendenza: niente PyTorch, niente scikit-learn. Il seme del generatore è fissato a 0 dove serve, quindi i numeri stampati sono riproducibili alla cifra. """ import math import numpy as np TOLLERANZA = 1e-9 def vicino(a, b, tol=TOLLERANZA): return a is not None and abs(a - b) <= tol * max(1.0, abs(b)) def softmax(s, asse=-1): """Softmax numericamente stabile: si sottrae il massimo prima di esponenziare.""" s = s - s.max(axis=asse, keepdims=True) e = np.exp(s) return e / e.sum(axis=asse, keepdims=True) # ---------------------------------------------------------------------- # ESERCIZIO 1 - La divisione per la radice di d_k # # Se q e k hanno componenti indipendenti, con media 0 e varianza 1, allora # il prodotto scalare q.k = somma di d_k prodotti indipendenti ha # varianza d_k, quindi deviazione standard sqrt(d_k). # # Due conseguenze da verificare con il codice: # a) la varianza empirica dei prodotti scalari si avvicina a d_k; # b) se due punteggi distano DELTA, il rapporto fra i loro pesi dopo il # softmax vale esattamente exp(DELTA) - indipendentemente dagli altri # punteggi, perché il denominatore si semplifica. # # Con d_k = 64 e due punteggi distanti due deviazioni standard, DELTA vale # 16 senza la scala e 2 con la scala. La differenza fra exp(16) ed exp(2) # è tutta la lezione. # ---------------------------------------------------------------------- def varianza_prodotti_scalari(d_k, n=200000, seme=0): """Varianza empirica di q.k su n coppie di vettori gaussiani standard. Attesa: un numero molto vicino a d_k. """ rng = np.random.default_rng(seme) # TODO: genera due matrici (n, d_k) di normali standard, calcola i # prodotti scalari riga per riga e restituisci la loro varianza (float). return None def rapporto_pesi(delta): """Rapporto fra i pesi di softmax di due punteggi che distano delta. Suggerimento: scrivi p_i / p_j con la definizione del softmax e semplifica il denominatore. Non serve NumPy. """ # TODO: restituisci il rapporto, un float. return None def esegui_esercizio_1(): print("Esercizio 1 - La divisione per la radice di d_k") var = varianza_prodotti_scalari(64) if var is None: print(" DA FARE: completa varianza_prodotti_scalari() (restituisce None).") return False rap = rapporto_pesi(16.0) if rap is None: print(" DA FARE: completa rapporto_pesi() (restituisce None).") return False print(f" Var(q.k) misurata con d_k = 64: {var:.3f} (attesa ~64)") print(f" deviazione standard dei punteggi grezzi: {math.sqrt(64):.3f}") print(f" due punteggi a due deviazioni, senza scala: delta = 16 -> rapporto {rap:.1f}") print(f" gli stessi due punteggi, con la scala: delta = 2 -> rapporto " f"{rapporto_pesi(2.0):.3f}") if abs(var - 64.0) > 1.5: print(" NUMERO SBAGLIATO: la varianza deve avvicinarsi a d_k = 64.") return False if not vicino(rap, math.exp(16.0), 1e-9): print(" NUMERO SBAGLIATO: il rapporto deve valere exp(delta).") return False print(" OK: senza la scala due punteggi vicinissimi producono pesi che") print(" differiscono di sette ordini di grandezza. Il softmax satura.") return True # ---------------------------------------------------------------------- # ESERCIZIO 2 - La maschera causale va applicata PRIMA del softmax # # Due implementazioni che sembrano equivalenti e non lo sono: # giusta softmax( where(maschera, punteggi, -inf) ) # sbagliata softmax( punteggi ) * maschera # # La seconda non rinormalizza: le righe non sommano più a 1, e la riga i # perde esattamente la massa di probabilità che era finita sul futuro. # La riga 0 è il caso estremo: vede una sola chiave su T. # ---------------------------------------------------------------------- PUNTEGGI = np.array([[0.0, 1.0, 2.0, 3.0], [1.0, 0.0, 1.0, 2.0], [2.0, 1.0, 0.0, 1.0], [3.0, 2.0, 1.0, 0.0]]) def pesi_mascherati_bene(punteggi): """Pesi di attenzione causale corretti: -inf sui punteggi, poi softmax.""" # TODO: costruisci la maschera triangolare inferiore con np.tril, # sostituisci -inf dove la maschera è False, poi applica softmax # sull'ultimo asse (le CHIAVI). return None def pesi_mascherati_male(punteggi): """L'errore da riconoscere: softmax prima, azzeramento dopo.""" # TODO: applica il softmax ai punteggi non mascherati e poi moltiplica # per la maschera triangolare. return None def esegui_esercizio_2(): print("Esercizio 2 - La maschera causale, prima e dopo il softmax") bene = pesi_mascherati_bene(PUNTEGGI) if bene is None: print(" DA FARE: completa pesi_mascherati_bene() (restituisce None).") return False male = pesi_mascherati_male(PUNTEGGI) if male is None: print(" DA FARE: completa pesi_mascherati_male() (restituisce None).") return False print(" somma delle righe, maschera prima: ", np.round(bene.sum(-1), 6).tolist()) print(" somma delle righe, maschera dopo: ", np.round(male.sum(-1), 6).tolist()) print(f" scarto massimo fra le due matrici: {np.abs(bene - male).max():.4f}") if not np.allclose(bene.sum(-1), 1.0): print(" NUMERO SBAGLIATO: con la maschera prima del softmax ogni riga") print(" deve sommare esattamente a 1.") return False if np.allclose(male.sum(-1), 1.0): print(" SOSPETTO: la versione sbagliata NON deve sommare a 1.") return False print(" OK: la riga 0 perde il 96,8% della sua massa, e nessuna eccezione") print(" viene sollevata: le forme tornano, i numeri no.") return True # ---------------------------------------------------------------------- # ESERCIZIO 3 - I parametri di un blocco Transformer # # Un blocco (attenzione multi-testa + rete feed-forward con espansione 4x + # due layer norm) contiene: # W_Q, W_K, W_V, W_O quattro matrici d x d -> 4 d^2 # i loro quattro bias -> 4 d # W_1 (d -> 4d) e W_2 (4d -> d) -> 8 d^2 # b_1 (4d) e b_2 (d) -> 5 d # due layer norm, gamma e beta ciascuna -> 4 d # # Totale: 12 d^2 + 13 d. Il numero di TESTE non compare: h teste da d/h # usano esattamente le stesse matrici di una testa da d. # ---------------------------------------------------------------------- def parametri_blocco_formula(d): """Il conto in forma chiusa, da scrivere come una sola espressione.""" # TODO: restituisci 12*d*d + 13*d (ma ricavalo, non copiarlo). return None def parametri_blocco_inventario(d): """Lo stesso conto, sommando le dimensioni di ogni singolo tensore.""" # TODO: costruisci una lista di tensori NumPy con le forme giuste # (np.zeros((d, d)) e simili) e somma i loro .size. return None def esegui_esercizio_3(): print("Esercizio 3 - I parametri di un blocco") for d in (64, 256, 512): a = parametri_blocco_formula(d) b = parametri_blocco_inventario(d) if a is None or b is None: print(" DA FARE: completa parametri_blocco_formula() e " "parametri_blocco_inventario().") return False print(f" d = {d:4d} formula {a:10d} inventario {b:10d} " f"{'coincidono' if a == b else 'DIVERGONO'}") if a != b: print(" NUMERO SBAGLIATO: i due conti devono coincidere esattamente.") return False if parametri_blocco_formula(512) != 3152384: print(" NUMERO SBAGLIATO: per d = 512 attesi 3152384 parametri.") return False print(" OK: 12 d^2 + 13 d, e il numero di teste non compare.") return True # ---------------------------------------------------------------------- # ESERCIZIO 4 - La memoria della cache chiave-valore # # In generazione si conservano K e V di ogni posizione già vista, per ogni # strato e per ogni testa di chiave-valore. Niente altro: non le query (che # servono solo al passo corrente) e non i pesi di attenzione. # # memoria = 2 * n_strati * n_teste_kv * d_testa * T * batch * byte # # Il 2 iniziale sono le due matrici, K e V. L'attenzione multi-query # (arXiv:1911.02150) condivide una sola coppia K,V fra tutte le teste: # divide questo numero per il numero di teste. # ---------------------------------------------------------------------- def memoria_cache(n_strati, n_teste_kv, d_testa, t, batch, byte): """Byte occupati dalla cache chiave-valore.""" # TODO: una riga. return None def esegui_esercizio_4(): print("Esercizio 4 - La memoria della cache chiave-valore") caso = dict(n_strati=24, n_teste_kv=8, d_testa=64, t=8192, batch=4, byte=2) tot = memoria_cache(**caso) if tot is None: print(" DA FARE: completa memoria_cache() (restituisce None).") return False mq = memoria_cache(**{**caso, "n_teste_kv": 1}) pesi = 24 * (12 * 512 ** 2 + 13 * 512) * 2 print(f" 24 strati, 8 teste KV, d_testa 64, T = 8192, batch 4, float16:") print(f" cache {tot / 2**30:8.3f} GiB") print(f" con multi-query {mq / 2**30:8.3f} GiB (un ottavo)") print(f" pesi dei blocchi {pesi / 2**20:8.1f} MiB (d = 512, float16)") print(f" rapporto cache / pesi: {tot / pesi:.1f}x") if tot != 1610612736: print(" NUMERO SBAGLIATO: attesi 1610612736 byte, cioè 1,5 GiB esatti.") return False print(" OK: in generazione lunga la cache pesa più del modello.") return True def main(): print("Corso Machine Learning Engineer - Modulo 10, Lezione 03") print("L'attenzione, derivata per intero") print("=" * 62) print() svolti = 0 for esercizio in (esegui_esercizio_1, esegui_esercizio_2, esegui_esercizio_3, esegui_esercizio_4): if esercizio(): svolti += 1 print() print("=" * 62) print(f"Esercizi svolti: {svolti} su 4.") if svolti < 4: print("Completa i TODO e rilancia. Le soluzioni commentate sono in fondo al file.") else: print("Tutti e quattro tornano. Ora riscrivi il blocco completo da zero,") print("senza guardare l'articolo, e verificalo con le differenze finite.") if __name__ == "__main__": main() # ====================================================================== # SOLUZIONI DI RIFERIMENTO - guardale dopo aver provato. # Per usarle, togli il commento e sostituisci i corpi delle funzioni sopra. # ====================================================================== # # def varianza_prodotti_scalari(d_k, n=200000, seme=0): # rng = np.random.default_rng(seme) # q = rng.normal(size=(n, d_k)) # k = rng.normal(size=(n, d_k)) # return float((q * k).sum(axis=1).var()) # # # def rapporto_pesi(delta): # # p_i / p_j = exp(s_i) / exp(s_j) = exp(s_i - s_j): il denominatore # # del softmax è lo stesso per i due pesi e si semplifica. # return math.exp(delta) # # # def pesi_mascherati_bene(punteggi): # t = punteggi.shape[-1] # maschera = np.tril(np.ones((t, t), dtype=bool)) # return softmax(np.where(maschera, punteggi, -np.inf), asse=-1) # # # def pesi_mascherati_male(punteggi): # t = punteggi.shape[-1] # maschera = np.tril(np.ones((t, t), dtype=bool)) # return softmax(punteggi, asse=-1) * maschera # # # def parametri_blocco_formula(d): # return 12 * d * d + 13 * d # # # def parametri_blocco_inventario(d): # tensori = [np.zeros((d, d)) for _ in range(4)] # W_Q W_K W_V W_O # tensori += [np.zeros(d) for _ in range(4)] # i loro bias # tensori += [np.zeros((d, 4 * d)), np.zeros(4 * d)] # W_1, b_1 # tensori += [np.zeros((4 * d, d)), np.zeros(d)] # W_2, b_2 # tensori += [np.zeros(d) for _ in range(4)] # gamma e beta x2 # return sum(t.size for t in tensori) # # # def memoria_cache(n_strati, n_teste_kv, d_testa, t, batch, byte): # return 2 * n_strati * n_teste_kv * d_testa * t * batch * byte # # NUMERI ATTESI # ------------- # Var(q.k) con d_k = 64, n = 200000, seme 0 63.767 # rapporto dei pesi con delta = 16 8886110.5 # rapporto dei pesi con delta = 2 7.389 # somma righe, maschera dopo il softmax [0.032059, 0.268941, 0.803388, 1.0] # scarto massimo fra le due matrici 0.9679 # parametri di un blocco, d = 512 3152384 # cache KV, 24 strati, 8 teste, T = 8192, b = 4 1610612736 byte = 1,500 GiB # la stessa con multi-query 0,188 GiB