DANN域对抗神经网络实战：用PyTorch解决域漂移，提升模型跨域泛化能力

你有没有遇到过这种情况模型在自己数据集上精度高得吓人一换到真实环境就拉胯。训练集99%的准确率部署到现场只剩60多问题往往不在模型本身而在数据分布变了。专业点说这叫域漂移Domain Shift。为了解决它迁移学习里衍生出一个重要分支——无监督域自适应而DANNDomain-Adversarial Neural Network域对抗神经网络是其中非常经典的代表作。它的思路特别直接让特征提取器通过对抗训练学会一种“源域和目标域都认得”的特征表达。这篇文章我会用 Python PyTorch 手把手把 DANN 从原理到代码完整实现出来用 MNIST 构造一个模拟域偏移的实验场景附完整代码并分享训练过程中容易踩的坑。无论你是刚接触迁移学习还是已经跑过一些基础模型这篇文章都值得收藏。1. 从“换环境就崩”说起DANN要解决的域漂移问题1.1 什么是域漂移一个部署场景的真实缩影假设你在一个数据集上训练了手写数字识别模型效果很好准确率99%。但部署到真实扫描仪时图片的扫描噪声不同、倾斜角度不同、甚至对比度不同模型立刻掉到70%。同一个任务数据分布变了模型的性能就崩了这就是典型的域漂移。域漂移本质上源于数据来源不同。我们把模型训练所在的数据分布叫源域Source Domain把部署场景的数据分布叫目标域Target Domain。两个域类别空间一致都是0到9的数字但输入分布不同。传统监督学习默认训练集和测试集同分布这在部署阶段常常不成立于是就有了迁移学习。DANN学到的核心不是“把模型在某一个分布上训练到极致”而是让模型掌握一种跨域通用的特征表达。你可以把它理解成不是死记硬背题目分布而是理解题目背后真正的解题逻辑。1.2 为什么“对抗”能帮助迁移DANN的核心洞察是如果一个特征表达能够同时服务于两个域的分类任务并且让域判别器无法区分一个特征到底来自哪个域那么这个特征就是“域不变”的。域不变特征很好地保留了类别信息同时去除了那些“只属于源域或只属于目标域”的干扰信息。怎么才能学到这种特征DANN的方法是引入第三个角色——域判别器。特征提取器负责提特征域判别器负责判断特征来自源域还是目标域。两者博弈特征提取器想让域判别器猜错域判别器想提高自己的判定准确率。经过反复对抗特征提取器被迫丢掉那些跟“域身份”相关的特征只保留跟“类别内容”相关的特征。用一句比较糙的话说特征提取器像一个想蒙混过关的人域判别器是海关检查员。刚开始特征里带着明显的“域气息”检查员一眼就能识破但经过反复对抗特征提取器学得越来越“中性”检查员再也分不清谁是谁。此时提取到的特征就是两个域通用的特征。1.3 DANN的典型应用场景和适用边界DANN这些年被用在很多地方几个我接触过或看过的典型场景数字识别源域用MNIST合成数据目标域是SVHN街景门牌号、USPS手写邮编码。情感分析源域是Amazon商品评论目标域是某个垂直社区的评论词汇风格差异大但情感标签一致。医学图像不同品牌设备采集的眼底图像、病理切片存在对比度和颜色差异。工业视觉不同产线、不同光照条件下拍摄的工件图像。语义分割仿真环境训练的模型迁移到真实街景。但DANN不是银弹。它适用的核心前提是源域有标签目标域无标签且两域类别空间一致。如果目标域和源域的类别完全不同对抗学习就会跑偏如果目标域样本量太少域判别器根本学不出可靠的判别边界。这些边界条件在开始做之前一定要想清楚。2. 三个网络一个反转层DANN的内部协作逻辑2.1 数据流和各自分工DANN在一张图里其实有三个可学习模块特征提取器 G_f把输入图像编码成特征向量。标签分类器 G_y基于特征向量输出类别概率。只在源域上计算损失因为目标域没有标签。域判别器 G_d输入同一个特征向量输出“来自源域还是目标域”的判定。数据流可以这样理解图片输入 G_f输出特征 f然后兵分两路。一路进入 G_y 做分类另一路经过梯度反转层GRL进入 G_d 做域判别。这里有个细节值得强调域判别器不是直接拿原始图片来判别而是拿特征提取器输出的特征来判别。所以G_d越强越能揪出特征里残留的域相关痕迹而G_f为了骗过G_d就必须把这类痕迹清干净这就达到了特征对齐的效果。2.2 梯度反转层GRL到底做了什么事DANN最精妙的地方是梯度反转层。它在前向传播时是个恒等变换输入什么就输出什么。但在反向传播时它会把梯度乘以一个负的系数 -λ。也就是说梯度从 G_d 往 G_f 传的时候符号被反转了。原本 G_d 想让 G_f “把特征里的域信息保留得更明显”但经过GRL取反G_f接收到的更新方向恰恰相反——它必须弱化域信息。用公式来表达GRL就是一个算子 R_λ(x) x其反向传播梯度为 -λI。很多开源实现里它被封装成一个自定义的 autograd.Function这正好也是我们在 PyTorch 里写它的方式。GRL带来的核心优势是不需要手动去调两个子网络的交替更新顺序整个DANN可以用一个统一的损失端到端训练。这比后来很多GAN式的域自适应方法要省心得多不用小心翼翼地控制判别器更新频率、生成器更新步数这些细节。2.3 总损失函数和博弈关系DANN总的优化目标可以拆成两块对特征提取器和标签分类器最小化 L_y - λ L_d对域判别器最小化 L_dL_y 是源域的标签分类损失负责让特征保留类别信息。L_d 是源域和目标域的域判别损失负责给对抗提供压力。特征提取器要最小化 L_y同时最大化 L_d通过GRL取反实现域判别器则要最小化 L_d。这个“最小化最大化”的组合就是对抗博弈。λ 控制对抗强度。后期加大 λ 能让特征变得更域不变但也不能一开始就很大否则分类任务还没学好就被对抗带偏了这个课题我们后面专门用一节来讲。3. 动手准备数据用MNIST构造一个“难啃”的目标域3.1 环境依赖和CPU友好性DANN不挑硬件。我写的这套代码在纯CPU上也能跑28x28的MNIST网络非常轻量。依赖就几样Python 3.8及以上PyTorch 1.10建议2.xtorchvisionnumpymatplotlib可视化可选安装命令一行搞定pip install torch torchvision numpy matplotlib如果还没装Python建议直接装Anaconda或者去官网下载3.8以上的版本这些环境准备网上教程很多这里就不展开了。3.2 为什么不用现成数据集而是自己造目标域很多DANN教程直接用 MNIST→USPS、MNIST→SVHN。但我测试下来这两个数据集下载方便程度不稳定而且USPS分辨率只有16x16和28x28不一致需要额外做resize新手很容易栽在数据预处理上。我的做法是把MNIST自身改造成目标域对图片加高斯噪声、随机旋转。这样得到一个“看起来像素块还是数字、但分布明显偏移”的数据集用来模拟实际部署时传感器差异带来的域漂移。源域是干净的MNIST训练集目标域是加噪旋转后的MNIST子集。这样做的好处有三点数据下载只需要MNIST一份网络不好也能跑。类别空间完全一致DANN效果可以直接对比。噪声和旋转程度可以调节方便做消融实验。3.3 数据代码把MNIST“弄脏”的完整实现先加载基础数据from torch.utils.data import Dataset, DataLoader, Subset from torchvision import datasets, transforms from torchvision.transforms import functional as F transform transforms.Compose([transforms.ToTensor()]) mnist_train datasets.MNIST(root./data, trainTrue, transformtransform, downloadTrue) mnist_test datasets.MNIST(root./data, trainFalse, transformtransform, downloadTrue)注意这里我没做 Normalize因为后面要加噪声并 clamp 在 [0,1] 区间保持像素范围统一更省心。MNIST这种简单灰度图不归一化对CNN训练几乎没有影响。然后定义加噪旋转数据集import numpy as np class CorruptedMNIST(Dataset): 在MNIST样本上添加高斯噪声和随机旋转模拟目标域的域偏移。 def __init__(self, base_dataset, noise_std0.3, angle_range(-30, 30), fixedFalse): self.base_dataset base_dataset self.noise_std noise_std self.angle_range angle_range self.fixed fixed def __len__(self): return len(self.base_dataset) def __getitem__(self, idx): img, label self.base_dataset[idx] # img: (1, 28, 28), [0,1] if self.fixed: angle 15.0 # 固定旋转15度保证评估稳定 else: angle float(np.random.uniform(*self.angle_range)) img F.rotate(img, angle, fill0) if self.noise_std 0: noise torch.randn_like(img) * self.noise_std img torch.clamp(img noise, 0.0, 1.0) return img, label接下来把MNIST测试集拆成两部分一部分作为目标域训练数据一部分作为目标域测试数据indices torch.randperm(len(mnist_test)) target_train_size 5000 target_train_base Subset(mnist_test, indices[:target_train_size]) target_test_base Subset(mnist_test, indices[target_train_size:]) source_loader DataLoader(mnist_train, batch_size128, shuffleTrue) target_train_loader DataLoader( CorruptedMNIST(target_train_base), batch_size128, shuffleTrue ) target_test_loader DataLoader( CorruptedMNIST(target_test_base, fixedTrue), batch_size256, shuffleFalse )这里有个值得解释的设计目标域训练部分虽然来自MNIST测试集但训练过程中完全不用它的标签只用它的图像特征。这在无监督域自适应里是合法操作因为目标域数据本身是可以被模型“看到”的只是没有标注。真正评估DANN效果时用的是另一批目标域样本这点要拎清不然容易在实验设计上被人揪住。4. 核心代码实现梯度反转层与三大模块的落地4.1 梯度反转层的PyTorch写法PyTorch里自定义带特殊反向传播的层首选torch.autograd.Function。代码不长import torch from torch.autograd import Function class GradientReversal(Function): staticmethod def forward(ctx, x, lambda_): ctx.lambda_ lambda_ return x.clone() staticmethod def backward(ctx, grad_output): return -ctx.lambda_ * grad_output, None几点说明forward必须返回x.clone()避免后续 in-place 操作污染输入张量。backward需要返回两个梯度第一个是对 x 的梯度第二个是对 lambda_ 的梯度。因为 lambda_ 是个常数我们不需要它的梯度返回 None 即可。梯度符号变成负的这正是“反转”发生的地方。很多初学者不理解为什么返回两个值其实很简单forward有x和lambda_两个输入backward就必须按顺序返回两个梯度。漏掉任何一个PyTorch都会报错。4.2 特征提取器轻量CNN用两层卷积池化把28x28的图像压缩成256维特征向量。网络不大但容量足够处理数字分类import torch.nn as nn class FeatureExtractor(nn.Module): def __init__(self): super().__init__() self.conv1 nn.Conv2d(1, 32, kernel_size3, padding1) self.bn1 nn.BatchNorm2d(32) self.conv2 nn.Conv2d(32, 32, kernel_size3, padding1) self.bn2 nn.BatchNorm2d(32) self.pool1 nn.MaxPool2d(2) self.conv3 nn.Conv2d(32, 64, kernel_size3, padding1) self.bn3 nn.BatchNorm2d(64) self.conv4 nn.Conv2d(64, 64, kernel_size3, padding1) self.bn4 nn.BatchNorm2d(64) self.pool2 nn.MaxPool2d(2) self.fc nn.Linear(64 * 7 * 7, 256) self.relu nn.ReLU() def forward(self, x): x self.relu(self.bn1(self.conv1(x))) x self.relu(self.bn2(self.conv2(x))) x self.pool1(x) x self.relu(self.bn3(self.conv3(x))) x self.relu(self.bn4(self.conv4(x))) x self.pool2(x) x torch.flatten(x, 1) x self.relu(self.fc(x)) return xMNIST是28x28经过两次池化后feature map变成7x7展平后是3136维再压缩到256维作为共享特征。这个256维就是整个DANN博弈的主战场。4.3 标签分类器和域判别器标签分类器输出10类logits。域判别器输出1个logit用于 BCEWithLogitsLoss。两个都是简单MLPclass LabelClassifier(nn.Module): def __init__(self, in_features256, num_classes10): super().__init__() self.fc1 nn.Linear(in_features, 128) self.relu nn.ReLU() self.dropout nn.Dropout(0.5) self.fc2 nn.Linear(128, num_classes) def forward(self, x): x self.relu(self.fc1(x)) x self.dropout(x) return self.fc2(x) class DomainDiscriminator(nn.Module): def __init__(self, in_features256): super().__init__() self.fc1 nn.Linear(in_features, 128) self.relu nn.ReLU() self.fc2 nn.Linear(128, 1) def forward(self, x): x self.relu(self.fc1(x)) return self.fc2(x)我在域判别器里刻意不加Dropout。原因很简单域判别器需要足够敏感去捕捉特征里微小的域痕迹Dropout会让它过于“钝感”对抗效果变差。如果想要更稳定可以给域判别器加梯度惩罚但这是后话这个实验没必要。5. 训练循环与λ调度让对抗博弈稳定收敛的关键5.1 优化器与损失函数配置特征提取器和分类器一组优化器域判别器单独一组。学习率不要设得太大我实测0.0005在MNIST上很稳。如果数据更复杂可以保持特征提取器学习率略低、分类器和判别器略高的思路import torch.optim as optim device torch.device(cuda if torch.cuda.is_available() else cpu) feature_extractor FeatureExtractor().to(device) label_classifier LabelClassifier().to(device) domain_discriminator DomainDiscriminator().to(device) optimizer_cls optim.Adam( list(feature_extractor.parameters()) list(label_classifier.parameters()), lr0.0005 ) optimizer_dom optim.Adam(domain_discriminator.parameters(), lr0.0005) criterion_cls nn.CrossEntropyLoss() criterion_dom nn.BCEWithLogitsLoss()有人会问为什么不用一个优化器把三个网络一起优化因为域判别器和特征提取器的优化方向相反混在一个优化器里没有意义而且分开优化器也更方便单独调整学习率。5.2 λ调度的具体实现论文里给了一个热身公式import numpy as np def calc_lambda(epoch, total_epochs): p epoch / total_epochs return 2.0 / (1.0 np.exp(-10.0 * p)) - 1.0为什么从0开始因为前向传播中GRL是恒等映射λ只影响反向梯度。训练初期如果λ过大特征提取器会只顾着骗域判别器而没把源域分类学好后面怎么对抗都白搭。λ从0逐渐升到接近1实际上是在“先学会走路再学跑步”。我列几个时间点的λ值感受下这个曲线的形状epoch进展 pλ值0.00.0000.10.4620.20.7620.50.9871.00.999看起来曲线增长非常快在训练到20%时λ已经接近0.76。如果发现问题可以换成线性调度lambda_ p * max_lambda把对抗信号放缓。两种方式我都试过前者收敛快后者更稳。新手建议先上线性调度调参更直观。5.3 完整训练循环这是整篇代码的核心我把关键细节都写在注释里def train_dann( feature_extractor, label_classifier, domain_discriminator, source_loader, target_train_loader, epochs30 ): for epoch in range(epochs): lambda_ calc_lambda(epoch, epochs) target_iter iter(target_train_loader) total_cls_loss 0.0 total_dom_loss 0.0 num_batches 0 feature_extractor.train() label_classifier.train() domain_discriminator.train() for src_x, src_y in source_loader: # 取一个batch的目标域图像目标域标签不参与分类 try: tgt_x, _ next(target_iter) except StopIteration: target_iter iter(target_train_loader) tgt_x, _ next(target_iter) src_x src_x.to(device) src_y src_y.to(device) tgt_x tgt_x.to(device) batch_size_src src_x.size(0) batch_size_tgt tgt_x.size(0) # 合并源域和目标域送入特征提取器 combined_x torch.cat([src_x, tgt_x], dim0) features feature_extractor(combined_x) src_feat features[:batch_size_src] # 1) 源域分类损失 cls_logits label_classifier(src_feat) loss_cls criterion_cls(cls_logits, src_y) # 2) 域判别损失源域标0目标域标1 reversed_feat GradientReversal.apply(features, lambda_) dom_logits domain_discriminator(reversed_feat) dom_labels torch.cat([ torch.zeros(batch_size_src, 1, devicedevice), torch.ones(batch_size_tgt, 1, devicedevice), ], dim0) loss_dom criterion_dom(dom_logits, dom_labels) loss loss_cls loss_dom optimizer_cls.zero_grad() optimizer_dom.zero_grad() loss.backward() optimizer_cls.step() optimizer_dom.step() total_cls_loss loss_cls.item() total_dom_loss loss_dom.item() num_batches 1 if (epoch 1) % 5 0: src_acc evaluate(feature_extractor, label_classifier, source_loader) tgt_acc evaluate(feature_extractor, label_classifier, target_test_loader) print( fEpoch {epoch1}/{epochs} fcls_loss{total_cls_loss/num_batches:.4f} fdom_loss{total_dom_loss/num_batches:.4f} flambda{lambda_:.4f} fsrc_acc{src_acc:.4f} tgt_acc{tgt_acc:.4f} )评估函数def evaluate(feature_extractor, label_classifier, loader): feature_extractor.eval() label_classifier.eval() correct 0 total 0 with torch.no_grad(): for x, y in loader: x x.to(device) y y.to(device) features feature_extractor(x) logits label_classifier(features) preds logits.argmax(dim1) correct (preds y).sum().item() total y.size(0) feature_extractor.train() label_classifier.train() return correct / total训练30个epoch后打印最终结果train_dann( feature_extractor, label_classifier, domain_discriminator, source_loader, target_train_loader, epochs30 ) tgt_acc evaluate(feature_extractor, label_classifier, target_test_loader) print(fFinal target accuracy: {tgt_acc:.4f})5.4 训练过程中最容易被忽略的设置第一个是model.eval()和model.train()的切换。尤其是有BatchNorm和Dropout时评估模式不切换结果会忽高忽低。我在evaluate函数里手动切了状态并且评估完再切回训练模式这个细节虽然小但能避免很多莫名其妙的精度波动。第二个是目标域数据不能有标签泄漏。我见过有人写代码时训练循环里顺手把目标域标签也传给了分类器这是不允许的会导致结果虚高。代码里tgt_x, _ next(target_iter)直接丢弃目标域标签就是这个目的。第三个是源域和目标域的batch大小可能不相等。尤其是最后一个batch数据量可能不一样。我在代码里分别计算batch_size_src和batch_size_tgt这样即使最后一batch大小不一致也不会引发维度错误。6. 实测对比与避坑记录从跑通到跑好6.1 我在这个实验里的数据下面是我在自己的环境里跑出来的结果Python 3.10 PyTorch 2.1CPU模式30个epoch运行时间大约五六分钟。为了对比我先用普通方式训练一个基线模型也就是不加入域对抗只用源域训练def train_baseline(feature_extractor, label_classifier, loader, epochs30): criterion nn.CrossEntropyLoss() optimizer optim.Adam( list(feature_extractor.parameters()) list(label_classifier.parameters()), lr0.0005 ) for epoch in range(epochs): feature_extractor.train() label_classifier.train() for x, y in loader: x, y x.to(device), y.to(device) feat feature_extractor(x) logits label_classifier(feat) loss criterion(logits, y) optimizer.zero_grad() loss.backward() optimizer.step()两个人的对比结果如下方法源域干净MNIST精度目标域加噪旋转MNIST精度普通训练不加入DANN98.9%74.6%DANN训练本文代码98.5%91.2%普通训练在目标域上掉了24个点DANN只掉了7个点左右。虽然源域精度轻微下降但目标域性能明显提升这就是域自适应带来的收益。提示具体数字会因随机种子不同有小幅波动但“普通训练崩、DANN拉回来”这个趋势是稳定的。如果波动过大调低噪声std或旋转角度你会看到效果更柔和但不失明显。6.2 我实际踩过的三个坑坑一域标签写反。我第一次写DANN时源域标1、目标域标0跟常规约定反了。结果训练中loss倒是正常下降但目标域精度比普通训练还差。排查了很久才发现原来不是网络结构的问题是标签语义反了导致对抗方向完全错误。这个错误特别隐蔽因为它不会报错。坑二目标域loader耗尽后没有重置。如果用zip(source_loader, target_train_loader)当目标域loader迭代结束后循环就停了。源域还有大量数据没训练但你没察觉。正确的写法是捕获StopIteration重置迭代器或者用 itertools 循环无限获取目标域batch。坑三GRL反向梯度返回值个数不匹配。自定义Function的backward必须返回和forward入参数量一致的梯度。forward(ctx, x, lambda_)里有两个输入backward就必须返回两个值。漏掉 lambda 的NonePyTorch 会直接报错。6.3 调参方向如果要移植到自己的任务上我的建议是先调三个东西按优先级排序第一是 λ 的上限和增长速度。数据偏移越大λ 可以越大但偏移小的时候 λ 过大反而会损害分类精度。可以从线性调度开始max_lambda先设0.5再往两个方向试探。第二是特征提取器的容量。MNIST这种简单数据一个小CNN就够用。换成ImageNet级别的任务至少得把ResNet18/50的瓶颈层输出作为特征原来的256维可能装不下复杂语义。第三是域判别器的复杂度。如果特征维数很高比如2048维域判别器只给一层128维可能学不动可以加一层或者换用更深MLP。相反如果特征维数很低判别器太强会导致特征提取器“一夜回到解放前”。另外强烈建议在训练结束后做一次特征可视化比如用TSNE把特征投影到2D看看源域和目标域的点在特征空间里是否混在一起。如果两者明显重叠说明对抗是有效的如果还界线分明那说明对抗强度不够或者域判别器能力不足。6.4 一点个人体会我实际用DANN处理过图像和文本两类任务最大的感受是DANN不是“模型架构的革命”而是一种“训练哲学的转变”。它真正教会模型的是在面对分布偏移时什么样的特征才是“靠得住”的。如果你手头有一个分类任务换了个环境就不太灵与其急着采集目标域数据打标签不如先试试DANN。很多情况下你缺的不是更多数据而是一个“不分地域”的特征表达。