小样本学习(Few-Shot Learning)（三）

文档中心

1. 前言

本文使用飞桨（PaddlePaddle）基于paddle.vision.datasets.Flowers数据集实践小样本学习问题的Pretraining+Fine Tuning解法。
小样本学习(Few-Shot Learning)（一）讲解了小样本学习问题的基本概念及基本思路，并介绍了使用孪生网络解决小样本学习问题的方法。
小样本学习(Few-Shot Learning)（二）讲解了小样本学习问题的Pretraining+Fine Tuning解法。
本人全部文章请参见：博客文章导航目录
本文归属于：元学习系列

2. 预训练神经网络（Pretraining）

使用Pretraining+Fine Tuning方法解决小样本分类问题，首先须预训练一个用于从图片中提取特征卷积神经网络。根据小样本学习(Few-Shot Learning)（二）一文所述预训练方法，预训练步骤如下：

在大规模数据集上使用传统分类监督学习方法预训练模型；
使用余弦相似度损失继续训练模型。

本文讲解的小样本分类实践，使用了PaddlePaddle内置预训练模型resnet18作为从图片中提取特征的网络模型，加载预训练模型参数代替预训练步骤1，并在paddle.vision.datasets.Flowers 数据训练集上生成预训练步骤2所需训练数据。

2.1 数据处理

从paddle.vision.datasets.Flowers数据训练集中加载前 $k$ 类数据，随机生成(anchor_image, positive_image, negative_image)图片对，构造使用余弦相似度训练特征提取模型的训练集。
定义create_emb_model_data_reader函数，依次选择anchor_image，positive_image和negative_image，并返回一个数据读取迭代器：

# -*- coding: utf-8 -*-# @Time    : 2021/9/2 20:22# @Author  : He Ruizhi# @File    : fewshot.py# @Software: PyCharmimport paddlefrom paddle.vision.models import resnet18import paddle.vision.transforms as transformersimport numpy as npfrom typing import Callablefrom collections import defaultdictimport randomimport warningswarnings.filterwarnings('ignore')print(paddle.__version__)  # 2.1.0def create_emb_model_data_reader(chosen_classes: int = 5,     batch_size: int = 32,     num_batches: int = 100) -> Callable:    """    生成用于训练特征提取网络的数据    :param chosen_classes: 从paddle.vision.datasets.Flowers数据集中选取前chosen_classes个类    :param batch_size: 正负样本总数    :param num_batches: 构建训练数据批次总数    :return: 一个Callable对象    """    assert chosen_classes >= 2, '训练数据中须包含相同类别图片和不同类别图片，因此类别数必须大于或等于2，' \    '但是接收chosen_classes参数值为：{}'.format(chosen_classes)    assert batch_size % 2 == 0, 'batch_size必须是偶数，但是接收的batch_size参数值为：{}'.format(batch_size)    trans = transformers.Compose([ transformers.Resize((224, 224)),  # 将图片大小调整成224x224 transformers.Transpose((2, 0, 1))  # 将图片数据调整成channel_first    ])    # 读取数据    flowers_data = paddle.vision.datasets.Flowers(mode='train', transform=trans)    # 筛选出一个小数据集，对整个数据集进行处理速度太慢    # 之所以不随机选择类别，是因为从flowers_data读取数据非常慢    print('正在读取特征提取模型训练数据集……')    mini_flowers_data = []    for flower_data in flowers_data: if flower_data[1] <= chosen_classes:     mini_flowers_data.append(flower_data) else:     break    print('读取数据完毕！')    # 将图片按类别分类    class_idx_to_image_idxs = defaultdict(list)    for image, label in mini_flowers_data: class_idx_to_image_idxs[label.tolist()[0]].append(image / 255.)    def reader(): for _ in range(num_batches):     # 定义存放一个batch数据的numpy数组     # 每条数据包含anchor图片、正样本图片、负样本图片各一张     x = np.empty((3, batch_size // 2, 3, 224, 224), dtype='float32')     for batch_image_idx in range(batch_size // 2):  # 选定抽取的两类图片类别id  images_class_idxs = list(class_idx_to_image_idxs.keys())  base_class_idx = random.choice(images_class_idxs)  negative_class_idx = random.choice(images_class_idxs)  while base_class_idx == negative_class_idx:      negative_class_idx = random.choice(images_class_idxs)  base_examples_for_class = class_idx_to_image_idxs[base_class_idx]  negative_examples_for_class = class_idx_to_image_idxs[negative_class_idx]  # 随机选择图片  anchor_image_idx = random.choice(range(len(base_examples_for_class)))  positive_image_idx = random.choice(range(len(base_examples_for_class)))  while positive_image_idx == anchor_image_idx:      positive_image_idx = random.choice(range(len(base_examples_for_class)))  negative_image_idx = random.choice(range(len(negative_examples_for_class)))  x[0, batch_image_idx] = base_examples_for_class[anchor_image_idx]  x[1, batch_image_idx] = base_examples_for_class[positive_image_idx]  x[2, batch_image_idx] = negative_examples_for_class[negative_image_idx]     yield x    return reader

