1 #coding=utf-8 2 import numpy as np 3 import sklearn.preprocessing as prep 4 import tensorflow as tf 5 from tensorflow.examples.tutorials.mnist import input_data 6 7 def xavier_init(fan_in,fan_out, constant = 1): 8 '''Yoshua Bengio指出深度学习模型的权重初始化太小，那么信号将在每层传递缩小而失去作用； 9 太大将导致发散.Xavier初始化器就是让权重被初始化得不大不小，正好合适。 10 从数学的角度看，Xavier就是满足均值为：0，方差为2/(n_in+n_out)的均匀或高斯分布''' 11 low = -constant*np.sqrt(6.0/(fan_in+fan_out)) 12 high = constant*np.sqrt(6.0/(fan_in+fan_out)) 13 return tf.random_uniform((fan_in,fan_out), minval=low, maxval=high,dtype = tf.float32) 14 #去噪声自编码class 15 class AdditiveGaussianNoiseAutoencoder(object): 16 17 def __init__(self, n_input, n_hidden, transfer_function=tf.nn.softplus, 18 optimizer=tf.train.AdamOptimizer(), scale=0.1): 19 ''' n_input:输入变量数 20 n_hidden:隐含层节点数 21 transfer_function:隐含层激活函数,默认为softplus 22 optimizer:优化器，默认为Adam 23 scale:高斯噪声系数，默认为0.1''' 24 self.n_input = n_input 25 self.n_hidden = n_hidden 26 self.transfer = transfer_function 27 self.scale = tf.placeholder(tf.float32) 28 self.training_scale = scale 29 #参数初始化使用_initialize_weights() 30 network_weights = self._initialize_weights() 31 self.weights = network_weights 32 33 #输入x 34 self.x = tf.placeholder(tf.float32,[None,self.n_input]) 35 36 '''隐藏层hidden,首先输入x加上噪声：self.x+scale*tf.random_normal((n_input,)) 37 然后tf.matmul上式与隐含层权重w1， 38 tf.add加上隐含层biases：b1， 39 最后使用self.transfer对结果进行激活函数处理''' 40 self.hidden = self.transfer(tf.add(tf.matmul( 41 self.x+scale*tf.random_normal((n_input,)), 42 self.weights['w1']), self.weights['b1'])) 43 '''输出层重构：reconstruction,不用激活函数 44 tf.matmul隐含层输出和输出层权重w2再加上输出层偏置b2''' 45 self.reconstruction = tf.add(tf.matmul(self.hidden, 46 self.weights['w2']), self.weights['b2']) 47 48 '''cost:直接使用平方误差即tf.substract计算输出self.reconstruction与self.x之差， 49 再使用tf.pow求差的平方''' 50 self.cost = 0.5*tf.reduce_sum(tf.pow(tf.subtract( 51 self.reconstruction, self.x), 2.0)) 52 53 54 self.optimizer = optimizer.minimize(self.cost) 55 56 init = tf.global_variables_initializer() 57 self.sess = tf.Session() 58 self.sess.run(init) 59 60 def _initialize_weights(self): 61 all_weights = dict() 62 '''w1使用xavier_init函数初始化，传入输入节点数和隐含层节点数， 63 它将返回一个比较适合softplus激活函数的权重初始分布''' 64 all_weights['w1'] = tf.Variable(xavier_init(self.n_input, 65 self.n_hidden)) 66 #b1，w2，b2使用tf.zeros全部为0 67 all_weights['b1'] = tf.Variable(tf.zeros([self.n_hidden], 68 dtype = tf.float32)) 69 70 all_weights['w2'] = tf.Variable(tf.zeros([self.n_hidden, 71 self.n_input], dtype = tf.float32)) 72 all_weights['b2'] = tf.Variable(tf.zeros([self.n_input], dtype = tf.float32)) 73 74 return all_weights 75 76 def partial_fit(self, X): 77 78 '''trian每一个batch数据并返回当年batch的cost 79 Session执行两个计算图的节点，cost和训练过程optimizer, 80 输入的feed_dict：输入数据x和噪声系数：scale''' 81 82 cost, opt = self.sess.run((self.cost, self.optimizer), 83 feed_dict = {self.x:X, self.scale: self.training_scale}) 84 85 return cost 86 87 88 def calc_total_cost(self, X): 89 #计算cost 90 return self.sess.run(self.cost, feed_dict = {self.x:X, 91 self.scale:self.training_scale 92 }) 93 94 def transform(self, X): 95 #计算抽象的特征，返回隐含层的输出结果 96 return self.sess.run(self.hidden, feed_dict = {self.x:X, 97 self.scale:self.training_scale 98 }) 99 100 def generate(self, hidden = None): 101 #将高阶抽象特征复原为原始数据 102 if hidden is None: 103 hidden = bp.random_normal(size = self.weights['b1']) 104 105 return self.sess.run(self.reconstruction, feed_dict = {self.hidden:hidden}) 106 107 108 def reconstruction(self, X): 109 '''整体运行一遍复原过程，包括提取高阶特征和用高阶特征复原原始数据 110 输入：原数据 输出：复原后的数据''' 111 return self.sess.run(self.reconstruction, feed_dict = {self.x:X, 112 self.scale:self.training_scale 113 }) 114 115 116 def getWeights(self): 117 #获取隐含层权重w1 118 return self.sess.run(self.weights['w1']) 119 120 def getBiases(self): 121 122 #获取隐含层的偏置系数b1 123 return self.sess.run(self.weights['b1']) 124 125 mnist = input_data.read_data_sets('MNIST_data', one_hot = True) 126 127 def standard_scale(X_train, X_test): 128 '''对训练、测试data进行标准化处理（让数据变成均值为0,标准差为1的分布）''' 129 preprocessor = prep.StandardScaler().fit(X_train) 130 X_train = preprocessor.transform(X_train) 131 X_test = preprocessor.transform(X_test) 132 return X_train, X_test 133 134 def get_random_block_from_data(data, batch_size): 135 '''随机获取block数据：取一个0到len(data)-batch_size之间的随机整数 136 再以这个随机数作为block的起始位置，然后顺序取batch_size的数据''' 137 start_index = np.random.randint(0,len(data)-batch_size) 138 139 return data[start_index:(start_index+batch_size)] 140 141 X_train, X_test = standard_scale(mnist.train.images, mnist.test.images) 142 143 n_samples = int(mnist.train.num_examples) 144 training_epochs = 20 145 batch_size = 128 146 display_step = 1 147 148 149 autoencoder = AdditiveGaussianNoiseAutoencoder(n_input = 784, 150 n_hidden = 200, 151 transfer_function = tf.nn.softplus, 152 optimizer = tf.train.AdamOptimizer(learning_rate = 0.001), 153 scale = 0.01 154 ) 155 156 for epoch in range(training_epochs): 157 158 avg_cost = 0. 159 total_batch = int(n_samples/batch_size) 160 161 for i in range(total_batch): 162 batch_xs = get_random_block_from_data(X_train, batch_size) 163 164 165 cost = autoencoder.partial_fit(batch_xs) 166 avg_cost += cost/n_samples*batch_size 167 168 if epoch%display_step == 0: 169 print("Epoch:", '%04d'%(epoch+1), "cost = ", 170 "{:.9f}".format(avg_cost)) 171 172 print("Total cost:"+str(autoencoder.calc_total_cost(X_test)))
Epoch: 0001 cost = 20001.253988636 Epoch: 0002 cost = 12866.668271591 Epoch: 0003 cost = 10933.510055682 Epoch: 0004 cost = 9885.109014205 Epoch: 0005 cost = 10337.800752273 Epoch: 0006 cost = 9621.243082386 Epoch: 0007 cost = 8365.464159659 Epoch: 0008 cost = 8419.876629545 Epoch: 0009 cost = 8918.941588636 Epoch: 0010 cost = 8069.571899432 Epoch: 0011 cost = 8420.874543182 Epoch: 0012 cost = 8921.646592614 Epoch: 0013 cost = 8175.836280682 Epoch: 0014 cost = 7681.573177273 Epoch: 0015 cost = 7809.688360227 Epoch: 0016 cost = 8018.361381250 Epoch: 0017 cost = 7809.863368750 Epoch: 0018 cost = 8121.875443750 Epoch: 0019 cost = 9078.917564773 Epoch: 0020 cost = 8115.000298295 Total cost:668358.0