lnp_ml/lnp_ml/modeling/layers/fusion.py

import torch
import torch.nn as nn
import torch.nn.functional as F
from typing import Dict, List, Literal, Optional, Tuple, Union


PoolingStrategy = Literal["concat", "avg", "max", "attention"]


class FusionLayer(nn.Module):
    """
    将多个 token 融合成单个向量。

    输入: Dict[str, Tensor] 或 [B, n_tokens, d_model]
    输出: [B, fusion_dim]

    策略:
        - concat: [B, n_tokens, d_model] -> [B, n_tokens * d_model]
        - avg: [B, n_tokens, d_model] -> [B, d_model]
        - max: [B, n_tokens, d_model] -> [B, d_model]
        - attention: [B, n_tokens, d_model] -> [B, d_model] (learnable attention pooling)
    """

    def __init__(
        self,
        d_model: int,
        n_tokens: int,
        strategy: PoolingStrategy = "attention",
    ) -> None:
        """
        Args:
            d_model: 每个 token 的维度
            n_tokens: token 数量（如 8）
            strategy: 融合策略
        """
        super().__init__()
        self.d_model = d_model
        self.n_tokens = n_tokens
        self.strategy = strategy

        if strategy == "concat":
            self.fusion_dim = n_tokens * d_model
        else:
            self.fusion_dim = d_model

        # Attention pooling: learnable query
        if strategy == "attention":
            self.attn_query = nn.Parameter(torch.randn(1, 1, d_model))
            self.attn_proj = nn.Linear(d_model, d_model)

    def forward(
        self,
        x: Union[Dict[str, torch.Tensor], torch.Tensor],
        return_attn_weights: bool = False,
    ) -> Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]]:
        """
        Args:
            x: Dict[str, Tensor] 每个 [B, d_model]，或已 stack 的 [B, n_tokens, d_model]
            return_attn_weights: 若为 True 且策略为 attention，额外返回 attn_weights [B, n_tokens]

        Returns:
            return_attn_weights=False: [B, fusion_dim]
            return_attn_weights=True:  ([B, fusion_dim], [B, n_tokens])
        """
        if isinstance(x, dict):
            x = torch.stack(list(x.values()), dim=1)

        if self.strategy == "concat":
            out = x.flatten(start_dim=1)
            return (out, None) if return_attn_weights else out

        elif self.strategy == "avg":
            out = x.mean(dim=1)
            return (out, None) if return_attn_weights else out

        elif self.strategy == "max":
            out = x.max(dim=1).values
            return (out, None) if return_attn_weights else out

        elif self.strategy == "attention":
            return self._attention_pooling(x, return_attn_weights)

        else:
            raise ValueError(f"Unknown strategy: {self.strategy}")

    def _attention_pooling(
        self, x: torch.Tensor, return_attn_weights: bool = False,
    ) -> Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]]:
        """
        Attention pooling: 用可学习 query 对 tokens 做加权求和

        Args:
            x: [B, n_tokens, d_model]
            return_attn_weights: 是否返回权重

        Returns:
            return_attn_weights=False: [B, d_model]
            return_attn_weights=True:  ([B, d_model], [B, n_tokens])
        """
        B = x.size(0)
        query = self.attn_query.expand(B, -1, -1)

        keys = self.attn_proj(x)
        scores = torch.bmm(query, keys.transpose(1, 2)) / (self.d_model ** 0.5)
        attn_weights = F.softmax(scores, dim=-1)  # [B, 1, n_tokens]

        out = torch.bmm(attn_weights, x).squeeze(1)  # [B, d_model]

        if return_attn_weights:
            return out, attn_weights.squeeze(1)  # [B, n_tokens]
        return out


class ResidualConcatFusion(nn.Module):
    """对真实 token 做 attention pooling，再用零初始化门把 MoE/LLM 旁路以残差方式加入。

    门为逐维向量（初始全零）：
      - 起点仍严格等于 baseline，"加了不会更差"的保证不变；
      - 标量门只能整体调音量，最优值是"有用维度的收益"与"噪声维度的损害"之间的妥协；
      - 标量门的梯度 ∂L/∂g = Σ_i u_i f_i 会自我抵消，逐维门 ∂L/∂g_i = u_i f_i 不会。
    """

    def __init__(self, d_model: int, strategy: PoolingStrategy = "attention") -> None:
        super().__init__()
        if strategy == "concat":
            raise ValueError("ResidualConcatFusion 不支持 concat（token 数随开关变化）")
        self.d_model = d_model
        self.pool = FusionLayer(d_model=d_model, n_tokens=1, strategy=strategy)
        self.fusion_dim = self.pool.fusion_dim
        # 零初始化逐维门控，旁路初始不参与
        self.g_moe = nn.Parameter(torch.zeros(d_model))
        self.g_llm = nn.Parameter(torch.zeros(d_model))
        self.g_retr = nn.Parameter(torch.zeros(d_model))

    def gate_norms(self) -> Dict[str, float]:
        """诊断用。标量门时代的 |g| 对应这里的 norm / sqrt(d_model)。"""
        return {
            "g_moe": float(self.g_moe.norm()),
            "g_llm": float(self.g_llm.norm()),
            "g_retr": float(self.g_retr.norm()),
        }

    def forward(
        self,
        chem: torch.Tensor,
        tab: torch.Tensor,
        f_moe: Optional[torch.Tensor] = None,
        f_llm: Optional[torch.Tensor] = None,
        f_retr: Optional[torch.Tensor] = None,
        return_attn_weights: bool = False,
    ) -> Union[torch.Tensor, Tuple[torch.Tensor, torch.Tensor]]:
        # 只对真实 token（chem + tab）做注意力池化，旁路不参与 softmax 竞争
        seq = torch.cat([chem, tab], dim=1)  # [B, n_chem + n_cond, d_model]
        pooled = self.pool(seq, return_attn_weights=return_attn_weights)
        if return_attn_weights:
            pooled, attn = pooled

        # [d_model] 与 [B, d_model] 自动广播
        out = pooled
        if f_moe is not None:
            out = out + self.g_moe * f_moe
        if f_llm is not None:
            out = out + self.g_llm * f_llm
        if f_retr is not None:
            out = out + self.g_retr * f_retr

        # 回归旁路：额外返回 pooled（纯 chem+tab，不含 f_llm/f_moe/f_retr）
        if return_attn_weights:
            return out, pooled, attn
        return out, pooled