每日基础课

一句话定义：多模态融合就是把不同感官的数据（比如听到的声音和看到的嘴型）组合起来做决策；在哪一步组合，决定了它是早期、晚期还是中间融合。

口诀：“前边拼输入，后边拼打分，中间拼特征。”

---

### 1. 生活类比：法官审案 假设法官（模型）要判断嫌疑人是否在撒谎（分类任务），手头有语音录音（音频）和面部表情（视频）两份证据。

- 早期融合（输入层融合）： 法官在审案前，直接把两份材料装订在同一本文件夹里（拼在一起），然后从头读到尾一次性判决。 - 晚期融合（决策层融合）： 法官找了两位助理。助理A只听录音判一个嫌疑度（80%撒谎），助理B只看录像判一个嫌疑度（60%撒谎）。法官最后把两个分数取个平均（70%撒谎）。 - 中间融合（特征层融合）： 助理A提炼出声音的语气特点（声音特征），助理B提炼出微表情特点（画面特征）。法官在开庭讨论中，让两人互相印证——“他说这句话时，嘴角有没有抽搐？”（交叉注意力/深度交互），最后综合得出结论。

---

### 2. 核心公式与符号解析

#### ① 早期融合 (Early Fusion) 把原始输入或底层特征直接拼接，送入单一网络： $$y = g([x_a, x_v])$$ - $x_a$：音频原始特征（如梅尔频谱图，维度 $[B, D_a]$）。 - $x_v$：视频原始特征（如唇部图像序列，维度 $[B, D_v]$）。 - $[\cdot, \cdot]$：拼接操作（Concatenation），拼成 $[B, D_a + D_v]$。 - $g(\cdot)$：主干神经网络。 - $y$：最终输出预测。

#### ② 晚期融合 (Late Fusion) 各模态独立走完全程，最后融合预测结果（如加权平均）： $$y = w_a \cdot f_a(x_a) + w_v \cdot f_v(x_v)$$ - $f_a(\cdot), f_v(\cdot)$：分别为音频和视频的独立分类网络。 - $f_a(x_a), f_v(x_v)$：两个模态各自输出的概率分布或得分（Logits）。 - $w_a, w_v$：可学习或手动设定的模态权重（通常满足 $w_a + w_v = 1$）。 - $y$：加权后的最终决策。

#### ③ 中间融合 (Intermediate Fusion) 各模态先提取中层特征，再通过交互模块融合，最后送入决策头： $$h_a = E_a(x_a), \quad h_v = E_v(x_v)$$ $$h_{fuse} = \text{FusionModule}(h_a, h_v)$$ $$y = \text{Classifier}(h_{fuse})$$ - $E_a, E_v$：音频和视频的独立特征编码器（Encoder）。 - $h_a, h_v$：编码后的中层特征向量（Embedding）。 - $\text{FusionModule}$：交互网络（最常见的是 Cross-Attention 交叉注意力或门控网络）。 - $y$：最终预测结果。

---

### 3. 它解决什么问题？没有它会怎样？ - 解决的问题：单模态信息往往存在盲区。嘈杂环境下声音听不清，但能看清嘴型；遮挡环境下看不清脸，但声音很清楚。融合能让多模态信息互补。 - 没有融合会怎样： - 如果只有单模态：在说话人日志（Diarization）中，重叠说话、噪声干扰时单靠声纹极易切分错误。 - 如果融合方式选错：在音视频采样率不一致（如音频 16kHz vs 视频 25fps）时强行做早期拼接，模型会很难对齐时序，直接训练崩溃。

### 实现思路 1. EarlyFusion：将 `Dim_Audio` 和 `Dim_Video` 在特征维度 `dim=-1` 直接 `torch.cat`，送入一个 MLP。 2. LateFusion：音频和视频各走一个独立的 MLP 输出 Logits（类别打分），最后通过加权和相加。 3. IntermediateFusion：音频和视频先通过 Linear 映射到相同的隐藏维度 $D_{mid}$，将音频作为 Query、视频作为 Key/Value 进行注意力融合，最后通过分类头输出。

---

### 完整可运行代码

import torch
import torch.nn as nn
import torch.nn.functional as F

# ==========================================
# 1. 早期融合 (Early Fusion)
# ==========================================
class EarlyFusionNet(nn.Module):
    def __init__(self, audio_dim: int, video_dim: int, num_classes: int = 2):
        super().__init__()
        # 直接把两个模态拼在一起，要求输入在同一时间步对齐
        in_dim = audio_dim + video_dim
        self.classifier = nn.Sequential(
            nn.Linear(in_dim, 128),
            nn.ReLU(),
            nn.Linear(128, num_classes)
        )

    def forward(self, audio_feat: torch.Tensor, video_feat: torch.Tensor) -> torch.Tensor:
        # [B, D_a], [B, D_v] -> [B, D_a + D_v]
        fused = torch.cat([audio_feat, video_feat], dim=-1)
        return self.classifier(fused)


