. 代码实践

2.1 迁移学习

这次就捡起来我们之前的猫狗二分类模型,之前我们尝试训练这个模型,多轮训练后验证集的最高准确率也只在 70% 上下波动,来看看对这个模型应用迁移学习的效果。

在对二分类模型应用迁移学习前,先简单介绍一下我们的迁移来源任务。

 

2.1.1 预训练模型 ResNet18(ImageNet 预训练)

我们使用的预训练模型叫ResNet18,ResNet全称为Residual Neural Network,中文翻译为残差神经网络,18是指网络深度。

这是一个非常经典且具有开创意义的模型结构,在大量的领域都能广泛应用。而它最初的使用,就是在图像分类上。

这里摆一张网络结构图,暂时先不介绍它的结构和原理,课程在下一部分才正式介绍图像学习的基本:卷积神经网络,现在,只需要知道我们借来了一个很厉害的模型就好了。

image.png

(注:这张图来自这里)

 

而对于ImageNet,我们之前也提到过:ImageNet 是一个包含 1400 万张图像、覆盖 1000 个类别 的大型图像分类数据集。每张图像都带有准确的标签,相当于给模型提供了大量“优质原材料”,让它先在通用视觉特征上打下坚实基础。

image.png

(注:这张则来自这里)

虽然ImageNet本身也包括了猫和狗的图像,有一些“透题”的感觉,但1000 类中“猫狗”类比例极小(不到 0.5%),超多的类别并不会让它特别偏向二者的识别,而是通用的纹理识别。

因此,经过ImageNet预训练的ResNet18模型仍很适合作为我们的迁移来源模型。

下面就看看如何改动我们之前的代码来引入它。

 

2.1.2 PyTorch引入预训练ResNet18

现在,我们就从代码层面看看如何使用预训练ResNet18。

 

(1)修改预处理以适应ResNet18输入层

首先,引入预训练ResNet18就代表数据从原本的输入我们创建的模型改为输入ResNet18。

所以,我们首先就应该更改数据预处理方法以匹配ResNet18输入层。

而在PyTorch里,负责数据预处理的就是transforms模块。

先看看我们原来的预处理:

 

transform = transforms.Compose([  

    transforms.Resize((128, 128)), 

    # 将图像的大小调整为 128x128 像素,保证输入图像的一致性 

    transforms.ToTensor(),

    # 将图像从 PIL 图像或 NumPy 数组转换为 PyTorch 张量,图像的像素值也会被从 [0, 255] 范围映射到 [0, 1 ]范围,这是使用 Pytorch 固定的一步。

    transforms.Normalize((0.5,), (0.5,))  

    # 标准化,原本在 [0, 1] 范围内的像素值会变换到 [-1, 1] 范围内。

])

·····中间代码

self.hidden1 = nn.Linear(128 * 128 * 3, 1024) # 模型输入层,对应输入维度和Resize大小对应,* 3是因为彩色图片有三个通道。

而现在,使用ResNet18,自然就要匹配他的一些输入设置,所以,我们修改成如下设置:

 

transform = transforms.Compose([  

    transforms.Resize((224, 224)), # ResNet 输入 224x224 这一步一定要有              

    transforms.ToTensor(),

    transforms.Normalize([0.485, 0.456, 0.406],  

                         [0.229, 0.224, 0.225]) # ResNet 官方均值方差  

    # 这些数字是基于 ImageNet 数据集 上的统计计算得到的,是针对每个通道的均值和方差。

])

如果你忘了标准化的是用来做什么的,我们第一次介绍它是在这里:归一化

现在,我们的数据经过预处理就能顺利输入ResNet18了,我们继续下一步。

 

(2)调用预训练ResNet18模型

同样,PyTorch 内置了 ResNet18 的模型结构,我们这样调用它:

 

model = models.resnet18(pretrained=True)

# pretrained 参数的T和F就代表是否使用预训练的模型参数,这里的 True 就代表使用。

# 而如果更改为 False ,就代表只使用 ResNet18 的网络结构。

