3d_parallel/step5_DP/model.py

import torch 
import torch.nn as nn
import torch.nn.functional as F
import math
import process_group_manager as pgm

def rotate_half(x):
    """Rotates half the hidden dims of the input."""
    x1 = x[..., :x.shape[-1] // 2]
    x2 = x[..., x.shape[-1] // 2:]
    return torch.cat((-x2, x1), dim=-1)

def apply_rotary_pos_emb(q, k, cos, sin):
    """
    Apply usage of rotary embeddings to q and k.
    Expects q, k in shape [batch, heads, seq, dim]
    Expects cos, sin in shape [seq, dim]
    """
    # Reshape cos/sin for broadcasting: [seq, dim] -> [1, 1, seq, dim]
    cos = cos.unsqueeze(0).unsqueeze(0)
    sin = sin.unsqueeze(0).unsqueeze(0)
    
    q_embed = (q * cos) + (rotate_half(q) * sin)
    k_embed = (k * cos) + (rotate_half(k) * sin)
    return q_embed, k_embed

class TritonRMSNorm(nn.Module):
    def __init__(self, hidden_size, eps=1e-5, device=None, dtype=None):
        super().__init__()
        self.eps = eps
        self.weight = nn.Parameter(torch.ones(hidden_size))

    def forward(self, x):
        # Standard RMSNorm implementation
        input_dtype = x.dtype
        x = x.to(torch.float32)
        variance = x.pow(2).mean(-1, keepdim=True)
        x = x * torch.rsqrt(variance + self.eps)
        return self.weight * x.to(input_dtype)

def get_cos_sin(seq_length, head_dim, base=500000.0):
    assert head_dim % 2 == 0
    # Frequency calculation on CPU
    theta = 1.0 / (base ** (torch.arange(0, head_dim, 2, dtype=torch.int64).float().to('cpu') / head_dim))
    
    dtype = torch.bfloat16
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    
    position = torch.arange(seq_length).to(device).unsqueeze(1).float() # [seq_length, 1]
    theta = theta.to(device)
    
    # Returns [seq_length, head_dim]
    cos = torch.cos(position.float() * theta.float()).to(dtype).repeat(1, 2)
    sin = torch.sin(position.float() * theta.float()).to(dtype).repeat(1, 2)
    return cos, sin

class Attention(nn.Module):
    def __init__(self, config, layer_idx):
        super().__init__()
        self.hidden_size = config.hidden_size
        self.num_heads = config.num_attention_heads
        self.num_key_values = config.num_key_value_heads
        self.head_dim = self.hidden_size // self.num_heads
        assert config.num_attention_heads % pgm.process_group_manager.tp_world_size == 0, "num_attention_heads should be divisible by tp world size"
        assert config.num_key_value_heads % pgm.process_group_manager.tp_world_size == 0, "num_key_value_heads should be divisible by  tp world size"
        self.num_local_heads = config.num_attention_heads // pgm.process_group_manager.tp_world_size # TP parallelism
        self.num_local_kv_heads = config.num_key_value_heads // pgm.process_group_manager.tp_world_size # TP parallelism
        
        self.q_proj = nn.Linear(config.hidden_size, self.num_heads * self.head_dim, bias=False)
        self.k_proj = nn.Linear(config.hidden_size, self.num_key_values * self.head_dim, bias=False)
        self.v_proj = nn.Linear(config.hidden_size, self.num_key_values * self.head_dim, bias=False)
        self.out_proj = nn.Linear(config.hidden_size, config.hidden_size, bias=False)
        self.layer_idx = layer_idx
        
    def forward(self, x, cos, sin, attention_mask=None, position_ids=None):
        batch_size, seq_length, hidden_dim = x.size()
        
        q = self.q_proj(x) 
        k = self.k_proj(x) 
        v = self.v_proj(x) 

        # Reshape to [batch, seq, heads, dim] -> Transpose to [batch, heads, seq, dim]
        q = q.view(batch_size, seq_length, self.num_local_heads, self.head_dim).transpose(1, 2)
        k = k.view(batch_size, seq_length, self.num_local_kv_heads, self.head_dim).transpose(1, 2)
        v = v.view(batch_size, seq_length, self.num_local_kv_heads, self.head_dim).transpose(1, 2)
        
        # Apply Rotary Embeddings (Manual implementation)
        # We slice cos/sin to match head_dim because get_cos_sin might produce larger caches in some implementations
        # (though in your code it matches exactly).
        q, k = apply_rotary_pos_emb(q, k, cos[:, :self.head_dim], sin[:, :self.head_dim])
        
        # Repeat KV heads if using Grouped Query Attention (GQA)
        k = k.repeat_interleave(self.num_local_heads // self.num_local_kv_heads, dim=1) 
        v = v.repeat_interleave(self.num_local_heads // self.num_local_kv_heads, dim=1) 
        
        # --- 3. Replacement for Flash Attention Func ---
        # If query and key lengths match, it implies training or full-sequence processing -> usually causal
        is_causal = True if q.size(2) == k.size(2) else False 
        
        out = F.scaled_dot_product_attention(
            q, k, v, 
            attn_mask=None, # SDPA handles causal masking via the is_causal flag
            dropout_p=0.0,
            is_causal=is_causal
        )
        
        # Reshape back: [batch, heads, seq, dim] -> [batch, seq, heads, dim] -> [batch, seq, hidden]
        out = out.transpose(1, 2).contiguous().reshape(batch_size, seq_length, self.num_local_heads * self.head_dim)
        out = self.out_proj(out) 
        return out

class MLP(nn.Module):
    def __init__(self, config) -> None:
        super().__init__()
        self.up_proj = nn.Linear(config.hidden_size, config.intermediate_size, bias=False)
        self.gate_proj = nn.Linear(config.hidden_size, config.intermediate_size, bias=False)
        self.down_proj = nn.Linear(config.intermediate_size, config.hidden_size, bias=False)

    def forward(self, x):
        return self.down_proj(F.silu(self.gate_proj(x)) * self.up_proj(x))

class DecoderLayer(nn.Module):
    # TritonRMSNorm (now standard RMSNorm) -> Attention -> Residual -> TritonRMSNorm -> MLP -> Residual
    def __init__(self, config, layer_idx):
        super().__init__()
        self.input_layernorm = TritonRMSNorm(config.hidden_size, eps=config.rms_norm_eps)
        self.post_attention_layernorm = TritonRMSNorm(config.hidden_size, eps=config.rms_norm_eps)
        self.attention = Attention(config, layer_idx = layer_idx)
        self.mlp = MLP(config)
        self.layer_idx = layer_idx
        head_dim = config.hidden_size // config.num_attention_heads
        # [max_position_embeddings, head_dim]
        self.cos, self.sin = get_cos_sin(config.max_position_embeddings, head_dim=head_dim , base=config.rope_theta) 

    def forward(self, x, attention_mask = None, position_ids = None):
        cos, sin = self.cos, self.sin 
        x = x + self.attention(self.input_layernorm(x), cos, sin, attention_mask, position_ids) # Attention 
        x = x + self.mlp(self.post_attention_layernorm(x)) # MLP
        return x
    
class Llama(nn.Module):
    def __init__(self, config) -> None:
        super().__init__()
        # sanity check 
        assert config.hidden_size % config.num_attention_heads == 0
        assert config.num_attention_heads % config.num_key_value_heads == 0 
        
        # params
        self.vocab_size = config.vocab_size
        self.hidden_size = config.hidden_size
        self.num_heads = config.num_attention_heads
        self.num_key_values = config.num_key_value_heads 
        self.head_dim = self.hidden_size // self.num_heads
        self.max_position_embeddings = config.max_position_embeddings
        self.num_layers = config.num_hidden_layers
        self.model_config = config
        
        # modules
        self.embedding = nn.Embedding(self.vocab_size, self.hidden_size)
        self.decoder_layers = nn.ModuleList([DecoderLayer(config, layer_idx=i) for i in range(self.num_layers)])
        self.final_proj = nn.Linear(self.hidden_size, self.vocab_size, bias=False)
        self.final_norm = TritonRMSNorm(self.hidden_size, eps=config.rms_norm_eps)
        
    def forward(self, input_ids, attention_mask=None, position_ids: torch.Tensor = None):
        x = self.embedding(input_ids)
        for layer in self.decoder_layers:
            x = layer(x)  # [batch_size, seq_length, hidden_dim]
        x = self.final_norm(x)
        logits = self.final_proj(x)
        
        return logits  # [batch_size, seq_length, vocab_size]