这里吐槽一下，飞桨内置数据集paddle.vision.datasets.Flowers中__getitem__方法，每次调用会执行image = self.data_tar.extractfile(img_ele).read()从压缩文件中读取数据，使得整个数据读取过程非常慢。

2.2 构建特征提取模型

定义继承自paddle.nn.Layer的FSCEmbNet类，加载PaddlePaddle内置预训练模型resnet18作为backbone，并自定义一个全连接层输出指定维度的特征向量：

class FSCEmbNet(paddle.nn.Layer):    """用于从图片中提取特征的网络    该网络利用预训练的resnet18骨干网络，再连接一个FC层，将图片Embedding层一个指定维度的特征向量    Args: embedding_dim: 指定将图片映射成多少维的特征向量    """    def __init__(self, embedding_dim=128): super(FSCEmbNet, self).__init__() rn18 = resnet18(pretrained=True) # 从预训练resnet18模型中将backbone提取出来 # 加载预训练模型的本质，是将模型中各模块参数进行了赋值。灵活地使用预训练模型，可以按照如下方式从模型中提取所需的模块 self.backbone = paddle.nn.Sequential(rn18.conv1, rn18.bn1, rn18.relu, rn18.maxpool,   rn18.layer1, rn18.layer2, rn18.layer3, rn18.layer4,   rn18.avgpool, paddle.nn.Flatten()) self.fc = paddle.nn.Linear(in_features=512, out_features=embedding_dim)    def forward(self, x): x = self.backbone(x) x = self.fc(x) # 将输出向量归一化 x = x / paddle.norm(x, axis=1, keepdim=True) return x

2.3 预训练特征提取模型

定义函数train_emb_model，实例化FSCEmbNet模型对象，定义Adam优化器，固定resnet18模型参数，传入特征提取模型中自定义全连接层参数作为优化参数。
使用FSCEmbNet分别提取anchor_image，positive_image和negative_image特征向量，并计算余弦相似度。构造训练数据集标签，并使用图片之间的余弦相似度与标签的均方误差作为损失函数。
计算损失函数关于模型参数的梯度，并优化相关参数：

