AI域适应折腾手记

域适应 迁移学习 深度学习
作者: 醉月思 发布日期: 2022-05-26 分类: 技术实践

后来折腾了一圈域适应（Domain Adaptation），有些心得，也踩了不少坑。

一开始以为都是中文文本，应该差不多，结果用BERT直接fine-tune，准确率只有不到70%。

到底什么是域适应问题

先说清楚一个事：域适应本质上就是怎么让模型在训练数据（源域）上学习后，还能在别的数据分布（目标域）上用得好。

听起来很简单，但实际操作起来坑不少。去年做个文本分类项目，源域是新闻数据，目标域是电商评论。一开始以为都是中文文本，应该差不多，结果用BERT直接fine-tune，准确率只有不到70%。

问题出在哪里？新闻用词偏正式，电商评论充满口语、表情符号、甚至火星文。虽然都是中文，但分布差异已经大到模型无法应对。

这就是典型的域偏移（Domain Shift）问题。域适应的目标就是想办法减小这个偏移，让源域上学到的知识能迁移到目标域。

域适应的几种常见做法

基于特征对齐的方法

特征对齐是最常见的思路，简单说就是让源域和目标域的特征分布在同一个空间里靠得更近。

最基础的实现用对抗训练，原理和GAN有点像：一个特征提取器不断提取特征，一个域分类器试图区分这些特征来自源域还是目标域。特征提取器要"欺骗"域分类器，让它分不出来，这样提取的特征就是域不变的。

下面是一个基于PyTorch的简单实现：

import torch
import torch.nn as nn
import torch.optim as optim

# 域分类器
class DomainClassifier(nn.Module):
    def __init__(self, input_dim):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(input_dim, 1024),
            nn.ReLU(),
            nn.Dropout(0.5),
            nn.Linear(1024, 1),
            nn.Sigmoid()
        )

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

# 特征提取器 + 任务分类器
class FeatureExtractor(nn.Module):
    def __init__(self):
        super().__init__()
        # 这里可以是预训练的BERT或其他backbone
        self.backbone = nn.Sequential(
            nn.Linear(768, 512),
            nn.ReLU(),
            nn.Dropout(0.3)
        )
        self.task_classifier = nn.Linear(512, num_classes)

    def forward(self, x):
        features = self.backbone(x)
        return features, self.task_classifier(features)

# 训练循环
def train_with_dann(source_loader, target_loader):
    feature_extractor = FeatureExtractor().cuda()
    domain_classifier = DomainClassifier(512).cuda()

    optimizer = optim.Adam(
        list(feature_extractor.parameters()) + list(domain_classifier.parameters()),
        lr=0.0001
    )

    criterion = nn.CrossEntropyLoss()
    domain_criterion = nn.BCELoss()

    for epoch in range(num_epochs):
        # 源域数据
        for source_data in source_loader:
            inputs, labels = source_data
            inputs, labels = inputs.cuda(), labels.cuda()

            # 任务学习
            features, task_outputs = feature_extractor(inputs)
            task_loss = criterion(task_outputs, labels)

            # 域分类器训练
            domain_labels = torch.ones(inputs.size(0), 1).cuda()
            domain_outputs = domain_classifier(features.detach())
            domain_loss = domain_criterion(domain_outputs, domain_labels)

            loss = task_loss + domain_loss
            loss.backward()
            optimizer.step()
            optimizer.zero_grad()

        # 目标域数据（反向梯度）
        for target_data in target_loader:
            inputs = target_data[0].cuda()

            features, _ = feature_extractor(inputs)
            domain_labels = torch.zeros(inputs.size(0), 1).cuda()
            domain_outputs = domain_classifier(features)
            domain_loss = domain_criterion(domain_outputs, domain_labels)

            # 反转梯度
            loss = domain_loss
            loss.backward()
            optimizer.step()
            optimizer.zero_grad()

