Create gradio_demo.py

src/gradio_demo.py

@@ -0,0 +1,224 @@
+import torch
+import torch.nn.functional as F
+from dataclasses import dataclass
+
+import tiktoken
+
+import safetensors.torch
+
+tokenizer = tiktoken.get_encoding("gpt2")
+
+import gradio as gr
+from huggingface_hub import hf_hub_download
+
+# Define the GPTConfig dataclass
+@dataclass
+class GPTConfig:
+    vocab_size : int = 50304
+    n_layer : int = 12
+    n_head : int = 6 # head dim 128 suggested by @Grad62304977
+    n_embd : int = 768
+
+# Define the Rotary class
+class Rotary(torch.nn.Module):
+
+    def __init__(self, dim, base=10000):
+        super().__init__()
+        self.inv_freq = 1.0 / (base ** (torch.arange(0, dim, 2).float() / dim))
+        self.seq_len_cached = None
+        self.cos_cached = None
+        self.sin_cached = None
+
+    def forward(self, x):
+        seq_len = x.shape[1]
+        if seq_len!= self.seq_len_cached:
+            self.seq_len_cached = seq_len
+            t = torch.arange(seq_len, device=x.device).type_as(self.inv_freq)
+            freqs = torch.outer(t, self.inv_freq).to(x.device)
+            self.cos_cached = freqs.cos().bfloat16()
+            self.sin_cached = freqs.sin().bfloat16()
+        return self.cos_cached[None, :, None, :], self.sin_cached[None, :, None, :]
+
+def apply_rotary_emb(x, cos, sin):
+    assert x.ndim == 4 # multihead attention
+    d = x.shape[3]//2
+    x1 = x[..., :d]
+    x2 = x[..., d:]
+    y1 = x1 * cos + x2 * sin
+    y2 = x1 * (-sin) + x2 * cos
+    return torch.cat([y1, y2], 3).type_as(x)
+
+# Define the CausalSelfAttention class
+class CausalSelfAttention(torch.nn.Module):
+
+    def __init__(self, config):
+        super().__init__()
+        self.n_head = config.n_head
+        self.n_embd = config.n_embd
+        self.head_dim = self.n_embd // self.n_head
+        assert self.n_embd % self.n_head == 0
+        self.c_q = torch.nn.Linear(self.n_embd, self.n_embd, bias=False)
+        self.c_k = torch.nn.Linear(self.n_embd, self.n_embd, bias=False)
+        self.c_v = torch.nn.Linear(self.n_embd, self.n_embd, bias=False)
+        # output projection
+        self.c_proj = torch.nn.Linear(self.n_embd, self.n_embd, bias=False)
+        self.c_proj.weight.data.zero_() # zero init suggested by @Grad62304977
+        self.rotary = Rotary(self.head_dim)
+        self.lamb = torch.nn.Parameter(torch.tensor(0.5)) # @Grad62304977
+
+    def forward(self, x, v1=None):
+        B, T, C = x.size() # batch size, sequence length, embedding dimensionality (n_embd)
+        q = self.c_q(x).view(B, T, self.n_head, self.head_dim)
+        k = self.c_k(x).view(B, T, self.n_head, self.head_dim)
+        v = self.c_v(x).view(B, T, self.n_head, self.head_dim)
+        if v1 is None:
+            v1 = v # This happens if we are in the first block. v needs to be accessed by subsequent blocks
+        v = (1 - self.lamb) * v + self.lamb * v1.view_as(v) # @Grad62304977
+        cos, sin = self.rotary(q)
+        q, k = F.rms_norm(q, (q.size(-1),)), F.rms_norm(k, (k.size(-1),)) # QK norm suggested by @Grad62304977
+        q, k = apply_rotary_emb(q, cos, sin), apply_rotary_emb(k, cos, sin)
+        y = F.scaled_dot_product_attention(q.transpose(1, 2), k.transpose(1, 2), v.transpose(1, 2), is_causal=True)
+        y = y.transpose(1, 2).contiguous().view_as(x) # re-assemble all head outputs side by side
+        y = self.c_proj(y)
+        return y, v1
+
+# Define the MLP class
+class MLP(torch.nn.Module):
+
+    def __init__(self, config):
+        super().__init__()
+        self.c_fc    = torch.nn.Linear(config.n_embd, 4 * config.n_embd, bias=False)
+        self.c_proj  = torch.nn.Linear(4 * config.n_embd, config.n_embd, bias=False)
+        self.c_proj.weight.data.zero_() # zero init suggested by @Grad62304977
+
+    def forward(self, x):
+        x = self.c_fc(x)
+        x = F.relu(x).square() # https://arxiv.org/abs/2109.08668v2; ~1-2% better than GELU; suggested by @SKYLINEZ007 and @Grad62304977
+        x = self.c_proj(x)
+        return x
+
+# Define the Block class
+class Block(torch.nn.Module):
+
+    def __init__(self, config):
+        super().__init__()
+        self.attn = CausalSelfAttention(config)
+        self.mlp = MLP(config)
+        self.lambdas = torch.nn.Parameter(torch.tensor([1., 0.]))
+
+    def forward(self, x, v1, x0):
+        x = self.lambdas[0] * x + self.lambdas[1] * x0
+        x1, v1 = self.attn(F.rms_norm(x, (x.size(-1),)), v1)
+        x = x + x1
+        x = x + self.mlp(F.rms_norm(x, (x.size(-1),)))
+        return x, v1
+
+# Define the GPT class
+class GPT(torch.nn.Module):
+
+    def __init__(self, config):
+        super().__init__()
+        self.config = config
+
+        self.transformer = torch.nn.ModuleDict(dict(
+            wte = torch.nn.Embedding(config.vocab_size, config.n_embd),
+            h = torch.nn.ModuleList([Block(config) for _ in range(config.n_layer)]),
+        ))
+        self.lm_head = torch.nn.Linear(config.n_embd, config.vocab_size, bias=False)
+        self.lm_head.weight.data.zero_() # @Grad62304977
+
+    def forward(self, idx, targets=None, return_logits=True):
+
+        # forward the GPT model itself
+        x = self.transformer.wte(idx) # token embeddings of shape (b, t, n_embd)
+        x = F.rms_norm(x, (x.size(-1),)) # @Grad62304977
+        x0 = x
+        v1 = None
+        for block in self.transformer.h:
+            x, v1 = block(x, v1, x0)
+        x = F.rms_norm(x, (x.size(-1),))
+
+        if targets is not None:
+            # if we are given some desired targets also calculate the loss
+            logits = self.lm_head(x)
+            logits = 30 * torch.tanh(logits / 30) # @Grad62304977
+            logits = logits.float() # use tf32/fp32 for logits
+            loss = F.cross_entropy(logits.view(-1, logits.size(-1)), targets.view(-1), ignore_index=-1)
+        else:
+            # inference-time mini-optimization: only forward the lm_head on the very last position
+            logits = self.lm_head(x[:, [-1], :]) # note: using list [-1] to preserve the time dim
+            logits = 30 * torch.tanh(logits / 30) # @Grad62304977
+            logits = logits.float() # use tf32/fp32 for logits
+            loss = None
+
+        # there are performance reasons why not returning logits is prudent, if not needed
+        if not return_logits:
+            logits = None
+
+        return logits, loss
+    
+    def generate(self, idx, max_new_tokens, temperature=1.0, top_k=None):
+        """
+        Take a conditioning sequence of indices idx (LongTensor of shape (b,t)) and complete
+        the sequence max_new_tokens times, feeding the predictions back into the model each time.
+        Most likely you'll want to make sure to be in model.eval() mode of operation for this.
+        """
+        for _ in range(max_new_tokens):
+            # if the sequence context is growing too long we must crop it at block_size
+            #idx_cond = idx if idx.size(1) <= self.config.block_size else idx[:, -self.config.block_size:]
+            # forward the model to get the logits for the index in the sequence
+            logits, _ = self(idx)
+            # pluck the logits at the final step and scale by desired temperature
+            logits = logits[:, -1, :] / temperature
+            # optionally crop the logits to only the top k options
+            if top_k is not None:
+                v, _ = torch.topk(logits, min(top_k, logits.size(-1)))
+                logits[logits < v[:, [-1]]] = -float('Inf')
+            # apply softmax to convert logits to (normalized) probabilities
+            probs = F.softmax(logits, dim=-1)
+            # sample from the distribution
+            idx_next = torch.multinomial(probs, num_samples=1)
+            # append sampled index to the running sequence and continue
+            idx = torch.cat((idx, idx_next), dim=1)
+
+        return idx
+
+
+# Run LLM inference
+def run_inference(model, input_ids, max_new_tokens, temperature):
+    input_ids = torch.tensor(input_ids).unsqueeze(0)
+    return model.generate(input_ids, max_new_tokens=max_new_tokens, temperature=temperature)
+
+# Main function
+def load_model():
+    config = GPTConfig()
+    model = GPT(config)
+    model_path = 'nanogpt-speedrun-baseline.safetensors'  # replace with your checkpoint path
+    hf_path = hf_hub_download("lemonteaa/nanogpt-speedrun", model_path)
+    missing, unexpected = safetensors.torch.load_model(model, hf_path)
+    model.eval()
+    return model
+
+nanogpt_model = load_model()
+
+def text_complete(prompt, max_new_tokens, temperature):
+    input_ids = tokenizer.encode_ordinary(prompt)
+    output_ids = run_inference(nanogpt_model, input_ids, max_new_tokens, temperature)
+    tmp = output_ids.squeeze().tolist()
+    return tokenizer.decode(tmp)
+
+
+demo = gr.Interface(
+    title="NanoGPT Speedrun Baseline Model Inference Demo",
+    fn=text_complete,
+    inputs=[
+        gr.Text(label="Prompt"),
+        gr.Slider(label="max new tokens", minimum=10, maximum=500, value=100, step=1),
+        gr.Slider(label="temperature", minimum=0.0, maximum=2.0, value=0.9, step=0.1)
+    ],
+    outputs=["text"],
+    examples=[["Once upon a time, "], ["A quick and delicious recipe for pancake: "], ["The role of IT in supply chain is "]]
+)
+
+demo.queue()
+demo.launch()