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
|
"""Generate TinyStories samples from a cascade (OLMo2-arch) checkpoint — plain forward, standard LLM inference."""
import argparse, pickle, sys
import torch, torch.nn as nn, torch.nn.functional as F
from pathlib import Path
ap = argparse.ArgumentParser()
ap.add_argument('--ckpt', required=True)
ap.add_argument('--n', type=int, default=3)
ap.add_argument('--len', type=int, default=180)
ap.add_argument('--temp', type=float, default=0.8)
ap.add_argument('--topk', type=int, default=40)
ap.add_argument('--prompt', default='Once upon a time')
args = ap.parse_args()
DD = Path('/home/yurenh2/ept/ep_run/data/tinystories_bpe')
vocab = pickle.load(open(DD / 'meta.pkl', 'rb'))['vocab_size']
from tokenizers import Tokenizer
tk = Tokenizer.from_file(str(DD / 'tokenizer.json'))
class RMSNorm(nn.Module):
def __init__(self, C, eps=1e-6):
super().__init__(); self.g = nn.Parameter(torch.ones(C)); self.eps = eps
def forward(self, x):
return x * torch.rsqrt(x.pow(2).mean(-1, keepdim=True) + self.eps) * self.g
class SwiGLU(nn.Module):
def __init__(self, C):
super().__init__()
h = ((8 * C // 3) + 63) // 64 * 64
self.w1 = nn.Linear(C, h, bias=False); self.w3 = nn.Linear(C, h, bias=False)
self.w2 = nn.Linear(h, C, bias=False)
def forward(self, x): return self.w2(F.silu(self.w1(x)) * self.w3(x))
class Olmo2Attn(nn.Module):
def __init__(self, C, H, T):
super().__init__()
self.H, self.hd = H, C // H
self.qkv = nn.Linear(C, 3 * C, bias=False); self.proj = nn.Linear(C, C, bias=False)
self.qn, self.kn = RMSNorm(C), RMSNorm(C)
inv = 1.0 / (500000.0 ** (torch.arange(0, self.hd, 2).float() / self.hd))
fr = torch.outer(torch.arange(T).float(), inv)
self.register_buffer('rc', fr.cos(), persistent=False)
self.register_buffer('rs', fr.sin(), persistent=False)
def rope(self, x):
T = x.shape[2]
x1, x2 = x[..., ::2], x[..., 1::2]
c, s = self.rc[None, None, :T], self.rs[None, None, :T]
return torch.stack((x1 * c - x2 * s, x1 * s + x2 * c), dim=-1).flatten(-2)
def forward(self, x):
B, T, C = x.shape
q, k, v = self.qkv(x).split(C, dim=2)
q, k = self.qn(q), self.kn(k)
q = self.rope(q.view(B, T, self.H, self.hd).transpose(1, 2))
k = self.rope(k.view(B, T, self.H, self.hd).transpose(1, 2))
v = v.view(B, T, self.H, self.hd).transpose(1, 2)
y = F.scaled_dot_product_attention(q, k, v, is_causal=True)
return self.proj(y.transpose(1, 2).contiguous().view(B, T, C))
class Olmo2Block(nn.Module):
def __init__(self, C, H, T):
super().__init__()
self.attn = Olmo2Attn(C, H, T); self.ff = SwiGLU(C)
self.na, self.nf = RMSNorm(C), RMSNorm(C)
def forward(self, z):
z = z + self.na(self.attn(z))
return z + self.nf(self.ff(z))
dev = 'cuda' if torch.cuda.is_available() else 'cpu'
ck = torch.load(args.ckpt, map_location=dev, weights_only=False)
cfg = ck['config']; C, H, T, L = cfg['C'], cfg['H'], cfg['T'], cfg['L']
print(f"[gen] {args.ckpt} | step {ck.get('step')} val {ck.get('val'):.4f} | L{L} C{C}", flush=True)
assert ck.get('wout') is not None, 'need untied head (olmo2 ckpt)'
tok = nn.Embedding(vocab, C).to(dev); tok.load_state_dict(ck['tok'])
blocks = nn.ModuleList([Olmo2Block(C, H, T) for _ in range(L)]).to(dev)
blocks.load_state_dict(ck['blocks'])
W_out = ck['wout'].to(dev)
ln_f = RMSNorm(C).to(dev); ln_f.load_state_dict(ck['lnf'])
for m in [tok, blocks, ln_f]: m.eval()
@torch.no_grad()
def gen_one(seed):
torch.manual_seed(seed)
ids = tk.encode(args.prompt).ids
for _ in range(args.len):
x = torch.tensor(ids[-T:], device=dev)[None]
z = tok(x)
for b in blocks: z = b(z)
logits = (ln_f(z[:, -1]) @ W_out.t()) / args.temp
v, _ = torch.topk(logits, args.topk)
logits[logits < v[:, -1:]] = -float('inf')
ids.append(int(torch.multinomial(F.softmax(logits, -1), 1)))
return tk.decode(ids)
for i in range(args.n):
print(f'--- sample {i+1} ---'); print(gen_one(1234 + i), flush=True)
|