def train_emb_model(epochs: int,      batch_size: int,      learning_rate: float,      data_reader: Callable,      embedding_dim: int = 128) -> None:    """    训练图片特征提取网络    :param epochs: 生成的训练数据集迭代的次数    :param batch_size: 每个batch数据量    :param learning_rate: 学习率    :param data_reader: 训练数据读取器，每次读取一个batch的数据    :param embedding_dim: 将图片转换成特征向量的维度    :return:    """    emb_model = FSCEmbNet(embedding_dim=embedding_dim)    emb_model.train()    # 定义优化器，只训练fc层参数    opt = paddle.optimizer.Adam(learning_rate=learning_rate, parameters=emb_model.fc.parameters())    for epoch in range(epochs): for batch_id, batch_data in enumerate(data_reader()):     # 分别获取anchor数据、正样本数据、负样本数据     anchors_data, positives_data, negatives_data = batch_data[0], batch_data[1], batch_data[2]     anchors = paddle.to_tensor(anchors_data)     positives = paddle.to_tensor(positives_data)     negatives = paddle.to_tensor(negatives_data)     # 得到三种数据对应的特征向量     anchors_embedding = emb_model(anchors)     positives_embedding = emb_model(positives)     negatives_embedding = emb_model(negatives)     # 生成标签     label = paddle.concat([paddle.ones([batch_size // 2, ]), paddle.zeros([batch_size // 2, ])])     # 计算余弦相似度     positives_similarity = paddle.nn.functional.cosine_similarity(anchors_embedding, positives_embedding)     negatives_similarity = paddle.nn.functional.cosine_similarity(anchors_embedding, negatives_embedding)     # 拼接成最终输出     similarity = paddle.concat([positives_similarity, negatives_similarity])     # 计算损失     loss = paddle.nn.functional.mse_loss(similarity, label)     if (batch_id + 1) % 50 == 0:  print("epoch: {}, batch_id: {}, loss is: {}".format(epoch+1, batch_id+1, loss.numpy()))     loss.backward()     opt.step()     opt.clear_grad()    # 训练完保存模型参数    paddle.save(emb_model.state_dict(), 'models/emb_model.pdparams')

指定emb_epochs = 10，emb_batch_size = 32，emb_learning_rate = 0.0005，chosen_classes = 25，调用create_emb_model_data_reader函数生成训练数据迭代器，并调用train_emb_model函数训练特征提取模型：

    # 训练特征提取网络    emb_epochs = 10    emb_batch_size = 32    emb_learning_rate = 0.0005    chosen_classes = 25    # 创建训练数据读取器    emb_data_reader = create_emb_model_data_reader(chosen_classes=chosen_classes, batch_size=emb_batch_size)    train_emb_model(epochs=emb_epochs, batch_size=emb_batch_size, learning_rate=emb_learning_rate,      data_reader=emb_data_reader)

训练过程打印信息如下：

正在读取特征提取模型训练数据集……读取数据完毕！epoch: 1, batch_id: 50, loss is: [0.15865126]epoch: 1, batch_id: 100, loss is: [0.11667006]epoch: 2, batch_id: 50, loss is: [0.11112603]epoch: 2, batch_id: 100, loss is: [0.09637958]epoch: 3, batch_id: 50, loss is: [0.09859052]epoch: 3, batch_id: 100, loss is: [0.08028176]epoch: 4, batch_id: 50, loss is: [0.05248487]epoch: 4, batch_id: 100, loss is: [0.07988761]epoch: 5, batch_id: 50, loss is: [0.07951481]epoch: 5, batch_id: 100, loss is: [0.07851607]epoch: 6, batch_id: 50, loss is: [0.07321084]epoch: 6, batch_id: 100, loss is: [0.05075514]epoch: 7, batch_id: 50, loss is: [0.06091946]epoch: 7, batch_id: 100, loss is: [0.06521406]epoch: 8, batch_id: 50, loss is: [0.07457791]epoch: 8, batch_id: 100, loss is: [0.06648796]epoch: 9, batch_id: 50, loss is: [0.0775561]epoch: 9, batch_id: 100, loss is: [0.08408293]epoch: 10, batch_id: 50, loss is: [0.06104364]epoch: 10, batch_id: 100, loss is: [0.05444841]

3. Fine Tuning

使用小样本学习(Few-Shot Learning)（二）一文所述Fine Tuning方法及A Good Initialization、Entropy Regularization和Cosine Similarity+Softmax Classifier技巧在Support Set中微调分类器参数。

3.1 生成Support Set与测试集

定义create_support_set_and_test_dataset函数，从paddle.vision.datasets.Flowers数据测试集前 $k$ 类中随机选择部分数据，并划分成support_set和test_set：

