『TensorFlow』降噪自編碼器設計

背景簡介

 

 

 

TensorFlow實現講解

 設計新思路：

1.使用類來記錄整個網絡：
　　使用_init_()屬性來記錄 網絡超參數 & 網絡框架 & 訓練過程
　　使用一個隱式方法初始化網絡參數
2.使用字典存儲初始化的各個參數（w&b）

 

參數初始化新思路：

 

主程序：

 圖結構實際實現

Version1：

導入包：

import numpy as np
import sklearn.preprocessing as prep
import tensorflow as tf
from tensorflow.examples.tutorials.mnist import input_data
import os

 關閉日志警告：

級別2是警告信息忽略，級別3是錯誤信息忽略

# 關閉tensorflow的警告信息
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'

 均勻分布參數生成函數：

# 標准均勻分布
def xavier_init(fan_in,fan_out,constant = 1):
    low   = -constant * np.sqrt(6.0/(fan_in+fan_out))
    hight =  constant * np.sqrt(6.0/(fan_in+fan_out))
    return tf.random_uniform((fan_in,fan_out),maxval=hight,minval=low,dtype=tf.float32)

 網絡類：

class AdditiveGuassianNoiseAutoencoder():
    def __init__(self,n_input,n_hidden,transfer_function=tf.nn.softplus,
                 optimzer=tf.train.AdamOptimizer(),scale=0.1):
        # 網絡參數
        self.n_input = n_input
        self.n_hidden = n_hidden
        self.transfer = transfer_function   # 激活函數
        self.training_scale = scale               # 噪聲水平
        network_weights = self._initialize_weights()
        self.weights = network_weights

        # 網絡結構
        self.x = tf.placeholder(tf.float32, [None, self.n_input])
        self.hidden = self.transfer(
            tf.add(tf.matmul(self.x + scale * tf.random_normal((n_input,)),
                             self.weights['w1']), self.weights['b1']))
        self.reconstruction = tf.add(tf.matmul(self.hidden, self.weights['w2']),
                                               self.weights['b2'])

        # 訓練部分
        self.cost = 0.5 * tf.reduce_sum(tf.pow(tf.subtract(self.reconstruction, self.x), 2))
        self.optimizer = optimzer.minimize(self.cost)
        init = tf.global_variables_initializer()
        self.sess = tf.Session()
        self.sess.run(init)

        print('begin to run session... ...')

    def _initialize_weights(self):
        all_weights = dict()
        all_weights['w1'] = tf.Variable(xavier_init(self.n_input, self.n_hidden))
        all_weights['b1'] = tf.Variable(tf.zeros([self.n_hidden], dtype=tf.float32))
        all_weights['w2'] = tf.Variable(xavier_init(self.n_hidden, self.n_input))
        all_weights['b2'] = tf.Variable(tf.zeros([self.n_input], dtype=tf.float32))
        return all_weights

 主函數：

AGN_AC = AdditiveGuassianNoiseAutoencoder(n_input=784, n_hidden=200,
                                          transfer_function=tf.nn.softplus,
                                          optimzer=tf.train.AdamOptimizer(learning_rate=0.01),
                                          scale=0.01)
writer = tf.summary.FileWriter(logdir='logs', graph=AGN_AC.sess.graph)
writer.close()

 圖：

VersionV2

添加了命名空間，使節點更為清晰，但實際圖結構顯得凌亂，由於W&b的節點沒有被划歸到層節點下的關系

import numpy as np
import sklearn.preprocessing as prep
import tensorflow as tf
from tensorflow.examples.tutorials.mnist import input_data
import os

# 關閉tensorflow的警告信息
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'

# 標准均勻分布
def xavier_init(fan_in,fan_out,constant = 1):
    low   = -constant * np.sqrt(6.0/(fan_in+fan_out))
    hight =  constant * np.sqrt(6.0/(fan_in+fan_out))
    return tf.random_uniform((fan_in,fan_out),maxval=hight,minval=low,dtype=tf.float32)