注意,这时我们原本设计的网络结构就被这一行替代了。

 

(3)第一次尝试:freeze迁移

我们先试试freeze迁移,也就是ResNet18的任何一层的参数都固定,不参与反向传播。

这样设置:

 

for param in model.parameters():  

    param.requires_grad = False # 默认为 True

# 显然,这个 for 循环就是在对现在模型每一层说:“不要参与反向传播”

(4)替换输出头以适配迁移目标任务

这一步的逻辑就是找到输出层,替换输出层。

要强调的一点是,这一步一定要在冻结之后进行。

替换层级会让层级初始化,并默认参与反向传播。 如果你把冻结放在了这步之后,那整个网络就不存在反向传播了。

来看看具体怎么做:

 

num_features = model.fc.in_features

# 这一行是在获取模型的一个叫 fc 的层的输入维度。

# 我们一般将网络的最后一层全连接层命名为 fc

# 也就是说,这一行是在获取模型的输出层的输入维度。  

model.fc = nn.Sequential(  

# Sequential 是 PyTorch 提供的一个容器类,它允许将多个层按顺序组合在一起。

    nn.Linear(num_features, 1),  

    nn.Sigmoid()  

)

# 很明显,我们把最后一层换成了只有一个神经元并经过Sigmoid激活来适配我们的任务要求。

我只是将原模型的输出层更改成适配猫狗二分类的结构,如果你想在最后增加更多自己的层级设置,只需要在 Sequential 里按顺序添加,并确保层间维度匹配即可。

 

好了,现在我们就完成了所有配置,来看看效果吧。

 

2.1.3 第一次运行:freeze迁移+目标任务数据较多

回忆一下,我们的猫狗数据集总共只有 2400 幅图像,这时一个相当小规模的数据集。在经过划分后,用于训练的数据就只有约 2000 个样本,我们是这样设置的:

 

train_size = int(0.8 * len(dataset))

我们先看看不改变这个划分,只应用迁移学习的效果,结果如下:

image.png

好家伙,强大无需多言,即使是只训练一轮的效果就已经强于我们之前的训练效果了。

简单分析一下原因:

 

ImageNet 的超大规模样本让 ResNet18 模型学习了对通用特征的识别。

卷积网络和残差网络本身在图学习的优越性也帮助了拟合。

简单打个比方就是:教材好+学生聪明。

 

2.1.4 第二次运行:freeze迁移+目标任务数据较少

之前我们在迁移学习的理论部分里提到过,迁移学习的出现原因还是因此迁移目标任务的数据不足。

我们再次严格一下这个条件试试:

 

train_size = int(0.1 * len(dataset))

现在,训练集,验证集,测试集都只有 240 个样本,我们再来看看效果:

image.png

可以看到和第一次运行的最大差别就在于最开始的轮次效果。

换句话说,这一次我们只给模型提供了很少的“练习题”,而模型只能依靠最后一层去适应猫狗分类这个任务。

 

继续打比方:这就像你找了一个学过大量数学知识的学霸来做一套小测验,他的思维能力依然很强,但题目太少,所以在前面几道题里发挥得并不稳定。

 

不过随着训练轮次增加,曲线还是逐渐稳定下来,这说明迁移学习确实提供了很好的“启动点”。

最终的结果也再次验证了一个关键结论:数据越少,迁移学习越有价值。

 

如果我们不迁移、从头训练,那么240 张训练图像几乎不可能产生有意义的分类能力,而 freeze 迁移学习却做到了“可用”。

 

现在,我们再来看看迁移学习的另一种形式。

 

2.1.5 第三次运行:fine-tuning

现在,我们再试试 fine-tuning 的效果,它是指在预训练基础上整体微调,也就是说,我们现在要”解冻“ 模型之前的层级。

如何解冻?你可能已经想到了:

 

for param in model.parameters():  

    param.requires_grad = False # 默认为 True

把这两行删掉或者注释掉就OK了。

现在来看看运行结果,这里我把训练集占比恢复到了0.8:

image.png

