自监督学习：自编码器

Posted 2023-01-09 ~hello world~

1、自编码原理

  先考虑有监督学习中神经网络的功能：

𝑑_in 是输入的特征向量长度，𝑑_out 是网络输出的向量长度。对于分类问题，网络模型通过把长度为 𝑑_in 输入特征向量 𝒙 变换到长度为 𝑑_out 的输出向量 𝒐，这个过程可以看成是特征降维的过程，即把原始的高维输入向量 𝒙 变换到低维的变量 𝒐。特征降维 (Dimensionality Reduction) 在机器学习中有广泛的应用，最常见的降维算法有主成分分析法 (Principal components analysis，简称 PCA)，通过对协方差矩阵进行特征分解而得到数据的主要成分，但是 PCA 本质上是一种线性变换，提取特征的能力极为有限。
  那么能不能利用神经网络的强大非线性表达能力去学习到低维的数据表示呢？问题的关键在于，训练神经网络一般需要一个显式的标签数据(或监督信号)，但是无监督的数据没有额外的标注信息，只有数据 𝒙 本身。于是，我们尝试利用数据 𝒙 本身作为监督信号来指导网络的训练，即希望神经网络能够学习到映射 𝑓𝜃: 𝒙 → 𝒙 。我们把网络 𝑓𝜃 切分为两个部分，前面的子网络尝试学习映射关系: 𝑔𝜃₁: 𝒙 → 𝒛，后面的子网络尝试学习映射关系 ℎ𝜃₂: 𝒛 → 𝒙，如下图 所示：

把 𝑔𝜃₁ 看成一个数据编码 (Encode) 的过程，把高维度的输入 𝒙 编码成低维度的隐变量 𝒛，称为 Encoder 网络(编码器)；ℎ𝜃2看成数据解码(Decode)的过程，把
编码过后的输入 𝒛 解码为高维度的 𝒙，称为 Decoder 网络(解码器)。

  编码器和解码器共同完成了输入数据 𝒙 的编码和解码过程，我们把整个网络模型 𝑓𝜃 叫做自动编码器(Auto-Encoder)，简称自编码器。如果使用深层神经网络来参数化 𝑔𝜃₁ 和 ℎ𝜃₂ 函数，则称为深度自编码器(Deep Auto-encoder)，如下图 所示：

  自编码器能够将输入变换到隐藏向量 𝒛，并通过解码器重建(Reconstruct，或恢复)出 𝒙 。我们希望解码器的输出能够完美地或者近似恢复出原来的输入，即$\overline{x}$ ≈ 𝒙，那么，自编码器的优化目标可以写成：

其中 dist(𝒙, $\overline{x}$) 表示 𝒙 和 $\overline{x}$ 的距离度量，称为重建误差函数。最常见的度量方法有欧氏距离 (Euclidean distance) 的平方，计算方法如下：

它和均方误差原理上是等价的。自编码器网络和普通的神经网络并没有本质的区别，只不过训练的监督信号由标签 𝒚 变成了自身 𝒙。借助于深层神经网络的非线性特征提取能力，自编码器可以获得良好的数据表示，相对于 PCA 等线性方法，自编码器性能更加优秀，甚至可以更加完美的恢复出输入𝒙。

2、Fashion MNIST 图片重建实战

2.1 Fashion MNIST 数据集

  在 TensorFlow 中，加载 Fashion MNIST 数据集同样非常方便，利用 keras.datasets.fashion_mnist.load_data() 函数即可在线下载、管理和加载。代码如下：

import keras
import numpy as np
import tensorflow as tf 
from tensorflow.keras import layers

# 加载 Fashion MNIST 图片数据集
(x_train, y_train), (x_test, y_test) = keras.datasets.fashion_mnist.load_data()

# 归一化
x_train, x_test = x_train.astype(np.float32) / 255., x_test.astype(np.float32) / 255.

# 只需要通过图片数据即可构建数据集对象，不需要标签
batchsz = 100
train_db = tf.data.Dataset.from_tensor_slices(x_train)
train_db = train_db.shuffle(batchsz * 5).batch(batchsz)

# 构建测试集对象
test_db = tf.data.Dataset.from_tensor_slices(x_test)
test_db = test_db.batch(batchsz)