class AdditiveGuassianNoiseAutoencoder():
    def __init__(self,n_input,n_hidden,transfer_function=tf.nn.softplus,
                 optimzer=tf.train.AdamOptimizer(),scale=0.1):
        # 網絡參數
        self.n_input = n_input
        self.n_hidden = n_hidden
        self.transfer = transfer_function         # 激活函數
        self.training_scale = scale               # 噪聲水平
        network_weights = self._initialize_weights()
        self.weights = network_weights

        # 網絡結構
        with tf.name_scope('Rawinput'): #<---
            self.x = tf.placeholder(tf.float32, [None, self.n_input])
        with tf.name_scope('NoiseAdder'): #<---
            self.scale = tf.placeholder(dtype=tf.float32) #<---使用占位符取代了固定的scale，增加了feed量
            self.noise_x = self.x + self.scale * tf.random_normal((n_input,)) #<---
        with tf.name_scope('Encoder'): #<---
            self.hidden = self.transfer(
                tf.add(tf.matmul(self.noise_x, self.weights['w1']), self.weights['b1']))
        with tf.name_scope('Reconstruction'): #<---
            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_sum(tf.pow(tf.subtract(self.reconstruction, self.x), 2))
        with tf.name_scope('Train'):
            self.optimizer = optimzer.minimize(self.cost)
        init = tf.global_variables_initializer()
        self.sess = tf.Session()
        self.sess.run(init)

        print('begin to run session... ...')

    def _initialize_weights(self):
        all_weights = dict()
        all_weights['w1'] = tf.Variable(xavier_init(self.n_input, self.n_hidden) ,name='weight1') #<---
        all_weights['b1'] = tf.Variable(tf.zeros([self.n_hidden], dtype=tf.float32), name='bias1')
        all_weights['w2'] = tf.Variable(xavier_init(self.n_hidden, self.n_input), name='weight2')
        all_weights['b2'] = tf.Variable(tf.zeros([self.n_input], dtype=tf.float32), name='bias2')
        return all_weights

AGN_AC = AdditiveGuassianNoiseAutoencoder(n_input=784, n_hidden=200,
                                          transfer_function=tf.nn.softplus,
                                          optimzer=tf.train.AdamOptimizer(learning_rate=0.01),
                                          scale=0.01)
writer = tf.summary.FileWriter(logdir='logs', graph=AGN_AC.sess.graph)
writer.close()

 圖：

 

Version3

保留字典結構存儲W&b

但是把字典key&value生成拆開放在了每一層中

import numpy as np
import sklearn.preprocessing as prep
import tensorflow as tf
from tensorflow.examples.tutorials.mnist import input_data
import os

# 關閉tensorflow的警告信息
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'

# 標准均勻分布
def xavier_init(fan_in,fan_out,constant = 1):
    low   = -constant * np.sqrt(6.0/(fan_in+fan_out))
    hight =  constant * np.sqrt(6.0/(fan_in+fan_out))
    return tf.random_uniform((fan_in,fan_out),maxval=hight,minval=low,dtype=tf.float32)

