元学习(三)--度量学习(原型神经网络)

原型神经网络

我们了解很多深度学习和机器学习的算法，但是有一个痛点，我们要求数据量极大，因为参数量极大，这个时候怎么办？我们还是期望能够使用较少的数据学到更大的信息量，这才是我们的目标，本章节中讲到的就是这类问题的学习方法，我们称之为小样本学习。

在本文的最后有一篇讲Prototypical Networks 的论文，大家如果还是不明白可以阅读这个论文，我会尽最大努力将Prototypical Networks的核心思想讲清楚。

C-way K-shot

形式化来说，few-shot 的训练集中包含了很多的类别，每个类别中有多个样本。在训练阶段，会在训练集中随机抽取 $C$ 个类别，每个类别 $K$ 个样本（总共 $C \times K$ 个数据），构建一个 meta-task，作为模型的支撑集（support-set）输入；再从这 $C$ 个类中剩余的数据中抽取一批（batch）样本作为模型的预测对象（batch set）。即要求模型从 $C \times K$ 个数据中学会如何区分这 C 个类别这样的任务被称为 C-way K-shot. 问题.

训练过程中，每次训练（episode）都会采样得到不同 meta-task，所以总体来看，训练包含了不同的类别组合，这种机制使得模型学会不同 meta-task 中的共性部分，比如如何提取重要特征及比较样本相似等，忘掉 meta-task 中 task相关部分。通过这种学习机制学到的模型，在面对新的未见过的 meta-task 时，也能较好地进行分类。 few-shot 一般指的是 k 不超过 20。

原型学习

假设我们有一个函数 $f(x)=X$ , $X \in R^M$ ,其中 $f$ 这个函数的目的是将所有的数据进行一个标准化，例如将任意维度的数据归一到 $M*1$ 的维度上，至于怎么做，其实就是一个embedded 的过程，现在我们暂且不说这个方法。

数据描述

首先我们将数据分为两份，我们在每个类别下抽取一部分数据作为支撑集，然后同类别的剩下的数据就叫做查询集。

我们有 $N$ 个带有标签的数据集合， $S={(x_1,y_1),..,(x_n,y_n)}$ ,带有标签的数据，这个集合可能很小。这个数据集合中具有 $K$ 类。

经过 $f(x)$ ,进行embedded以后输出的结果为 $c_k$ .根据每个标签可以计算每个类别的期望。当然这里说的都还是在支撑集中。

c_k=\frac{1}{|S_k|} \sum_i^N f(x_i)

(x_i,y_i) \in S_k

计算距离函数，一般我们使用欧几里得距离作为这个距离函数 $d(a,b)$ .我们这时候定义在查询集上的分类概率。 $x$ 是查询集上的一条待检测数据。

p(y=k|x)=\frac{exp(-d(f(x),c_k))}{\sum_k exp(-d(f(x), c_k))}

我们的损失函数也可以定义为

J=-log^{p(y=k|x)}

def conv_block(inputs, out_channels, name='conv'):
    with tf.variable_scope(name):
        conv = tf.layers.conv2d(inputs, out_channels, kernel_size=3, padding='SAME')
        conv = tf.contrib.layers.batch_norm(conv, updates_collections=None, decay=0.99, scale=True, center=True)
        conv = tf.nn.relu(conv)
        conv = tf.contrib.layers.max_pool2d(conv, 2)
        return conv
def encoder(x, h_dim, z_dim, reuse=False):
    with tf.variable_scope('encoder', reuse=reuse):
        net = conv_block(x, h_dim, name='conv_1')
        net = conv_block(net, h_dim, name='conv_2')
        net = conv_block(net, h_dim, name='conv_3')
        net = conv_block(net, z_dim, name='conv_4')
        net = tf.contrib.layers.flatten(net)
        return net

def euclidean_distance(a, b):
    # a.shape = N x D
    # b.shape = M x D
    N, D = tf.shape(a)[0], tf.shape(a)[1]
    M = tf.shape(b)[0]
    a = tf.tile(tf.expand_dims(a, axis=1), (1, M, 1))
    b = tf.tile(tf.expand_dims(b, axis=0), (N, 1, 1))
    return tf.reduce_mean(tf.square(a - b), axis=2)