2.2 编码器

  利用编码器可以将输入图片 𝒙 ∈ 𝑅⁷⁸⁴ 降维到较低维度的隐藏向量：h ∈ 𝑅²⁰，并基于隐藏向量 h 利用解码器重建图片，自编码器模型如下图所示。

编码器由 3 层全连接层网络组成，输出节点数分别为 256、128、20，解码器同样由 3 层全连接网络组成，输出节点数分别为 128、256、784。
  编码器子网络的实现：利用 3 层的神经网络将长度为 784 的图片向量数据依次降维到 256、128，最后降维到 h_dim 维度，每层使用 ReLU 激活函数，最后一层不使用激活函数。代码如下：

# 创建 Encoders 网络，实现在自编码器类的初始化函数中
self.encoder = Sequential([
    layers.Dense(256, activation=tf.nn.relu),
    layers.Dense(128, activation=tf.nn.relu),
    layers.Dense(20)
])

2.3 解码器

  解码器子网络的创建：基于隐藏向量 h_dim 依次升维到 128、256、784 长度，除最后一层，激活函数使用 ReLU 函数。解码器的输出为 784 长度的向量，代表了打平后的28 × 28大小图片，通过 Reshape 操作即可恢复为图片矩阵。代码如下：

# 创建 Decoders 网络
self.decoder = Sequential([
    layers.Dense(128, activation=tf.nn.relu),
    layers.Dense(256, activation=tf.nn.relu),
    layers.Dense(784)
])

2.4 网络训练

  自编码器的训练过程与分类器的基本一致，通过误差函数计算出重建向量 𝒙 与原始输 $\overline{x}$ 入向量𝒙之间的距离，再利用 TensorFlow 的自动求导机制同时求出 encoder 和 decoder 的梯度，循环更新即可。
首先创建自编码器实例和优化器，并设置合适的学习率。

# 指定输入大小
model.build(input_shape=(4, 784))
# 打印网络信息
model.summary()
# 创建优化器，并设置学习率
optimizer = optimizers.Adam(lr=lr)

  本例固定训练 100 个 Epoch，每次通过前向计算获得重建图片向量，并利用tf.nn.sigmoid_cross_entropy_with_logits 损失函数计算重建图片与原始图片直接的误差，实际上利用 MSE 误差函数也是可行的。代码如下：

for epoch in range(100): # 训练 100 个 Epoch
    for step, x in enumerate(train_db): # 遍历训练集
        # 打平，[b, 28, 28] => [b, 784]
        x = tf.reshape(x, [-1, 784])
        # 构建梯度记录器
        with tf.GradientTape() as tape:
            # 前向计算获得重建的图片
            x_rec_logits = model(x)
            # 计算重建图片与输入之间的损失函数
            rec_loss = tf.nn.sigmoid_cross_entropy_with_logits(labels=x, logits=x_rec_logits)
            # 计算均值
            rec_loss = tf.reduce_mean(rec_loss)
        # 自动求导，包含了 2 个子网络的梯度
        grads = tape.gradient(rec_loss, model.trainable_variables)
        # 自动更新，同时更新 2 个子网络
        optimizer.apply_gradients(zip(grads, model.trainable_variables))
        
        if step % 100 ==0:
            # 间隔性打印训练误差
            print(epoch, step, float(rec_loss))

2.5 图片重建

  自编码器的模型性能一般不好量化评价，因此一般需要根据具体问题来讨论自编码器的学习效果，比如对于图片重建，一般依赖于人工主观评价图片生成的质量，或利用某些图片逼真度计算方法(如 Inception Score 和Frechet Inception Distance)来辅助评估。
  为了测试图片重建效果，我们把数据集切分为训练集与测试集，其中测试集不参与训练。我们从测试集中随机采样测试图片𝒙 ∈ 𝔻test，经过自编码器计算得到重建后的图片，然后将真实图片与重建图片保存为图片阵列，并可视化，方便比对。代码如下：

