文章目录
- 论文介绍
- 步骤具体讲解
- 自定义矢量量化层
- 获取最近距离的码字的索引计算推导
- 损失函数
- 相关参考
- 矢量量化层的代码实现
- 完整代码实现
论文介绍
- 不同于变分编码器和自动编码器，vq-vae中的latent space是离散的，并不是连续的，这能够避免后验塌陷。除此之外，vq-vae中的先验概率分布并不是静态的，是可以通过训练进行学习的。
- 编码器是使用卷积网络生成对应的特征，然后计算欧式距离，将是映射到离散的码本中
- 解码器是根据最近的码字索引抽取对应的码本中的编码，并且使用这个码进行数据生成。这个就不用与自编码器中的latent space是一个连续的空间。
- 前向很简单，但是反向传播就很苦难，因为在正向过程中，获取最近码字的过程，是不可导的。可以使用straight-through estimator因为编码器和解码器的大小是相同的。通过复制梯度，调整对应的编码向量的方向，实现编码器的输出不断向最近的编码靠近。
- 这里是实现将ze编码器的输出，找到最近的一个码本中的码字进行映射，然后保留对应的索引。
- 因为对于等式2是没有真实梯度的，而且该等式是不可导的，所以我们通过straght-through estimator近似梯度，简单来说就是使用解码器的输入梯度，复制给编码器的输出梯度。
- 因为编码器的输出和解码器的输入是具有相同的尺寸的，所以解码器的梯度信息对于编码器而言也是有用的，可以让编码器了解如何改变输出，才能降低重建损失。
- 上述图片为损失函数的构成，分别是三个部分
- 第一个部分是重建损失函数，保证重建之后和原来的之间的差异，解码器和编码器都会优化 这部分损失函数。
- 第二部分是优化码本中的编码embedding ，是码本中码字不断向输入进行靠拢。学习embeddings.
- 第三部分是优化编码器输出，向码本中的码进行逼近的损失函数。调整编码器的输出。
步骤具体讲解
自定义矢量量化层
* 左边矩阵是编码器的输出，进行reshape之后的一个二维矩阵，每一行表示一个特征编码，然后在右边码本中找到欧式距离最近的一列特征进行替换映射。替换如下，每一行的f特征，换成了码本中对应E码字，这是最终结果，中间省略了很多计算过程。
具体计算过程如下
- 首先计算一行特征到码字中各个码字的距离，形成一个新的向量，每一个位置，表示对应行的特征到对应列的向量之间的码字距离。
- 选取每一行最小值，并记录对应的索引，生成一个新的矩阵，记录的是每一行最小值对应的索引。
- 根据索引，生成对应的独热编码矩阵，根据索引形成对应的矩阵。原来获取最近的索引的时候，就是在所有的码本中找最近的码进行计算的，所以这里生成对应同宽度的独热编码矩阵。和原来的码本进行相乘，获得结果就是对应最近的码字
- 直接和对应的码本进行相乘，获取最近的码字。
获取最近距离的码字的索引计算推导
- 这里是距离计算的具体方式，对于两个1x3的向量，对于更加复杂维度，计算的方式也相同，这里结合具体的代码来看一下
- 计算两个矩阵的相乘，就是上图中的xy
similarity = tf.matmul(falttened_input,self.embeddings)
- 计算两个两个矩阵的平方和，对于第一个矩阵而言，每一列是一个特征向量，对于第二个矩阵而言，每一行是个向量，是计算这两个向量的平方和。
distances = (
tf.reduce_sum(falttened_input**2,axis=1,keepdims=True)
+ tf.reduce_sum(self.embeddings ** 2,axis=0)
- 2 * similarity
)
- 选出最近的索引
# 获取最小距离的索引，最终返回的是每一个样本在对应行中最小值，返回的是一个（Batch_size*H*W,1）的数组，
# 每一行的特征对应最小的码的索引
encoding_indices = tf.argmin(distances,axis=1)
- 具体的矩阵的形状的变化，见上一个个章节的内容。
损失函数
-
损失函数主要由里那个部分构成，分别是量化损失（commitment loss）和码本损失（codebook loss）
-
量化损失
- 衡量量化之前的原始输入和量化之后的输出的差异
- 通过计算原始输入和停止梯度的量化输出之间的均方差来衡量
- 使模型学习将输入的数据映射到码本中最接近的向量，是控制输入x的，所以不需要对码本进行梯度调整。
-
码本损失
- 衡量量化输出和原始输入之间的反向重构误差
- 通过计算量化输出和停止梯度的原始输入之间的方差衡量。
- 使模型学习将量化输出映射回原始输入空间，是控制码本的梯度，所以不需要改变输入x的参数
-
最终的损失函数‘
- 两部分的损失进行加权想家，权重有 β \beta β进行控制
- 使用stop_gradient是为了控制损失函数对于解码器输出的影响，保证解码器能够正确反量化码本向量，不受梯度的干扰。
-
具体实现代码，说实话，这部分理解的不够透彻
# 计算矢量化的损失，并且将之加到当前层上，
# reduce_mean:计算张量指定维度上的平均值
# tf.stop_gradient:不计算输入变量的梯度
commitment_loss = tf.reduce_mean((tf.stop_gradient(quantized) - x ) ** 2)
codebook_loss = tf.reduce_mean((quantized - tf.stop_gradient(x) ) ** 2)
self.add_loss(self.beta * commitment_loss + codebook_loss)
- 最终的输出，总的前向传播中，码本是离散的，无法求导，需要直接将梯度传到输入续
# straight-through estimator:直接将参数的梯度作为对应浮点型参数的梯度
# 直接使用激活函数之后的梯度，代替之前的梯度
quantized = x + tf.stop_gradient(quantized - x )
相关参考
Straight——through Estimator的解释
stop_gradient的解释
矢量量化层的代码实现
from tensorflow.keras import Model
# 一般进行版本更新都是改变的包的导向，或者改变包的方法名
from tensorflow.keras.layers import Input,Conv2D,ReLU,BatchNormalization,Flatten,Dense,\
Reshape,Conv2DTranspose,Activation,Lambda
from tensorflow.keras import layers,metrics
# 引入backend,这个用来自己定义层，将一些函数定义成特定的层
from tensorflow.keras import backend as K
# 导入numpy,对三维数据进行操作
import numpy as np
# 导入对应优化器，注意，这里已经改变了包的地址
from tensorflow.keras.optimizers.legacy import Adam
# 导入损失函数
from tensorflow.keras.losses import MeanSquaredError
# 导入系统模块
import os
# 序列加载模块
import pickle
import tensorflow as tf
tf.compat.v1.disable_eager_execution()
# 自定义层,自定义矢量量化层
# 这是在编码器和解码器之间自定义的一个层。
# 输入为编码器的输出，形状是（batch_size, height, width,num_filters）
# 矢量量化器将会输入进行flatten,仅仅保证过滤核filter的尺寸不变，
# 最终的尺寸是batch_size * height * width,num_filters)
# 作用：将滤波器的总数当作是潜在embedding的大小
# 然后embedding 表格是被初始化，用来学习码本code book
# 我们我们通过计算展平之后的编码器输出和码本码字的欧式距离
# 我们选择距离最小的码字，然后应用独热编码实现量化效果，借此实现了码字和相应距离最近的编码器输出作为关联
#
# 因为矢量量化是不可导的，所以直接用解码器的梯度作为编码器的梯度
#
#
class VectorQuantizer(layers.Layer):
def __init__(self,num_embeddings,embedding_dim,beta=0.25,**kwargs):
'''
确定矢量量化的码字尺寸，
:param num_embeddings:batch_size * height * width
:param embedding_dim:num_filters
:param beta:beta参数设置为0.25
:param kwargs:
'''
super().__init__(**kwargs)
self.embedding_dim = embedding_dim
self.num_embedding = num_embeddings
self.beta = beta
# 初始化将要量化的embedding
# 生成具有均匀分布的张量的初始化器
w_init = tf.random_uniform_initializer()
# 定义tensorflow中的图片变量并且命名为embedding_vqvae
self.embeddings = tf.Variable(
initial_value=w_init(
shape = (self.embedding_dim,self.num_embedding),
dtype = "float32"
),
trainable=True,
name = "embeddings_vqvae",
)
def cal(self,x):
"""
步骤：
1、将input进行flatten,仅仅保留embedding_dim这一个维度
2、使用独热编码对embedding进行量化
3、将量化的值，还原回原来的输入形状
4、计算量化层的损失函数
5、将解码器的梯度传到编码器
:param x: 当前层的输入
:return: 一个展平之后的embeddings,并且保存了对应的filter_num
"""
# 1、将input进行flatten,仅仅保留embedding_dim这一个维度，（Batch_size, H,W,filter_num）
input_shape = tf.shape(x)
# 将原来的处理过后的特征改变为特定的形状，最后一个维度指定，前两个维度自动调整,(Batch_size*H*W,filter_num)
flattened = tf.reshape(x,[-1,self.embedding_dim])
# 2、使用独热编码对embedding进行量化,获取距离最近的
encoding_indices = self.get_code_indices(flattened)
# 这里是生成对应的独热编码，位数是embedding的维度，索引是embedding的编号
# 创建一个独热编码矩阵形状为(Batch_size*H*W,self.num_embeddings),根据索引将对应的列置为1
encodings = tf.one_hot(encoding_indices,self.num_embedding)
# 矩阵连乘，除了对应的独热编码对应的embedding可以保留，其余的都归零
# matmul矩阵相乘，将之原来的矩阵(Batch_size*H*W,self.num_embeddings)，
# 映射为一个(Batch_size*H*W,self.embedding_dim)的矩阵
quantized = tf.matmul(encodings,self.embeddings,transpose_b=True)
# 将经过独热编码处理之后的embedding还原成原始大小的数据
quantized = tf.reshape(quantized,input_shape)
# 计算矢量化的损失，并且将之加到当前层上，
# reduce_mean:计算张量指定维度上的平均值
# tf.stop_gradient:不计算输入变量的梯度
commitment_loss = tf.reduce_mean((tf.stop_gradient(quantized) - x ) ** 2)
codebook_loss = tf.reduce_mean((quantized - tf.stop_gradient(x) ) ** 2)
self.add_loss(self.beta * commitment_loss + codebook_loss)
# straight-through estimator:直接将参数的梯度作为对应浮点型参数的梯度
# 直接使用激活函数之后的梯度，代替之前的梯度
quantized = x + tf.stop_gradient(quantized - x )
return quantized
def get_code_indices(self,falttened_input):
"""
计算输入和不同码字的欧式距离，判定对应的值
:param falttened_input:平展之后的输入
:return:
"""
# (Batch_size*H*W,filter_num) * （self.embedding_dim,self.num_embedding）
similarity = tf.matmul(falttened_input,self.embeddings)
# 这部分就是xx+yy-2xy
distances = (
tf.reduce_sum(falttened_input**2,axis=1,keepdims=True)
+ tf.reduce_sum(self.embeddings ** 2,axis=0)
- 2 * similarity
)
# 这里的最终结果是（Batch_size*H*W，num_embeddings）其中每一个元素为每一个输入样本和码本之间的欧式距离
# 获取最小距离的索引，最终返回的是每一个样本在对应行中最小值，返回的是一个（Batch_size*H*W,1）的数组，
# 每一行的特征对应最小的码的索引
encoding_indices = tf.argmin(distances,axis=1)
return encoding_indices
完整代码实现
from tensorflow.keras import Model
# 一般进行版本更新都是改变的包的导向，或者改变包的方法名
from tensorflow.keras.layers import Input,Conv2D,ReLU,BatchNormalization,Flatten,Dense,\
Reshape,Conv2DTranspose,Activation,Lambda
from tensorflow.keras import layers,metrics
# 引入backend,这个用来自己定义层，将一些函数定义成特定的层
from tensorflow.keras import backend as K
# 导入numpy,对三维数据进行操作
import numpy as np
# 导入对应优化器，注意，这里已经改变了包的地址
from tensorflow.keras.optimizers.legacy import Adam
# 导入损失函数
from tensorflow.keras.losses import MeanSquaredError
# 导入系统模块
import os
# 序列加载模块
import pickle
import tensorflow as tf
tf.compat.v1.disable_eager_execution()
# 自定义层,自定义矢量量化层
# 这是在编码器和解码器之间自定义的一个层。
# 输入为编码器的输出，形状是（batch_size, height, width,num_filters）
# 矢量量化器将会输入进行flatten,仅仅保证过滤核filter的尺寸不变，
# 最终的尺寸是batch_size * height * width,num_filters)
# 作用：将滤波器的总数当作是潜在embedding的大小
# 然后embedding 表格是被初始化，用来学习码本code book
# 我们我们通过计算展平之后的编码器输出和码本码字的欧式距离
# 我们选择距离最小的码字，然后应用独热编码实现量化效果，借此实现了码字和相应距离最近的编码器输出作为关联
#
# 因为矢量量化是不可导的，所以直接用解码器的梯度作为编码器的梯度
#
#
class VectorQuantizer(layers.Layer):
def __init__(self,num_embeddings,embedding_dim,beta=0.25,**kwargs):
'''
确定矢量量化的码字尺寸，
:param num_embeddings:batch_size * height * width
:param embedding_dim:num_filters
:param beta:beta参数设置为0.25
:param kwargs:
'''
super().__init__(**kwargs)
self.embedding_dim = embedding_dim
self.num_embedding = num_embeddings
self.beta = beta
# 初始化将要量化的embedding
# 生成具有均匀分布的张量的初始化器
w_init = tf.random_uniform_initializer()
# 定义tensorflow中的图片变量并且命名为embedding_vqvae
self.embeddings = tf.Variable(
initial_value=w_init(
shape = (self.embedding_dim,self.num_embedding),
dtype = "float32"
),
trainable=True,
name = "embeddings_vqvae",
)
def cal(self,x):
"""
步骤：
1、将input进行flatten,仅仅保留embedding_dim这一个维度
2、使用独热编码对embedding进行量化
3、将量化的值，还原回原来的输入形状
4、计算量化层的损失函数
5、将解码器的梯度传到编码器
:param x: 当前层的输入
:return: 一个展平之后的embeddings,并且保存了对应的filter_num
"""
# 1、将input进行flatten,仅仅保留embedding_dim这一个维度，（Batch_size, H,W,filter_num）
input_shape = tf.shape(x)
# 将原来的处理过后的特征改变为特定的形状，最后一个维度指定，前两个维度自动调整,(Batch_size*H*W,filter_num)
flattened = tf.reshape(x,[-1,self.embedding_dim])
# 2、使用独热编码对embedding进行量化,获取距离最近的
encoding_indices = self.get_code_indices(flattened)
# 这里是生成对应的独热编码，位数是embedding的维度，索引是embedding的编号
# 创建一个独热编码矩阵形状为(Batch_size*H*W,self.num_embeddings),根据索引将对应的列置为1
encodings = tf.one_hot(encoding_indices,self.num_embedding)
# 矩阵连乘，除了对应的独热编码对应的embedding可以保留，其余的都归零
# matmul矩阵相乘，将之原来的矩阵(Batch_size*H*W,self.num_embeddings)，
# 映射为一个(Batch_size*H*W,self.embedding_dim)的矩阵
quantized = tf.matmul(encodings,self.embeddings,transpose_b=True)
# 将经过独热编码处理之后的embedding还原成原始大小的数据
quantized = tf.reshape(quantized,input_shape)
# 计算矢量化的损失，并且将之加到当前层上，
# reduce_mean:计算张量指定维度上的平均值
# tf.stop_gradient:不计算输入变量的梯度
commitment_loss = tf.reduce_mean((tf.stop_gradient(quantized) - x ) ** 2)
codebook_loss = tf.reduce_mean((quantized - tf.stop_gradient(x) ) ** 2)
self.add_loss(self.beta * commitment_loss + codebook_loss)
# straight-through estimator:直接将参数的梯度作为对应浮点型参数的梯度
# 直接使用激活函数之后的梯度，代替之前的梯度
quantized = x + tf.stop_gradient(quantized - x )
return quantized
def get_code_indices(self,falttened_input):
"""
计算输入和不同码字的欧式距离，判定对应的值
:param falttened_input:平展之后的输入
:return:
"""
# (Batch_size*H*W,filter_num) * （self.embedding_dim,self.num_embedding）
similarity = tf.matmul(falttened_input,self.embeddings)
# 这部分就是xx+yy-2xy
distances = (
tf.reduce_sum(falttened_input**2,axis=1,keepdims=True)
+ tf.reduce_sum(self.embeddings ** 2,axis=0)
- 2 * similarity
)
# 这里的最终结果是（Batch_size*H*W，num_embeddings）其中每一个元素为每一个输入样本和码本之间的欧式距离
# 获取最小距离的索引，最终返回的是每一个样本在对应行中最小值，返回的是一个（Batch_size*H*W,1）的数组，
# 每一行的特征对应最小的码的索引
encoding_indices = tf.argmin(distances,axis=1)
return encoding_indices
class VQVAE:
"""
Autoencoder: 变分自动编码器
这个是在原来的自动编码器上做了如下的修改
1、去除原来的bottleneck部分，换成量化层进行实现
2、更新损失函数，加入KL散度以及修改原来的MSE
"""
# 1、当前类别的构造函数，
def __init__(self,
input_shape,
conv_filters,
conv_kernels,
conv_strides,
latent_space_dim,
num_embeddings
):
# 2、 将所有的属性都赋值给对应实体属性
self.input_shape = input_shape # [28,28,1]这里是使用minst手写数据集进行测试的
self.conv_filters = conv_filters # [2,4,8]
self.conv_kernels = conv_kernels # [3,5,3]
self.conv_strides = conv_strides # [1,2,2]
self.latent_space_dim = latent_space_dim # 码本中单个码字的维度
self.num_embeddings = num_embeddings # 码本中码字的个数
# 3、这里的习惯就很好，将一个大模型拆解成两个小模型，编程的时候，只需要对应进行修改就行了
self.encoder = None
self.decoder = None
self.vq_layer = None
self.model = None
# 4、将部分变量声明为隐私变量,前置单下划线，私有变量
self._num_conv_layers = len(conv_filters)
# 5、设置build函数，这里是实例化类的时候进行调用
self._build()
# 9、后续添加的变量
self._shape_encoder = None
# 3.1 添加整体模型的输入,和encoder的输入是相同的
self._model_input = None
# 4.2 损失函数权重
self.reconstruction_weight = 0.5
# 这个方法在三个模块都是需要修改的
def summary(self):
""" 做测试，判定模型是否成功 """
self.encoder.summary()
self.decoder.summary()
self.model.summary()
# 3.2 增加编译函数
def compile(self,learning_rate = 0.0001):
""" 指定损失函数和优化器，并对模型进行优化 """
optimizer = Adam(learning_rate = learning_rate)
self.model.compile(
optimizer = optimizer,
loss = self._calculate_reconstruction_loss,
)
self.total_loss_tracker = metrics.Mean(name="total_loss")
self.reconstruction_loss_tracker = metrics.Mean(
name="reconstruction_loss"
)
self.vq_loss_tracker = metrics.Mean(name="vq_loss")
@property
def metrics(self):
return [
self.total_loss_tracker,
self.reconstruction_loss_tracker,
self.vq_loss_tracker,
]
# 3.3 增加训练函数
def train(self,x_train,batch_size,num_epochs):
self.model.fit(
x_train,
x_train,
batch_size = batch_size,
epochs = num_epochs,
shuffle = True
)
# 3.4 模型保存部分
def save(self,save_folder = "."):
""" 保存模型,需要创建文件，分别保存参数和权重"""
self._create_folder_if_not_exist(save_folder)
self._save_parameters(save_folder)
self._save_weights(save_folder)
# 3.5 模型加载部分,这部分要注意，是声明为类方法，不用实例化，直接可以调用
@classmethod
def load(cls,save_folder ="."):
""" 加载模型,包括模型的参数设置和模型的训练权重 """
parameters_path = os.path.join(save_folder,"parameters.pkl")
with open(parameters_path,"rb") as f:
parameters = pickle.load(f)
vae = VQVAE(*parameters)
weight_path = os.path.join(save_folder,"weights.h5")
vae.load_weights(weight_path)
return vae
def load_weights(self,weight_path):
self.model.load_weights(weight_path)
def reconstruct(self,image):
""" 重建图片，并返回生成之后的图片以及对应的特征空间 """
latent_space = self.encoder.predict(image)
reconstruct_image = self.decoder.predict(latent_space)
return reconstruct_image,latent_space
# 4.2 将两种损失函数进行综
# 4.2 损失函数重建
def _calculate_reconstruction_loss(self,y_target,y_predict):
""" 模型重建损失函数，加上了对应alpha """
error = y_predict - y_target
reconstruction_loss = K.mean(K.square(error),axis = [1,2,3]) # 注意，这里只需要返回除了第一个图片序号的后两个维度
return reconstruction_loss
# 3.4 分别实现上述方法
def _create_folder_if_not_exist(self,save_folder):
if not os.path.exists(save_folder):
os.makedirs(save_folder)
# 3.4 分别实现上述方法
def _save_parameters(self,save_folder):
""" 主要是保存模型对应的参数，包括每一层具体的设置 """
parameters = [
self.input_shape ,
self.conv_filters,
self.conv_kernels,
self.conv_strides,
self.latent_space_dim
]
save_path = os.path.join(save_folder,"parameters.pkl")
with open(save_path,"wb") as f:
pickle.dump(parameters,f)
# 3.4 实现save的子方法
def _save_weights(self,save_folder):
save_path = os.path.join(save_folder, "weights.h5")
self.model.save_weights(save_path)
# 6、具体实现相关的方法，这个是总的build函数，需要构建三个模块，分别是encoder、decoder和model
def _build(self):
self._build_encoder()
self._build_quantizer_layer()
self._build_decoder()
self._build_VAE()
def _build_quantizer_layer(self):
# 获取编码器的输出，并生成对应的中间层
# 感觉这里有问题，所有的层应该都是事先已经声明好的
vq_layer = VectorQuantizer(self.num_embeddings,self.latent_space_dim,name = "vector_quantizer")
# vq_output = vq_layer(quantizer_input)
self.vq_layer = vq_layer
# 7、从上到下，逐个子方法进行实现
def _build_encoder(self):
# 8、按照网络的层次，将模型串联起来，按照模块进行组装
encoder_input = self._add_encoder_input()
conv_layers = self._add_conv_layers(encoder_input)
# 调整通道维度，使之和特征空间相适应，
encoder_output = layers.Conv2D(self.latent_space_dim,1,padding="same")(conv_layers)
self._shape_encoder = encoder_output.shape
self._model_input = encoder_input
self.encoder = Model(encoder_input,encoder_output,name="encoder")
# 8、从上到下，按照顺序，逐个实现_build_encoder模块中所有方法
def _add_encoder_input(self):
return Input(shape = self.input_shape,name= "encoder_input")
# 8、从上到下，按照顺序，逐个实现_build_encoder模块中所有方法
def _add_conv_layers(self,encoder_input):
""" 在编码器中增加卷积模块 """
x = encoder_input
# 9、这部分是按照层的顺序逐渐叠加网络层
for layer_index in range(self._num_conv_layers):
# 尽量将自己的模块封装在别的人的模块上
x = self._add_conv_layer(layer_index,x)
return x
# 8、从里到外，完成对应的卷积模块
def _add_conv_layer(self,layer_index,x):
""" 增加卷积模块，每一部分构成如下，conv2d + relu + batch normalization """
layer_num = layer_index + 1
conv_layer = Conv2D(
filters = self.conv_filters[layer_index],
kernel_size = self.conv_kernels[layer_index],
strides = self.conv_strides[layer_index],
padding = "same",
name = f"encoder_conv_layer_{layer_num}"
)
x = conv_layer(x)
x = ReLU(name = f"encoder_relu_{layer_num}")(x)
x = BatchNormalization(name = f"encoder_bn_{layer_num}")(x)
return x
# 7、从上到下，逐个子方法进行实现
# 2.1 完成解码器的大部分框架
def _build_decoder(self):
""" 创建解码器，输入层、全连阶层、恢复成三维、进行反卷积、输出层 """
decoder_input = self._add_decoder_input()
conv_transpose_layers = self._add_conv_transpose_layers(decoder_input)
decoder_output = self._add_decoder_output(conv_transpose_layers)
self.decoder = Model(decoder_input,decoder_output,name = "decoder")
# 2.2 具体实现各个子函数,下述函数都是按照顺序完成并实现的
def _add_decoder_input(self):
""" 解码器的输入 """
return Input(shape = self._shape_encoder[1:],name = "decoder_input")
def _add_dense_layer(self,decoder_input):
""" 解码器的全连阶层，输出数据是二维的，这里并不知道怎么设置？？"""
# 这部分设置神经元的数量，和输出的维度而数量相同
num_neurons = np.prod(self._shape_before_bottleneck) # 将数据恢复原始的数据[1,2,4]=>8,现在是将8转成三维的数组
dense_layer = Dense(num_neurons,name = "decoder_dense_layer")(decoder_input)
return dense_layer
def _add_reshape_layer(self,dense_layer):
""" 增加对应的调整形状层，将全链接层的输出，恢复成三维数组 """
# 这里并不知道调用什么层进行设计
reshape_layer = Reshape(self._shape_before_bottleneck)(dense_layer)
return reshape_layer
def _add_conv_transpose_layers(self,x):
""" 增加反卷积模块 """
# 按照相反的顺序遍历所有的卷积层，并且在第一层停下
for layers_index in reversed(range(1,self._num_conv_layers)):
# 理解：原来的卷积层标记[0,1,2]，翻转之后的输出为[2,1,0]
x = self._add_conv_transpose_layer(x,layers_index)
return x
def _add_conv_transpose_layer(self,x,layer_index):
# 注意，这里的层序号是按照倒序来的，需要还原成正常序号
# 一个卷积模块：卷积层+ReLu+batchnormalization
layer_num = self._num_conv_layers - layer_index
conv_transpose_layer =Conv2DTranspose(
filters = self.conv_filters[layer_index],
kernel_size = self.conv_kernels[layer_index],
strides = self.conv_strides[layer_index],
padding = "same",
name = f"decoder_conv_transpose_layer_{layer_num}"
)
x = conv_transpose_layer(x)
x =ReLU(name=f"decoder_ReLu_{layer_num}")(x)
x = BatchNormalization(name = f"decoder_BN_{layer_num}")(x)
return x
def _add_decoder_output(self,x):
""" 增加模型的输出层 """
# 这部分要和encoder是一个完全的逆过程，而且之前的反卷积模块是少了最后一层
# ，所以这里需要额外设置一层
conv_transpose_layer = Conv2DTranspose(
filters=1, # filters 对应图片中的channel.最终生成图片是一个[28,28,1]的灰度图片
kernel_size=self.conv_kernels[0],
strides=self.conv_strides[0],
padding="same",
name=f"decoder_conv_transpose_layer_{self._num_conv_layers}"
)
x = conv_transpose_layer(x)
output_layer = Activation("sigmoid",name = "sigmoid_layer")(x)
return output_layer
# 3.1 实现整个模型而自动编码器
# 7、从上到下，逐个子方法进行实现
def _build_VAE(self):
""" 对于自动编码器的识别，链接编码器和解码器 """
model_input = self._model_input
encoder_output = self.encoder(model_input)
quantizered = self.vq_layer(encoder_output)
# model_output = self.decoder(quantized_latent)
model_output = self.decoder(quantizered)
self.model = Model(model_input,model_output,name = "VQVAE")
if __name__ == '__main__':
VAE = VQVAE(
input_shape= [28,28,1],
conv_filters = [32,64,64,64],
conv_kernels = [3,3,3,3],
conv_strides= [1, 2, 2, 1],
latent_space_dim=16,
num_embeddings=64
)
VAE.summary()
![[MySQL]一文带你学明白数据库控制语言——DCL](https://img-blog.csdnimg.cn/a1651f2a1e5b4f35a9e56de9aecdb402.png)
