Construir un transformer

import warnings
warnings.filterwarnings("ignore")
import math, time
import numpy as np
import matplotlib.pyplot as plt
import torch
import torch.nn as nn
import torch.nn.functional as F

np.random.seed(0); torch.manual_seed(0)
np.set_printoptions(precision=3, suppress=True)

def softmax(x, axis=-1):
    x = x - x.max(axis=axis, keepdims=True)
    e = np.exp(x)
    return e / e.sum(axis=axis, keepdims=True)

def show_matrix(ax, M, xlabels, ylabels, title, cmap="Blues", vmin=None, vmax=None):
    ax.imshow(M, cmap=cmap, vmin=vmin, vmax=vmax)
    ax.set_xticks(range(len(xlabels))); ax.set_xticklabels(xlabels, rotation=45, ha="right")
    ax.set_yticks(range(len(ylabels))); ax.set_yticklabels(ylabels)
    ax.set_title(title, fontsize=10)

def sinusoidal_pe(n_pos, d):
    pos = np.arange(n_pos)[:, None]
    i = np.arange(0, d, 2)[None, :]
    angle = pos / (10000 ** (i / d))
    pe = np.zeros((n_pos, d))
    pe[:, 0::2], pe[:, 1::2] = np.sin(angle), np.cos(angle)
    return pe

pe = sinusoidal_pe(64, 64)
fig, axes = plt.subplots(1, 3, figsize=(18, 4))
axes[0].imshow(pe.T, cmap="RdBu", aspect="auto"); axes[0].set_xlabel("posición"); axes[0].set_ylabel("dimensión")
axes[0].set_title("Codificación posicional sinusoidal")
for dim in [0, 10, 30]:
    axes[1].plot(pe[:, dim], label=f"dim. {dim}")
axes[1].legend(); axes[1].set_xlabel("posición"); axes[1].set_title("Cada dimensión es una onda de distinta frecuencia")
axes[2].imshow(pe @ pe.T, cmap="viridis"); axes[2].set_title("PE · PE: la similitud solo depende de la distancia")
axes[2].set_xlabel("posición"); axes[2].set_ylabel("posición")
plt.tight_layout(); plt.show()

tokens = ["the", "cat", "sat", "on", "the", "mat"]
vocab = sorted(set(tokens))
d_model, d_k = 8, 4
E = np.random.randn(len(vocab), d_model)                     # tabla de embeddings
X = np.array([E[vocab.index(t)] for t in tokens]) + sinusoidal_pe(len(tokens), d_model)
W_Q, W_K, W_V = (np.random.randn(d_model, d_k) / np.sqrt(d_model) for _ in range(3))

Q, K, V = X @ W_Q, X @ W_K, X @ W_V
scores = Q @ K.T / np.sqrt(d_k)                              # (6, 6): cada consulta frente a cada clave
A = softmax(scores, axis=-1)                                  # cada fila suma 1
out = A @ V                                                   # (6, 4): nueva representación por token

print("X:", X.shape, " Q,K,V:", Q.shape, " puntuaciones:", scores.shape, " salida:", out.shape)
print("sumas por fila de A:", A.sum(axis=1))
fig, axes = plt.subplots(1, 3, figsize=(17, 4.5))
show_matrix(axes[0], scores, tokens, tokens, "1) puntuaciones  QKᵀ/√d_k", cmap="RdBu_r")
show_matrix(axes[1], A, tokens, tokens, "2) pesos  softmax(puntuaciones)  (fila = consulta)", vmin=0, vmax=1)
show_matrix(axes[2], out, [f"v{i}" for i in range(d_k)], tokens, "3) salida  A·V", cmap="RdBu_r")
plt.tight_layout(); plt.show()

dims = [4, 16, 64, 256, 1024]
ent_raw, ent_scaled = [], []
for d in dims:
    q, k = np.random.randn(1000, d), np.random.randn(1000, 32, d)
    s = np.einsum("nd,nkd->nk", q, k)
    for store, ss in [(ent_raw, s), (ent_scaled, s / np.sqrt(d))]:
        p = softmax(ss)
        store.append(-(p * np.log(p + 1e-12)).sum(1).mean())
