2. 域适应中的特征分布对齐

任务说明

本案例面向Sim2Real视觉任务中的经典问题：在合成数据（源域）上训练的模型，直接部署到真实数据（目标域）时性能显著下降。目标是通过最大均值差异（Maximum Mean Discrepancy, MMD）在特征空间中对齐源域与目标域的分布，使模型在目标域上获得更好的分类效果。我们将构建一个包含特征提取器、分类器和MMD损失的最小可复现训练框架，并在两个分布偏移的合成数据集上验证对齐效果。

方法原理与实现

MMD是一种基于核函数的分布距离度量。给定源域特征 X_s 和目标域特征 X_t，其经验估计为：

MMD² = E[k(x_s, x_s')] - 2E[k(x_s, x_t)] + E[k(x_t, x_t')]

其中 k 为高斯核 k(x,y) = exp(-||x-y||²/(2σ²))。我们使用多核MMD（多个带宽取平均）以增强表达能力。训练时总损失为 L = L_cls + λ·MMD²，其中 L_cls 仅在源域有标签数据上计算。

为在有限篇幅内完整验证域适应效果，我们构造两个可复现的合成视觉特征数据集：源域特征来自以 [2,2] 为中心的高斯分布（类别0）和 [-2,-2] 为中心的高斯分布（类别1）；目标域特征由源域特征施加线性变换 x_t = R·x_s + t 得到，模拟光照/视角变化导致的协变量偏移。模型为两层全连接网络，在源域训练后直接测试目标域（无适应基线），再使用MMD联合训练后测试（有适应）。

代码实现


import torch
import torch.nn as nn
import numpy as np
from torch.utils.data import TensorDataset, DataLoader

# ---------- 1. 数据生成（可复现） ----------
def generate_data(seed=42):
    torch.manual_seed(seed)
    np.random.seed(seed)
    n_per_class = 500
    # 源域：两个类别的高斯簇
    X_s0 = torch.randn(n_per_class, 2) * 0.5 + torch.tensor([2.0, 2.0])
    X_s1 = torch.randn(n_per_class, 2) * 0.5 + torch.tensor([-2.0, -2.0])
    X_s = torch.cat([X_s0, X_s1], dim=0)
    y_s = torch.cat([torch.zeros(n_per_class), torch.ones(n_per_class)]).long()
    # 目标域：对源域做线性变换（旋转+平移），模拟域偏移
    theta = np.radians(30)
    R = torch.tensor([[np.cos(theta), -np.sin(theta)],
                      [np.sin(theta), np.cos(theta)]], dtype=torch.float32)
    t = torch.tensor([1.0, -0.5])
    X_t = X_s @ R.T + t
    # 目标域标签仅用于评估，不参与训练
    y_t = y_s.clone()
    return X_s, y_s, X_t, y_t

# ---------- 2. 模型定义 ----------
class FeatureExtractor(nn.Module):
    def __init__(self, in_dim=2, feat_dim=8):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(in_dim, 16),
            nn.ReLU(),
            nn.Linear(16, feat_dim)
        )
    def forward(self, x):
        return self.net(x)

class Classifier(nn.Module):
    def __init__(self, feat_dim=8, n_class=2):
        super().__init__()
        self.fc = nn.Linear(feat_dim, n_class)
    def forward(self, feat):
        return self.fc(feat)

# ---------- 3. MMD损失（多核高斯） ----------
def compute_mmd(X_s, X_t, bandwidths=[0.1, 1.0, 10.0]):
    """计算多核MMD²，输入为特征矩阵 (N, D)"""
    def rbf_kernel(A, B, sigma):
        # A: (N1, D), B: (N2, D) -> (N1, N2)
        dist2 = torch.cdist(A, B) ** 2
        return torch.exp(-dist2 / (2 * sigma ** 2))
    mmd = 0.0
    for bw in bandwidths:
        K_ss = rbf_kernel(X_s, X_s, bw)
        K_tt = rbf_kernel(X_t, X_t, bw)
        K_st = rbf_kernel(X_s, X_t, bw)
        n_s = X_s.size(0)
        n_t = X_t.size(0)
        # 无偏估计
        term1 = K_ss.sum() / (n_s * (n_s - 1))
        term2 = K_tt.sum() / (n_t * (n_t - 1))
        term3 = 2 * K_st.sum() / (n_s * n_t)
        mmd += (term1 + term2 - term3)
    return mmd / len(bandwidths)

