models/multimodal_model.py

"""Multi-modal sentiment model combining audio and text."""

import logging
from typing import Any, Dict, Optional

import torch
import torch.nn as nn

from .base_model import BaseModel

logger = logging.getLogger(__name__)


class MultiModalSentimentModel(BaseModel):
    """Multi-modal model combining text and audio features."""
    
    def __init__(
        self,
        text_dim: int = 768,
        audio_dim: int = 8,
        hidden_dim: int = 256,
        num_classes: int = 4,
        dropout: float = 0.2,
    ):
        """
        Args:
            text_dim: Text embedding dimension
            audio_dim: Audio feature dimension
            hidden_dim: Hidden layer dimension
            num_classes: Number of sentiment classes
            dropout: Dropout rate
        """
        super().__init__()
        
        self.text_dim = text_dim
        self.audio_dim = audio_dim
        
        # Text encoder
        self.text_encoder = nn.Sequential(
            nn.Linear(text_dim, hidden_dim),
            nn.ReLU(),
            nn.Dropout(dropout),
        )
        
        # Audio encoder
        self.audio_encoder = nn.Sequential(
            nn.Linear(audio_dim, hidden_dim // 2),
            nn.ReLU(),
            nn.Dropout(dropout),
        )
        
        # Fusion layer
        self.fusion = nn.Sequential(
            nn.Linear(hidden_dim + hidden_dim // 2, hidden_dim),
            nn.ReLU(),
            nn.Dropout(dropout),
        )
        
        # Classification head
        self.classifier = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim // 2),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(hidden_dim // 2, num_classes),
        )
    
    def forward(
        self,
        text_features: torch.Tensor,
        audio_features: torch.Tensor,
    ) -> torch.Tensor:
        """
        Forward pass.
        
        Args:
            text_features: Text embeddings (batch, text_dim)
            audio_features: Audio features (batch, audio_dim)
        
        Returns:
            Logits (batch, num_classes)
        """
        # Encode each modality
        text_encoded = self.text_encoder(text_features)
        audio_encoded = self.audio_encoder(audio_features)
        
        # Concatenate and fuse
        fused = torch.cat([text_encoded, audio_encoded], dim=-1)
        fused = self.fusion(fused)
        
        # Classify
        logits = self.classifier(fused)
        
        return logits
    
    def predict(
        self,
        text_features: torch.Tensor,
        audio_features: Optional[torch.Tensor] = None,
    ) -> Dict[str, Any]:
        """Make predictions."""
        self.eval()
        
        if audio_features is None:
            # Text-only mode
            audio_features = torch.zeros(
                text_features.size(0), self.audio_dim
            ).to(text_features.device)
        
        with torch.no_grad():
            logits = self.forward(text_features, audio_features)
            probs = torch.softmax(logits, dim=-1)
        
        sentiment_labels = ["negative", "neutral", "positive", "sarcastic"]
        
        predictions = []
        for i, probs_i in enumerate(probs):
            pred_idx = probs_i.argmax().item()
            predictions.append({
                "sentiment": sentiment_labels[pred_idx],
                "confidence": probs_i[pred_idx].item(),
                "probabilities": {
                    label: probs_i[j].item()
                    for j, label in enumerate(sentiment_labels)
                },
            })
        
        return {"predictions": predictions}


class CrossModalAttention(nn.Module):
    """Cross-modal attention for audio-text fusion."""
    
    def __init__(
        self,
        query_dim: int,
        key_dim: int,
        hidden_dim: int,
        num_heads: int = 4,
    ):
        super().__init__()
        
        self.num_heads = num_heads
        self.head_dim = hidden_dim // num_heads
        
        self.query = nn.Linear(query_dim, hidden_dim)
        self.key = nn.Linear(key_dim, hidden_dim)
        self.value = nn.Linear(key_dim, hidden_dim)
        
        self.output = nn.Linear(hidden_dim, hidden_dim)
        self.scale = self.head_dim ** -0.5
    
    def forward(
        self,
        query: torch.Tensor,
        key_value: torch.Tensor,
    ) -> torch.Tensor:
        """Cross-attention forward pass."""
        batch_size = query.size(0)
        
        # Linear projections
        Q = self.query(query).view(batch_size, -1, self.num_heads, self.head_dim).transpose(1, 2)
        K = self.key(key_value).view(batch_size, -1, self.num_heads, self.head_dim).transpose(1, 2)
        V = self.value(key_value).view(batch_size, -1, self.num_heads, self.head_dim).transpose(1, 2)
        
        # Attention scores
        scores = torch.matmul(Q, K.transpose(-2, -1)) * self.scale
        attention = torch.softmax(scores, dim=-1)
        
        # Apply attention to values
        context = torch.matmul(attention, V)
        context = context.transpose(1, 2).contiguous().view(batch_size, -1, self.num_heads * self.head_dim)
        
        return self.output(context)


if __name__ == "__main__":
    print("Testing MultiModalSentimentModel...")
    
    model = MultiModalSentimentModel(
        text_dim=768,
        audio_dim=8,
        hidden_dim=256,
        num_classes=4,
    )
    
    # Mock inputs
    text_features = torch.randn(2, 768)
    audio_features = torch.randn(2, 8)
    
    logits = model(text_features, audio_features)
    print(f"Output shape: {logits.shape}")
    print(f"Total parameters: {model.get_num_parameters():,}")