迁移学习（一）：基础与核心概念

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

"""
特征迁移与域适应示例
方法：特征提取 + MMD 对齐 + 微调
"""

import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
from sklearn.metrics import accuracy_score
from sklearn.svm import SVC

np.random.seed(42)
torch.manual_seed(42)

# --- 数据生成：模拟域偏移 ---------------------------------------------------

def generate_source_domain(n_samples=1000):
    """源域：两个分得很开的高斯簇。"""
    X0 = np.random.randn(n_samples // 2, 2) * 0.5 + np.array([-2, -2])
    X1 = np.random.randn(n_samples // 2, 2) * 0.5 + np.array([2, 2])
    X = np.vstack([X0, X1])
    y = np.hstack([np.zeros(n_samples // 2), np.ones(n_samples // 2)])
    return X, y

def generate_target_domain(n_samples=200):
    """目标域：旋转 45 度并平移（协变量偏移）。"""
    theta = np.pi / 4
    rotation = np.array([[np.cos(theta), -np.sin(theta)],
                         [np.sin(theta),  np.cos(theta)]])
    X0 = (np.random.randn(n_samples // 2, 2) * 0.6 + np.array([-1, -1])) @ rotation.T
    X1 = (np.random.randn(n_samples // 2, 2) * 0.6 + np.array([ 1,  1])) @ rotation.T
    X = np.vstack([X0, X1])
    y = np.hstack([np.zeros(n_samples // 2), np.ones(n_samples // 2)])
    return X, y

X_source,        y_source        = generate_source_domain(1000)
X_target_train,  y_target_train  = generate_target_domain(50)   # 少量标注数据
X_target_test,   y_target_test   = generate_target_domain(200)  # 测试集

print(f"源域: {X_source.shape}  |  目标训练: {X_target_train.shape}  "
      f"|  目标测试: {X_target_test.shape}")

# --- 方法 1：仅在少量目标样本上从头训练 -----------------------------------

clf_scratch = SVC(kernel="rbf", gamma="auto").fit(X_target_train, y_target_train)
acc_scratch = accuracy_score(y_target_test, clf_scratch.predict(X_target_test))
print(f"[从头训练]   准确率: {acc_scratch:.4f}")

# --- 方法 2：在源域训练，直接套用到目标域（无适应） -----------------------

clf_direct = SVC(kernel="rbf", gamma="auto").fit(X_source, y_source)
acc_direct = accuracy_score(y_target_test, clf_direct.predict(X_target_test))
print(f"[直接迁移]   准确率: {acc_direct:.4f}")

# --- 方法 3：特征迁移 + MMD 对齐 + 微调 -----------------------------------

class FeatureExtractor(nn.Module):
    def __init__(self, input_dim=2, hidden_dim=32, output_dim=16):
        super().__init__()
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, hidden_dim), nn.ReLU(),
            nn.Linear(hidden_dim, output_dim), nn.ReLU(),
        )

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

class Classifier(nn.Module):
    def __init__(self, input_dim=16, num_classes=2):
        super().__init__()
        self.fc = nn.Linear(input_dim, num_classes)

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

def compute_mmd(x_source, x_target, gamma=1.0):
    """计算两个 batch 之间的 RBF 核 MMD（有偏估计，SGD 下足够用）。"""
    xx = torch.sum(x_source ** 2, dim=1, keepdim=True)
    yy = torch.sum(x_target ** 2, dim=1, keepdim=True)
    K_ss = torch.exp(-gamma * (xx + xx.t() - 2 * x_source @ x_source.t()))
    K_tt = torch.exp(-gamma * (yy + yy.t() - 2 * x_target @ x_target.t()))
    K_st = torch.exp(-gamma * (xx + yy.t() - 2 * x_source @ x_target.t()))
    n_s, n_t = x_source.size(0), x_target.size(0)
    return K_ss.sum() / n_s ** 2 + K_tt.sum() / n_t ** 2 - 2 * K_st.sum() / (n_s * n_t)

def train_with_mmd(X_source, y_source, X_target_unlabeled,
                   X_target_labeled, y_target_labeled,
                   epochs=100, lambda_mmd=0.5):
    """两阶段训练：源域分类 + MMD 对齐，然后在目标域上微调。"""
    X_s   = torch.FloatTensor(X_source)
    y_s   = torch.LongTensor(y_source.astype(int))
    X_t_u = torch.FloatTensor(X_target_unlabeled)
    X_t_l = torch.FloatTensor(X_target_labeled)
    y_t_l = torch.LongTensor(y_target_labeled.astype(int))

    feat = FeatureExtractor()
    clf  = Classifier()
    optimizer = optim.Adam(list(feat.parameters()) + list(clf.parameters()), lr=1e-3)
    criterion = nn.CrossEntropyLoss()

    loader = DataLoader(TensorDataset(X_s, y_s), batch_size=32, shuffle=True)

    # 阶段 1：源域分类 + 在无标注目标上做 MMD 对齐
    for _ in range(epochs):
        for X_batch, y_batch in loader:
            optimizer.zero_grad()
            loss_cls = criterion(clf(feat(X_batch)), y_batch)
            loss_mmd = compute_mmd(feat(X_batch), feat(X_t_u))
            (loss_cls + lambda_mmd * loss_mmd).backward()
            optimizer.step()

    # 阶段 2：在少量标注目标样本上做轻量微调
    for _ in range(50):
        optimizer.zero_grad()
        criterion(clf(feat(X_t_l)), y_t_l).backward()
        optimizer.step()

    return feat, clf

feat, clf_mmd = train_with_mmd(
    X_source, y_source,
    X_target_unlabeled=X_target_test,
    X_target_labeled=X_target_train,
    y_target_labeled=y_target_train,
)

feat.eval(); clf_mmd.eval()
with torch.no_grad():
    preds = torch.argmax(clf_mmd(feat(torch.FloatTensor(X_target_test))), dim=1).numpy()

acc_transfer = accuracy_score(y_target_test, preds)
print(f"[特征 + MMD] 准确率: {acc_transfer:.4f}")
print(f"\n相比从头训练提升: "
      f"{(acc_transfer - acc_scratch) / acc_scratch * 100:.1f}%")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

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

def _train_linear_probe(features, labels, num_classes, n_epochs=5, lr=1e-2):
    probe = nn.Linear(features.size(1), num_classes)
    opt   = optim.Adam(probe.parameters(), lr=lr)
    crit  = nn.CrossEntropyLoss()
    for _ in range(n_epochs):
        opt.zero_grad()
        crit(probe(features), labels).backward()
        opt.step()
    return probe

@torch.no_grad()
def _extract(model, loader):
    feats, labs = [], []
    for x, y in loader:
        feats.append(model(x)); labs.append(y)
    return torch.cat(feats), torch.cat(labs)

def transfer_feasibility_check(source_model, target_train_loader,
                               target_test_loader, num_classes, n_epochs=5):
    """比较源冻结 vs 随机初始化线性探针在目标域上的表现。"""
    source_model.eval()
    rand_model = type(source_model)()      # 同架构，全新初始化
    rand_model.eval()

    f_tr_s, y_tr = _extract(source_model, target_train_loader)
    f_te_s, y_te = _extract(source_model, target_test_loader)
    f_tr_r, _    = _extract(rand_model,   target_train_loader)
    f_te_r, _    = _extract(rand_model,   target_test_loader)

    probe_s = _train_linear_probe(f_tr_s, y_tr, num_classes, n_epochs)
    probe_r = _train_linear_probe(f_tr_r, y_tr, num_classes, n_epochs)

    acc_s = (probe_s(f_te_s).argmax(1) == y_te).float().mean().item()
    acc_r = (probe_r(f_te_r).argmax(1) == y_te).float().mean().item()
    rec   = ("transfer" if acc_s > acc_r + 0.05 else
             "marginal" if acc_s > acc_r - 0.02 else "scratch")
    return {"source_frozen_acc": acc_s, "random_init_acc": acc_r,
            "recommendation": rec}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

import copy, torch, torch.nn as nn, torch.optim as optim

def _freeze_first_k(model, k):
    """就地冻结 model.children() 的前 k 个 '模块'。"""
    for i, child in enumerate(model.children()):
        for p in child.parameters():
            p.requires_grad = (i >= k)

def _train_one(model, loader, n_epochs=10, lr=1e-3):
    params = [p for p in model.parameters() if p.requires_grad]
    opt    = optim.Adam(params, lr=lr)
    crit   = nn.CrossEntropyLoss()
    model.train()
    for _ in range(n_epochs):
        for x, y in loader:
            opt.zero_grad()
            crit(model(x), y).backward()
            opt.step()
    return model

@torch.no_grad()
def _eval(model, loader):
    model.eval(); correct = total = 0
    for x, y in loader:
        correct += (model(x).argmax(1) == y).sum().item(); total += y.size(0)
    return correct / total

def compare_freezing_strategies(model, target_train_loader, target_test_loader,
                                depths=(0, 1, 2, 3, 4), n_epochs=10):
    """对每个冻结深度运行一次微调，返回 {depth: test_acc}。"""
    results = {}
    for k in depths:
        m = copy.deepcopy(model)
        _freeze_first_k(m, k)
        _train_one(m, target_train_loader, n_epochs=n_epochs)
        results[k] = _eval(m, target_test_loader)
    return results