TensorFlow實戰1——TensorFlow實現Autoencoder

  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