from dataclasses import dataclass import torch import torch.nn as nn from torch.nn import functional as F import math def precompute(head_dim : int, seq_len : int, device : str, base : float = 10000.0): assert head_dim % 2 == 0 theta_pair = torch.arange(0,head_dim,2).float() theta = 1.0 / (base ** (theta_pair / head_dim)).to(device) m = torch.arange(seq_len,device=device) freqs = torch.outer(m,theta).float() freqs_complex = torch.polar(torch.ones_like(freqs),freqs) return freqs_complex def apply_rope(x : torch.tensor, freqs_complex : torch.tensor, device:str): x_complex = torch.view_as_complex(x.float().reshape(*x.shape[:-1],-1,2)) freqs_complex = freqs_complex.unsqueeze(0).unsqueeze(0) x_rotated = x_complex * freqs_complex x_out = torch.view_as_real(x_rotated) x_out = x_out.reshape(*x.shape) return x_out.type_as(x) class CasualSelfAttention(nn.Module): def __init__(self,config): super().__init__() self.c_attn = nn.Linear(config.n_embd, config.n_embd * 3,bias=False) self.c_proj = nn.Linear(config.n_embd,config.n_embd,bias=False) self.n_head = config.n_head self.n_embd = config.n_embd def forward(self,x,freqs): B,T,C = x.size() qkv = self.c_attn(x) q,k,v = qkv.split(self.n_embd,dim=2) q = q.view(B,T, self.n_head, C // self.n_head).transpose(1,2) k = k.view(B,T, self.n_head, C // self.n_head).transpose(1,2) v = v.view(B,T , self.n_head, C // self.n_head).transpose(1,2) q = apply_rope(q, freqs, x.device) k = apply_rope(k, freqs, x.device) y = F.scaled_dot_product_attention(q,k,v,is_causal=True) y = y.transpose(1,2).contiguous().view(B,T,C) y = self.c_proj(y) return y class FeedForward(nn.Module): def __init__(self,config): super().__init__() hidden_dim = int(2 * config.n_embd / 3) hidden_dim = config.multiple_of * ((hidden_dim + config.multiple_of - 1) // config.multiple_of) self.w1 = nn.Linear(config.n_embd,hidden_dim, bias=False) self.w2 = nn.Linear(hidden_dim,config.n_embd, bias = False) self.w3 = nn.Linear(config.n_embd, hidden_dim, bias= False) def forward(self,x : torch.tensor): return self.w2(F.silu(self.w1(x)) * self.w3(x)) class Block(nn.Module): def __init__(self,config): super().__init__() self.sa = CasualSelfAttention(config) self.mlp = FeedForward(config) self.ln1 = nn.RMSNorm(config.n_embd,eps=1e-05) self.ln2 = nn.RMSNorm(config.n_embd,eps =1e-05) def forward(self,x,freqs): x = x + self.sa(self.ln1(x),freqs) x = x + self.mlp(self.ln2(x)) return x class Model(nn.Module): def __init__(self,config): super().__init__() self.register_buffer( 'freqs_complex',precompute( config.n_embd // config.n_head, config.block_size, device=config.device ) ) self.config = config self.transformer = nn.ModuleDict(dict( wte = nn.Embedding(config.vocab_size,config.n_embd), h = nn.ModuleList([Block(config) for _ in range(config.n_layer)]), ln_f = nn.RMSNorm(config.n_embd,eps = 1e-05) )) self.lm_head = nn.Linear(config.n_embd, config.vocab_size,bias=False) self.transformer.wte.weight = self.lm_head.weight def forward(self,idx,targets=None): B,T = idx.size() x = self.transformer.wte(idx) for block in self.transformer.h: x = block(x,self.freqs_complex[:T]) x = self.transformer.ln_f(x) logits = self.lm_head(x) loss = None if targets is not None: loss = F.cross_entropy(logits.view(-1,logits.size(-1)),targets.view(-1)) return logits,loss def sample_top_p(self,probs,p): probs_sort, probs_idx = torch.sort(probs,dim=-1,descending=True) probs_sum = torch.cumsum(probs_sort,dim=-1) mask = probs_sum - probs_sort > p probs_sort[mask] = 0.0 probs_sort.div_(probs_sort.sum(dim=-1,keepdim=True)) next_token = torch.multinomial(probs_sort,num_samples=1) next_token = torch.gather(probs_idx,-1,next_token) return next_token @torch.no_grad() def generate(self, idx, max_new_tokens, temperature=1.0, top_k=None,top_p = None): for _ in range(max_new_tokens): idx_cond = idx if idx.size(1) <= self.config.block_size else idx[:, -self.config.block_size:] logits, _ = self(idx_cond) logits = logits[:, -1, :] / temperature 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) if top_p is not None: idx_next = self.sample_top_p(probs,top_p) else: idx_next = torch.multinomial(probs, num_samples=1) idx = torch.cat((idx, idx_next), dim=1) return idx