models_best/encoder.py

"""
Transformer Encoder with Pre-Layer Normalization
"""

import torch
import torch.nn as nn
from typing import Optional

from .attention import MultiHeadAttention
from .feed_forward import FeedForward
from .layer_norm import LayerNorm
from .embeddings import ScaledEmbedding
from .positional_encoding import PositionalEncoding


class EncoderLayer(nn.Module):
    """
    Transformer Encoder Layer with Pre-Layer Normalization
    
    Structure:
    1. LayerNorm -> Self-Attention -> Dropout -> Residual
    2. LayerNorm -> Feed-Forward -> Dropout -> Residual
    """
    
    def __init__(self, d_model: int, n_heads: int, d_ff: int,
                 dropout: float = 0.1, attention_dropout: float = 0.1,
                 activation_dropout: float = 0.0):
        """
        Args:
            d_model: Model dimension
            n_heads: Number of attention heads
            d_ff: Feed-forward dimension
            dropout: Dropout rate
            attention_dropout: Dropout rate for attention
            activation_dropout: Dropout rate for FFN activation
        """
        super().__init__()
        
        # Pre-layer normalization
        self.norm1 = LayerNorm(d_model)
        self.norm2 = LayerNorm(d_model)
        
        # Multi-head self-attention
        self.self_attn = MultiHeadAttention(
            d_model, n_heads, dropout, attention_dropout
        )
        
        # Feed-forward network
        self.ffn = FeedForward(d_model, d_ff, dropout, activation_dropout)
        
        # Dropout for residual connections
        self.dropout1 = nn.Dropout(dropout)
        self.dropout2 = nn.Dropout(dropout)
    
    def forward(self, x, mask: Optional[torch.Tensor] = None,
                return_attention: bool = False):
        """
        Args:
            x: [batch_size, seq_len, d_model]
            mask: [batch_size, 1, 1, seq_len]
            return_attention: Whether to return attention weights
        
        Returns:
            output: [batch_size, seq_len, d_model]
            attention: [batch_size, n_heads, seq_len, seq_len] (if return_attention)
        """
        # Self-attention block with pre-norm
        residual = x
        x = self.norm1(x)
        
        if return_attention:
            attn_out, attn_weights = self.self_attn(x, x, x, mask, return_attention=True)
            x = residual + self.dropout1(attn_out)
        else:
            x = residual + self.dropout1(self.self_attn(x, x, x, mask))
            attn_weights = None
        
        # Feed-forward block with pre-norm
        residual = x
        x = self.norm2(x)
        x = residual + self.dropout2(self.ffn(x))
        
        if return_attention:
            return x, attn_weights
        return x


class TransformerEncoder(nn.Module):
    """
    Complete Transformer Encoder
    """
    
    def __init__(self, vocab_size: int, d_model: int, n_heads: int,
                 d_ff: int, n_layers: int, max_len: int = 5000,
                 dropout: float = 0.1, attention_dropout: float = 0.1,
                 activation_dropout: float = 0.0, pad_idx: int = 0,
                 scale_embedding: bool = True):
        """
        Args:
            vocab_size: Source vocabulary size
            d_model: Model dimension
            n_heads: Number of attention heads
            d_ff: Feed-forward dimension
            n_layers: Number of encoder layers
            max_len: Maximum sequence length
            dropout: Dropout rate
            attention_dropout: Dropout rate for attention
            activation_dropout: Dropout rate for FFN activation
            pad_idx: Padding token index
            scale_embedding: Whether to scale embeddings
        """
        super().__init__()
        
        self.d_model = d_model
        self.pad_idx = pad_idx
        
        # Embedding layer
        self.embedding = ScaledEmbedding(
            vocab_size, d_model, pad_idx, scale=scale_embedding, dropout=0.0
        )
        
        # Positional encoding
        self.pos_encoding = PositionalEncoding(d_model, max_len, dropout)
        
        # Stack of encoder layers
        self.layers = nn.ModuleList([
            EncoderLayer(
                d_model, n_heads, d_ff, dropout,
                attention_dropout, activation_dropout
            )
            for _ in range(n_layers)
        ])
        
        # Final layer norm (important for Pre-LN)
        self.final_norm = LayerNorm(d_model)
        
        self.dropout = nn.Dropout(dropout)
    
    def forward(self, src, src_mask: Optional[torch.Tensor] = None,
                return_attention: bool = False):
        """
        Args:
            src: [batch_size, src_len]
            src_mask: [batch_size, 1, 1, src_len]
            return_attention: Whether to return attention weights
        
        Returns:
            output: [batch_size, src_len, d_model]
            attentions: List of attention weights (if return_attention)
        """
        # Embedding + positional encoding
        x = self.embedding(src)
        x = self.pos_encoding(x)
        
        # Pass through encoder layers
        attentions = [] if return_attention else None
        
        for layer in self.layers:
            if return_attention:
                x, attn = layer(x, src_mask, return_attention=True)
                attentions.append(attn)
            else:
                x = layer(x, src_mask, return_attention=False)
        
        # Final layer normalization
        x = self.final_norm(x)
        
        if return_attention:
            return x, attentions
        return x
    
    def init_weights(self, init_std: float = 0.02):
        """Initialize model weights"""
        # Initialize embeddings
        self.embedding.init_weights(init_std)
        
        # Initialize linear layers
        for module in self.modules():
            if isinstance(module, nn.Linear):
                nn.init.normal_(module.weight, mean=0.0, std=init_std)
                if module.bias is not None:
                    nn.init.zeros_(module.bias)