# Import PyTorch core library
import torch # Deep learning framework: provides tensor operations and automatic differentiation
import torch.nn as nn # Neural network modules: Linear, Embedding, LayerNorm, ModuleList, etc.
import torch.nn.functional as F # Functional interface: provides stateless operations such as softmax
import torch.optim as optim # Optimizer module: provides parameter update algorithms such as Adam
import math # Math library: used for sqrt to compute scaling factor
# ============================================================
# 0. vocab — Vocabulary definition
# ============================================================
# The example uses only 4 Chinese characters as a minimal vocabulary for understanding each step of attention computation
chars = ["你", "好", "世", "界"] # Vocabulary: 4 characters
vocab_size = len(chars) # Vocabulary size = 4
# Forward mapping: character → integer index, for model input
char2idx = {c: i for i, c in enumerate(chars)}
# Reverse mapping: integer index → character, for restoring Chinese characters during inference
idx2char = {i: c for c, i in char2idx.items()}
# ============================================================
# 1. Transformer hyperparameters (standard Transformer style)
# ============================================================
d_model = 512 # Model hidden dimension (embedding dimension), classic Transformer configuration
n_heads = 8 # Number of multi-head attention heads
n_layers = 6 # Number of stacked Transformer blocks
d_ff = 2048 # Feed-forward network intermediate dimension (typically 4x d_model)
# Ensure d_model is divisible by n_heads, so each head dimension is an integer
assert d_model % n_heads == 0
d_head = d_model // n_heads # Dimension per attention head = 512 / 8 = 64
# ============================================================
# 2. Training data — Autoregressive sequence construction
# ============================================================
# Transformer automatically learns the complete process of token → embedding → attention → output logits,
# without manually specifying target_vectors.
#
# Input sequence: [你, 好, 世, 界] → indices [0, 1, 2, 3]
# Target sequence: [好, 世, 界, 你] → indices [1, 2, 3, 0]
# This is "autoregressive": given the first t characters, predict the (t+1)-th character
data = torch.tensor([[0, 1, 2, 3]]) # Input tensor, shape (batch=1, seq_len=4)
target = torch.tensor([[1, 2, 3, 0]]) # Target tensor, shape (batch=1, seq_len=4)
# ============================================================
# 3. Multi-Head Attention — Multi-head attention layer
# ============================================================
class MultiHeadAttention(nn.Module):
"""
Standard Scaled Dot-Product Multi-Head Self-Attention.
Mathematical definition:
Attention(Q, K, V) = softmax(Q·Kᵀ / √d_k) · V
Process: input x → linear projection to obtain Q/K/V → split into multiple heads → compute attention scores →
scale → causal mask → softmax → weighted sum of V → merge heads → output projection
"""
def __init__(self, d_model, n_heads):
"""
Parameters:
d_model: Total model dimension, also the projection dimension for Q/K/V
n_heads: Number of attention heads
"""
super().__init__()
self.d_model = d_model # Model dimension, e.g., 512
self.n_heads = n_heads # Number of attention heads, e.g., 8
self.d_head = d_model // n_heads # Dimension per head = 512 / 8 = 64
# Four bias-free linear projection matrices (all with shape d_model × d_model):
self.W_Q = nn.Linear(d_model, d_model, bias=False) # Query projection: x → Q
self.W_K = nn.Linear(d_model, d_model, bias=False) # Key projection: x → K
self.W_V = nn.Linear(d_model, d_model, bias=False) # Value projection: x → V
self.W_O = nn.Linear(d_model, d_model, bias=False) # Output projection: fusion after concatenation
def forward(self, x, return_attention=False):
"""
Forward pass.
Parameters:
x: Input tensor, shape (B, T, D)
return_attention: Whether to return intermediate computation results (Q/K/V/scores/alpha) for interpretability analysis
Returns:
if return_attention=False: out, shape (B, T, D)
if return_attention=True: (out, Q, K, V, scores, scores_scaled, alpha)
"""
B, T, D = x.shape # B=batch size, T=sequence length, D=d_model
# ---------- Step 1: Linear projection x → Q, K, V ----------
Q = self.W_Q(x) # (B, T, D) → (B, T, D), query vector for each token
K = self.W_K(x) # (B, T, D) → (B, T, D), key vector for each token
V = self.W_V(x) # (B, T, D) → (B, T, D), value vector for each token
# ---------- Step 2: Split into multiple heads ----------
# view: (B, T, D) → (B, T, n_heads, d_head)
# transpose: swap dimensions 1 and 2 → (B, n_heads, T, d_head)
# After this, each head independently has a T × d_head Q/K/V subspace
Q = Q.view(B, T, self.n_heads, self.d_head).transpose(1, 2)
K = K.view(B, T, self.n_heads, self.d_head).transpose(1, 2)
V = V.view(B, T, self.n_heads, self.d_head).transpose(1, 2)
# ---------- Step 3: Compute attention scores S = Q · Kᵀ ----------
# Q: (B, n_heads, T, d_head), Kᵀ: (B, n_heads, d_head, T)
# scores: (B, n_heads, T, T) — raw similarity between position i and position j
scores = Q @ K.transpose(-2, -1)
# ---------- Step 4: Scale S / √d_k ----------
# Divide by √d_k to prevent dot product values from becoming too large, avoiding softmax gradient saturation
# With standard configuration d_head = 64, the scaling factor = 8
scores_scaled = scores / math.sqrt(self.d_head)
# ---------- Step 5: Causal Mask ----------
# Autoregressive language models require position i to only attend to tokens at positions ≤ i (no future information)
# torch.triu(..., diagonal=1) generates an upper triangular matrix (True above the diagonal)
# For example, when T=4:
# [[F, T, T, T],
# [F, F, T, T],
# [F, F, F, T],
# [F, F, F, F]]
mask = torch.triu(torch.ones(T, T), diagonal=1).bool()
# Fill masked positions with -inf; after softmax, weights ≈ 0, i.e., "attention forbidden"
scores_scaled = scores_scaled.masked_fill(mask, float("-inf"))
# ---------- Step 6: Softmax normalization ----------
# Apply softmax along the last dimension (key direction) to obtain attention weight distribution α
# Sum of weights in each row = 1
alpha = F.softmax(scores_scaled, dim=-1)
# ---------- Step 7: Weighted sum output = α · V ----------
# α: (B, n_heads, T, T) — attention weights
# V: (B, n_heads, T, d_head) — value vectors
# out: (B, n_heads, T, d_head) — weighted context representation
out = alpha @ V
# ---------- Step 8: Merge multiple heads ----------
# transpose: (B, n_heads, T, d_head) → (B, T, n_heads, d_head)
# contiguous + view: flatten to (B, T, n_heads * d_head) = (B, T, D)
out = out.transpose(1, 2).contiguous()
out = out.view(B, T, D)
# ---------- Step 9: Output projection ----------
out = self.W_O(out) # W_O fuses information from different heads
# Determine return content based on return_attention flag
if return_attention:
return out, Q, K, V, scores, scores_scaled, alpha
return out
# ============================================================
# 4. FeedForward — Feed-forward network
# ============================================================
class FeedForward(nn.Module):
"""
Position-wise Feed-Forward Network.
Independently applies two fully connected layers + ReLU activation to each position's representation:
FFN(x) = ReLU(x·W₁ + b₁)·W₂ + b₂
The intermediate dimension d_ff is typically 4x d_model (512 → 2048), expanding then compressing back to original dimension.
"""
def __init__(self, d_model, d_ff):
"""
Parameters:
d_model: Input/output dimension
d_ff: Intermediate hidden layer dimension (typically 4x d_model)
"""
super().__init__()
self.net = nn.Sequential(
nn.Linear(d_model, d_ff), # Up-projection: d_model → d_ff (512 → 2048)
nn.ReLU(), # Nonlinear activation function, introduces nonlinearity
nn.Linear(d_ff, d_model) # Down-projection: d_ff → d_model (2048 → 512)
)
def forward(self, x):
return self.net(x)
# ============================================================
# 5. TransformerBlock — Transformer block
# ============================================================
class TransformerBlock(nn.Module):
"""
A complete Transformer block using Pre-Norm residual structure:
x → LayerNorm → MultiHeadAttention → + → x'
x' → LayerNorm → FeedForward → + → x"
Pre-Norm (normalize before sublayer) is more stable for training than Post-Norm.
"""
def __init__(self, d_model, n_heads, d_ff):
"""
Parameters:
d_model: Model hidden dimension
n_heads: Number of attention heads
d_ff: Feed-forward network intermediate dimension
"""
super().__init__()
# First Pre-Norm sublayer: LayerNorm + MultiHeadAttention
self.ln1 = nn.LayerNorm(d_model) # Layer normalization, standardizes along the last dimension
self.attn = MultiHeadAttention(d_model, n_heads)
# Second Pre-Norm sublayer: LayerNorm + FeedForward
self.ln2 = nn.LayerNorm(d_model)
self.ffn = FeedForward(d_model, d_ff)
def forward(self, x, return_attention=False):
"""
Forward pass.
Parameters:
x: Input tensor (B, T, D)
return_attention: Whether to pass to attention layer to capture intermediate values
Returns:
Normal mode: output tensor (B, T, D)
Attention mode: (output, Q, K, V, scores, scores_scaled, alpha)
"""
if return_attention:
# Need to capture attention intermediate values:
# First LayerNorm, then attention (also returning attention intermediate values), finally residual connection
attn_out, Q, K, V, scores, scores_scaled, alpha = \
self.attn(self.ln1(x), return_attention=True)
x = x + attn_out # Residual connection 1: x + Attention(LayerNorm(x))
x = x + self.ffn(self.ln2(x)) # Residual connection 2: x + FFN(LayerNorm(x))
return x, Q, K, V, scores, scores_scaled, alpha
else:
# Normal forward: do not capture intermediate values
x = x + self.attn(self.ln1(x)) # Residual connection 1
x = x + self.ffn(self.ln2(x)) # Residual connection 2
return x
# ============================================================
# 6. GPT — Autoregressive language model
# ============================================================
class GPT(nn.Module):
"""
Miniature GPT (Generative Pre-trained Transformer) model.
Architecture composition:
Token Embedding → added with Position Embedding
→ N TransformerBlocks
→ Final LayerNorm
→ Linear projection head (outputs vocabulary-dimension logits)
Through return_attention=True, intermediate attention values of the first TransformerBlock can be captured
for visualization analysis and teaching demonstrations.
"""
def __init__(
self,
vocab_size, # Vocabulary size, here 4
d_model=512, # Hidden dimension
n_heads=8, # Number of attention heads
n_layers=6, # Number of Transformer blocks
d_ff=2048, # FFN intermediate dimension
max_len=128 # Maximum supported sequence length
):
super().__init__()
# Token Embedding: maps character indices to d_model-dimensional dense vectors
self.token_embedding = nn.Embedding(vocab_size, d_model)
# Position Embedding: assigns a learnable embedding vector for each position 0~max_len-1
# Enables the model to perceive relative/absolute positions of tokens (Transformer itself has no sequence awareness)
self.position_embedding = nn.Embedding(max_len, d_model)
# Use ModuleList instead of Sequential, to traverse by index and extract specific layer outputs
self.blocks = nn.ModuleList([
TransformerBlock(d_model, n_heads, d_ff)
for _ in range(n_layers) # Stack n_layers=6 Transformer blocks
])
# Final layer normalization: stabilizes distribution after all Transformer blocks, before output projection
self.ln_f = nn.LayerNorm(d_model)
# Output projection head: maps d_model-dimensional hidden states back to vocab_size dimensions, obtaining logits for each token
self.head = nn.Linear(d_model, vocab_size)
def forward(self, idx, return_attention=False):
"""
Forward pass.
Parameters:
idx: Input token indices, shape (B, T)
return_attention: Whether to return intermediate attention values of the first TransformerBlock
Returns:
if return_attention=False: logits, shape (B, T, vocab_size)
if return_attention=True: (logits, (Q, K, V, scores, scores_scaled, alpha))
"""
B, T = idx.shape # B=batch size, T=sequence length
# Generate position indices [0, 1, 2, ..., T-1], shape (1, T), moved to same device as idx
positions = torch.arange(T).unsqueeze(0).to(idx.device)
# Token embedding + Position embedding (element-wise addition)
token_emb = self.token_embedding(idx) # (B, T) → (B, T, d_model)
pos_emb = self.position_embedding(positions) # (1, T) → (1, T, d_model) → broadcast
x = token_emb + pos_emb # Embedding fusion: (B, T, d_model)
saved = None # For saving intermediate results of the first layer's attention
for i, block in enumerate(self.blocks):
# Iterate through each TransformerBlock
if return_attention and i == 0:
# Only capture attention intermediate values at the first layer (i==0) for analysis
x, Q, K, V, scores, scores_scaled, alpha = \
block(x, return_attention=True)
saved = (Q, K, V, scores, scores_scaled, alpha)
else:
x = block(x) # Other layers proceed with normal forward pass
x = self.ln_f(x) # Final LayerNorm
logits = self.head(x) # (B, T, d_model) → (B, T, vocab_size)
if return_attention:
return logits, saved # Return predictions + first layer attention data
return logits
# ============================================================
# 7. Model instantiation and optimizer/loss function configuration
# ============================================================
model = GPT( # Instantiate GPT model
vocab_size=vocab_size, # Vocabulary size = 4
d_model=d_model, # Dimension = 512
n_heads=n_heads, # 8 attention heads
n_layers=n_layers, # 6 Transformer layers
d_ff=d_ff # FFN intermediate dimension = 2048
)
optimizer = optim.Adam(model.parameters(), lr=1e-4) # Adam optimizer, learning rate 0.0001
loss_fn = nn.CrossEntropyLoss() # Cross-entropy loss: measures the gap between logits and target distribution
# ============================================================
# 8. Training loop
# ============================================================
for epoch in range(1000): # Train for 1000 epochs
optimizer.zero_grad() # Clear gradient cache from the previous round
logits = model(data) # Forward pass: input [0,1,2,3] → logits (1,4,4)
# Compute cross-entropy loss
# logits.view(-1, vocab_size): (1*4, 4) = (4, 4) — flatten to 4 samples, each with 4 classes
# target.view(-1): (1*4,) = (4,) — flatten to 4 target labels
loss = loss_fn(
logits.view(-1, vocab_size),
target.view(-1)
)
loss.backward() # Backward pass: compute gradients for all parameters
optimizer.step() # Parameter update: optimize along gradient direction
if epoch % 100 == 0: # Output current loss every 100 epochs
print(f"\nEpoch {epoch}, loss={loss.item():.6f}")
# ============================================================
# 9. Inference and attention visualization
# ============================================================
with torch.no_grad(): # Disable gradient computation during inference to save memory
# Set return_attention=True during inference to capture intermediate attention values of the first layer
logits, saved = model(data, return_attention=True)
# argmax takes the index with the highest logits at each position, i.e., the model's predicted next character
pred = torch.argmax(logits, dim=-1) # (1, 4, 4) → (1, 4)
# Unpack saved attention intermediate values
Q, K, V, scores, scores_scaled, alpha = saved
# ---------- Print input/output ----------
print("\n==============================")
print("Input")
print("==============================")
print([idx2char[i.item()] for i in data[0]]) # Restore indices to Chinese characters
print("\n==============================")
print("Prediction")
print("==============================")
print([idx2char[i.item()] for i in pred[0]])
# ---------- Visualize first layer attention ----------
# The following takes data from batch=0, head=0 to show the complete attention computation process
print("\n==============================")
print("Q")
print("==============================")
print(Q[0, 0].detach().numpy().round(3)) # Batch 0, Head 0: (T, d_head)
print("\n==============================")
print("K")
print("==============================")
print(K[0, 0].detach().numpy().round(3))
print("\n==============================")
print("V")
print("==============================")
print(V[0, 0].detach().numpy().round(3))
print("\n==============================")
print("scores = QK^T") # S = Q · Kᵀ, raw similarity
print("==============================")
print(scores[0, 0].detach().numpy().round(3)) # (T, T) matrix, each row is the score of a query against all keys
print("\n==============================")
print("scores_scaled") # S / √d_k, after scaling
print("==============================")
print(scores_scaled[0, 0].detach().numpy().round(3))
print("\n==============================")
print("softmax alpha") # α = softmax(S_scaled)
print("==============================")
print(alpha[0, 0].detach().numpy().round(3)) # Each row sums ≈ 1, causal mask makes α[i][j>i]=0