很显然,效果仍然不错,就不再详细解释了。

 

最后要说的是,迁移学习是我们在面对当前任务数据不足时的一种选择,但它的效果不一定会像我们现在展示的这么好。

虽然我们之前训练猫狗二分类的结果一直不太好,但实际上是因为课程把卷积网络安排在了后面介绍,我们之前使用全连接网络训练更像是"狗拿耗子",自然不会有很好的结果。

实际上,猫狗二分类只是一个图学习中入门级的任务,加上和 ResNet18 的适配,所以在这次演示中起到了很好的效果,如果要应用迁移学习,还是要视具体任务选择。

在调优过程中,不变的还是不断的尝试。

 

下面就来看看这周了解的另一种学习方式:多任务学习。

 

2.2 多任务学习

多任务学习在代码逻辑上的重点在于数据集标签和网络结构上,其他部分不会产生太大的变化。

这部分就不再找专门的数据集来进行演示了,我们重点看看如何实现多任务学习的“前面共享、后面分头”的结构。

还是先拿我们之前的老结构拿出来晒晒:

 

class NeuralNetwork(nn.Module):  

    def __init__(self):  

        super().__init__()  

        self.flatten = nn.Flatten()  

        self.hidden1 = nn.Linear(128 * 128 * 3, 1024)  

        self.hidden2 = nn.Linear(1024, 512)  

        self.hidden3 = nn.Linear(512, 128)  

        self.hidden4 = nn.Linear(128, 32)  

        self.hidden5 = nn.Linear(32, 8)  

        self.hidden6 = nn.Linear(8, 3)  

        self.relu = nn.ReLU()  

        self.output = nn.Linear(3, 1)  

        self.sigmoid = nn.Sigmoid()  

  

    def forward(self, x):  

        x = self.flatten(x)  

        x = self.relu(self.hidden1(x))  

        x = self.relu(self.hidden2(x))  

        x = self.relu(self.hidden3(x))  

        x = self.relu(self.hidden4(x))  

        x = self.relu(self.hidden5(x))  

        x = self.relu(self.hidden6(x))  

        x = self.sigmoid(self.output(x))  

        return x

如果不使用这种一条路走到底的线性结构,而是实现允许“分叉” 的多任务学习树形结构,就是这部分内容。

 

2.2.1 多任务学习的网络结构

下面给出一个简单的例子:

任务A:二分类(猫狗)

任务B:图像亮度回归(0~1)

网络结构就可以变成这样:

 

class MultiTaskNet(nn.Module):

    def __init__(self):

        super().__init__()

        self.flatten = nn.Flatten()

        self.relu = nn.ReLU()

 

        # —— 前面共享部分 ——

        self.shared1 = nn.Linear(128 * 128 * 3, 1024)

        self.shared2 = nn.Linear(1024, 256)

 

        # —— 后面分头部分 ——

        # 任务 A:猫狗二分类

        self.headA = nn.Linear(256, 1)

        # 任务 B:亮度回归

        self.headB = nn.Linear(256, 1)

        self.sigmoid = nn.Sigmoid()

 

    def forward(self, x):

        x = self.flatten(x)

        x = self.relu(self.shared1(x))

        x = self.relu(self.shared2(x))

        outA = self.sigmoid(self.headA(x)) # 分类

        outB = self.headB(x) # 回归

        return outA, outB # 返回量增加为两个

你可以看到,整个结构的前半部分的参数是共享的;后半部分两个任务分头走,最后的两个返回值就是对两个任务的预测。

 

2.2.2 多任务学习的损失函数怎么写?

因为现在有两个输出,就需要计算两个任务的损失,再把它们加起来:

 

lossA = criterionA(outA, labelA) # 比如二分类 BCE

lossB = criterionB(outB, labelB) # 比如 MSE 回归损失

loss = lossA + lossB

loss.backward()

如果两个任务重要程度不同,也可以加权:

 

loss = 0.7 * lossA + 0.3 * lossB

比如主任务是猫狗分类,辅助任务是亮度预测,那就让“分类任务”权重大一点。

 