x = tf.placeholder(tf.float32, [None, None, im_height, im_width, channels])
q = tf.placeholder(tf.float32, [None, None, im_height, im_width, channels])
x_shape = tf.shape(x)
q_shape = tf.shape(q)
num_classes, num_support = x_shape[0], x_shape[1]
num_queries = q_shape[1]
y = tf.placeholder(tf.int64, [None, None])
y_one_hot = tf.one_hot(y, depth=num_classes)
emb_x = encoder(tf.reshape(x, [num_classes * num_support, im_height, im_width, channels]), h_dim, z_dim)
emb_dim = tf.shape(emb_x)[-1]
emb_x = tf.reduce_mean(tf.reshape(emb_x, [num_classes, num_support, emb_dim]), axis=1)
emb_q = encoder(tf.reshape(q, [num_classes * num_queries, im_height, im_width, channels]), h_dim, z_dim, reuse=True)
dists = euclidean_distance(emb_q, emb_x) 【1】
log_p_y = tf.reshape(tf.nn.log_softmax(-dists), [num_classes, num_queries, -1])
ce_loss = -tf.reduce_mean(tf.reshape(tf.reduce_sum(tf.multiply(y_one_hot, log_p_y), axis=-1), [-1]))【2】
acc = tf.reduce_mean(tf.to_float(tf.equal(tf.argmax(log_p_y, axis=-1), y)))

for ep in range(n_epochs):【3】
    for epi in range(n_episodes):
        epi_classes = np.random.permutation(n_classes)[:n_way]
        support = np.zeros([n_way, n_shot, im_height, im_width], dtype=np.float32)
        query = np.zeros([n_way, n_query, im_height, im_width], dtype=np.float32)
        for i, epi_cls in enumerate(epi_classes):
            selected = np.random.permutation(n_examples)[:n_shot + n_query]
            support[i] = train_dataset[epi_cls, selected[:n_shot]]
            query[i] = train_dataset[epi_cls, selected[n_shot:]]
        support = np.expand_dims(support, axis=-1)
        query = np.expand_dims(query, axis=-1)
        labels = np.tile(np.arange(n_way)[:, np.newaxis], (1, n_query)).astype(np.uint8)
        _, ls, ac = sess.run([train_op, ce_loss, acc], feed_dict={x: support, q: query, y:labels})
        if (epi+1) % 50 == 0:
            print('[epoch {}/{}, episode {}/{}] => loss: {:.5f}, acc: {:.5f}'.format(ep+1, n_epochs, epi+1, n_episodes, ls, ac))

上面的代码是一份简易的代码，伪代码在下面的文章中是有的，它的训练过程是用支撑集中数据计算均值，使用查询集合中的数据和支撑集合的数据完成距离的衡量，在【1】处就是制定了距离的函数，在【2】处就还原了文章中的损失函数。【3】处就开始遍历片段，随机选取数据，然后放到网络中进行学习和拟合。

训练过程

每次 iteration 从 964 个类中随机的选择 20 个类，从每个类中的 20 个样本采样 5 个作为 support set，5 个作为 query set。（选择的这个数目可以自行改变）。

直观理解

我们暂且不讨论数学上的解释，你可能会有一个比较大的疑问是，为什么这样的方式能够在小样本中学习到东西呢？

其实我是这样理解的，如果你做一个二分类，最后模型输出一个概率分布，我们也可以理解成是一种embedded，但是我们用更高维度的向量来表达，就变成了不是非A即B的分类问题，而是一个表示学习，这样泛化性能做的更好，因为我们仅仅是将同类的实体使用了类似的表达而已，并没有直接指定一个类别，这样我们更加容易学习到样本中间结果，而非最终结果。

还有比较重要的一点是，大家发现没有，我们embedded的结果不是用我们支撑集作为拟合，而是选择查询集的数据来度量，这样就从一定程度上避免了过拟合的风险。

如下内容仅供学习，请勿传播
Prototypical Networks for Few-shot Learning

# general AI

支付宝捐赠