CFT对比学习原理剖析与实战：如何解决小样本学习中的特征表示难题

1次阅读

共计 2545 个字符，预计需要花费 7 分钟才能阅读完成。

传统小样本学习方法（如 MAML、Prototypical Nets）在跨域任务中常面临两个核心问题：

特征表示不足：当目标域与源域分布差异较大时，预训练特征无法有效迁移。例如在医疗影像分类中，用自然图像预训练的模型常需要大量目标域样本微调
泛化性缺陷：标准交叉熵损失会迫使模型过度拟合少量样本，导致在测试集上表现波动剧烈。实验数据显示，在 MiniImageNet 5-way 1-shot 任务中，传统方法的跨域准确率可能骤降 15%-20%

SimCLR：依赖大批量（通常≥4096）构建负样本对，在小样本场景下难以满足

\mathcal{L}_{SimCLR} = -\log\frac{\exp(sim(z_i,z_j)/\tau)}{\sum_{k=1}^{2N}\mathbb{1}_{k\neq i}\exp(sim(z_i,z_k)/\tau)}

MoCo：虽然通过队列机制降低计算量，但静态字典会导致特征更新滞后

动态特征迁移：在投影头后添加域适配层（Domain Adaptor Layer），其参数更新遵循：
```
\theta_{dal} \leftarrow \alpha\theta_{dal} + (1-\alpha)\frac{1}{m}\sum_{i=1}^m \phi(x_i^t)
```
双向对比损失：同时优化源域和目标域的特征对齐

import torch
import torch.nn as nn
from typing import Tuple

class CFTProjectionHead(nn.Module):
    """特征投影头（含 LayerNorm 防止特征坍缩）"""
    def __init__(self, input_dim: int = 2048, hidden_dim: int = 512, output_dim: int = 128):
        super().__init__()
        self.layers = nn.Sequential(nn.Linear(input_dim, hidden_dim, bias=False),
            nn.LayerNorm(hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, output_dim, bias=False)
        )

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return self.layers(x)

class CFTLoss(nn.Module):
    """改进的 InfoNCE 损失（支持跨域对比）"""
    def __init__(self, temperature: float = 0.1):
        super().__init__()
        self.temp = temperature
        self.cross_entropy = nn.CrossEntropyLoss()

    def forward(self, 
                z_src: torch.Tensor, 
                z_tgt: torch.Tensor, 
                neg_queue: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor]:
        # 计算正样本相似度
        pos_sim = torch.einsum('i d, i d -> i', z_src, z_tgt).unsqueeze(-1) / self.temp
        # 计算负样本相似度
        neg_sim = torch.einsum('i d, k d -> i k', z_src, neg_queue) / self.temp
        # 组合 logits
        logits = torch.cat([pos_sim, neg_sim], dim=1)
        labels = torch.zeros(z_src.size(0), dtype=torch.long).to(z_src.device)
        return self.cross_entropy(logits, labels)

梯度解耦 ：使用neg_queue.detach() 避免反向传播更新队列

动量更新：采用 MoCo-style 的队列更新策略，保持特征一致性

@torch.no_grad()
def update_queue(self, new_keys: torch.Tensor):
    self.queue = torch.cat([new_keys, self.queue[:-new_keys.size(0)]], dim=0)

奇异值分析：定期计算特征矩阵的 SVD，若最大奇异值占比 >90% 则可能出现坍缩

def check_collapse(features: torch.Tensor, threshold: float = 0.9) -> bool:
    _, s, _ = torch.svd(features.float())
    return (s[0] / s.sum()).item() > threshold