Delete models/multimodal_attention.py with huggingface_hub

models/multimodal_attention.py

@@ -1,470 +0,0 @@
-"""multimodal_attention.py — Multi-head Attention Multimodal para CNN-BiGRU.
-
-Implementa atenção multi-cabeça cross-modal que combina informações de
-múltiplas modalidades (texto A, texto B, imagem, áudio) via mecanismo
-de atenção cruzada hierárquica.
-
-==============================================================================
-ANÁLISE MATEMÁTICA E LÓGICA
-==============================================================================
-
-Dadas as representações de N modalidades:
-    H_1, H_2, ..., H_N   onde H_i ∈ R^{B×T_i×d_i}
-
-A multimodal multi-head attention computa:
-
-    Para cada par (i, j) com i != j:
-        Q_i = H_i W_Q^i          [B, T_i, d]
-        K_j = H_j W_K^j          [B, T_j, d]
-        V_j = H_j W_V^j          [B, T_j, d]
-
-        A_{i->j} = softmax(Q_i K_j^T / sqrt(d_h)) V_j   [B, T_i, d]
-
-    Output para modalidade i:
-        H_i' = H_i + sum_{j != i} A_{i->j}
-
-Em seguida, uma camada de fusão final agrega todas as H_i':
-    H_fused = sum_i alpha_i * Pool(H_i')   onde alpha_i é um gate de modalidade
-
-VANTAGENS:
-    1. Cada modalidade "olha" para todas as outras — descoberta de correlações
-    2. Atenção multi-cabeça permite capturar diferentes tipos de relações
-    3. Gate de modalidade permite PESAR dinamicamente cada modalidade
-       (e.g., imagem é mais importante para pergunta visual)
-
-==============================================================================
-INTEGRAÇÃO
-==============================================================================
-
-- Recebe as saídas do CooperativeCNNBiGRU (seq_A, seq_B) e dos encoders
-  multimodais (img_seq, audio_seq)
-- Substitui a MultimodalFusion simples (gated) por uma versão com atenção
-- Mantém compatibilidade com a interface existente
-
-Autor: CNN-BiGRU Project
-"""
-from __future__ import annotations
-
-import logging
-import math
-from dataclasses import dataclass, field
-from typing import Dict, List, Optional, Tuple
-
-import torch
-import torch.nn as nn
-import torch.nn.functional as F
-
-logger = logging.getLogger(__name__)
-
-
-# ============================================================================
-# Configuração
-# ============================================================================
-
-@dataclass
-class MultimodalAttentionConfig:
-    """Configuração da atenção multimodal multi-cabeça."""
-    # Dimensões por modalidade (input)
-    d_text_a: int = 128
-    d_text_b: int = 128
-    d_image: int = 32
-    d_audio: int = 32
-    # Dimensão interna (todas modalidades são projetadas para d_model)
-    d_model: int = 128
-    # Número de cabeças
-    n_heads: int = 4
-    # Dropout
-    dropout: float = 0.1
-    # Modos de fusão final
-    fusion_mode: str = "attention"  # "attention" | "gated" | "concat"
-    # Usar gate de modalidade (alpha_i)
-    use_modality_gate: bool = True
-    # Device
-    device: str = "cpu"
-
-
-# ============================================================================
-# Cross-Modal Attention Block
-# ============================================================================
-
-class CrossModalAttention(nn.Module):
-    """Atenção cruzada entre duas modalidades.
-
-    A modalidade "query" atende sobre a modalidade "key_value".
-    """
-
-    def __init__(
-        self,
-        d_query: int,
-        d_kv: int,
-        d_model: int,
-        n_heads: int = 4,
-        dropout: float = 0.1,
-    ):
-        super().__init__()
-        assert d_model % n_heads == 0, \
-            f"d_model ({d_model}) deve ser divisível por n_heads ({n_heads})"
-        self.d_model = d_model
-        self.n_heads = n_heads
-        self.head_dim = d_model // n_heads
-
-        # Projeções para Query (da modalidade query)
-        self.q_proj = nn.Linear(d_query, d_model)
-        # Projeções para Key, Value (da modalidade kv)
-        self.k_proj = nn.Linear(d_kv, d_model)
-        self.v_proj = nn.Linear(d_kv, d_model)
-        # Output projection
-        self.out_proj = nn.Linear(d_model, d_query)
-
-        self.dropout = nn.Dropout(dropout)
-        self.scale = 1.0 / math.sqrt(self.head_dim)
-
-        # Init
-        for layer in (self.q_proj, self.k_proj, self.v_proj, self.out_proj):
-            nn.init.xavier_uniform_(layer.weight)
-            nn.init.zeros_(layer.bias)
-
-    def forward(
-        self,
-        query: torch.Tensor,
-        kv: torch.Tensor,
-        kv_mask: Optional[torch.Tensor] = None,
-    ) -> torch.Tensor:
-        """
-        Args:
-            query: [B, T_q, d_query]
-            kv: [B, T_kv, d_kv]
-            kv_mask: [B, T_kv] (1=valid, 0=pad)
-
-        Returns:
-            output: [B, T_q, d_query] — query enriquecida com info de kv
-        """
-        B, Tq, _ = query.shape
-        Tkv = kv.size(1)
-
-        # Projeções
-        q = self.q_proj(query)  # [B, T_q, d_model]
-        k = self.k_proj(kv)     # [B, T_kv, d_model]
-        v = self.v_proj(kv)     # [B, T_kv, d_model]
-
-        # Reshape para multi-head: [B, n_heads, T, head_dim]
-        q = q.view(B, Tq, self.n_heads, self.head_dim).transpose(1, 2)
-        k = k.view(B, Tkv, self.n_heads, self.head_dim).transpose(1, 2)
-        v = v.view(B, Tkv, self.n_heads, self.head_dim).transpose(1, 2)
-
-        # Scores: [B, n_heads, T_q, T_kv]
-        scores = torch.matmul(q, k.transpose(-2, -1)) * self.scale
-
-        # Máscara (padding em kv)
-        if kv_mask is not None:
-            scores = scores.masked_fill(
-                kv_mask.unsqueeze(1).unsqueeze(1) == 0, -1e9
-            )
-
-        # Softmax
-        attn = F.softmax(scores, dim=-1)
-        attn = self.dropout(attn)
-
-        # Contexto: [B, n_heads, T_q, head_dim]
-        ctx = torch.matmul(attn, v)
-        # Reshape: [B, T_q, d_model]
-        ctx = ctx.transpose(1, 2).contiguous().view(B, Tq, self.d_model)
-
-        # Output projection de volta para d_query
-        return self.out_proj(ctx)
-
-
-# ============================================================================
-# Modality Gate (peso dinâmico por modalidade)
-# ============================================================================
-
-class ModalityGate(nn.Module):
-    """Computa pesos alpha_i ∈ [0, 1] para cada modalidade.
-
-    Usa a representação pooled de cada modalidade para decidir o peso.
-    """
-
-    def __init__(
-        self,
-        d_text_a: int,
-        d_text_b: int,
-        d_image: int,
-        d_audio: int,
-        d_hidden: int = 64,
-    ):
-        super().__init__()
-        # Projeta cada modalidade para d_hidden
-        self.proj_a = nn.Linear(d_text_a, d_hidden)
-        self.proj_b = nn.Linear(d_text_b, d_hidden)
-        self.proj_img = nn.Linear(d_image, d_hidden)
-        self.proj_aud = nn.Linear(d_audio, d_hidden)
-
-        # Score: d_hidden -> 1
-        self.scorer = nn.Linear(d_hidden, 1)
-        nn.init.zeros_(self.scorer.bias)
-
-    def forward(
-        self,
-        h_a: torch.Tensor,
-        h_b: torch.Tensor,
-        h_img: torch.Tensor,
-        h_aud: torch.Tensor,
-    ) -> Dict[str, torch.Tensor]:
-        """
-        Args:
-            h_a, h_b, h_img, h_aud: [B, d_i] representações pooled
-
-        Returns:
-            dict com alpha_i ∈ [B, 1] para cada modalidade
-        """
-        # Projeta e computa score
-        s_a = self.scorer(torch.tanh(self.proj_a(h_a)))
-        s_b = self.scorer(torch.tanh(self.proj_b(h_b)))
-        s_img = self.scorer(torch.tanh(self.proj_img(h_img)))
-        s_aud = self.scorer(torch.tanh(self.proj_aud(h_aud)))
-
-        # Stack e softmax sobre as 4 modalidades
-        scores = torch.cat([s_a, s_b, s_img, s_aud], dim=-1)  # [B, 4]
-        weights = F.softmax(scores, dim=-1)  # [B, 4]
-
-        return {
-            "alpha_a": weights[:, 0:1],
-            "alpha_b": weights[:, 1:2],
-            "alpha_img": weights[:, 2:3],
-            "alpha_aud": weights[:, 3:4],
-            "weights": weights,
-        }
-
-
-# ============================================================================
-# Multimodal Multi-Head Attention
-# ============================================================================
-
-class MultimodalMultiHeadAttention(nn.Module):
-    """Atenção multi-cabeça cross-modal completa.
-
-    Combina 4 modalidades (texto A, texto B, imagem, áudio) via atenção
-    cruzada par a par + gate de modalidade.
-    """
-
-    def __init__(self, config: MultimodalAttentionConfig):
-        super().__init__()
-        self.config = config
-
-        # Cross-modal attention: cada par (i, j) com i != j
-        # Total: 4*3 = 12 pares direcionais
-        # Simplificação: projetar todas para d_model, usar 4 blocos de cross-attn
-        # Texto A atende B, img, aud
-        self.attn_a_to_b = CrossModalAttention(
-            config.d_text_a, config.d_text_b, config.d_model, config.n_heads, config.dropout
-        )
-        self.attn_a_to_img = CrossModalAttention(
-            config.d_text_a, config.d_image, config.d_model, config.n_heads, config.dropout
-        )
-        self.attn_a_to_aud = CrossModalAttention(
-            config.d_text_a, config.d_audio, config.d_model, config.n_heads, config.dropout
-        )
-        # Texto B atende A, img, aud
-        self.attn_b_to_a = CrossModalAttention(
-            config.d_text_b, config.d_text_a, config.d_model, config.n_heads, config.dropout
-        )
-        self.attn_b_to_img = CrossModalAttention(
-            config.d_text_b, config.d_image, config.d_model, config.n_heads, config.dropout
-        )
-        self.attn_b_to_aud = CrossModalAttention(
-            config.d_text_b, config.d_audio, config.d_model, config.n_heads, config.dropout
-        )
-        # Imagem atende A, B, aud (opcional — pode ser pesado)
-        self.attn_img_to_a = CrossModalAttention(
-            config.d_image, config.d_text_a, config.d_model, config.n_heads, config.dropout
-        )
-        self.attn_img_to_b = CrossModalAttention(
-            config.d_image, config.d_text_b, config.d_model, config.n_heads, config.dropout
-        )
-        # Áudio atende A, B
-        self.attn_aud_to_a = CrossModalAttention(
-            config.d_audio, config.d_text_a, config.d_model, config.n_heads, config.dropout
-        )
-        self.attn_aud_to_b = CrossModalAttention(
-            config.d_audio, config.d_text_b, config.d_model, config.n_heads, config.dropout
-        )
-
-        # LayerNorm para cada modalidade (estabilidade)
-        self.ln_a = nn.LayerNorm(config.d_text_a)
-        self.ln_b = nn.LayerNorm(config.d_text_b)
-        self.ln_img = nn.LayerNorm(config.d_image)
-        self.ln_aud = nn.LayerNorm(config.d_audio)
-
-        # Modality gate
-        self.gate = ModalityGate(
-            config.d_text_a, config.d_text_b, config.d_image, config.d_audio
-        ) if config.use_modality_gate else None
-
-        # Pooling projections (para fusão final)
-        d_total = config.d_text_a + config.d_text_b + config.d_image + config.d_audio
-        if config.fusion_mode == "attention":
-            self.fusion_proj = nn.Linear(d_total, config.d_model)
-        elif config.fusion_mode == "gated":
-            self.fusion_proj = nn.Linear(d_total, config.d_model)
-        else:  # concat
-            self.fusion_proj = nn.Linear(d_total, config.d_model)
-
-    def forward(
-        self,
-        seq_a: torch.Tensor,
-        seq_b: torch.Tensor,
-        seq_img: Optional[torch.Tensor] = None,
-        seq_aud: Optional[torch.Tensor] = None,
-        mask_a: Optional[torch.Tensor] = None,
-        mask_b: Optional[torch.Tensor] = None,
-        mask_img: Optional[torch.Tensor] = None,
-        mask_aud: Optional[torch.Tensor] = None,
-    ) -> Dict[str, torch.Tensor]:
-        """Forward pass completo.
-
-        Args:
-            seq_a: [B, T_a, d_a]
-            seq_b: [B, T_b, d_b]
-            seq_img: [B, T_img, d_img] ou None
-            seq_aud: [B, T_aud, d_aud] ou None
-            masks: máscaras de padding correspondentes
-
-        Returns:
-            dict com:
-                fused: [B, d_model] representação fundida
-                seq_a_out, seq_b_out, seq_img_out, seq_aud_out: saídas enriquecidas
-                modality_weights: pesos do gate (se ativo)
-        """
-        B = seq_a.size(0)
-        device = seq_a.device
-
-        # Default para modalidades ausentes: zeros
-        if seq_img is None:
-            seq_img = torch.zeros(B, 1, self.config.d_image, device=device)
-            mask_img = torch.ones(B, 1, device=device)
-        if seq_aud is None:
-            seq_aud = torch.zeros(B, 1, self.config.d_audio, device=device)
-            mask_aud = torch.ones(B, 1, device=device)
-
-        # Cross-modal attention (com residual + LayerNorm)
-        # Texto A atende B, img, aud
-        a_to_b = self.attn_a_to_b(seq_a, seq_b, mask_b)
-        a_to_img = self.attn_a_to_img(seq_a, seq_img, mask_img)
-        a_to_aud = self.attn_a_to_aud(seq_a, seq_aud, mask_aud)
-        seq_a_out = self.ln_a(seq_a + a_to_b + a_to_img + a_to_aud)
-
-        # Texto B atende A, img, aud
-        b_to_a = self.attn_b_to_a(seq_b, seq_a, mask_a)
-        b_to_img = self.attn_b_to_img(seq_b, seq_img, mask_img)
-        b_to_aud = self.attn_b_to_aud(seq_b, seq_aud, mask_aud)
-        seq_b_out = self.ln_b(seq_b + b_to_a + b_to_img + b_to_aud)
-
-        # Imagem atende A, B
-        img_to_a = self.attn_img_to_a(seq_img, seq_a, mask_a)
-        img_to_b = self.attn_img_to_b(seq_img, seq_b, mask_b)
-        seq_img_out = self.ln_img(seq_img + img_to_a + img_to_b)
-
-        # Áudio atende A, B
-        aud_to_a = self.attn_aud_to_a(seq_aud, seq_a, mask_a)
-        aud_to_b = self.attn_aud_to_b(seq_aud, seq_b, mask_b)
-        seq_aud_out = self.ln_aud(seq_aud + aud_to_a + aud_to_b)
-
-        # Pooling (mean sobre a sequência, com máscara)
-        def pool(seq, mask):
-            if mask is None:
-                return seq.mean(dim=1)
-            m = mask.unsqueeze(-1).float()
-            return (seq * m).sum(dim=1) / m.sum(dim=1).clamp(min=1.0)
-
-        pooled_a = pool(seq_a_out, mask_a)        # [B, d_a]
-        pooled_b = pool(seq_b_out, mask_b)        # [B, d_b]
-        pooled_img = pool(seq_img_out, mask_img)  # [B, d_img]
-        pooled_aud = pool(seq_aud_out, mask_aud)  # [B, d_aud]
-
-        # Modality gate (opcional)
-        if self.gate is not None:
-            gate_out = self.gate(pooled_a, pooled_b, pooled_img, pooled_aud)
-            alpha_a = gate_out["alpha_a"]
-            alpha_b = gate_out["alpha_b"]
-            alpha_img = gate_out["alpha_img"]
-            alpha_aud = gate_out["alpha_aud"]
-            modality_weights = gate_out["weights"]
-        else:
-            alpha_a = alpha_b = alpha_img = alpha_aud = 1.0
-            modality_weights = None
-
-        # Fusão final
-        # Concat + projeção (com gate aplicado)
-        if self.gate is not None:
-            pooled_a_g = pooled_a * alpha_a
-            pooled_b_g = pooled_b * alpha_b
-            pooled_img_g = pooled_img * alpha_img
-            pooled_aud_g = pooled_aud * alpha_aud
-        else:
-            pooled_a_g = pooled_a
-            pooled_b_g = pooled_b
-            pooled_img_g = pooled_img
-            pooled_aud_g = pooled_aud
-
-        concat = torch.cat([pooled_a_g, pooled_b_g, pooled_img_g, pooled_aud_g], dim=-1)
-        fused = self.fusion_proj(concat)
-
-        return {
-            "fused": fused,
-            "seq_a_out": seq_a_out,
-            "seq_b_out": seq_b_out,
-            "seq_img_out": seq_img_out,
-            "seq_aud_out": seq_aud_out,
-            "pooled_a": pooled_a,
-            "pooled_b": pooled_b,
-            "pooled_img": pooled_img,
-            "pooled_aud": pooled_aud,
-            "modality_weights": modality_weights,
-        }
-
-
-# ============================================================================
-# Self-test
-# ============================================================================
-
-def _self_test():
-    """Teste rápido do módulo multimodal attention."""
-    torch.manual_seed(42)
-    config = MultimodalAttentionConfig(
-        d_text_a=32, d_text_b=32, d_image=16, d_audio=16,
-        d_model=64, n_heads=4, dropout=0.1,
-        use_modality_gate=True,
-    )
-    mha = MultimodalMultiHeadAttention(config)
-    print(f"Params: {sum(p.numel() for p in mha.parameters())}")
-
-    B = 2
-    seq_a = torch.randn(B, 8, 32)
-    seq_b = torch.randn(B, 6, 32)
-    seq_img = torch.randn(B, 4, 16)
-    seq_aud = torch.randn(B, 5, 16)
-    mask_a = torch.ones(B, 8)
-    mask_b = torch.ones(B, 6)
-    mask_img = torch.ones(B, 4)
-    mask_aud = torch.ones(B, 5)
-
-    out = mha(seq_a, seq_b, seq_img, seq_aud, mask_a, mask_b, mask_img, mask_aud)
-    print(f"Fused: {out['fused'].shape}")
-    print(f"seq_a_out: {out['seq_a_out'].shape}")
-    print(f"Modality weights: {out['modality_weights']}")
-
-    # Test sem modalidades opcionais
-    out2 = mha(seq_a, seq_b, None, None, mask_a, mask_b)
-    print(f"Fused (text only): {out2['fused'].shape}")
-
-
-if __name__ == "__main__":
-    _self_test()
-
-
-__all__ = [
-    "MultimodalAttentionConfig",
-    "CrossModalAttention",
-    "ModalityGate",
-    "MultimodalMultiHeadAttention",
-]