# 重建图片，从测试集采样一批图片
x = next(iter(test_db))
logits = model(tf.reshape(x, [-1, 784])) # 打平并送入自编码器
x_hat = tf.sigmoid(logits) # 将输出转换为像素值，使用 sigmoid 函数
# 恢复为 28x28,[b, 784] => [b, 28, 28]
x_hat = tf.reshape(x_hat, [-1, 28, 28])
# 输入的前 50 张+重建的前 50 张图片合并，[b, 28, 28] => [2b, 28, 28]
x_concat = tf.concat([x[:50], x_hat[:50]], axis=0)
x_concat = x_concat.numpy() * 255. # 恢复为 0~255 范围
x_concat = x_concat.astype(np.uint8) # 转换为整型
save_images(x_concat, 'ae_images/rec_epoch_%d.png'%epoch) # 保存图片

2.6 完整代码

import keras
import numpy as np
import tensorflow as tf 
from matplotlib import pyplot as plt
from PIL import Image
from tensorflow.keras import layers
from tensorflow.keras import losses,Sequential,optimizers,layers,datasets

# 加载 Fashion MNIST 图片数据集
(x_train, y_train), (x_test, y_test) = keras.datasets.fashion_mnist.load_data()

# 归一化
x_train, x_test = x_train.astype(np.float32) / 255., x_test.astype(np.float32) / 255.

# 只需要通过图片数据即可构建数据集对象，不需要标签
batchsz = 100
train_db = tf.data.Dataset.from_tensor_slices(x_train)
train_db = train_db.shuffle(batchsz * 5).batch(batchsz)

# 构建测试集对象
test_db = tf.data.Dataset.from_tensor_slices(x_test)
test_db = test_db.batch(batchsz)

class AE(keras.Model):
    def __init__(self):
        super(AE, self).__init__()
 
        # 创建Encoders网络
        self.encoder = keras.Sequential([
            keras.layers.Dense(256, activation=tf.nn.relu), # 参数量784*256+256
            keras.layers.Dense(128, activation=tf.nn.relu), # 参数量256*128+128
            keras.layers.Dense(20)                          # 参数量128*20+20
        ])
 
        # 创建Decoders网络
        self.decoder = keras.Sequential([
            keras.layers.Dense(128, activation=tf.nn.relu), # 参数量20*128+128
            keras.layers.Dense(256, activation=tf.nn.relu), # 参数量128*256+256
            keras.layers.Dense(784)                         # 参数量256*784+784
        ])
 
    def call(self, inputs, training=None):
        # 前向传播
        # 编码获取隐藏向量h
        h = self.encoder(inputs)
        # 解码获取重建图片
        out = self.decoder(h)
        return out

model = AE()
model.build(input_shape=(4,784))
model.summary()
 
optimizer = keras.optimizers.Adam(learning_rate=1e-3)

for epoch in range(100): 
    for step, x in enumerate(train_db):
        x = tf.reshape(x,[-1,784])
        with tf.GradientTape() as tape:
            xx = model(x)
            loss = tf.nn.sigmoid_cross_entropy_with_logits(labels=x, logits=xx)
            loss = tf.reduce_mean(loss)

        grads = tape.gradient(loss, model.trainable_variables)
        optimizer.apply_gradients(zip(grads, model.trainable_variables))

        if step%100 == 0:
            print(step, 'loss: ', float(loss))

#  save_images 函数负责将多张图片合并并保存为一张大图，这部分代码使用 PIL 图片库完成图片阵列逻辑
def save_images(imgs, name):
    new_im = Image.new('L', (280, 280))
    index = 0
    for i in range(0, 280, 28): # 10 行图片阵列
        for j in range(0, 280, 28): # 10 列图片阵列
            im = imgs[index]
            im = Image.fromarray(im, mode='L')
            new_im.paste(im, (i, j)) # 写入对应位置
            index += 1
    # 保存图片阵列
    new_im.save(name)
 
 
x = next(iter(test_db))
logits = model(tf.reshape(x, [-1,784]))
x_hat = tf.sigmoid(logits)
x_hat = tf.reshape(x_hat, [-1,28,28])
 
x_concat = tf.concat([x[:50], x_hat[:50]], axis=0)
x_concat = x_concat.numpy() * 255. 
x_concat = x_concat.astype(np.uint8)
save_images(x_concat, 'ae_images/rec_epoch_%d.png'%epoch)