model.py

import torch
import torch.nn as nn
import torch.nn.functional as F
from typing import Optional, Tuple
import math

class RMSNorm(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):
        return self._norm(x.float()).type_as(x) * self.weight

class SSMBlock(nn.Module):
    """
    Una implementación simplificada de una capa de State Space Model (SSM) Selectivo.
    A diferencia del Transformer, esta capa tiene memoria LINEAL.
    """
    def __init__(self, dim: int, state_dim: int = 16, expand: int = 2):
        super().__init__()
        self.dim = dim
        self.state_dim = state_dim
        self.inner_dim = expand * dim
        
        # Proyecciones de entrada
        self.in_proj = nn.Linear(dim, self.inner_dim * 2, bias=False)
        
        # Convolución 1D para capturar contexto local (estilo Mamba)
        self.conv1d = nn.Conv1d(
            in_channels=self.inner_dim,
            out_channels=self.inner_dim,
            kernel_size=4,
            groups=self.inner_dim,
            padding=3
        )
        
        # Parámetros del SSM (Simplificado)
        # Delta (dt): El paso de tiempo reactivo
        self.dt_proj = nn.Linear(self.inner_dim, self.inner_dim, bias=True)
        
        # matrices A y B (Simplificadas para ser selectivas)
        self.A_log = nn.Parameter(torch.log(torch.arange(1, state_dim + 1).float().repeat(self.inner_dim, 1)))
        self.B_proj = nn.Linear(self.inner_dim, state_dim, bias=False)
        self.C_proj = nn.Linear(self.inner_dim, state_dim, bias=False)
        
        # Proyección de salida
        self.out_proj = nn.Linear(self.inner_dim, dim, bias=False)

    def forward(self, x: torch.Tensor):
        # x: [Batch, SeqLen, Dim]
        b, l, d = x.shape
        
        # 1. Proyección inicial y split
        x_and_res = self.in_proj(x) # [B, L, 2 * InnerDim]
        x_inner, res = x_and_res.split(self.inner_dim, dim=-1)
        
        # 2. Convolución local
        x_inner = x_inner.transpose(1, 2) # [B, InnerDim, L]
        x_inner = self.conv1d(x_inner)[:, :, :l] # Recortar padding
        x_inner = x_inner.transpose(1, 2) # [B, L, InnerDim]
        x_inner = F.silu(x_inner) # Activación Swish
        
        # 3. Mecanismo SSM Selectivo (Simplificado)
        # En una implementación real (como Mamba), esto se hace con un kernel de GPU 
        # para ser ultra rápido. Aquí usamos un bucle o aproximación para entenderlo.
        
        dt = F.softplus(self.dt_proj(x_inner)) # [B, L, InnerDim]
        A = -torch.exp(self.A_log) # [InnerDim, StateDim]
        B = self.B_proj(x_inner) # [B, L, StateDim]
        C = self.C_proj(x_inner) # [B, L, StateDim]
        
        # El "Estado Oculto" del modelo (Memory)
        # Aquí es donde ocurre la magia: el estado tiene tamaño fijo [B, InnerDim, StateDim]
        state = torch.zeros(b, self.inner_dim, self.state_dim, device=x.device)
        y = torch.zeros(b, l, self.inner_dim, device=x.device)
        
        # Escaneo Secuencial (Selective Scan)
        # Esto reemplaza a la Atención del Transformer.
        for t in range(l):
            # dt_t: [B, InnerDim]
            dt_t = dt[:, t, :].unsqueeze(-1)
            # x_t: [B, InnerDim]
            x_t = x_inner[:, t, :].unsqueeze(-1)
            # B_t: [B, StateDim]
            B_t = B[:, t, :].unsqueeze(1)
            
            # Discretización (Aproximación de Euler)
            A_bar = torch.exp(A.unsqueeze(0) * dt_t) # [B, InnerDim, StateDim]
            B_bar = dt_t * B_t
            
            # Actualizar Estado: h = A*h + B*x
            state = A_bar * state + B_bar * x_t
            
            # Salida: y = C*h
            C_t = C[:, t, :].unsqueeze(-1) # [B, StateDim, 1]
            y[:, t, :] = torch.matmul(state, C_t).squeeze(-1)
        
        # 4. Combinar con el residuo y proyectar salida
        out = y * F.silu(res)
        return self.out_proj(out)

class TransformerKiller(nn.Module):
    """
    Arquitectura basada en SSM (Mamba-like).
    No usa Atención. Contexto teóricamente infinito.
    """
    def __init__(self, vocab_size: int, dim: int, n_layers: int, state_dim: int = 16):
        super().__init__()
        self.tok_embeddings = nn.Embedding(vocab_size, dim)
        
        self.layers = nn.ModuleList([
            nn.ModuleDict({
                'norm': RMSNorm(dim),
                'ssm': SSMBlock(dim, state_dim=state_dim)
            }) for _ in range(n_layers)
        ])
        
        self.norm_f = RMSNorm(dim)
        self.output = nn.Linear(dim, vocab_size, bias=False)

    def forward(self, tokens: torch.Tensor):
        x = self.tok_embeddings(tokens)
        
        for layer in self.layers:
            # Conexión residual + SSM
            x = x + layer['ssm'](layer['norm'](x))
            
        x = self.norm_f(x)
        return self.output(x)

if __name__ == "__main__":
    # Test de dimensiones
    model = TransformerKiller(vocab_size=100, dim=128, n_layers=4)
    test_input = torch.randint(0, 100, (2, 50))
    output = model(test_input)
    print(f"Input shape: {test_input.shape}")
    print(f"SSM Output shape: {output.shape}")
    # Nota como no hay límite de BLOCK_SIZE en el forward, 
    # solo el que dicte tu RAM, pero el coste es Lineal!