Tensorflow学习笔记（四）降噪自动编码器—运行会话，训练模型

import numpy as np
import sklearn.preprocessing as prep
import tensorflow as tf
from tensorflow.examples.tutorials.mnist import input_data
import os
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'

# Xavier均匀初始化
def xavier_init(f_in, f_out):
'''
如果网络的权重初始化太小，信号在层间传递时逐渐缩小会难以产生作用，如果权重初始化过大，那信号在层间传递时会被逐渐放大，最终导致发散。
Xavier均匀初始化，这种权重初始化方法会根据某一层网络的输入输出节点数量自动调整最合适的随机分布，让初始化权重不大不小，正好合适。
从数学的角度讲，就是让权重满足0均值，方差为2/(f_in+f_out)，随机分布可以使均匀分布或者高斯分布。

'''
low = -1 * np.sqrt(6.0 / (f_in + f_out))
high = 1 * np.sqrt(6.0 / (f_in + f_out))
return tf.random_uniform((f_in, f_out), minval=low, maxval=high, dtype=tf.float32)  # 建立了一个均匀分布

# 加性高斯噪声的自动编码器
class AdditiveGaussianNoiseAutoencoder(object):
def __init__(self, n_input, n_hidden, transfer_function=tf.nn.softplus, optimizer=tf.train.AdamOptimizer(),gauss_scale=0.1):
# 输入变量数
self.n_input = n_input
# 隐藏层节点数
self.n_hidden = n_hidden
# 隐藏层激活函数
self.transfer_function = transfer_function
self.training_scale=gauss_scale
# 隐含层和重构层参数
self.weights = dict()
with tf.name_scope('Input'):
# 输入信号
self.x = tf.placeholder(tf.float32, [None, self.n_input])
with tf.name_scope('NoiseAdder'):
self.scale=tf.placeholder(tf.float32)
self.noise_input=self.x+self.scale * tf.random_normal((n_input,))
with tf.name_scope('Encoder'):
# 隐藏层
self.weights['w1'] = tf.Variable(xavier_init(self.n_input, self.n_hidden), name='weight1')
self.weights['b1'] = tf.Variable(tf.zeros([self.n_hidden]), dtype=tf.float32, name='bias1')
self.hidden = self.transfer_function(tf.add(tf.matmul(self.noise_input, self.weights['w1']),self.weights['b1']))
with tf.name_scope('Reconstruction'):
# 重构层
self.weights['w2'] = tf.Variable(tf.zeros([self.n_hidden, self.n_input]), dtype=tf.float32, name='weight2')
self.weights['b2'] = tf.Variable(tf.zeros([self.n_input]), dtype=tf.float32, name='bias2')
self.reconstruction = tf.add(tf.matmul(self.hidden, self.weights['w2']), self.weights['b2'])
with tf.name_scope('Loss'):
# 将重构的结果与输入做对比，计算损失
self.cost = 0.5 * tf.reduce_mean(tf.pow(tf.subtract(self.reconstruction, self.x), 2))
with tf.name_scope('Train'):
# 优化器，用于减小损失
self.optimizer = optimizer.minimize(self.cost)
# 初始化所有变量
init = tf.global_variables_initializer()
self.sess = tf.Session()
self.sess.run(init)
print('begin to run session')
#在一个批次上训练
def train_batch(self,X):
cost,optimizer=self.sess.run((self.cost,self.optimizer),feed_dict={self.x:X,self.scale:self.training_scale})
return cost
#在给定样本上计算损失（用于测试）
def get_test_loss(self,X):
cost= self.sess.run(self.cost,feed_dict={self.x: X, self.scale: self.training_scale})
return cost
#返回自编码器隐含层的输出结果，获得抽样后的高阶特征表示
def hidden_out(self,X):
return self.sess.run(self.hidden,feed_dict={self.x: X, self.scale: self.training_scale})
#将隐藏层的高阶特征作为输入，重建为原始类似输入数据
def generate(self,hidden=None):
if hidden==None:
hidden=np.random.normal(size=self.weights['b1'])
return self.sess.run(self.reconstruction,feed_dict={self.hidden:hidden})
#整体运行一遍复原过程，包括提取高阶特征以及重建原始数据，输入原始数据，输出复员后的数据
def reconstruction(self,X):
return self.sess.run(self.reconstruction,feed_dict={self.x: X, self.scale: self.training_scale})
#获取隐含层的权重
def get_weights(self):
return self.sess.run(self.weights['w1'])
#获取隐含层的偏置
def get_biases(self):
return self.sess.run(self.weights['b1'])

# 实例化AdditiveGaussianNoiseAutoencoder类对象实例
AGN_AutoEncoder = AdditiveGaussianNoiseAutoencoder(n_input=784,
n_hidden=200,
transfer_function=tf.nn.softplus,
optimizer=tf.train.AdamOptimizer(learning_rate=0.01))
# 把计算图写入事件文件，在TensorBoard里面查看
writer = tf.summary.FileWriter(logdir='logs', graph=AGN_AutoEncoder.sess.graph)
writer.close()

#使用sklearn.preprocess的数据标准化操作（0均值，标准差为1）预处理数据
def standard_scale(X_train,X_test):
preprocessor=prep.StandardScaler().fit(X_train)    #从训练集上估计出均值和方差
X_train=preprocessor.transform(X_train)            #将估计出的均值和方差作用在训练集上
X_test=preprocessor.transform(X_test)              #将估计出的均值和方差作用在测试集上，对测试集做变换，保持变换的一致性
return X_train,X_test

#在[0,len(data)-batchsize]区间内取一个随机数，以这个随机整数为起始索引，抽取batchsize个样本
def get_batchsize_data(data,batch_size):
start_index=np.random.randint(0,len(data)-batch_size)
return data[start_index:(start_index+batch_size)]

#读取数据集
mnist = input_data.read_data_sets('../MNIST_data/', one_hot=True )

#使用标准化预处理操作变换数据集
X_train,X_test=standard_scale(mnist.train.images,mnist.test.images)   #将图像的像素点变成784维标准分布

#定义训练参数
n_samples=int(mnist.train.num_examples)     #训练样本总数
train_epochs=20                             #训练20个（n_samples/batchsize）批次
batch_size=128                              #每个批次的样本数量
dispaly_step=1                              #输出训练结果的间隔

#开始训练过程，每一轮epoch训练开始时，将平均损失avg_cost设为0，每轮要训练（n_samples/batchsize）批次，
#这里采用的是有放回抽样，不能保证每个样本都被抽到，并参与训练
for epoch in range(train_epochs):
avg_cost=0
total_batch=int(n_samples/batch_size)
for i in range(total_batch):
batch_xs=get_batchsize_data(X_train,batch_size)
cost=AGN_AutoEncoder.train_batch(batch_xs)
avg_cost+=cost/batch_size
# print('隐藏层特征：{}'.format(AGN_AutoEncoder.hidden_out(X_train)))
# print('隐含层的权重{}'.format(AGN_AutoEncoder.get_weights()))
# print('隐含层的偏置{}'.format(AGN_AutoEncoder.get_biases()))
avg_cost/=total_batch
if epoch%dispaly_step==0:
print('epoch:%d,loss=%.9f'%(epoch+1,avg_cost))
#计算测试集上的损失
print('test total loss:',str(AGN_AutoEncoder.get_test_loss(X_test)))