Tensorflow中提供了通過變量名稱來創建和獲取一個變量的機制。通過這個機制,在不同的函數中可以直接通過變量的名字來使用變量,而不需要將變量通過參數的形式到處傳遞。該機制主要是通過tf.get_variable和tf.variable_scope函數來實現的。下面將分別介紹兩個函數的使用。
如果需要通過tf.get_variable獲取一個已經創建的變量,需要通過tf.variable_scope函數來生成一個上下文管理器,並明確指定在這個上下文管理器中,tf.get_variable將直接獲取已經生成的變量。代碼實例如下:
# 在名字為foo的命名空間內創建名字為v的變量 with tf.variable_scope("foo"): v=tf.get_variable('v',[1],initializer=tf.constant_initializer(1.0)) # 因為在命名空間foo中已經存在名字為v的變量,所有下面的代碼將會報錯: with tf.variable_scope('foo'): v=tf.get_variable('v',[1]) # 在生成上下文管理器時,將參數reuse設置為True。這樣tf.get_variable函數將直接獲取 # 已經聲明的變量 with tf.variable_scope('foo',reuse=True): v1=tf.get_variable('v',[1]) print(v==v1) # 將參數reuse設置為True時,tf.variable_scope將只能獲取已經創建過的變量。因為在命名 # 空間bar中還沒有創建變量v,所以下面的代碼將會報錯: with tf.variable_scope('bar',reuse=True): v=tf.get_variable('v',[1])
總結:當tf.variable_scope函數設置參數reuse=True時,這個上下文管理器內所有的tf.get_variable函數會直接獲取已經創建的變量。如果變量不存在,tf.get_variable將會報錯;相反,如果tf.variable_scope函數使用參數reuse=None或reuse=False創建上下文管理器,tf.get_variable將創建新的變量。如果同名的變量已經存在,則tf.get_variable報錯。
tensorflow中tf.variable_scope函數是可以嵌套的。下面將說明當tf.variable_scope函數嵌套時,reuse參數的取值將如何確定。
with tf.variable_scope('root'): # 通過如下方法獲取當前上下文管理器中reuse參數的取值 print(tf.get_variable_scope().reuse) # 輸出False with tf.variable_scope('foo',reuse=True): print(tf.get_variable_scope().reuse) # 輸出True with tf.variable_scope('bar'): # 不指定reuse時,reuse取值和上層保持一致 print(tf.get_variable_scope().reuse) # 輸出True print(tf.get_variable_scope().reuse) # 輸出False
tf.variable_scope函數生成的上下文管理器會創建一個tensorflow中 的命名空間,在命名空間內創建的變量名稱都會帶上這個命名空間名作為前綴。所以,tf.variable_scope函數除了可以控制tf.get_variable執行的功能之外,這個函數也提供了一個管理變量命名空間的方式。代碼如下:
import tensorflow as tf v1=tf.get_variable('v',[1]) print(v1.name) # 輸出v:0。'v'為變量的名稱,':0'表示這個變量是生成變量這個運算的 # 第一個結果 with tf.variable_scope('foo'): v2=tf.get_variable('v',[1]) print(v2.name) # 輸出foo/v:0。在tf.variable_scope中創建變量,變量名稱前面會加入 # 命名空間的名稱,並通過/來分割命名空間的名稱和變量的名稱 with tf.variable_scope('foo'): with tf.variable_scope('bar'): v3=tf.get_variable('v',[1]) print(v3.name) # 輸出foo/bar/v:0。命名空間可以嵌套 v4=tf.get_variable('v1',[1]) print(v4.name) # 輸出foo/v1:0 # 創建一個名稱為空的命令空間,並設置reuse=True with tf.variable_scope('',reuse=True): v5=tf.get_variable('foo/var/v',[1]) print(v5==v3) # 輸出True v6=tf.get_variable('foo/v1',[1]) print(v6==v4) # 輸出True
通過tf.variable_scope和tf.get_variable函數,對tensorflow 基礎學習五中的前向傳播函數進行改進:
import tensorflow as tf from tensorflow.examples.tutorials.mnist import input_data # MNIST數據集相關的常用 INPUT_NODE=784 OUTPUT_NODE=10 # 配置神經網絡的參數 LAYER1_NODE=500 # 隱藏層節點個數 BATCH_SIZE=100 # 一個訓練batch中的數據個數。 LEARNING_RATE_BASE=0.8 # 基礎的學習率 LEARNING_RATE_DECAY=0.99 # 學習率的衰減率 REGULARIZATION_RATE=0.0001 # 描述模型復雜度的正則化項在損失函數中的系統 TRAINING_STEPS=30000 # 訓練輪數 MOVING_AVERAGE_DECAY=0.99 # 滑動平均衰減率 def get_weight_variable(shape, regularizer): weights = tf.get_variable("weights", shape, initializer=tf.truncated_normal_initializer(stddev=0.1)) if regularizer != None: tf.add_to_collection('losses', regularizer(weights)) return weights # 定義一個三層神經網絡。 def inference(input_tensor,avg_class,regularizer): # 當沒有提供滑動平均類時,直接使用參數當前的取值。 if avg_class==None: with tf.variable_scope('layer1'): weights = get_weight_variable([INPUT_NODE, LAYER1_NODE], regularizer) biases = tf.get_variable("biases", [LAYER1_NODE], initializer=tf.constant_initializer(0.0)) layer1 = tf.nn.relu(tf.matmul(input_tensor, weights) + biases) with tf.variable_scope('layer2'): weights = get_weight_variable([LAYER1_NODE, OUTPUT_NODE], regularizer) biases = tf.get_variable("biases", [OUTPUT_NODE], initializer=tf.constant_initializer(0.0)) layer2 = tf.matmul(layer1, weights) + biases return layer2 else: # 首先使用avg_class.average函數來計算出變量的滑動平均值, # 然后再計算相應的神經網絡前向傳播結果 with tf.variable_scope('layer1',reuse=True): weights = get_weight_variable([INPUT_NODE, LAYER1_NODE], regularizer) biases = tf.get_variable("biases", [LAYER1_NODE], initializer=tf.constant_initializer(0.0)) layer1 = tf.nn.relu(tf.matmul(input_tensor,avg_class.average(weights))+avg_class.average(biases)) with tf.variable_scope('layer2',reuse=True): weights = get_weight_variable([LAYER1_NODE, OUTPUT_NODE], regularizer) biases = tf.get_variable("biases", [OUTPUT_NODE], initializer=tf.constant_initializer(0.0)) layer2 = tf.matmul(layer1,avg_class.average(weights))+avg_class.average(biases) return layer2 # 訓練模型的過程 def train(mnist): x=tf.placeholder(tf.float32,shape=[None,INPUT_NODE],name='x-input') y_=tf.placeholder(tf.float32,shape=[None,OUTPUT_NODE],name='y-input') # 計算L2正則化損失函數 regularizer=tf.contrib.layers.l2_regularizer(REGULARIZATION_RATE) y=inference(x,None,regularizer) # 定義存儲訓練輪數的變量。該變量為不可訓練變量(trainable=False) global_step=tf.Variable(0,trainable=False) # 給定滑動平均衰減率和訓練輪數的變量,初始化滑動平均類。給定訓練輪數的變量 # 可以加快訓練早期變量的更新速度。 variable_averages=tf.train.ExponentialMovingAverage(MOVING_AVERAGE_DECAY,global_step) # 在所有代表神經網絡參數的變量上使用滑動平均。tf.trainable_variables返回的就是圖上集合 # GraphKeys.TRAINABLE_VARIABLES中的元素。這個集合的元素就是所有沒有指定 # trainable=False的參數。 variables_averages_op=variable_averages.apply(tf.trainable_variables()) # 計算使用了滑動平均之后的前向傳播結果。滑動平均不會改變變量本身的值,而是維護一個影子 # 變量來記錄其滑動平均值。所以當需要使用這個滑動平均值時,需要明確調用average函數 average_y=inference(x,variable_averages,None) # 使用交叉熵作為損失函數。這里使用 # sparse_softmax_cross_entropy_with_logits函數來計算交叉熵。因為標准答案是一個長度為 # 10的一維數組,而該函數需要提供的是一個正確答案的數字,所以需要使用tf.argmax函數來 # 得到正確答案對應的類別編號 cross_entropy=tf.nn.sparse_softmax_cross_entropy_with_logits(logits=y,labels=tf.argmax(y_,1)) # 計算在當前batch中所有樣例的交叉熵平均值 cross_entropy_mean=tf.reduce_mean(cross_entropy) # 總損失等於交叉熵損失和正則化損失的和 loss=cross_entropy_mean+tf.add_n(tf.get_collection('losses')) # 設置指數衰減的學習率。 learning_rate=tf.train.exponential_decay( LEARNING_RATE_BASE, # 基礎的學習率,隨着迭代的進行,更新變量時使用的學習率 # 在這個基礎上遞減 global_step, # 當前迭代的輪數 mnist.train.num_examples/BATCH_SIZE, # 過完所有的訓練數據需要的迭代次數 LEARNING_RATE_DECAY # 學習率衰減速度 ) # 使用tf.train.GradientDescentOptimizer優化算法來優化損失函數。這里的損失函數 # 包含了交叉熵損失和L2正則化損失。 train_step=tf.train.GradientDescentOptimizer(learning_rate)\ .minimize(loss,global_step=global_step) # 在訓練神經網絡模型時,每過一遍數據既需要通過反向傳播來更新網絡中的參數, # 又要更新每一個參數的滑動平均值。為了一次完成多個操作,tensorflow提供了 # tf.control_dependencies和tf.group兩種機制。下面兩行程序和 # train_op=tf.group(train_step,variables_averages_op)是等價的。 with tf.control_dependencies([train_step,variables_averages_op]): train_op=tf.no_op(name='train') # 檢驗使用了滑動平均模型的神經網絡前向傳播結果是否正確。tf.argmax(average_y,1) # 計算每一個樣例的預測答案。其中average_y是一個batch_size*10的二維數組,每一行 # 表示一個樣例的前向傳播結果。tf.argmax的第二個參數‘1’表示選取最大值的操作僅在 # 第一個維度中進行,也就是說,只在每一行選取最大值對應的下標。於是得到的結果是 # 一個長度為batch的以為數組,這個一維數組中的值就表示了每一個樣例對應的數字識別結果。 # tf.equal判斷兩個張量的每一維是否相等。 correct_prediction=tf.equal(tf.argmax(average_y,1),tf.argmax(y_,1)) accuracy=tf.reduce_mean(tf.cast(correct_prediction,tf.float32)) # 初始化會話並開始訓練過程。 with tf.Session() as sess: init_op=tf.global_variables_initializer() sess.run(init_op) # 准備驗證數據。一般在神經網絡的訓練過程中會通過驗證數據來大致判斷停止的 # 條件和評判訓練的效果。 validate_feed={x:mnist.validation.images,y_:mnist.validation.labels} # 准備測試數據 test_feed={x:mnist.test.images,y_:mnist.test.labels} # 迭代的訓練神經網絡 for i in range(TRAINING_STEPS): # 每1000輪輸出一次在驗證數據集上的測試結果 if i%1000==0: validate_acc=sess.run(accuracy,feed_dict=validate_feed) print('After {} training step(s),validation accuracy' 'using average model is {}'.format(i,validate_acc)) # 產生這一輪使用的一個batch的訓練數據,並運行訓練過程。 xs,ys=mnist.train.next_batch(BATCH_SIZE) sess.run(train_op,feed_dict={x:xs,y_:ys}) # 在訓練結束后,在測試數據上檢測神經網絡模型的最終正確率 test_acc=sess.run(accuracy,feed_dict=test_feed) print('After {} training step(s), test accuracy using average ' 'model is {}'.format(TRAINING_STEPS,test_acc)) def main(argv=None): # 聲明處理MNIST數據集的類,這個類在初始化時會自動下載數據 mnist=input_data.read_data_sets('.',one_hot=True) train(mnist) # tensorflow 提供的一個主程序入口,tf.app.run會調用上面定義的main函數 if __name__=='__main__': tf.app.run()