plt.figure(figsize=(7, 4))
plt.semilogx(dims, ent_raw, "o-", label="sin escalar  q·k"); plt.semilogx(dims, ent_scaled, "o-", label="escalada  q·k/√d")
plt.axhline(np.log(32), color="gray", ls="--", label="uniforme sobre 32 claves")
plt.xlabel("d_k"); plt.ylabel("entropía de los pesos de atención"); plt.legend(); plt.grid(alpha=0.3)
plt.title("Sin escalar, la atención colapsa a one-hot al crecer d"); plt.show()

def attention(Xin):
    q, k, v = Xin @ W_Q, Xin @ W_K, Xin @ W_V
    return softmax(q @ k.T / np.sqrt(d_k)) @ v

X_nopos = np.array([E[vocab.index(t)] for t in tokens])
perm = np.random.permutation(len(tokens))
print("permutación:", perm)
print("atención(X barajada) == atención(X) barajada:", np.allclose(attention(X_nopos[perm]), attention(X_nopos)[perm]))

n = len(tokens)
causal = np.tril(np.ones((n, n), dtype=bool))
A_causal = softmax(np.where(causal, scores, -np.inf))

pad = np.array([True, True, True, True, False, False])      # supongamos que los dos últimos tokens son relleno
A_pad = softmax(np.where(pad[None, :], scores, -np.inf))

fig, axes = plt.subplots(1, 3, figsize=(17, 4.5))
show_matrix(axes[0], causal.astype(float), tokens, tokens, "máscara causal (1 = permitido)", cmap="Greys")
show_matrix(axes[1], A_causal, tokens, tokens, "atención causal: triangular inferior", vmin=0, vmax=1)
show_matrix(axes[2], A_pad, tokens, tokens, "máscara de relleno: se ignoran las 2 últimas columnas", vmin=0, vmax=1)
plt.tight_layout(); plt.show()

def multi_head_attention(X, Wq, Wk, Wv, Wo, n_heads, mask=None):
    n, d = X.shape
    dh = d // n_heads
    def split(M):                                   # (n, d) -> (heads, n, dh)
        return M.reshape(n, n_heads, dh).transpose(1, 0, 2)
    Q, K, V = split(X @ Wq), split(X @ Wk), split(X @ Wv)
    scores = Q @ K.transpose(0, 2, 1) / np.sqrt(dh)  # (heads, n, n)
    if mask is not None:
        scores = np.where(mask, scores, -np.inf)
    A = softmax(scores)
    heads = A @ V                                   # (heads, n, dh)
    concat = heads.transpose(1, 0, 2).reshape(n, d) # de vuelta a (n, d)
    return concat @ Wo, A

d_model, n_heads = 8, 4
Wq, Wk, Wv, Wo = (np.random.randn(d_model, d_model) / np.sqrt(d_model) for _ in range(4))
Y, A_heads = multi_head_attention(X, Wq, Wk, Wv, Wo, n_heads)
print("entrada", X.shape, "-> salida", Y.shape, "| mapas de atención", A_heads.shape, "(cabezas, consulta, clave)")
fig, axes = plt.subplots(1, n_heads, figsize=(18, 4))
for h, ax in enumerate(axes):
    show_matrix(ax, A_heads[h], tokens, tokens, f"cabeza {h}", vmin=0, vmax=1)
plt.tight_layout(); plt.show()

def gelu(x):
    return 0.5 * x * (1 + np.tanh(np.sqrt(2 / np.pi) * (x + 0.044715 * x ** 3)))   # la aproximación de GPT-2

d_ff = 4 * d_model
W1, b1 = np.random.randn(d_model, d_ff) / np.sqrt(d_model), np.zeros(d_ff)
W2, b2 = np.random.randn(d_ff, d_model) / np.sqrt(d_ff), np.zeros(d_model)
ffn = lambda x: gelu(x @ W1 + b1) @ W2 + b2

whole = ffn(X)
one_by_one = np.stack([ffn(X[i:i + 1])[0] for i in range(len(X))])
print("FFN aplicada a la secuencia == FFN aplicada a cada token por separado:", np.allclose(whole, one_by_one))
print(f"parámetros de la FFN: {W1.size + b1.size + W2.size + b2.size}  frente a  parámetros de la atención: {4 * d_model * d_model}")

def layer_norm(x, gamma, beta, eps=1e-5):
    mu = x.mean(axis=-1, keepdims=True)
    var = x.var(axis=-1, keepdims=True)
    return gamma * (x - mu) / np.sqrt(var + eps) + beta

