code/model.py

import math
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.checkpoint import checkpoint
from dataclasses import dataclass

@dataclass
class ModelConfig:
    """Configuration for ViuResonance100M - T4 Optimized"""
    vocab_size: int = 64000
    d_model: int = 1024
    n_layers: int = 12
    n_heads: int = 16
    dropout: float = 0.1
    chunk_size: int = 256  # T4 ke liye 256, 512 se OOM aayega
    max_seq_len: int = 4096

class RMSNorm(nn.Module):
    def __init__(self, dim: int, eps: float = 1e-6):
        super().__init__()
        self.weight = nn.Parameter(torch.ones(dim))
        self.eps = eps
        
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        # T4 fp16 me stable rakhne ke liye float32 me norm
        try:
            # PyTorch 2.4+ me available
            if hasattr(F, 'rms_norm'):
                return F.rms_norm(x.float(), (x.shape[-1],), self.weight.float(), self.eps).to(x.dtype)
        except Exception:
            pass
        norm_x = x.float() * torch.rsqrt(x.float().pow(2).mean(-1, keepdim=True) + self.eps)
        return (norm_x * self.weight.float()).to(x.dtype)
        # return (norm_x * self.weight.float()).to(x.dtype)

class ResonanceLayerKaggle(nn.Module):
    def __init__(self, config: ModelConfig):
        super().__init__()
        if config.d_model % config.n_heads != 0:
            raise ValueError(f"d_model ({config.d_model}) must be divisible by n_heads ({config.n_heads})")
            
        self.n_heads = config.n_heads
        self.head_dim = config.d_model // config.n_heads
        self.chunk_size = config.chunk_size
        
        self.to_amp = nn.Linear(config.d_model, config.d_model, bias=False)
        self.to_freq = nn.Linear(config.d_model, config.n_heads, bias=False)
        self.to_phase = nn.Linear(config.d_model, config.n_heads, bias=False)
        self.to_v = nn.Linear(config.d_model, config.d_model, bias=False)
        self.to_out = nn.Linear(config.d_model, config.d_model, bias=False)
        self.norm = RMSNorm(config.d_model)
        
        # T4: pos cache - har forward me arange banane se bachao
        self.register_buffer("pos_cache", torch.arange(config.max_seq_len, dtype=torch.long), persistent=False)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        B, T, C = x.shape
        h = self.norm(x)
        A = self.to_amp(h).view(B, T, self.n_heads, self.head_dim)
        V = self.to_v(h).view(B, T, self.n_heads, self.head_dim)
        freq = torch.tanh(self.to_freq(h)) * 2.0  # B,T,H
        phase = torch.tanh(self.to_phase(h)) * math.pi

        # pos cache use karo, T tak slice
        if T > self.pos_cache.shape[0]:
            # Agar kabhi bada T aaye to dynamically banao
            pos = torch.arange(T, device=x.device, dtype=torch.long)
        else:
            pos = self.pos_cache[:T]

        out = torch.empty(B, T, C, device=x.device, dtype=x.dtype)
        
        for i in range(0, T, self.chunk_size):
            end = min(i+self.chunk_size, T)
            chunk_len = end - i
            
            phase_i = phase[:, i:end, :].permute(0,2,1).unsqueeze(-1) # B,H,chunk,1
            phase_j = phase.permute(0,2,1).unsqueeze(-2) # B,H,1,T
            
            row_idx = torch.arange(i, end, device=x.device, dtype=torch.long)[:, None]
            col_idx = pos[None, :]
            dist_long = (row_idx - col_idx).clamp(min=0)
            dist = dist_long.to(freq.dtype).view(1, 1, chunk_len, T)
            
            freq_i = freq[:, i:end, :].permute(0,2,1).unsqueeze(-1)
            angle = (phase_i - phase_j) + freq_i * dist * 0.05
            
            score = torch.cos(angle) # B,H,chunk,T
            
            causal_bool = (row_idx >= col_idx).view(1,1,chunk_len,T)
            score = score.masked_fill(~causal_bool, 0.0)
            score = score / math.sqrt(self.head_dim)
            
            amp = A.norm(dim=-1).permute(0,2,1) # B,H,T
            amp_i = amp[:,:,i:end].unsqueeze(-1)
            amp_j = amp.unsqueeze(-2)
            score = score * torch.sigmoid(amp_i * amp_j)
            
            V_t = V.permute(0,2,1,3).contiguous() # B,H,T,D
            Bh = B * self.n_heads
            
            s = score.reshape(Bh, chunk_len, T)
            v = V_t.reshape(Bh, T, self.head_dim)
            o = torch.bmm(s, v).view(B, self.n_heads, chunk_len, self.head_dim).permute(0,2,1,3).reshape(B, chunk_len, C)
            out[:, i:end] = o

        return self.to_out(out)

class ViuResonance100M(nn.Module):
    def __init__(self, config: ModelConfig):
        super().__init__()
        self.config = config
        self.emb = nn.Embedding(config.vocab_size, config.d_model)
        self.layers = nn.ModuleList([ResonanceLayerKaggle(config) for _ in range(config.n_layers)])
        self.norm = RMSNorm(config.d_model)
        self.head = nn.Linear(config.d_model, config.vocab_size, bias=False)
        self.resid_dropout = nn.Dropout(config.dropout)
        
        # Sahi order: pehle init, phir tie
        self.apply(self._init_weights)
        self.head.weight = self.emb.weight

    def _init_weights(self, module: nn.Module):
        if isinstance(module, nn.Linear):
            torch.nn.init.normal_(module.weight, mean=0.0, std=0.02)
        elif isinstance(module, nn.Embedding):
            torch.nn.init.normal_(module.weight, mean=0.0, std=0.02)
        elif isinstance(module, RMSNorm):
            torch.nn.init.ones_(module.weight)

    def forward(self, idx: torch.Tensor, targets: torch.Tensor = None, use_checkpoint: bool = True) -> tuple[torch.Tensor, torch.Tensor | None]:
        x = self.emb(idx)
        for layer in self.layers:
            if self.training and use_checkpoint:
                x = x + self.resid_dropout(checkpoint(layer, x, use_reentrant=False))
            else:
                x = x + self.resid_dropout(layer(x))
                
        logits = self.head(self.norm(x))
        loss = None
        if targets is not None:
            loss = F.cross_entropy(logits.view(-1, logits.size(-1)), targets.view(-1), ignore_index=-100)
        return logits, loss