def create_support_set_and_test_dataset(chosen_classes: int = 5,     num_shot: int = 2,     num_test: int = 5):    """    从paddle.vision.datasets.Flowers数据集的测试集中选取support set和测试数据    :param chosen_classes: 选取的类别数    :param num_shot: Support set中每个类别选取多少个样本    :param num_test: 测试数据中每个类别选取多少个样本    :return:    """    trans = transformers.Compose([ transformers.Resize((224, 224)),  # 将图片大小调整成224x224 transformers.Transpose((2, 0, 1))  # 将图片数据调整成channel_first    ])    flowers_data = paddle.vision.datasets.Flowers(mode='test', transform=trans)    # 筛选出一个小数据集，对整个数据集进行处理速度太慢    # 之所以不随机选择类别，是因为从flowers_data读取数据非常慢    print('正在读取类别预测模型训练数据集……')    mini_flowers_data = []    for flower_data in flowers_data: if flower_data[1] <= chosen_classes:     mini_flowers_data.append(flower_data) else:     break    print('数据读取完毕！')    # 将图片按类别分类    class_idx_to_images = defaultdict(list)    for image, label in mini_flowers_data: class_idx_to_images[label.tolist()[0]].append(image / 255.)    # 检测选择的各类别中时候样本数均比num_shot+num_test大    # 之所以不随机选择类别，是因为从flowers_data读取数据非常慢    for key in class_idx_to_images.keys(): assert num_shot + num_test < len(class_idx_to_images[key]), \     'num_shot和num_test值过大，检测到某类数据图片数为：{}'.format(len(class_idx_to_images[key]))    support_set = defaultdict(list)    test_set = defaultdict(list)    for class_id in class_idx_to_images.keys(): total_samples = random.sample(class_idx_to_images[class_id], num_shot + num_test) support_set[class_id] = total_samples[:num_shot] test_set[class_id] = total_samples[-num_test:]    return support_set, test_set

3.2 构建小样本分类问题预测网络

定义继承自paddle.nn.Layer的FSCPredictor类，自定义小样本学习(Few-Shot Learning)（二）一文所述 $S o f t m a x$ 分类器参数 $W$ 和 $b$ ，并使用A Good Initialization技巧初始化参数矩阵 $W$ 。在forward函数中使用Cosine Similarity+Softmax Classifier计算 $w_i$ 和 $q$ 的Cosine Similarity：

class FSCPredictor(paddle.nn.Layer):    """小样本分类问题预测网络    Args: embedding_net: 将图片映射成特征向量的网络 support_set: 小样本分类问题的Support Set    """    def __init__(self, embedding_net, support_set): super(FSCPredictor, self).__init__() self.embedding_net = embedding_net self.embedding_net.eval() # 将Support Set中各类别图片全部转换成特征向量，并求平均，得到M矩阵 matrix_m = [] for class_id in support_set.keys():     one_class = paddle.to_tensor(support_set[class_id])     class_embs = embedding_net(one_class)     this_emb = paddle.mean(class_embs, axis=0)     matrix_m.append(this_emb.numpy()) matrix_m = np.array(matrix_m, dtype='float32') # 创建Softmax分类器中的参数W和b softmax_classifier_w = self.create_parameter(shape=list(matrix_m.shape),    default_initializer=paddle.nn.initializer.Assign(matrix_m)) softmax_classifier_b = self.create_parameter(shape=[matrix_m.shape[0]], is_bias=True) self.add_parameter('softmax_classifier_w', softmax_classifier_w) self.add_parameter('softmax_classifier_b', softmax_classifier_b)    def forward(self, x): x = self.embedding_net(x) # 根据fine tuning技巧，计算w和b的余弦相似度，需要将self.softmax_classifier_w进行归一化 normed_w = self.softmax_classifier_w / paddle.norm(self.softmax_classifier_w, axis=1, keepdim=True) normed_w = paddle.transpose(normed_w, [1, 0]) # x不需要进行归一化，因为embedding_net输出已经进行归一化了 x = paddle.matmul(x, normed_w) x = x + self.softmax_classifier_b return x

3.3 训练小样本分类问题预测网络

定义函数train_predictor，实例化FSCPredictor模型对象，定义Adam优化器，固定embedding_model模型参数，传入 $W$ 和 $b$ 作为优化参数。
使用预测网络输出与Support Set中图片类别标签的 $C r o s s E n t r o p y$ 作为损失函数，优化相关参数：

def train_predictor(embedding_model: paddle.nn.Layer,      support_set: defaultdict,      epochs: int,      learning_rate: float,      use_entropy_regularization: bool = True) -> None:    """    训练小样本分类问题预测器    :param embedding_model: 用于提取图片特征的网络    :param support_set: 小样本分类问题的Support set    :param epochs: 训练预测器的数据迭代轮数    :param learning_rate: 训练分类器的学习率    :param use_entropy_regularization: 是否使用entropy regularization    :return:    """    predictor = FSCPredictor(embedding_net=embedding_model, support_set=support_set)    predictor.train()    # 只学习Softmax分类器中的参数W和b    opt = paddle.optimizer.Adam(learning_rate=learning_rate,    parameters=[predictor.softmax_classifier_w, predictor.softmax_classifier_b])    for epoch in range(epochs): # 将整个Support set作为训练数据 input_datas = [] labels = [] for class_id in support_set:     input_datas += support_set[class_id]     labels += [class_id - 1] * len(support_set[class_id]) input_datas = paddle.to_tensor(input_datas) labels = paddle.to_tensor(labels, dtype='int64') logits = predictor(input_datas) loss = paddle.nn.functional.cross_entropy(logits, labels) if use_entropy_regularization:     # 计算Entropy Regularization，向量p的entropy等于自己和自己的cross_entropy     softmaxed_logits = paddle.nn.functional.softmax(logits)     entropy_regularization = paddle.nn.functional.cross_entropy(softmaxed_logits, softmaxed_logits,  use_softmax=False, soft_label=True)     loss += entropy_regularization if (epoch + 1) % 20 == 0:     print("epoch: {}, loss is: {}".format(epoch+1, loss.numpy())) loss.backward() opt.step() opt.clear_grad()    # 训练完保存参数    paddle.save(predictor.state_dict(), 'models/predictor.pdparams')

指定predictor_epochs = 200，predictor_learning_rate = 0.0005，调用create_support_set_and_test_dataset函数生成Support Set与训练集，实例化emb_model，并调用train_predictor函数训练预测网络：

    # 训练小样本分类问题预测器    predictor_epochs = 200    predictor_learning_rate = 0.0005    support_set, test_set = create_support_set_and_test_dataset()    # 加载训练好的特征提取网络    emb_model = FSCEmbNet()    emb_model_state_dict = paddle.load('models/emb_model.pdparams')    emb_model.set_state_dict(emb_model_state_dict)    # 开始训练    train_predictor(emb_model, support_set, epochs=predictor_epochs, learning_rate=predictor_learning_rate)

训练过程打印信息如下：

正在读取类别预测模型训练数据集……数据读取完毕！epoch: 20, loss is: [2.5626867]epoch: 40, loss is: [2.5274563]epoch: 60, loss is: [2.5005512]epoch: 80, loss is: [2.4801893]epoch: 100, loss is: [2.4648228]epoch: 120, loss is: [2.4532337]epoch: 140, loss is: [2.4444957]epoch: 160, loss is: [2.4379153]epoch: 180, loss is: [2.4329672]epoch: 200, loss is: [2.4292505]

由于使用了A Good Initialization技巧，采用Support Set中各类均值向量初始化训练参数 $W$ ，因此损失下降幅度不大。

4. 小样本分类模型测试

定义test函数，分别计算测试集中各类别准确率及在整个测试集上的小样本分类准确率：

def test(predictor: paddle.nn.Layer,  test_set: defaultdict) -> None:    """    测试小样本分类模型    :param predictor: 小样本分类问题预测器    :param test_set: 测试集    :return:     """    total_test_samples_num = 0    total_right_samples_num = 0    print('============================================================')    for class_id in test_set.keys(): this_class_data = test_set[class_id] this_class_samples_num = len(this_class_data) total_test_samples_num += this_class_samples_num this_class_data = paddle.to_tensor(this_class_data) predict_this_class = predictor(this_class_data) predict_label = paddle.argmax(predict_this_class, axis=1).numpy() this_right_samples_num = np.sum(predict_label == (class_id - 1)) total_right_samples_num += this_right_samples_num print('当前类别标签：{}，该类样本数：{}，预测正确数：{}，该类预测准确率：{:.2f}'.format(class_id,   this_class_samples_num, this_right_samples_num,   this_right_samples_num/this_class_samples_num))    print('测试样本总数：{}，预测正确总数：{}，预测准确率：{:.2f}'.format(total_test_samples_num, total_right_samples_num,    total_right_samples_num / total_test_samples_num))

实例化小样本分类预测模型，加载训练参数，并调用test函数：

if __name__ == '__main__':    # # 训练特征提取网络    # emb_epochs = 10    # emb_batch_size = 32    # emb_learning_rate = 0.0005    # chosen_classes = 25    # # 创建训练数据读取器    # emb_data_reader = create_emb_model_data_reader(chosen_classes=chosen_classes, batch_size=emb_batch_size)    # train_emb_model(epochs=emb_epochs, batch_size=emb_batch_size, learning_rate=emb_learning_rate,    #   data_reader=emb_data_reader)    #    # # 训练小样本分类问题预测器    # predictor_epochs = 200    # predictor_learning_rate = 0.0005    # support_set, test_set = create_support_set_and_test_dataset()    # # 加载训练好的特征提取网络    # emb_model = FSCEmbNet()    # emb_model_state_dict = paddle.load('models/emb_model.pdparams')    # emb_model.set_state_dict(emb_model_state_dict)    # # 开始训练    # train_predictor(emb_model, support_set, epochs=predictor_epochs, learning_rate=predictor_learning_rate)    # 判断测试集中图片类别    support_set, test_set = create_support_set_and_test_dataset()    predictor = FSCPredictor(FSCEmbNet(), support_set)    predictor_state_dict = paddle.load('models/predictor.pdparams')    predictor.set_state_dict(predictor_state_dict)    test(predictor, test_set)

打印测试信息如下：

当前类别标签：1，该类样本数：5，预测正确数：5，该类预测准确率：1.00当前类别标签：2，该类样本数：5，预测正确数：5，该类预测准确率：1.00当前类别标签：3，该类样本数：5，预测正确数：5，该类预测准确率：1.00当前类别标签：4，该类样本数：5，预测正确数：4，该类预测准确率：0.80当前类别标签：5，该类样本数：5，预测正确数：5，该类预测准确率：1.00测试样本总数：25，预测正确总数：24，预测准确率：0.96

本文使用飞桨（PaddlePaddle）基于paddle.vision.datasets.Flowers数据集实践了小样本学习(Few-Shot Learning)（二）一文讲解的小样本学习问题的Pretraining+Fine Tuning解法。
在本文小样本分类问题实践中，各模型训练超参数设置均没有做精细的调节，但是在测试集上仍然能够取得比较不错的测试准确率。
这种小样本分类问题的Pretraining+Fine Tuning解法解决方法非常简单，而且准确率基本与小样本学习领域最好的方法相当，具有非常重要的实践价值。

2021年9月16日更新：
本文所述小样本分类实践，虽然在Pretraining及Fine Tuning阶段分别使用了paddle.vision.datasets.Flowers数据训练及测试集部分数据，但是Support Set及测试样本类别均包含在预训练阶段数据集中。之所以没有严格按照小样本学习定义设置数据集，是因为从paddle.vision.datasets.Flowers读取数据较慢，故在生成Support Set和测试数据时没有跳过前 $k$ 类( $k$ 表示Pretraining阶段从数据集中选择的前 $k$ 个类别)。
不过可以使用一个方法生成训练数据集，Support Set与测试数据集。由于在进行该小样本分类实践时候，个人状态不是很好，因此程序设计上存在一点问题，读者可以自行优化相关程序。
从去年九月份，本人业余时间一直在学习机器学习理论知识并进行实践，几乎没有彻底的放松过。至【小样本学习(Few-Shot Learning)（三）】一文发布，今年八九月份，总有一种精神恍惚，身体疲倦之感。
目前正按计划进行休息和放松，休息调整完后预计会全身心投入智能围棋程序开发，预计会有较长一段时间停止更新。
智能围棋程序开发完成后，预计会在CSDN上发布专栏讲解智能围棋程序原理及实现方法，敬请期待~

5. 参考资料链接

https://www.paddlepaddle.org.cn/documentation/docs/zh/tutorial/cv_case/convnet_image_classification/convnet_image_classification.html