def transformer_block(x, p, mask=None):
    a, attn = multi_head_attention(layer_norm(x, *p["ln1"]), *p["attn"], n_heads=p["heads"], mask=mask)
    x = x + a                                         # residual 1: la atención suma al flujo
    x = x + p["ffn"](layer_norm(x, *p["ln2"]))        # residual 2: el MLP suma al flujo
    return x, attn

params = {"ln1": (np.ones(d_model), np.zeros(d_model)), "ln2": (np.ones(d_model), np.zeros(d_model)),
          "attn": (Wq, Wk, Wv, Wo), "heads": n_heads, "ffn": ffn}
h = X
for layer in range(3):                               # pesos compartidos por brevedad; los modelos reales usan pesos distintos en cada capa
    h_new, _ = transformer_block(h, params, mask=causal)
    print(f"bloque {layer}: forma {h_new.shape}, tamaño de la actualización ‖Δ‖ = {np.linalg.norm(h_new - h):.2f}")
    h = h_new

# Comparamos nuestra LayerNorm con la de PyTorch
x_t = torch.randn(5, 8)
ours = layer_norm(x_t.numpy(), np.ones(8), np.zeros(8))
print("LayerNorm coincide con PyTorch:", np.allclose(ours, F.layer_norm(x_t, (8,)).numpy(), atol=1e-5))

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, n_heads):
        super().__init__()
        assert d_model % n_heads == 0
        self.h, self.dh = n_heads, d_model // n_heads
        self.Wq, self.Wk, self.Wv, self.Wo = (nn.Linear(d_model, d_model) for _ in range(4))
        self.last_attn = None
    def forward(self, x_q, x_kv, mask=None):
        B, Tq, D = x_q.shape
        Tk = x_kv.shape[1]
        q = self.Wq(x_q).view(B, Tq, self.h, self.dh).transpose(1, 2)    # (B, h, Tq, dh)
        k = self.Wk(x_kv).view(B, Tk, self.h, self.dh).transpose(1, 2)
        v = self.Wv(x_kv).view(B, Tk, self.h, self.dh).transpose(1, 2)
        scores = q @ k.transpose(-2, -1) / math.sqrt(self.dh)            # (B, h, Tq, Tk)
        if mask is not None:
            scores = scores.masked_fill(~mask, float("-inf"))
        attn = scores.softmax(-1)
        self.last_attn = attn.detach()
        out = (attn @ v).transpose(1, 2).reshape(B, Tq, D)
        return self.Wo(out)

class FeedForward(nn.Module):
    def __init__(self, d_model, mult=4):
        super().__init__()
        self.net = nn.Sequential(nn.Linear(d_model, mult * d_model), nn.GELU(), nn.Linear(mult * d_model, d_model))
    def forward(self, x):
        return self.net(x)

class EncoderBlock(nn.Module):
    def __init__(self, d, h):
        super().__init__()
        self.ln1, self.attn, self.ln2, self.ffn = nn.LayerNorm(d), MultiHeadAttention(d, h), nn.LayerNorm(d), FeedForward(d)
    def forward(self, x):
        y = self.ln1(x)
        x = x + self.attn(y, y)
        return x + self.ffn(self.ln2(x))

class DecoderBlock(nn.Module):
    def __init__(self, d, h):
        super().__init__()
        self.ln1, self.self_attn = nn.LayerNorm(d), MultiHeadAttention(d, h)
        self.ln2, self.cross_attn = nn.LayerNorm(d), MultiHeadAttention(d, h)
        self.ln3, self.ffn = nn.LayerNorm(d), FeedForward(d)
    def forward(self, x, memory, causal_mask):
        y = self.ln1(x)
        x = x + self.self_attn(y, y, causal_mask)                  # solo mira las salidas anteriores
        x = x + self.cross_attn(self.ln2(x), memory)               # mira el origen codificado
        return x + self.ffn(self.ln3(x))