这个实现的问题在哪里？实际跑起来发现，域分类器如果太强，特征提取器就学不好；域分类器太弱，又起不到对齐作用。

后来发现一个技巧：给域分类器加个梯度反转层（Gradient Reversal Layer），在反向传播时把梯度符号反转。这样源域和目标域的优化方向就一致了。

基于实例加权的方法

另一种思路是对源域样本加权，让模型更关注那些和目标域"像"的样本。

实际项目中用了这个方法后，效果有提升但不明显。后来想明白一个问题：如果源域和目标域差异太大，加权再怎么调整，基础分布就是不一样，效果也就有限。

import numpy as np
from sklearn.utils import class_weight

def compute_domain_weights(source_features, target_features):
    # 计算源域和目标域的分布差异
    # 这里简化处理，实际可以用更复杂的方法
    source_mean = source_features.mean(axis=0)
    target_mean = target_features.mean(axis=0)

    # 计算每个源域样本与目标域的相似度
    similarities = 1 / (1 + np.linalg.norm(
        source_features - target_mean, axis=1
    ))

    # 归一化得到权重
    weights = similarities / similarities.sum()
    return weights

# 训练时加权
weights = compute_domain_weights(source_features, target_features)
weighted_criterion = nn.CrossEntropyLoss(
    weight=torch.FloatTensor(weights).cuda()
)

基于自训练的方法

自训练的思路是用源域训练的模型给目标域打伪标签，再用这些带标签的数据继续训练。

def self_training_loop(initial_model, target_loader):
    model = initial_model.cuda()

    for iteration in range(self_train_iterations):
        # 生成伪标签
        pseudo_labels = []
        confident_samples = []

        model.eval()
        with torch.no_grad():
            for data in target_loader:
                inputs = data[0].cuda()
                outputs = model(inputs)
                probs = torch.softmax(outputs, dim=1)
                max_probs, preds = torch.max(probs, dim=1)

                # 只用高置信度的样本
                confident_mask = max_probs > confidence_threshold
                if confident_mask.any():
                    pseudo_labels.append(preds[confident_mask])
                    confident_samples.append(inputs[confident_mask])

        # 用伪标签继续训练
        model.train()
        for samples, labels in zip(confident_samples, pseudo_labels):
            outputs = model(samples)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()
            optimizer.zero_grad()

    return model

这个方法效果不错，但有个大坑：伪标签质量差的话，会加剧错误。第一次跑的时候置信度阈值设太高，能用的样本太少；设太低，又引入了很多噪声。

后来用了渐进式策略：前期用高阈值保证质量，后期逐步降低阈值增加样本量。

实际踩过的坑

坑一：数据预处理不一致

这个问题看起来很基础，但很容易忽略。做文本分类时，源域是标准的中文分词，目标域数据里夹杂了英文、emoji、特殊符号。预处理如果不统一，特征本身就对不上。

最惨的一次是源域用jieba分词，目标域用的是其他分词工具，同一个词的分法都不一样。后续所有特征对齐方法基本都失效了。

解决方法很简单但很重要：所有域的数据预处理步骤要完全一致。

# 统一预处理
def preprocess_text(text):
    # 去除特殊字符
    text = re.sub(r'[^一-龥a-zA-Z0-9\s]', '', text)

    # 统一分词
    tokens = jieba.lcut(text)

    # 停用词过滤
    tokens = [t for t in tokens if t not in stop_words and len(t) > 1]

    return ' '.join(tokens)

# 对源域和目标域都使用相同的预处理
source_data = [preprocess_text(t) for t in source_data]
target_data = [preprocess_text(t) for t in target_data]

坑二：模型容量不匹配

源域数据量大，用了一个大模型；目标域数据少，同样用大模型结果过拟合严重。

后来想明白了：域适应不是简单地用更强的模型，而是要找到合适的模型容量。数据量少的时候，小模型反而更稳健。