这就是本篇的全部内容了,下一章就到计算机视觉部分了,也就终于可以展开之前一直在提的卷积网络了。

 

3.附录

3.1 Pytorch版 迁移学习代码

import torch  

import torch.nn as nn  

import torch.optim as optim  

from torchvision import datasets, transforms, models  

from torch.utils.data import DataLoader, random_split  

import matplotlib.pyplot as plt  

  

transform = transforms.Compose([  

    transforms.Resize((224, 224)), # ResNet 输入 224x224 transforms.ToTensor(),  

    transforms.Normalize([0.485, 0.456, 0.406],  

                         [0.229, 0.224, 0.225]) # ResNet 官方均值方差  

])  

  

dataset = datasets.ImageFolder(root='./cat_dog', transform=transform)  

  

train_size = int(0.8 * len(dataset))  

val_size = int(0.1 * len(dataset))  

test_size = len(dataset) - train_size - val_size  

train_dataset, val_dataset, test_dataset = random_split(dataset, [train_size, val_size, test_size])  

  

train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)  

val_loader = DataLoader(val_dataset, batch_size=32, shuffle=False)  

test_loader = DataLoader(test_dataset, batch_size=32, shuffle=False)  

  

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")  

  

model = models.resnet18(pretrained=True)  

  

# 是否冻结预训练参数  

for param in model.parameters():  

    param.requires_grad = False  

  

# 替换最后一层  

num_features = model.fc.in_features  

model.fc = nn.Sequential(  

    nn.Linear(num_features, 1),  

    nn.Sigmoid()  

)  

  

model = model.to(device)  

  

criterion = nn.BCELoss()  

optimizer = optim.Adam(model.fc.parameters(), lr=0.001)  

  

epochs = 10  

train_losses = []  

train_accs = []  

val_accs = []  

  

for epoch in range(epochs):  

    model.train()  

    epoch_train_loss = 0  

    correct_train = 0  

    total_train = 0  

  

    for images, labels in train_loader:  

        images, labels = images.to(device), labels.to(device).float().unsqueeze(1)  

  

        outputs = model(images)  

        loss = criterion(outputs, labels)  

  

        optimizer.zero_grad()  

        loss.backward()  

        optimizer.step()  

  

        epoch_train_loss += loss.item()  

        preds = (outputs > 0.5).int()  

        correct_train += (preds == labels.int()).sum().item()  

        total_train += labels.size(0)  

  

    avg_train_loss = epoch_train_loss / len(train_loader)  

    train_acc = correct_train / total_train  

  

    train_losses.append(avg_train_loss)  

    train_accs.append(train_acc)  

  

    model.eval()  

    correct_val = 0  

    total_val = 0  

  

    with torch.no_grad():  

        for images, labels in val_loader:  

            images, labels = images.to(device), labels.to(device).float().unsqueeze(1)  

            outputs = model(images)  

            preds = (outputs > 0.5).int()  

            correct_val += (preds == labels.int()).sum().item()  

            total_val += labels.size(0)  

  

    val_acc = correct_val / total_val  

    val_accs.append(val_acc)  

  

    print(f"轮次 [{epoch+1}/{epochs}] "  

          f"训练损失: {avg_train_loss:.4f} "  

          f"训练准确率: {train_acc:.4f} "  

          f"验证准确率: {val_acc:.4f}")  

  

  

plt.rcParams['font.sans-serif'] = ['SimHei']  

plt.rcParams['axes.unicode_minus'] = False  

plt.figure(figsize=(10,5))  

plt.plot(train_losses, label='训练损失')  

plt.plot(train_accs, label='训练准确率')  

plt.plot(val_accs, label='验证准确率')  

plt.legend()  

plt.grid(True)  

plt.show()  

  

  

model.eval()  

correct = 0  

total = 0  

  

with torch.no_grad():  

    for images, labels in test_loader:  

        images, labels = images.to(device), labels.to(device).float().unsqueeze(1)  

        outputs = model(images)  

        preds = (outputs > 0.5).int()  

        correct += (preds == labels.int()).sum().item()  

        total += labels.size(0)  

  