# ---------- 4. 训练与评估 ----------
def train_eval(use_mmd=True, epochs=100, lambda_mmd=1.0, seed=42):
    X_s, y_s, X_t, y_t = generate_data(seed)
    # 划分源域训练/验证（目标域全部用于测试）
    n_train = int(0.8 * X_s.size(0))
    perm = torch.randperm(X_s.size(0))
    train_idx, val_idx = perm[:n_train], perm[n_train:]
    train_loader = DataLoader(TensorDataset(X_s[train_idx], y_s[train_idx]),
                              batch_size=64, shuffle=True)
    val_s = (X_s[val_idx], y_s[val_idx])
    test_t = (X_t, y_t)

    feat_net = FeatureExtractor()
    clf = Classifier()
    params = list(feat_net.parameters()) + list(clf.parameters())
    opt = torch.optim.Adam(params, lr=1e-3)
    loss_cls = nn.CrossEntropyLoss()

    for epoch in range(epochs):
        feat_net.train(); clf.train()
        for xb, yb in train_loader:
            opt.zero_grad()
            feat_s = feat_net(xb)
            out_s = clf(feat_s)
            L_cls = loss_cls(out_s, yb)
            if use_mmd:
                # 从目标域采样与当前batch相同数量的样本
                idx_t = torch.randint(0, X_t.size(0), (xb.size(0),))
                feat_t = feat_net(X_t[idx_t])
                L_mmd = compute_mmd(feat_s, feat_t)
                loss = L_cls + lambda_mmd * L_mmd
            else:
                loss = L_cls
            loss.backward()
            opt.step()

    # 评估
    feat_net.eval(); clf.eval()
    def accuracy(X, y):
        with torch.no_grad():
            feat = feat_net(X)
            pred = clf(feat).argmax(dim=1)
            return (pred == y).float().mean().item()
    acc_s_val = accuracy(val_s[0], val_s[1])
    acc_t = accuracy(test_t[0], test_t[1])
    return acc_s_val, acc_t

# ---------- 5. 运行对比 ----------
if __name__ == "__main__":
    print("=== 无域适应（仅源域训练） ===")
    acc_s_no, acc_t_no = train_eval(use_mmd=False, seed=42)
    print(f"源域验证准确率: {acc_s_no:.4f}")
    print(f"目标域测试准确率: {acc_t_no:.4f}")

    print("\n=== 有MMD域适应 ===")
    acc_s_mmd, acc_t_mmd = train_eval(use_mmd=True, seed=42)
    print(f"源域验证准确率: {acc_s_mmd:.4f}")
    print(f"目标域测试准确率: {acc_t_mmd:.4f}")

运行与验证

依赖：Python 3.8+，PyTorch 1.9+，NumPy。无需额外数据集，代码自动生成合成数据。

运行方式：将上述代码保存为 mmd_da.py，在终端执行 python mmd_da.py。

判定标准：观察输出的两组准确率。若MMD域适应有效，则“有MMD”的目标域测试准确率应明显高于“无MMD”的目标域准确率（例如提升超过5个百分点），同时源域验证准确率保持基本稳定（不显著下降）。由于数据生成和训练过程固定随机种子，结果可复现。注意：由于目标域由源域线性变换而来，特征分布存在系统性偏移，MMD对齐应能有效缓解该偏移。

MMD域适应训练与验证流程 数据生成 源域: 高斯簇 (有标签) 目标域: 线性变换 (无标签) 特征提取器 两层全连接网络 输出8维特征 源域特征 f(X_s) 用于分类损失 目标域特征 f(X_t) 仅用于MMD计算 分类器 交叉熵损失 L_cls MMD² 分布对齐 多核高斯核 (σ=0.1,1,10) 总损失 L L_cls + λ·MMD² 反向传播更新 特征提取器 + 分类器 验证 目标域测试 准确率对比 有/无MMD 图例 数据/特征 网络模块 MMD损失

预期输出示例（基于固定随机种子，逻辑上可复现）：无MMD时目标域准确率约0.75～0.85，有MMD时目标域准确率提升至0.90以上，源域验证准确率保持在0.95左右。实际数值以运行输出为准，重点观察相对提升趋势。

← 上一章 📖 返回目录 下一章 →