import math from dataclasses import dataclass import torch import torch.nn as nn import torch.nn.functional as F @dataclass class ModelConfig: vocab_size: int = 32768 max_seq_len: int = 512 hidden_dim: int = 768 intermediate_dim: int = 2048 # SwiGLU dimension num_layers: int = 12 num_heads: int = 12 eps: float = 1e-5 class RMSNorm(nn.Module): def __init__(self, dim: int, eps: float = 1e-5): super().__init__() self.eps = eps self.weight = nn.Parameter(torch.ones(dim)) def forward(self, x: torch.Tensor) -> torch.Tensor: variance = x.pow(2).mean(-1, keepdim=True) return x * torch.rsqrt(variance + self.eps) * self.weight def precompute_rope_cis(dim: int, max_seq_len: int, theta: float = 10000.0) -> torch.Tensor: freqs = 1.0 / (theta ** (torch.arange(0, dim, 2)[: (dim // 2)].float() / dim)) t = torch.arange(max_seq_len, dtype=torch.float32) freqs = torch.outer(t, freqs) cis = torch.polar(torch.ones_like(freqs), freqs) return cis # [max_seq_len, dim // 2] def apply_rope(x: torch.Tensor, cis: torch.Tensor) -> torch.Tensor: B, T, H, D = x.shape x_complex = torch.view_as_complex(x.float().reshape(B, T, H, -1, 2)) cis = cis[:T].unsqueeze(0).unsqueeze(2) # Broadcast to [1, T, 1, D // 2] x_rotated = torch.view_as_real(x_complex * cis).reshape(B, T, H, D) return x_rotated.type_as(x) class CausalSelfAttention(nn.Module): def __init__(self, cfg: ModelConfig): super().__init__() self.num_heads = cfg.num_heads self.head_dim = cfg.hidden_dim // cfg.num_heads self.q_proj = nn.Linear(cfg.hidden_dim, cfg.hidden_dim, bias=False) self.k_proj = nn.Linear(cfg.hidden_dim, cfg.hidden_dim, bias=False) self.v_proj = nn.Linear(cfg.hidden_dim, cfg.hidden_dim, bias=False) self.out_proj = nn.Linear(cfg.hidden_dim, cfg.hidden_dim, bias=False) def forward(self, x: torch.Tensor, cis: torch.Tensor) -> torch.Tensor: B, T, C = x.shape q = self.q_proj(x).view(B, T, self.num_heads, self.head_dim) k = self.k_proj(x).view(B, T, self.num_heads, self.head_dim) v = self.v_proj(x).view(B, T, self.num_heads, self.head_dim) q = apply_rope(q, cis) k = apply_rope(k, cis) q = q.transpose(1, 2) # [B, H, T, D] k = k.transpose(1, 2) v = v.transpose(1, 2) # Dispatches automatically to FlashAttention or Cutlass out = F.scaled_dot_product_attention(q, k, v, is_causal=True) out = out.transpose(1, 2).contiguous().view(B, T, C) return self.out_proj(out) class SwiGLUMLP(nn.Module): def __init__(self, cfg: ModelConfig): super().__init__() self.w1 = nn.Linear(cfg.hidden_dim, cfg.intermediate_dim, bias=False) self.w2 = nn.Linear(cfg.hidden_dim, cfg.intermediate_dim, bias=False) self.w3 = nn.Linear(cfg.intermediate_dim, cfg.hidden_dim, bias=False) def forward(self, x: torch.Tensor) -> torch.Tensor: return self.w3(F.silu(self.w1(x)) * self.w2(x)) class TransformerBlock(nn.Module): def __init__(self, cfg: ModelConfig): super().__init__() self.attn_norm = RMSNorm(cfg.hidden_dim, eps=cfg.eps) self.attn = CausalSelfAttention(cfg) self.mlp_norm = RMSNorm(cfg.hidden_dim, eps=cfg.eps) self.mlp = SwiGLUMLP(cfg) def forward(self, x: torch.Tensor, cis: torch.Tensor) -> torch.Tensor: x = x + self.attn(self.attn_norm(x), cis) x = x + self.mlp(self.mlp_norm(x)) return x class DecoderLM(nn.Module): def __init__(self, cfg: ModelConfig): super().__init__() self.cfg = cfg self.tok_embeddings = nn.Embedding(cfg.vocab_size, cfg.hidden_dim) self.layers = nn.ModuleList([TransformerBlock(cfg) for _ in range(cfg.num_layers)]) self.norm = RMSNorm(cfg.hidden_dim, eps=cfg.eps) self.lm_head = nn.Linear(cfg.hidden_dim, cfg.vocab_size, bias=False) # Weight tying self.lm_head.weight = self.tok_embeddings.weight # Precompute RoPE table rope_cis = precompute_rope_cis(cfg.hidden_dim // cfg.num_heads, cfg.max_seq_len) self.register_buffer("rope_cis", rope_cis, persistent=False) self.apply(self._init_weights) def _init_weights(self, module): if isinstance(module, nn.Linear): torch.nn.init.normal_(module.weight, mean=0.0, std=0.02) elif isinstance(module, nn.Embedding): torch.nn.init.normal_(module.weight, mean=0.0, std=0.02) def forward(self, input_ids: torch.Tensor, targets: torch.Tensor | None = None): B, T = input_ids.shape x = self.tok_embeddings(input_ids) cis = self.rope_cis[:T] for layer in self.layers: x = layer(x, cis) x = self.norm(x) logits = self.lm_head(x) loss = None if targets is not None: loss = F.cross_entropy(logits.view(-1, self.cfg.vocab_size), targets.view(-1)) return logits, loss if __name__ == "__main__": cfg = ModelConfig() m = DecoderLM(cfg) total_params = sum(p.numel() for p in m.parameters()) print(f"Architecture loaded: {total_params / 1e6:.2f}M parameters")