class Transformer(nn.Module):
    def __init__(self, vocab, d=64, heads=4, n_enc=2, n_dec=2, max_len=32):
        super().__init__()
        self.src_emb, self.tgt_emb = nn.Embedding(vocab, d), nn.Embedding(vocab, d)
        self.register_buffer("pe", torch.tensor(sinusoidal_pe(max_len, d), dtype=torch.float32))
        self.enc = nn.ModuleList(EncoderBlock(d, heads) for _ in range(n_enc))
        self.dec = nn.ModuleList(DecoderBlock(d, heads) for _ in range(n_dec))
        self.ln_enc, self.ln_dec = nn.LayerNorm(d), nn.LayerNorm(d)
        self.out = nn.Linear(d, vocab)
        self.d = d
    def encode(self, src):
        x = self.src_emb(src) * math.sqrt(self.d) + self.pe[:src.shape[1]]
        for blk in self.enc:
            x = blk(x)
        return self.ln_enc(x)
    def decode(self, tgt_in, memory):
        T = tgt_in.shape[1]
        causal = torch.tril(torch.ones(T, T, dtype=torch.bool, device=tgt_in.device))
        x = self.tgt_emb(tgt_in) * math.sqrt(self.d) + self.pe[:T]
        for blk in self.dec:
            x = blk(x, memory, causal)
        return self.out(self.ln_dec(x))
    def forward(self, src, tgt_in):
        return self.decode(tgt_in, self.encode(src))

N_DIGITS, SEQ, SOS = 10, 10, 10
VOCAB = N_DIGITS + 1

def make_batch(B):
    src = torch.randint(0, N_DIGITS, (B, SEQ))
    tgt = src.sort(dim=1).values
    tgt_in = torch.cat([torch.full((B, 1), SOS), tgt[:, :-1]], dim=1)     # desplazamiento a la derecha: teacher forcing
    return src, tgt_in, tgt

torch.manual_seed(0)
model = Transformer(VOCAB)
print(f"parámetros: {sum(p.numel() for p in model.parameters()):,}")
src, tgt_in, tgt = make_batch(2)
print("origen de ejemplo       :", src[0].tolist())
print("entrada del decodificador:", tgt_in[0].tolist(), "(10 = <sos>)")
print("objetivo                :", tgt[0].tolist())

opt = torch.optim.AdamW(model.parameters(), lr=1e-3, weight_decay=0.01)
steps = 1200
sched = torch.optim.lr_scheduler.OneCycleLR(opt, max_lr=1e-3, total_steps=steps, pct_start=0.1)   # calentamiento y después decaimiento
t0, losses = time.time(), []
for step in range(steps):
    src, tgt_in, tgt = make_batch(128)
    logits = model(src, tgt_in)
    loss = F.cross_entropy(logits.reshape(-1, VOCAB), tgt.reshape(-1))
    opt.zero_grad(); loss.backward(); nn.utils.clip_grad_norm_(model.parameters(), 1.0); opt.step(); sched.step()
    losses.append(loss.item())
    if step % 300 == 0 or step == steps - 1:
        print(f"paso {step:4d}  pérdida {loss.item():.4f}  ({time.time() - t0:.0f} s)")
plt.figure(figsize=(7, 3.5)); plt.semilogy(losses); plt.xlabel("paso"); plt.ylabel("pérdida"); plt.grid(alpha=0.3)
plt.title("Entrenando el transformer para ordenar"); plt.show()

@torch.no_grad()
def greedy_sort(model, src):
    memory = model.encode(src)
    out = torch.full((src.shape[0], 1), SOS)
    for _ in range(SEQ):                                   # autorregresivo: un dígito cada vez
        nxt = model.decode(out, memory)[:, -1].argmax(-1, keepdim=True)
        out = torch.cat([out, nxt], dim=1)
    return out[:, 1:]

model.eval()
torch.manual_seed(123)
src, _, tgt = make_batch(2000)
pred = greedy_sort(model, src)
print(f"secuencias totalmente correctas: {(pred == tgt).all(1).float().mean():.1%}   exactitud por dígito: {(pred == tgt).float().mean():.2%}")
for i in range(4):
    print("entrada", src[i].tolist(), "\nsalida ", pred[i].tolist(), "✓" if (pred[i] == tgt[i]).all() else "✗")

example = torch.tensor([[7, 2, 9, 0, 4, 4, 1, 8, 3, 6]])
with torch.no_grad():
    memory = model.encode(example)
    res = greedy_sort(model, example)
    tgt_in = torch.cat([torch.tensor([[SOS]]), res[:, :-1]], dim=1)
    model.decode(tgt_in, memory)                         # rellena last_attn para toda la secuencia decodificada