print(f"测试准确率: {correct / total:.4f}")

3.2 Tensorflow版 迁移学习代码

注:TF没有内置ResNet的18版本,而是ResNet50以及更高的版本,而TF和提供第三方 ResNet18 库的兼容性又不太好,因此这里实际上使用的是ResNet50。

 

import tensorflow as tf  

from tensorflow import keras  

from tensorflow.keras import layers, models  

import matplotlib.pyplot as plt  

  

IMG_SIZE = (224, 224)  

BATCH_SIZE = 32  

train_ds = keras.preprocessing.image_dataset_from_directory(  

    "./cat_dog",  

    validation_split=0.2,  

    subset="training",  

    seed=42,  

    image_size=IMG_SIZE,  

    batch_size=BATCH_SIZE  

)  

  

val_test_ds = keras.preprocessing.image_dataset_from_directory(  

    "./cat_dog",  

    validation_split=0.2,  

    subset="validation",  

    seed=42,  

    image_size=IMG_SIZE,  

    batch_size=BATCH_SIZE  

)  

  

 

val_ds = val_test_ds.take(len(val_test_ds) // 2)  

test_ds = val_test_ds.skip(len(val_test_ds) // 2)  

  

preprocess = keras.applications.resnet50.preprocess_input  

  

def preprocess_fn(image, label):  

    return preprocess(tf.cast(image, tf.float32)), tf.cast(label, tf.float32)  

  

train_ds = train_ds.map(preprocess_fn)  

val_ds = val_ds.map(preprocess_fn)  

test_ds = test_ds.map(preprocess_fn)  

  

train_ds = train_ds.prefetch(tf.data.AUTOTUNE)  

val_ds = val_ds.prefetch(tf.data.AUTOTUNE)  

test_ds = test_ds.prefetch(tf.data.AUTOTUNE)  

  

 

# 构建迁移学习模型(冻结 ResNet50)  

base_model = keras.applications.ResNet50(  

    weights="imagenet",  

    include_top=False,  

    input_shape=(224, 224, 3),  

    pooling="avg"  

)  

  

base_model.trainable = False # 冻结  

  

inputs = keras.Input(shape=(224, 224, 3))  

x = base_model(inputs, training=False)  

outputs = layers.Dense(1, activation="sigmoid")(x)  

model = keras.Model(inputs, outputs)  

  

model.compile(  

    optimizer=keras.optimizers.Adam(learning_rate=0.001),  

    loss="binary_crossentropy",  

    metrics=["accuracy"]  

)  

  

 

# 训练  

history = model.fit(  

    train_ds,  

    validation_data=val_ds,  

    epochs=10  

)  

  

plt.figure(figsize=(10, 5))  

plt.plot(history.history["loss"], label="训练损失")  

plt.plot(history.history["accuracy"], label="训练准确率")  

plt.plot(history.history["val_accuracy"], label="验证准确率")  

plt.legend()  

plt.grid(True)  

plt.show()  

  

test_loss, test_acc = model.evaluate(test_ds)  

print("测试准确率:", test_acc)

3.3 Tensorflow版 多任务学习网络结构

class MultiTaskNet(tf.keras.Model):

    def __init__(self):

        super(MultiTaskNet, self).__init__()

        # —— 前面共享部分 ——

        self.flatten = layers.Flatten()

        self.shared1 = layers.Dense(1024, activation='relu')

        self.shared2 = layers.Dense(256, activation='relu')

        # —— 后面分头部分 ——

        # 任务 A:猫狗二分类

        self.headA = layers.Dense(1, activation='sigmoid')

        # 任务 B:亮度回归

        self.headB = layers.Dense(1, activation=None)

 

    def call(self, inputs, training=False):

        x = self.flatten(inputs)

        x = self.shared1(x)

        x = self.shared2(x)

        outA = self.headA(x) # 分类输出

        outB = self.headB(x) # 回归输出

        return outA, outB

更多推荐