Create model_classes.py

model_classes.py

@@ -0,0 +1,208 @@
+# model_classes.py
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+import json
+
+# ========================= RWKV-Mamba Hybrid =========================
+class RWKVMambaHybrid(nn.Module):
+    """Combines RWKV time-mixing with Mamba state-space dynamics"""
+    def __init__(self, d_model, d_state=64):
+        super().__init__()
+        self.d_model = d_model
+        self.d_state = d_state
+        self.w_mix = nn.Parameter(torch.ones(d_model) * 0.5)
+        self.A = nn.Parameter(torch.randn(d_state, d_state) * 0.01)
+        self.B = nn.Parameter(torch.randn(d_state, d_model) * 0.01)
+        self.C = nn.Parameter(torch.randn(d_model, d_state) * 0.01)
+        self.D = nn.Parameter(torch.ones(d_model) * 0.1)
+
+    def forward(self, x):
+        B, T, C = x.shape
+        h = torch.zeros(B, C, device=x.device)
+        s = torch.zeros(B, self.d_state, device=x.device)
+        outputs = []
+
+        for t in range(T):
+            x_t = x[:, t, :]
+            h = self.w_mix * h + (1 - self.w_mix) * x_t
+            s = s @ self.A.T + x_t @ self.B.T
+            y_t = s @ self.C.T + h * self.D
+            outputs.append(y_t)
+
+        return torch.stack(outputs, dim=1)
+
+
+# ========================= Full Attention =========================
+class FullAttention(nn.Module):
+    """Standard Multi-Head Attention"""
+    def __init__(self, d_model, n_heads=16):
+        super().__init__()
+        self.d_model = d_model
+        self.n_heads = n_heads
+        self.head_dim = d_model // n_heads
+        assert d_model % n_heads == 0, "d_model must be divisible by n_heads"
+
+        self.qkv = nn.Linear(d_model, d_model * 3)
+        self.out_proj = nn.Linear(d_model, d_model)
+
+    def forward(self, x, mask=None):
+        B, T, C = x.shape
+        qkv = self.qkv(x)
+        q, k, v = qkv.chunk(3, dim=-1)
+
+        q = q.view(B, T, self.n_heads, self.head_dim).transpose(1, 2)
+        k = k.view(B, T, self.n_heads, self.head_dim).transpose(1, 2)
+        v = v.view(B, T, self.n_heads, self.head_dim).transpose(1, 2)
+
+        attn = (q @ k.transpose(-2, -1)) / (self.head_dim ** 0.5)
+        if mask is not None:
+            mask = mask.expand(B, self.n_heads, T, T).bool()
+            attn = attn.masked_fill(mask == 0, float('-inf'))
+
+        attn = F.softmax(attn, dim=-1)
+        out = attn @ v
+        out = out.transpose(1, 2).contiguous().view(B, T, C)
+        return self.out_proj(out)
+
+
+# ========================= i3 Hybrid Block =========================
+class i3HybridBlock(nn.Module):
+    """Single hybrid block with RWKV-Mamba + FFN"""
+    def __init__(self, d_model, d_state=64, ffn_mult=4):
+        super().__init__()
+        self.ln1 = nn.LayerNorm(d_model)
+        self.hybrid = RWKVMambaHybrid(d_model, d_state)
+        self.ln2 = nn.LayerNorm(d_model)
+        d_ff = d_model * ffn_mult
+        self.ffn = nn.Sequential(
+            nn.Linear(d_model, d_ff),
+            nn.GELU(),
+            nn.Linear(d_ff, d_model)
+        )
+
+    def forward(self, x, mask=None):
+        x = x + self.hybrid(self.ln1(x))
+        x = x + self.ffn(self.ln2(x))
+        return x
+
+
+# ========================= i3 Attention Block =========================
+class i3AttentionBlock(nn.Module):
+    """Single attention block with MHA + FFN"""
+    def __init__(self, d_model, n_heads=16, ffn_mult=4):
+        super().__init__()
+        self.ln1 = nn.LayerNorm(d_model)
+        self.attn = FullAttention(d_model, n_heads)
+        self.ln2 = nn.LayerNorm(d_model)
+        d_ff = d_model * ffn_mult
+        self.ffn = nn.Sequential(
+            nn.Linear(d_model, d_ff),
+            nn.GELU(),
+            nn.Linear(d_ff, d_model)
+        )
+
+    def forward(self, x, mask=None):
+        x = x + self.attn(self.ln1(x), mask)
+        x = x + self.ffn(self.ln2(x))
+        return x
+
+
+# ========================= i3 Model =========================
+class i3Model(nn.Module):
+    """Full hybrid LLM: 10 Hybrid + 6 Attention blocks"""
+    def __init__(self, vocab_size, d_model=512, n_heads=16,
+                 max_seq_len=256, d_state=32):
+        super().__init__()
+        self.vocab_size = vocab_size
+        self.d_model = d_model
+        self.max_seq_len = max_seq_len
+
+        self.embed = nn.Embedding(vocab_size, d_model)
+        self.pos_embed = nn.Embedding(max_seq_len, d_model)
+
+        hybrid_layers = [i3HybridBlock(d_model, d_state=d_state) for _ in range(10)]
+        attention_layers = [i3AttentionBlock(d_model, n_heads=n_heads) for _ in range(6)]
+        self.layers = nn.ModuleList(hybrid_layers + attention_layers)
+
+        self.ln_f = nn.LayerNorm(d_model)
+        self.head = nn.Linear(d_model, vocab_size)
+        self.apply(self._init_weights)
+
+    def _init_weights(self, module):
+        if isinstance(module, (nn.Linear, nn.Embedding)):
+            module.weight.data.normal_(mean=0.0, std=0.02)
+            if isinstance(module, nn.Linear) and module.bias is not None:
+                module.bias.data.zero_()
+
+    def forward(self, idx, targets=None):
+        B, T = idx.shape
+        assert T <= self.max_seq_len
+        pos = torch.arange(0, T, device=idx.device).unsqueeze(0)
+        x = self.embed(idx) + self.pos_embed(pos)
+        mask = torch.tril(torch.ones(T, T, device=idx.device)).view(1, 1, T, T)
+
+        for layer in self.layers:
+            x = layer(x, mask)
+
+        x = self.ln_f(x)
+        logits = self.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
+
+    @torch.no_grad()
+    def generate(self, idx, max_new_tokens, temperature=1.0, top_k=None):
+        for _ in range(max_new_tokens):
+            idx_cond = idx if idx.size(1) <= self.max_seq_len else idx[:, -self.max_seq_len:]
+            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)
+            idx_next = torch.multinomial(probs, num_samples=1)
+            idx = torch.cat((idx, idx_next), dim=1)
+        return idx
+
+
+# ========================= ChunkTokenizer =========================
+class ChunkTokenizer:
+    """Memory-efficient 2-3 character chunk tokenizer"""
+    def __init__(self):
+        self.chunk_to_idx = {}
+        self.idx_to_chunk = {}
+        self.vocab_size = 0
+        self.unk_token = '<UNK>'
+        self.unk_idx = 0
+
+    def load(self, path):
+        with open(path, 'r') as f:
+            data = json.load(f)
+        self.chunk_to_idx = data['chunk_to_idx']
+        self.idx_to_chunk = {int(k): v for k, v in data['idx_to_chunk'].items()}
+        self.vocab_size = data['vocab_size']
+        self.unk_token = data.get('unk_token', '<UNK>')
+        self.unk_idx = data.get('unk_idx', 0)
+
+    def encode(self, text):
+        text = text.lower()
+        pos = 0
+        indices = []
+        while pos < len(text):
+            for chunk_len in [3, 2, 1]:
+                chunk = text[pos:pos+chunk_len]
+                if chunk in self.chunk_to_idx:
+                    indices.append(self.chunk_to_idx[chunk])
+                    pos += chunk_len
+                    break
+            else:
+                indices.append(self.unk_idx)
+                pos += 1
+        return indices
+
+    def decode(self, indices):
+        return ''.join([self.idx_to_chunk.get(int(i), self.unk_token) for i in indices])