|
import torch |
|
import torch.nn as nn |
|
import torch.nn.functional as F |
|
import math |
|
|
|
from dataclasses import dataclass |
|
from typing import Optional, Tuple |
|
|
|
import tiktoken |
|
import os |
|
import numpy as np |
|
|
|
torch.set_float32_matmul_precision('high') |
|
|
|
def load_tokens(filename): |
|
npt = np.load(filename) |
|
npt = npt.astype(np.int32) |
|
ptt = torch.tensor(npt, dtype=torch.long) |
|
return ptt |
|
|
|
class DataLoaderLite: |
|
def __init__(self, B, T, split): |
|
self.B = B |
|
self.T = T |
|
assert split in {'train', 'val'} |
|
|
|
|
|
data_root = "tinystories" |
|
shards = os.listdir(data_root) |
|
shards = [s for s in shards if split in s] |
|
shards = sorted(shards) |
|
shards = [os.path.join(data_root, s) for s in shards] |
|
self.shards = shards |
|
assert len(shards) > 0, f"no shards found for split {split}" |
|
print(f"found {len(shards)} shards for split {split}") |
|
self.reset() |
|
|
|
def reset(self): |
|
|
|
self.current_shard = 0 |
|
self.tokens = load_tokens(self.shards[self.current_shard]) |
|
self.current_position = self.B * self.T |
|
|
|
def next_batch(self): |
|
B, T = self.B, self.T |
|
buf = self.tokens[self.current_position : self.current_position+B*T+1] |
|
x = (buf[:-1]).view(B, T) |
|
y = (buf[1:]).view(B, T) |
|
|
|
self.current_position += B * T |
|
|
|
if self.current_position + (B * T ) > len(self.tokens): |
|
self.current_shard = (self.current_shard + 1) % len(self.shards) |
|
self.tokens = load_tokens(self.shards[self.current_shard]) |
|
self.current_position = B * T |
|
return x, y |
|
|
|
@dataclass |
|
class ModelArgs: |
|
dim: int = 768 |
|
n_layers: int = 16 |
|
n_heads: int = 16 |
|
n_kv_heads: Optional[int] = 4 |
|
vocab_size: int = 50304 |
|
multiple_of: int = 256 |
|
ffn_dim_multiplier: Optional[float] = None |
|
norm_eps: float = 1e-5 |
|
rope_theta: float = 50000 |
|
max_batch_size: int = 4 |
|
max_seq_len: int = 1024 |
|
device: str = 'cuda' if torch.cuda.is_available() else 'cpu' |
|
dropout_rate: float = 0.1 |
|
|
|
params = ModelArgs() |
|
|
|
class RMSNorm(torch.nn.Module): |
|
def __init__(self, dim: int, eps: float = 1e-6): |
|
super().__init__() |
|
self.eps = eps |
|
self.weight = nn.Parameter(torch.ones(dim)) |
|
|
|
def _norm(self, x): |
|
return x * torch.rsqrt(x.pow(2).mean(-1, keepdim=True) + self.eps) |
|
|
|
def forward(self, x): |
|
output = self._norm(x.float()).type_as(x) |
|
return output * self.weight |
|
|
|
|
|
def precompute_freqs_cis(dim: int, end: int, theta: float = 10000.0): |
|
freqs = 1.0 / (theta ** (torch.arange(0, dim, 2)[: (dim // 2)].float() / dim)) |
|
t = torch.arange(end, device=freqs.device, dtype=torch.float32) |
|
freqs = torch.outer(t, freqs) |
|
freqs_cis = torch.polar(torch.ones_like(freqs), freqs) |
|
return freqs_cis.to(params.device) |
|
|
|
def reshape_for_broadcast(freqs_cis: torch.Tensor, x: torch.Tensor): |
|
ndim = x.ndim |
|
assert 0 <= 1 < ndim |
|
assert freqs_cis.shape == (x.shape[1], x.shape[-1]), f'freqs_cis.shape {freqs_cis.shape} != (x.shape[1], x.shape[-1]) {(x.shape[1], x.shape[-1])}' |
|
shape = [d if i == 1 or i == ndim - 1 else 1 for i, d in enumerate(x.shape)] |
|
return freqs_cis.view(*shape) |
|
|
|
def apply_rotary_emb( |
|
xq: torch.Tensor, |
|
xk: torch.Tensor, |
|
freqs_cis: torch.Tensor, |
|
) -> Tuple[torch.Tensor, torch.Tensor]: |
|
xq_ = torch.view_as_complex(xq.float().reshape(*xq.shape[:-1], -1, 2)) |
|
xk_ = torch.view_as_complex(xk.float().reshape(*xk.shape[:-1], -1, 2)) |
|
freqs_cis = reshape_for_broadcast(freqs_cis, xq_) |
|
xq_out = torch.view_as_real(xq_ * freqs_cis).flatten(3) |
|
xk_out = torch.view_as_real(xk_ * freqs_cis).flatten(3) |
|
return xq_out.type_as(xq), xk_out.type_as(xk) |
|
|
|
def repeat_kv(x: torch.Tensor, n_rep: int) -> torch.Tensor: |
|
"""torch.repeat_interleave(x, dim=2, repeats=n_rep)""" |
|
bs, seqlen, n_kv_heads, head_dim = x.shape |
|
if n_rep == 1: |
|
return x |
|
return ( |
|
x[:, :, :, None, :] |
|
.expand(bs, seqlen, n_kv_heads, n_rep, head_dim) |
|
.reshape(bs, seqlen, n_kv_heads * n_rep, head_dim) |
|
) |
|
|
|
class Attention(nn.Module): |
|
def __init__(self, args: ModelArgs): |
|
super().__init__() |
|
self.n_heads = args.n_heads |
|
self.n_kv_heads = args.n_heads if args.n_kv_heads is None else args.n_kv_heads |
|
self.n_rep = args.n_heads // self.n_kv_heads |
|
self.head_dim = args.dim // args.n_heads |
|
|
|
self.wq = nn.Linear(args.dim, args.n_heads * self.head_dim, bias=False) |
|
self.wk = nn.Linear(args.dim, self.n_kv_heads * self.head_dim, bias=False) |
|
self.wv = nn.Linear(args.dim, self.n_kv_heads * self.head_dim, bias=False) |
|
self.wo = nn.Linear(args.n_heads * self.head_dim, args.dim, bias=False) |
|
|
|
self.cache_k = torch.zeros( |
|
(args.max_batch_size, args.max_seq_len, self.n_kv_heads, self.head_dim), |
|
requires_grad = False |
|
).to(args.device) |
|
self.cache_v = torch.zeros( |
|
(args.max_batch_size, args.max_seq_len, self.n_kv_heads, self.head_dim), |
|
requires_grad = False |
|
).to(args.device) |
|
|
|
def forward( |
|
self, |
|
x: torch.Tensor, |
|
freqs_cis: torch.Tensor, |
|
mask: Optional[torch.Tensor], |
|
start_pos: int = None, |
|
): |
|
bsz, seqlen, _ = x.shape |
|
xq, xk, xv = self.wq(x), self.wk(x), self.wv(x) |
|
|
|
xq = xq.view(bsz, seqlen, self.n_heads, self.head_dim) |
|
xk = xk.view(bsz, seqlen, self.n_kv_heads, self.head_dim) |
|
xv = xv.view(bsz, seqlen, self.n_kv_heads, self.head_dim) |
|
|
|
xq, xk = apply_rotary_emb(xq, xk, freqs_cis=freqs_cis) |
|
|
|
if start_pos is not None: |
|
self.cache_k = self.cache_k.to(xq) |
|
self.cache_v = self.cache_v.to(xq) |
|
|
|
|
|
self.cache_k[:bsz, start_pos : start_pos + seqlen] = xk |
|
self.cache_v[:bsz, start_pos : start_pos + seqlen] = xv |
|
|
|
|
|
keys = self.cache_k[:bsz, : start_pos + seqlen] |
|
values = self.cache_v[:bsz, : start_pos + seqlen] |
|
else: |
|
|
|
keys, values = xk, xv |
|
|
|
keys = repeat_kv(keys, self.n_rep) |
|
values = repeat_kv(values, self.n_rep) |
|
|
|
xq = xq.transpose(1, 2) |
|
keys = keys.transpose(1, 2) |
|
values = values.transpose(1, 2) |
|
|
|
scores = torch.matmul(xq, keys.transpose(2, 3)) / math.sqrt(self.head_dim) |
|
if mask is not None: |
|
scores = scores + mask |
|
scores = F.softmax(scores.float(), dim=-1).type_as(xq) |
|
|
|
output = torch.matmul(scores, values) |
|
output = output.transpose(1, 2).contiguous().view(bsz, seqlen, -1) |
|
return self.wo(output) |
|
|
|
class FeedForward(nn.Module): |
|
def __init__( |
|
self, |
|
dim: int, |
|
hidden_dim: int, |
|
multiple_of: int, |
|
ffn_dim_multiplier: Optional[float], |
|
): |
|
super().__init__() |
|
|
|
hidden_dim = int(2 * hidden_dim / 3) |
|
if ffn_dim_multiplier is not None: |
|
hidden_dim = int(ffn_dim_multiplier * hidden_dim) |
|
hidden_dim = multiple_of * ((hidden_dim + multiple_of - 1) // multiple_of) |
|
|
|
self.w1 = nn.Linear(dim, hidden_dim, bias=False) |
|
self.w2 = nn.Linear(hidden_dim, dim, bias=False) |
|
self.w3 = nn.Linear(dim, hidden_dim, bias=False) |
|
|
|
def forward(self, x): |
|
return self.w2(F.silu(self.w1(x)) * self.w3(x)) |
|
|
|
class TransformerBlock(nn.Module): |
|
def __init__(self, args: ModelArgs): |
|
super().__init__() |
|
self.n_heads = args.n_heads |
|
self.dim = args.dim |
|
self.head_dim = args.dim // args.n_heads |
|
self.attention = Attention(args) |
|
self.feed_forward = FeedForward( |
|
dim=args.dim, |
|
hidden_dim=4 * args.dim, |
|
multiple_of=args.multiple_of, |
|
ffn_dim_multiplier=args.ffn_dim_multiplier, |
|
) |
|
self.attention_norm = RMSNorm(args.dim, eps=args.norm_eps) |
|
self.ffn_norm = RMSNorm(args.dim, eps=args.norm_eps) |
|
self.dropout_rate = args.dropout_rate |
|
|
|
def forward( |
|
self, |
|
x: torch.Tensor, |
|
freqs_cis: torch.Tensor, |
|
mask: Optional[torch.Tensor], |
|
start_pos: int = None, |
|
training = False, |
|
): |
|
h = x + F.dropout(self.attention(self.attention_norm(x), freqs_cis, mask, start_pos), p=self.dropout_rate, training=training) |
|
out = h + F.dropout(self.feed_forward(self.ffn_norm(h)), p=self.dropout_rate, training=training) |
|
return out |
|
|
|
class Moose(nn.Module): |
|
def __init__(self, params: ModelArgs): |
|
super().__init__() |
|
self.params = params |
|
self.vocab_size = params.vocab_size |
|
self.n_layers = params.n_layers |
|
self.max_seq_len = params.max_seq_len |
|
|
|
self.tok_embeddings = nn.Embedding(params.vocab_size, params.dim) |
|
|
|
self.layers = torch.nn.ModuleList() |
|
for _ in range(params.n_layers): |
|
self.layers.append(TransformerBlock(params)) |
|
|
|
self.norm = RMSNorm(params.dim, eps=params.norm_eps) |
|
self.output = nn.Linear( |
|
params.dim, |
|
params.vocab_size, |
|
bias=False) |
|
|
|
self.freqs_cis = precompute_freqs_cis( |
|
params.dim // params.n_heads, |
|
params.max_seq_len * 2, |
|
params.rope_theta,) |
|
|
|
|
|
mask = torch.full((params.max_seq_len, params.max_seq_len), |
|
float("-inf"), |
|
device=params.device) |
|
mask = torch.triu(mask, diagonal=1) |
|
self.register_buffer('mask', mask) |
|
|
|
self.criterion = nn.CrossEntropyLoss() |
|
|
|
def forward(self, |
|
tokens: torch.Tensor, |
|
targets: torch.Tensor): |
|
bsz, seqlen = tokens.shape |
|
assert tokens.shape == targets.shape |
|
assert seqlen == self.max_seq_len |
|
|
|
h = self.tok_embeddings(tokens) |
|
|
|
freqs_cis = self.freqs_cis.to(h.device) |
|
freqs_cis = self.freqs_cis[:seqlen] |
|
|
|
for layer in self.layers: |
|
h = layer( |
|
h, |
|
freqs_cis, |
|
self.mask, |
|
start_pos = None, |
|
training = True |
|
) |
|
|
|
h = self.norm(h) |
|
logits = self.output(h).float() |
|
|
|
loss = self.criterion( |
|
logits.view(bsz * seqlen, self.vocab_size), |
|
targets.reshape(bsz * seqlen)) |
|
|
|
return logits, loss |
|
|
|
@torch.inference_mode() |
|
def forward_inference(self, |
|
tokens: torch.Tensor, |
|
start_pos: int, |
|
max_context_window: int, |
|
): |
|
_bsz, seqlen = tokens.shape |
|
h = self.tok_embeddings(tokens) |
|
self.freqs_cis = self.freqs_cis.to(h.device) |
|
freqs_cis = self.freqs_cis[start_pos : start_pos + seqlen] |
|
|
|
mask = self.mask[:seqlen, :seqlen] |
|
mask = torch.hstack( |
|
[torch.zeros((seqlen, start_pos), device=tokens.device), mask] |
|
).type_as(h) |
|
|
|
for layer in self.layers: |
|
h = layer( |
|
h, |
|
freqs_cis, |
|
mask, |
|
start_pos = start_pos |
|
) |
|
h = self.norm(h) |
|
logits = self.output(h).float() |
|
return logits |
|
|
|
@torch.inference_mode() |
|
def Sampler( |
|
self, |
|
logits: torch.Tensor, |
|
temperature: float, |
|
top_p: float |
|
) -> torch.Tensor: |
|
logits = logits[:,-1,:] |
|
|
|
logits.div_(temperature) |
|
|
|
probs = torch.softmax(logits, dim=-1, dtype=torch.float) |
|
|
|
probs_sort, probs_idx = torch.sort(probs, dim=-1, descending=True) |
|
|
|
probs_sum = torch.cumsum(probs_sort, dim=-1) |
|
top_ps_mask = (probs_sum - probs_sort) > top_p |
|
probs_sort = torch.where(top_ps_mask, 0, probs_sort) |
|
probs_sort.div_(probs_sort.sum(dim=-1, keepdim=True)) |
|
probs = torch.gather(probs_sort, dim=-1, index=torch.argsort(probs_idx, dim=-1)) |
|
next_token_id = torch.multinomial(probs, num_samples=1) |
|
|
|
return next_token_id |
|
|
|
@torch.inference_mode() |
|
def generate( |
|
self, |
|
prompt: str, |
|
max_gen_len: int = 100, |
|
memory_saver_div: int = 1, |
|
temperature: float = 0.6, |
|
top_p: float = 0.9, |
|
) -> str: |
|
assert ((memory_saver_div & (memory_saver_div-1)) == 0) & (memory_saver_div > 0), f'memory_saver_div {memory_saver_div} must be power of 2' |
|
max_context_window = self.max_seq_len // memory_saver_div |
|
|
|
enc = tiktoken.get_encoding('gpt2') |
|
tokens = enc.encode(prompt) |
|
|
|
tokens = torch.tensor(tokens, device=self.params.device) |
|
tokens = tokens.unsqueeze(0) if len(tokens.shape)==1 else tokens |
|
|
|
start_pos = max(tokens.shape[1] - max_context_window, 0) |
|
eot = enc._special_tokens['<|endoftext|>'] |
|
|
|
while True: |
|
logits = self.forward_inference( |
|
tokens[:,-max_context_window:], |
|
start_pos = start_pos, |
|
max_context_window = max_context_window |
|
) |
|
|
|
next_token = self.Sampler( |
|
logits = logits, |
|
temperature = temperature, |
|
top_p = top_p, |
|
) |
|
|
|
tokens = torch.cat((tokens, next_token), dim=1) |
|
|
|
if next_token.item() == eot: |
|
break |
|
|
|
if tokens.shape[1] >= max_context_window: |
|
start_pos += 1 |
|
|
|
output = enc.decode(tokens.squeeze(0).tolist()) |
|
|
|
return output[:-13] |