# 根据数据量选择模型大小
def select_model_by_dataset_size(num_samples, input_dim):
    if num_samples < 1000:
        # 小数据用小模型
        hidden_dims = [128, 64]
        dropout = 0.5
    elif num_samples < 10000:
        # 中等数据用中等模型
        hidden_dims = [256, 128, 64]
        dropout = 0.3
    else:
        # 大数据用大模型
        hidden_dims = [512, 256, 128]
        dropout = 0.2

    return build_model(input_dim, hidden_dims, dropout)

坑三：评估标准不合理

一开始用源域测试集来评估域适应效果，这完全不对。域适应的最终目标是在目标域上表现好，所以必须用目标域的真实标签来评估。

但问题来了：目标域的真实标签可能很少。这时候就需要半监督评估或者小样本验证。

def evaluate_domain_adaptation(model, source_test, target_test, target_labeled):
    # 在源域测试集上评估（作为基准）
    source_acc = evaluate(model, source_test)

    # 在目标域测试集上评估（目标）
    target_acc = evaluate(model, target_test)

    # 域差异度量
    domain_discrepancy = compute_domain_discrepancy(model, source_test, target_test)

    # 迁移效率：目标域提升 / 源域下降
    transfer_efficiency = (target_acc - baseline_target) / (baseline_source - source_acc)

    return {
        'source_accuracy': source_acc,
        'target_accuracy': target_acc,
        'domain_discrepancy': domain_discrepancy,
        'transfer_efficiency': transfer_efficiency
    }

实际效果如何

做了几个实际项目，效果有好有坏。最成功的案例是一个产品评论分类项目：

  • 源域：新闻评论，标注数据10万条
  • 目标域：电商评论，标注数据1000条
  • 不做域适应：目标域准确率65%
  • 用DANN特征对齐：目标域准确率78%
  • 特征对齐 + 自训练：目标域准确率85%

另一个失败案例是医疗影像诊断：

  • 源域：公开数据集
  • 目标域：医院私有数据
  • 不管用什么方法，准确率都上不去

后来发现，公开数据集的成像设备和医院的设备差异太大，预处理都很难对齐。这种情况下，域适应能做的事情就很有限了。

什么时候该考虑域适应

不是所有场景都适合域适应。根据实际经验，满足这些条件可以考虑：

  1. 源域和目标域有足够相似性，都是图像、文本、音频等同一类型数据
  2. 源域有足够标注数据，目标域标注数据少或没有
  3. 域差异主要来源于采集方式、设备、环境等可对齐的因素
  4. 有一定的计算资源，域适应方法通常比普通训练慢

如果源域和目标域差异太大（比如训练图像识别模型要做音频识别），或者目标域完全没有相似性特征，这时候域适应可能不是最佳方案，不如考虑重新收集数据或者换方法。

还有些开放问题

域适应还有很多没解决好的问题：

  • 多源域适应：怎么同时从多个源域学习？
  • 无监督域适应：目标域完全无标注的情况下怎么保证效果？
  • 在线域适应：目标域数据分布随时间变化，怎么持续适应？
  • 域泛化：模型在未见过的域上怎么保持性能？

这些问题的解决可能需要更深的理解，也可能等不来通用方法。实际项目中，还是得根据具体场景选择合适的策略，没有银弹。

写在后面

域适应不是什么魔法，它只是帮我们从一个角度解决数据分布差异问题。有时候有用，有时候没用，这很正常。

技术实践就是不断试错和积累经验的过程。这篇文章记录了一些实际的做法和教训，希望能给遇到类似问题的人一点参考。

下次再遇到训练集好得不行、测试集掉链子的情况，至少知道可以从这个角度想想解决办法。

版权声明: 本文首发于 指尖魔法屋-AI域适应折腾手记（https://blog.thinkmoon.cn/post/230-domain-adaptation-practice-source-to-target/） 转载或引用必须申明原指尖魔法屋来源及源地址！

相关阅读

根据标签、分类和主题为你挑选的相近文章