src_lbl = [f"{d}" for d in example[0].tolist()]
out_lbl = [f"→{d}" for d in res[0].tolist()]
print("entrada:", example[0].tolist(), "\nsalida :", res[0].tolist())

H = model.enc[0].attn.h
fig, axes = plt.subplots(2, H, figsize=(4 * H, 8))
for l, blk in enumerate(model.enc):
    for h in range(H):
        show_matrix(axes[l, h], blk.attn.last_attn[0, h], src_lbl, src_lbl, f"codificador C{l} cabeza {h}", vmin=0, vmax=1)
fig.suptitle("Autoatención del codificador (filas = dígito de consulta, columnas = dígito clave)"); plt.tight_layout(); plt.show()

fig, axes = plt.subplots(2, H, figsize=(4 * H, 8))
for l, blk in enumerate(model.dec):
    for h in range(H):
        show_matrix(axes[l, h], blk.cross_attn.last_attn[0, h], src_lbl, out_lbl, f"decodificador C{l} at. cruzada cabeza {h}", cmap="Greens", vmin=0, vmax=1)
fig.suptitle("Atención cruzada del decodificador (filas = paso de salida, columnas = posición de entrada)"); plt.tight_layout(); plt.show()

def rollout(attn_maps):
    R = torch.eye(attn_maps[0].shape[-1])
    for A in attn_maps:
        A_bar = A.mean(0)                                       # promedio de las cabezas
        A_res = 0.5 * A_bar + 0.5 * torch.eye(A_bar.shape[0])   # tiene en cuenta el camino residual
        A_res = A_res / A_res.sum(-1, keepdim=True)
        R = A_res @ R
    return R

enc_maps = [blk.attn.last_attn[0] for blk in model.enc]
fig, axes = plt.subplots(1, 3, figsize=(16, 4.8))
show_matrix(axes[0], enc_maps[0].mean(0), src_lbl, src_lbl, "capa 0 (media de las cabezas)", cmap="Purples")
show_matrix(axes[1], enc_maps[1].mean(0), src_lbl, src_lbl, "capa 1 (media de las cabezas)", cmap="Purples")
show_matrix(axes[2], rollout(enc_maps), src_lbl, src_lbl, "rollout a través de ambas capas", cmap="Purples")
plt.tight_layout(); plt.show()

model.zero_grad()
emb = (model.src_emb(example) * math.sqrt(model.d)).detach().requires_grad_(True)
x = emb + model.pe[:SEQ]
for blk in model.enc:
    x = blk(x)
memory = model.ln_enc(x)
logits = model.decode(tgt_in, memory)                        # (1, SEQ, VOCAB)

saliency = torch.zeros(SEQ, SEQ)
for k in range(SEQ):
    chosen = logits[0, k, res[0, k]]                         # el logit del dígito que realmente escribió
    g, = torch.autograd.grad(chosen, emb, retain_graph=True)
    saliency[k] = g[0].norm(dim=-1)
saliency = saliency / saliency.sum(1, keepdim=True)

fig, axes = plt.subplots(1, 2, figsize=(12, 4.8))
show_matrix(axes[0], saliency, src_lbl, out_lbl, "saliencia por gradiente (normalizada por fila)", cmap="Oranges")
show_matrix(axes[1], model.dec[-1].cross_attn.last_attn[0].mean(0), src_lbl, out_lbl, "atención cruzada de la última capa (media de las cabezas)", cmap="Greens")
plt.tight_layout(); plt.show()

def count(m): return sum(p.numel() for p in m.parameters())
blk = model.enc[0]
attn_p, ffn_p, ln_p = count(blk.attn), count(blk.ffn), count(blk.ln1) + count(blk.ln2)
print(f"bloque codificador (d=64): atención {attn_p:,}  FFN {ffn_p:,}  LayerNorm {ln_p:,}")
print(f"proporción de la FFN: {ffn_p / (attn_p + ffn_p + ln_p):.0%},  12·d² = {12 * 64 ** 2:,}")
for name, d, L in [("GPT-2 small", 768, 12), ("GPT-2 XL", 1600, 48), ("GPT-3", 12288, 96)]:
    print(f"{name:12s} d={d:5d} layers={L:3d}  ->  12·L·d² ≈ {12 * L * d * d / 1e9:6.2f} mil millones de parámetros en los bloques")