BPACO对比学习：从零开始掌握对比学习核心原理与实践

1次阅读

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

在机器学习领域，获取带标签数据往往成本高昂。对比学习作为无监督学习的重要分支，通过让模型学习 ” 哪些样本相似 / 不相似 ” 来代替直接学习标签，显著降低对标注数据的依赖。传统无监督方法（如 PCA、K-means）难以捕捉复杂特征关系，而对比学习通过构建正负样本对，能更好地学习数据的内在表示。

BPACO（Bidirectional Positive-Aware Contrastive Learning）通过两个关键改进提升对比学习效果：

双向正样本关注机制：不仅考虑锚点与正样本的关系，还显式建模正样本间的共性特征
渐进式负样本筛选：训练过程中动态调整负样本难度，避免简单负样本主导学习过程

以图像数据为例，标准构建流程：

对每个样本（锚点）应用两种不同数据增强（如裁剪 + 颜色抖动）得到正样本对
同一 batch 内其他样本的自然增强版本作为负样本
引入 memory bank 保留历史样本特征扩展负样本库

特殊处理：

对困难正样本（增强后差异大的样本）给予更高权重
对模糊负样本（可能与锚点相似的样本）进行二次验证

BPACO 的损失函数由三部分组成：

def bpaco_loss(anchor, positive, negatives, temperature=0.1):
    # 计算锚点 - 正样本相似度
    pos_sim = torch.cosine_similarity(anchor, positive, dim=-1)  

    # 计算锚点 - 负样本相似度
    neg_sim = torch.cosine_similarity(anchor.unsqueeze(1), negatives, dim=-1)

    # 双向正样本约束
    reverse_pos = torch.cosine_similarity(positive, anchor, dim=-1)

    # 组合损失项
    numerator = torch.exp(pos_sim / temperature)
    denominator = numerator + torch.exp(neg_sim / temperature).sum()
    loss = -torch.log(numerator / denominator) - 0.3 * reverse_pos.mean()
    return loss

import torchvision.transforms as transforms

# 定义对比学习特有的双重增强
class DualAugment:
    def __init__(self):
        self.aug1 = transforms.Compose([transforms.RandomResizedCrop(224),
            transforms.RandomHorizontalFlip(),
            transforms.ColorJitter(0.4, 0.4, 0.4, 0.1),
            transforms.ToTensor()])

        self.aug2 = transforms.Compose([transforms.RandomResizedCrop(224),
            transforms.RandomGrayscale(p=0.2),
            transforms.GaussianBlur(3),
            transforms.ToTensor()])

    def __call__(self, x):
        return self.aug1(x), self.aug2(x)

import torch.nn as nn

class ProjectionHead(nn.Module):
    def __init__(self, input_dim=2048, hidden_dim=512, output_dim=128):
        super().__init__()
        self.layers = nn.Sequential(nn.Linear(input_dim, hidden_dim),
            nn.BatchNorm1d(hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, output_dim)
        )

    def forward(self, x):
        return self.layers(x)

# 以 ResNet50 为例的完整模型
class BPACO_Model(nn.Module):
    def __init__(self, backbone):
        super().__init__()
        self.backbone = backbone
        self.projector = ProjectionHead()

    def forward(self, x):
        features = self.backbone(x)
        projections = self.projector(features)
        return nn.functional.normalize(projections, dim=1)

from torch.utils.data import DataLoader

dataset = YourDataset(transform=DualAugment())
train_loader = DataLoader(dataset, batch_size=64, shuffle=True)

model = BPACO_Model(backbone=resnet50())
optimizer = torch.optim.AdamW(model.parameters(), lr=3e-4)

for epoch in range(100):
    for batch in train_loader:
        x1, x2 = batch  # 两种增强版本

        # 获取特征向量
        z1 = model(x1)
        z2 = model(x2)

        # 计算损失 - 对称计算
        loss = 0.5 * (bpaco_loss(z1, z2, negatives=z2) + 
                      bpaco_loss(z2, z1, negatives=z1))

        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

在 CIFAR-10 数据集上的线性评估结果：