class AdditiveGuassianNoiseAutoencoder():
    def __init__(self,n_input,n_hidden,transfer_function=tf.nn.softplus,
                 optimzer=tf.train.AdamOptimizer(),scale=0.1):
        # 網絡參數
        self.n_input = n_input
        self.n_hidden = n_hidden
        self.transfer = transfer_function         # 激活函數
        self.training_scale = scale               # 噪聲水平
        self.weights = dict() #<---
        # network_weights = self._initialize_weights()
        # self.weights = network_weights

        # 網絡結構
        with tf.name_scope('Rawinput'):
            self.x = tf.placeholder(tf.float32, [None, self.n_input])
        with tf.name_scope('NoiseAdder'):
            self.scale = tf.placeholder(dtype=tf.float32)
            self.noise_x = 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(
                tf.add(tf.matmul(self.noise_x, self.weights['w1']), self.weights['b1']))
        with tf.name_scope('Reconstruction'):
            self.weights['w2'] = tf.Variable(xavier_init(self.n_hidden, self.n_input), 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_sum(tf.pow(tf.subtract(self.reconstruction, self.x), 2))
        with tf.name_scope('Train'):
            self.optimizer = optimzer.minimize(self.cost)
        init = tf.global_variables_initializer()
        self.sess = tf.Session()
        self.sess.run(init)

        print('begin to run session... ...')

    # def _initialize_weights(self):
    #     all_weights = dict()
    #     all_weights['w1'] = tf.Variable(xavier_init(self.n_input, self.n_hidden) ,name='weight1')
    #     all_weights['b1'] = tf.Variable(tf.zeros([self.n_hidden], dtype=tf.float32), name='bias1')
    #     all_weights['w2'] = tf.Variable(xavier_init(self.n_hidden, self.n_input), name='weight2')
    #     all_weights['b2'] = tf.Variable(tf.zeros([self.n_input], dtype=tf.float32), name='bias2')
    #     return all_weights

AGN_AC = AdditiveGuassianNoiseAutoencoder(n_input=784, n_hidden=200,
                                          transfer_function=tf.nn.softplus,
                                          optimzer=tf.train.AdamOptimizer(learning_rate=0.01),
                                          scale=0.01)
writer = tf.summary.FileWriter(logdir='logs', graph=AGN_AC.sess.graph)
writer.close()

 圖：

總結：

1.參數變量使用字典保存提升規整性；

2.參數變量生成仍然要放在層中，可視化效果更好。

降噪自編碼器完整程序

import numpy as np
import sklearn.preprocessing as prep
import tensorflow as tf
from tensorflow.examples.tutorials.mnist import input_data
import os

# 關閉tensorflow的警告信息
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'

# 標准均勻分布
def xavier_init(fan_in,fan_out,constant = 1):
    low = -constant * np.sqrt(6.0/(fan_in+fan_out))
    hight = constant * np.sqrt(6.0/(fan_in+fan_out))
    return tf.random_uniform((fan_in,fan_out),maxval=hight,minval=low,dtype=tf.float32)

class AdditiveGuassianNoiseAutoencoder():
    def __init__(self,n_input,n_hidden,transfer_function=tf.nn.softplus,
                 optimzer=tf.train.AdamOptimizer(),scale=0.1):
        # 網絡參數
        self.n_input = n_input
        self.n_hidden = n_hidden
        self.transfer = transfer_function         # 激活函數
        self.training_scale = scale               # 噪聲水平
        self.weights = dict()
        # network_weights = self._initialize_weights()
        # self.weights = network_weights

        # 網絡結構
        with tf.name_scope('Rawinput'):
            self.x = tf.placeholder(tf.float32, [None, self.n_input])
        with tf.name_scope('NoiseAdder'):
            self.scale = tf.placeholder(dtype=tf.float32)
            self.noise_x = 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(
                tf.add(tf.matmul(self.noise_x, self.weights['w1']), self.weights['b1']))
        with tf.name_scope('Reconstruction'):
            self.weights['w2'] = tf.Variable(xavier_init(self.n_hidden, self.n_input), 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_sum(tf.pow(tf.subtract(self.reconstruction, self.x), 2))
        with tf.name_scope('Train'):
            self.optimizer = optimzer.minimize(self.cost)
        init = tf.global_variables_initializer()
        self.sess = tf.Session()
        self.sess.run(init)

        print('begin to run session... ...')


    def partial_fit(self, X):
        '''
        訓練並計算cost
        :param X: 
        :return: 
        '''
        cost, opt = self.sess.run([self.cost, self.optimizer],
                                  feed_dict={self.x:X, self.scale:self.training_scale})
        return cost

    def calc_cost(self, X):
        '''
        不訓練，只計算cost
        :param X: 
        :return: 
        '''
        return self.sess.run(self.cost, feed_dict={self.x: X, self.scale: self.training_scale})


# 數據集預處理
def standard_scale(X_train, X_test): #<-----數據集預處理部分
    '''
    0均值，1標准差
    :param X_train: 
    :param X_test: 
    :return: 
    '''
    # 根據預估的訓練集的參數生成預處理器
    preprocessor = prep.StandardScaler().fit(X_train)
    X_train = preprocessor.transform(X_train)
    X_test  = preprocessor.transform(X_test)
    return X_train, X_test


def get_random_block_from_data(data, batch_size):
    '''
    隨機取一個batch的數據
    :param data: 
    :param batch_size: 
    :return: 
    '''
    start_index = np.random.randint(0, len(data) - batch_size)
    return data[start_index:(start_index+batch_size)]

# 展示計算圖
AGN_AC = AdditiveGuassianNoiseAutoencoder(n_input=784, n_hidden=200,
                                          transfer_function=tf.nn.softplus,
                                          optimzer=tf.train.AdamOptimizer(learning_rate=0.01),
                                          scale=0.01)
writer = tf.summary.FileWriter(logdir='logs', graph=AGN_AC.sess.graph)
writer.close()


# 讀取數據
mnist = input_data.read_data_sets('../Mnist_data/', one_hot=True)
X_train, X_test = standard_scale(mnist.train.images, mnist.test.images)

n_samples = int(mnist.train.num_examples)     # 訓練樣本總數
training_epochs = 20                          # 訓練輪數，1輪等於n_samples/batch_size
batch_size = 128                              # batch容量
display_step = 1                              # 展示間隔

# 訓練
for epoch in range(training_epochs):
    avg_cost = 0                              # 平均損失
    total_batch = int(n_samples/batch_size)   # 每一輪中step總數
    for i in range(total_batch):
        batch_xs = get_random_block_from_data(X_train, batch_size)
        cost = AGN_AC.partial_fit(batch_xs)
        avg_cost += cost / batch_size
    avg_cost /= total_batch

    if epoch % display_step == 0:
        print('epoch : %04d, cost = %.9f' % (epoch+1, avg_cost))

# 計算測試集上的cost
print('Total coat:', str(AGN_AC.calc_cost(X_test)))

 引入了數據預處理機制：

import sklearn.preprocessing as prep

# 數據集預處理
def standard_scale(X_train, X_test): #<-----一個新的嘗試
    '''
    0均值，1標准差
    :param X_train: 
    :param X_test: 
    :return: 
    '''
    # 根據預估的訓練集的參數生成預處理器
    preprocessor = prep.StandardScaler().fit(X_train)
    X_train = preprocessor.transform(X_train)
    X_test  = preprocessor.transform(X_test)
    return X_train, X_test

 因為就是個范例而已，所以並沒有加入更多的步驟。

 輸出：

Python 3.6.0 |Anaconda 4.3.1 (64-bit)| (default, Dec 23 2016, 12:22:00)
Type "copyright", "credits" or "license" for more information.
IPython 5.1.0 -- An enhanced Interactive Python.
?         -> Introduction and overview of IPython's features.
%quickref -> Quick reference.
help      -> Python's own help system.
object?   -> Details about 'object', use 'object??' for extra details.
PyDev console: using IPython 5.1.0
Running /home/hellcat/PycharmProjects/data_analysis/TensorFlow/autoencodes/denoise.py
Backend Qt5Agg is interactive backend. Turning interactive mode on.
begin to run session... ...
Extracting ../Mnist_data/train-images-idx3-ubyte.gz
Extracting ../Mnist_data/train-labels-idx1-ubyte.gz
Extracting ../Mnist_data/t10k-images-idx3-ubyte.gz
Extracting ../Mnist_data/t10k-labels-idx1-ubyte.gz
epoch : 0001, cost = 172728.840323210
epoch : 0002, cost = 384090.340043217
epoch : 0003, cost = 1424137.733514817
epoch : 0004, cost = 252195.476644165
epoch : 0005, cost = 1989602.406287275
epoch : 0006, cost = 82078.567135613
epoch : 0007, cost = 4571607.288953234
epoch : 0008, cost = 12936386.999440582
epoch : 0009, cost = 192551.124642752
epoch : 0010, cost = 40848.185927740
epoch : 0011, cost = 2998.114711095
epoch : 0012, cost = 15210.583374379
epoch : 0013, cost = 38411.792979990
epoch : 0014, cost = 5556.733809144
epoch : 0015, cost = 35625.806443790
epoch : 0016, cost = 1274942.135287910
epoch : 0017, cost = 214436.171889868
epoch : 0018, cost = 29740.634501637
epoch : 0019, cost = 1136.356513888
epoch : 0020, cost = 2248.695473340
Total coat: 4.2886e+06