# ==========================================
# 2. 晚期融合 (Late Fusion)
# ==========================================
class LateFusionNet(nn.Module):
    def __init__(self, audio_dim: int, video_dim: int, num_classes: int = 2):
        super().__init__()
        # 两个独立的模态网络，分别输出最终判决 Logits
        self.audio_net = nn.Sequential(
            nn.Linear(audio_dim, 64),
            nn.ReLU(),
            nn.Linear(64, num_classes)
        )
        self.video_net = nn.Sequential(
            nn.Linear(video_dim, 64),
            nn.ReLU(),
            nn.Linear(64, num_classes)
        )
        # 可学习的融合权重系数
        self.weight_audio = nn.Parameter(torch.tensor(0.5))
        self.weight_video = nn.Parameter(torch.tensor(0.5))

    def forward(self, audio_feat: torch.Tensor, video_feat: torch.Tensor) -> torch.Tensor:
        logit_a = self.audio_net(audio_feat) # [B, num_classes]
        logit_v = self.video_net(video_feat) # [B, num_classes]
        
        # 归一化权重后加权融合决策
        weights = F.softmax(torch.stack([self.weight_audio, self.weight_video]), dim=0)
        return weights[0] * logit_a + weights[1] * logit_v


# ==========================================
# 3. 中间融合 (Intermediate Fusion - 门控交互)
# ==========================================
class IntermediateFusionNet(nn.Module):
    def __init__(self, audio_dim: int, video_dim: int, hidden_dim: int = 64, num_classes: int = 2):
        super().__init__()
        # 阶段 1: 单模态特征投影到统一维度
        self.proj_audio = nn.Linear(audio_dim, hidden_dim)
        self.proj_video = nn.Linear(video_dim, hidden_dim)

        # 阶段 2: 交互与门控融合 (Gated Fusion)
        # 计算音频对视频的门控权重：通过声音决定关注多少画面特征
        self.gate = nn.Sequential(
            nn.Linear(hidden_dim * 2, hidden_dim),
            nn.Sigmoid()
        )
        self.fusion_layer = nn.Linear(hidden_dim * 2, hidden_dim)

        # 阶段 3: 融合后的共享分类头
        self.classifier = nn.Sequential(
            nn.ReLU(),
            nn.Linear(hidden_dim, num_classes)
        )

    def forward(self, audio_feat: torch.Tensor, video_feat: torch.Tensor) -> torch.Tensor:
        h_a = F.relu(self.proj_audio(audio_feat)) # [B, hidden_dim]
        h_v = F.relu(self.proj_video(video_feat)) # [B, hidden_dim]

        # 拼接中层特征
        concat_h = torch.cat([h_a, h_v], dim=-1) # [B, hidden_dim * 2]
        
        # 门控交互：动态调整视觉特征强度
        g = self.gate(concat_h) # [B, hidden_dim]
        h_v_gated = g * h_v

        # 再次融合
        fused_h = self.fusion_layer(torch.cat([h_a, h_v_gated], dim=-1))
        return self.classifier(fused_h)


# ==========================================
# 4. 验证测试
# ==========================================
if __name__ == "__main__":
    torch.manual_seed(42)
    
    # 模拟数据：4个样本，音频维度128，视频维度256
    batch_size = 4
    audio_dim = 128
    video_dim = 256
    num_classes = 2

    dummy_audio = torch.randn(batch_size, audio_dim)
    dummy_video = torch.randn(batch_size, video_dim)

    # 实例化三个模型
    model_early = EarlyFusionNet(audio_dim, video_dim, num_classes)
    model_late = LateFusionNet(audio_dim, video_dim, num_classes)
    model_inter = IntermediateFusionNet(audio_dim, video_dim, hidden_dim=64, num_classes=num_classes)

    # 前向传播
    out_early = model_early(dummy_audio, dummy_video)
    out_late = model_late(dummy_audio, dummy_video)
    out_inter = model_inter(dummy_audio, dummy_video)

    # 验证输出形状
    assert out_early.shape == (batch_size, num_classes), "Early fusion output shape mismatch!"
    assert out_late.shape == (batch_size, num_classes), "Late fusion output shape mismatch!"
    assert out_inter.shape == (batch_size, num_classes), "Intermediate fusion output shape mismatch!"

    print("✅ 全部通过！输出示例：")
    print(f"Early Fusion Logits (Shape {out_early.shape}):\n{out_early.detach()}")
    print(f"Late Fusion Logits (Shape {out_late.shape}):\n{out_late.detach()}")
    print(f"Intermediate Fusion Logits (Shape {out_inter.shape}):\n{out_inter.detach()}")