File size: 5,519 Bytes
ddd581d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 | import math
from dataclasses import dataclass
import torch
import torch.nn as nn
import torch.nn.functional as F
@dataclass
class GPTConfig:
vocab_size: int = 50257
block_size: int = 256
n_layer: int = 6
n_head: int = 6
n_embd: int = 384
dropout: float = 0.1
bias: bool = False
class CausalSelfAttention(nn.Module):
def __init__(self, cfg: GPTConfig):
super().__init__()
assert cfg.n_embd % cfg.n_head == 0
self.n_head = cfg.n_head
self.qkv = nn.Linear(cfg.n_embd, 3 * cfg.n_embd, bias=cfg.bias)
self.proj = nn.Linear(cfg.n_embd, cfg.n_embd, bias=cfg.bias)
self.attn_drop = cfg.dropout
self.resid_drop = nn.Dropout(cfg.dropout)
def forward(self, x):
B, T, C = x.shape
q, k, v = self.qkv(x).split(C, dim=2)
q, k, v = (t.view(B, T, self.n_head, C // self.n_head).transpose(1, 2) for t in (q, k, v))
y = F.scaled_dot_product_attention(
q, k, v,
dropout_p=self.attn_drop if self.training else 0.0,
is_causal=True,
)
y = y.transpose(1, 2).contiguous().view(B, T, C)
return self.resid_drop(self.proj(y))
class MLP(nn.Module):
def __init__(self, cfg: GPTConfig):
super().__init__()
self.fc = nn.Linear(cfg.n_embd, 4 * cfg.n_embd, bias=cfg.bias)
self.proj = nn.Linear(4 * cfg.n_embd, cfg.n_embd, bias=cfg.bias)
self.drop = nn.Dropout(cfg.dropout)
def forward(self, x):
return self.drop(self.proj(F.gelu(self.fc(x))))
class Block(nn.Module):
"""Pre-LayerNorm transformer block."""
def __init__(self, cfg: GPTConfig):
super().__init__()
self.ln1 = nn.LayerNorm(cfg.n_embd, bias=cfg.bias)
self.attn = CausalSelfAttention(cfg)
self.ln2 = nn.LayerNorm(cfg.n_embd, bias=cfg.bias)
self.mlp = MLP(cfg)
def forward(self, x):
x = x + self.attn(self.ln1(x))
x = x + self.mlp(self.ln2(x))
return x
class GPT(nn.Module):
def __init__(self, cfg: GPTConfig):
super().__init__()
self.cfg = cfg
self.tok_emb = nn.Embedding(cfg.vocab_size, cfg.n_embd)
self.pos_emb = nn.Embedding(cfg.block_size, cfg.n_embd)
self.drop = nn.Dropout(cfg.dropout)
self.blocks = nn.ModuleList(Block(cfg) for _ in range(cfg.n_layer))
self.ln_f = nn.LayerNorm(cfg.n_embd, bias=cfg.bias)
self.lm_head = nn.Linear(cfg.n_embd, cfg.vocab_size, bias=False)
self.lm_head.weight = self.tok_emb.weight # weight tying
self.apply(self._init_weights)
for name, p in self.named_parameters():
if name.endswith("proj.weight"):
nn.init.normal_(p, mean=0.0, std=0.02 / math.sqrt(2 * cfg.n_layer))
def _init_weights(self, m):
if isinstance(m, nn.Linear):
nn.init.normal_(m.weight, mean=0.0, std=0.02)
if m.bias is not None:
nn.init.zeros_(m.bias)
elif isinstance(m, nn.Embedding):
nn.init.normal_(m.weight, mean=0.0, std=0.02)
def num_params(self):
return sum(p.numel() for p in self.parameters())
def forward(self, idx, targets=None):
B, T = idx.shape
assert T <= self.cfg.block_size, f"Sequence length {T} > block_size {self.cfg.block_size}"
pos = torch.arange(T, device=idx.device)
x = self.drop(self.tok_emb(idx) + self.pos_emb(pos))
for block in self.blocks:
x = block(x)
x = self.ln_f(x)
if targets is not None:
logits = self.lm_head(x)
loss = F.cross_entropy(logits.view(-1, logits.size(-1)), targets.view(-1), ignore_index=-1)
return logits, loss
return self.lm_head(x[:, [-1], :]), None
def configure_optimizer(self, lr=3e-4, weight_decay=0.1, betas=(0.9, 0.95)):
params = [p for p in self.parameters() if p.requires_grad]
decay = [p for p in params if p.dim() >= 2]
no_decay = [p for p in params if p.dim() < 2]
groups = [
{"params": decay, "weight_decay": weight_decay},
{"params": no_decay, "weight_decay": 0.0},
]
return torch.optim.AdamW(groups, lr=lr, betas=betas)
@torch.no_grad()
def generate(self, idx, max_new_tokens, temperature=1.0, top_k=None):
self.eval()
eos_token_id = 50256
for _ in range(max_new_tokens):
idx_cond = idx[:, -self.cfg.block_size:]
logits, _ = self(idx_cond)
logits = logits[:, -1, :] / max(temperature, 1e-8)
if top_k is not None:
v, _ = torch.topk(logits, min(top_k, logits.size(-1)))
logits[logits < v[:, [-1]]] = float("-inf")
probs = F.softmax(logits, dim=-1)
next_token = torch.multinomial(probs, 1)
idx = torch.cat([idx, next_token], dim=1)
if torch.all(next_token == eos_token_id):
break
return idx
if __name__ == "__main__":
cfg = GPTConfig(vocab_size=1000, block_size=64, n_layer=2, n_head=2, n_embd=64)
model = GPT(cfg)
print(f"params: {model.num_params():,}")
x = torch.randint(0, cfg.vocab_size, (4, 64))
logits, loss = model(x, targets=x)
print("loss:", loss.item())
print(model.generate(x[:1, :8], max_new_tokens=10).shape) |