十、將 TensorFlow 投入生產 · ApacheCN 深度學習譯文集

# 十、將 TensorFlow 投入生產在本章中，我們將介紹以下主題： * 實現單元測試 * 使用多個執行器 * 并行化 TensorFlow * 將 TensorFlow 投入生產 * 生產環境 TensorFlow 的一個例子 * 使用 TensorFlow 服務 # 介紹到目前為止，我們已經介紹了如何在 TensorFlow 中訓練和評估各種模型。因此，在本章中，我們將向您展示如何編寫可供生產使用的代碼。生產就緒代碼有各種定義，但對我們來說，生產代碼將被定義為具有單元測試的代碼，分離訓練和評估代碼，并有效地保存，并加載數據管道和圖會話的各種所需部分。 > 本章提供的 Python 腳本應該從命令行運行。這允許運行測試，并將設備位置記錄到屏幕上。 # 實現單元測試測試代碼可以加快原型設計速度，提高調試效率，加快更改速度，并且可以更輕松地共享代碼。在 TensorFlow 中有許多簡單的方法可以實現單元測試，我們將在本文中介紹它們。 ## 準備在編寫 TensorFlow 模型時，有助于進行單元測試以檢查程序的功能。這有助于我們，因為當我們想要對程序單元進行更改時，測試將確保這些更改不會以未知方式破壞模型。在這個秘籍中，我們將創建一個依賴于`MNIST`數據的簡單 CNN 網絡。有了它，我們將實現三種不同類型的單元測試來說明如何在 TensorFlow 中編寫它們。 > 請注意，Python 有一個很棒的測試庫，名為 Nose。 TensorFlow 還具有內置測試功能，我們將在其中查看，這樣可以更輕松地測試 Tensor 對象的值，而無需評估會話中的值。 1. 首先，我們需要加載必要的庫并格式化數據，如下所示： ```py import sys import numpy as np import tensorflow as tf from tensorflow.python.framework import ops ops.reset_default_graph() # Start a graph session sess = tf.Session() # Load data data_dir = 'temp' mnist = tf.keras.datasets.mnist (train_xdata, train_labels), (test_xdata, test_labels) = mnist.load_data() train_xdata = train_xdata / 255.0 test_xdata = test_xdata / 255.0 # Set model parameters batch_size = 100 learning_rate = 0.005 evaluation_size = 100 image_width = train_xdata[0].shape[0] image_height = train_xdata[0].shape[1] target_size = max(train_labels) + 1 num_channels = 1 # greyscale = 1 channel generations = 100 eval_every = 5 conv1_features = 25 conv2_features = 50 max_pool_size1 = 2 # NxN window for 1st max pool layer max_pool_size2 = 2 # NxN window for 2nd max pool layer fully_connected_size1 = 100 dropout_prob = 0.75 ``` 1. 然后，我們需要聲明我們的占位符，變量和模型公式，如下所示： ```py # Declare model placeholders x_input_shape = (batch_size, image_width, image_height, num_channels) x_input = tf.placeholder(tf.float32, shape=x_input_shape) y_target = tf.placeholder(tf.int32, shape=(batch_size)) eval_input_shape = (evaluation_size, image_width, image_height, num_channels) eval_input = tf.placeholder(tf.float32, shape=eval_input_shape) eval_target = tf.placeholder(tf.int32, shape=(evaluation_size)) dropout = tf.placeholder(tf.float32, shape=()) # Declare model parameters conv1_weight = tf.Variable(tf.truncated_normal([4, 4, num_channels, conv1_features], stddev=0.1, dtype=tf.float32)) conv1_bias = tf.Variable(tf.zeros([conv1_features], dtype=tf.float32)) conv2_weight = tf.Variable(tf.truncated_normal([4, 4, conv1_features, conv2_features], stddev=0.1, dtype=tf.float32)) conv2_bias = tf.Variable(tf.zeros([conv2_features], dtype=tf.float32)) # fully connected variables resulting_width = image_width // (max_pool_size1 * max_pool_size2) resulting_height = image_height // (max_pool_size1 * max_pool_size2) full1_input_size = resulting_width * resulting_height * conv2_features full1_weight = tf.Variable(tf.truncated_normal([full1_input_size, fully_connected_size1], stddev=0.1, dtype=tf.float32)) full1_bias = tf.Variable(tf.truncated_normal([fully_connected_size1], stddev=0.1, dtype=tf.float32)) full2_weight = tf.Variable(tf.truncated_normal([fully_connected_size1, target_size], stddev=0.1, dtype=tf.float32)) full2_bias = tf.Variable(tf.truncated_normal([target_size], stddev=0.1, dtype=tf.float32)) # Initialize Model Operations def my_conv_net(input_data): # First Conv-ReLU-MaxPool Layer conv1 = tf.nn.conv2d(input_data, conv1_weight, strides=[1, 1, 1, 1], padding='SAME') relu1 = tf.nn.relu(tf.nn.bias_add(conv1, conv1_bias)) max_pool1 = tf.nn.max_pool(relu1, ksize=[1, max_pool_size1, max_pool_size1, 1], strides=[1, max_pool_size1, max_pool_size1, 1], padding='SAME') # Second Conv-ReLU-MaxPool Layer conv2 = tf.nn.conv2d(max_pool1, conv2_weight, strides=[1, 1, 1, 1], padding='SAME') relu2 = tf.nn.relu(tf.nn.bias_add(conv2, conv2_bias)) max_pool2 = tf.nn.max_pool(relu2, ksize=[1, max_pool_size2, max_pool_size2, 1], strides=[1, max_pool_size2, max_pool_size2, 1], padding='SAME') # Transform Output into a 1xN layer for next fully connected layer final_conv_shape = max_pool2.get_shape().as_list() final_shape = final_conv_shape[1] * final_conv_shape[2] * final_conv_shape[3] flat_output = tf.reshape(max_pool2, [final_conv_shape[0], final_shape]) # First Fully Connected Layer fully_connected1 = tf.nn.relu(tf.add(tf.matmul(flat_output, full1_weight), full1_bias)) # Second Fully Connected Layer final_model_output = tf.add(tf.matmul(fully_connected1, full2_weight), full2_bias) # Add dropout final_model_output = tf.nn.dropout(final_model_output, dropout) return final_model_output model_output = my_conv_net(x_input) test_model_output = my_conv_net(eval_input) ``` 1. 接下來，我們創建我們的損失函數以及我們的預測和精確操作。然后，我們初始化以下模型變量： ```py # Declare Loss Function (softmax cross entropy) loss = tf.reduce_mean(tf.nn.sparse_softmax_cross_entropy_with_logits(model_output, y_target)) # Create a prediction function prediction = tf.nn.softmax(model_output) test_prediction = tf.nn.softmax(test_model_output) # Create accuracy function def get_accuracy(logits, targets): batch_predictions = np.argmax(logits, axis=1) num_correct = np.sum(np.equal(batch_predictions, targets)) return 100\. * num_correct/batch_predictions.shape[0] # Create an optimizer my_optimizer = tf.train.MomentumOptimizer(learning_rate, 0.9) train_step = my_optimizer.minimize(loss) # Initialize Variables init = tf.global_variables_initializer() sess.run(init) ``` 1. 對于我們的第一個單元測試，我們使用類`tf.test.TestCase`并創建一種方法來測試占位符（或變量）的值。對于此測試用例，我們確保損失概率（用于保持）大于`0.25`，因此模型不會更改為嘗試訓練超過 75% 的損失，如下所示： ```py # Check values of tensors! class DropOutTest(tf.test.TestCase): # Make sure that we don't drop too much def dropout_greaterthan(self): with self.test_session(): self.assertGreater(dropout.eval(), 0.25) ``` 1. 接下來，我們需要測試我們的`accuracy`函數是否按預期運行。為此，我們創建一個概率樣本數組和我們期望的樣本，然后確保測試精度返回 100% ，如下所示： ```py # Test accuracy function class AccuracyTest(tf.test.TestCase): # Make sure accuracy function behaves correctly def accuracy_exact_test(self): with self.test_session(): test_preds = [[0.9, 0.1],[0.01, 0.99]] test_targets = [0, 1] test_acc = get_accuracy(test_preds, test_targets) self.assertEqual(test_acc.eval(), 100.) ``` 1. 我們還可以確保`Tensor`對象是我們期望的形狀。要通過`target_size`測試模型輸出是`batch_size`的預期形狀，請輸入以下代碼： ```py # Test tensorshape class ShapeTest(tf.test.TestCase): # Make sure our model output is size [batch_size, num_classes] def output_shape_test(self): with self.test_session(): numpy_array = np.ones([batch_size, target_size]) self.assertShapeEqual(numpy_array, model_output) ``` 1. 現在我們需要在腳本中使用`main()`函數告訴 TensorFlow 我們正在運行哪個應用。腳本如下： ```py def main(argv): # Start training loop train_loss = [] train_acc = [] test_acc = [] for i in range(generations): rand_index = np.random.choice(len(train_xdata), size=batch_size) rand_x = train_xdata[rand_index] rand_x = np.expand_dims(rand_x, 3) rand_y = train_labels[rand_index] train_dict = {x_input: rand_x, y_target: rand_y, dropout: dropout_prob} sess.run(train_step, feed_dict=train_dict) temp_train_loss, temp_train_preds = sess.run([loss, prediction], feed_dict=train_dict) temp_train_acc = get_accuracy(temp_train_preds, rand_y) if (i + 1) % eval_every == 0: eval_index = np.random.choice(len(test_xdata), size=evaluation_size) eval_x = test_xdata[eval_index] eval_x = np.expand_dims(eval_x, 3) eval_y = test_labels[eval_index] test_dict = {eval_input: eval_x, eval_target: eval_y, dropout: 1.0} test_preds = sess.run(test_prediction, feed_dict=test_dict) temp_test_acc = get_accuracy(test_preds, eval_y) # Record and print results train_loss.append(temp_train_loss) train_acc.append(temp_train_acc) test_acc.append(temp_test_acc) acc_and_loss = [(i + 1), temp_train_loss, temp_train_acc, temp_test_acc] acc_and_loss = [np.round(x, 2) for x in acc_and_loss] print('Generation # {}. Train Loss: {:.2f}. Train Acc (Test Acc): {:.2f} ({:.2f})'.format(*acc_and_loss)) ``` 1. 要讓我們的腳本執行測試或訓練，我們需要以不同的方式從命令行調用它。以下代碼段是主程序代碼。如果程序收到參數`test`，它將執行測試；否則，它將運行訓練： ```py if __name__ == '__main__': cmd_args = sys.argv if len(cmd_args) > 1 and cmd_args[1] == 'test': # Perform unit-tests tf.test.main(argv=cmd_args[1:]) else: # Run the TensorFlow app tf.app.run(main=None, argv=cmd_args) ``` 1. 如果我們在命令行上運行程序，我們應該得到以下輸出： ```py $ python3 implementing_unit_tests.py test ... ---------------------------------------------------------------------- Ran 3 tests in 0.001s OK ``` 前面步驟中描述的完整程序可以在[書籍的 GitHub 倉庫](https://github.com/nfmcclure/tensorflow_cookbook/)和 [Packt 倉庫](https://github.com/PacktPublishing/TensorFlow-Machine-Learning-Cookbook-Second-Edition)中找到。 ## 工作原理在本節中，我們實現了三種類型的單元測試：張量值，操作輸出和張量形狀。 TensorFlow 有更多類型的單元測試函數，[可在此處找到](https://www.tensorflow.org/versions/master/api_docs/python/test.html) 。請記住，單元測試有助于確保代碼能夠按預期運行，為共享代碼提供信心，并使再現性更易于訪問。 # 使用多個執行器您將意識到 TensorFlow 有許多功能，包括計算圖，它們可以自然地并行計算。計算圖可以分為不同的處理器以及處理不同的批量。我們將討論如何在此秘籍中訪問同一臺機器上的不同處理器。 ## 準備對于此秘籍，我們將向您展示如何在同一系統上訪問多個設備并對其進行訓練。這是一種非常常見的情況：與 CPU 一起，機器可能具有一個或多個可以共享計算負載的 GPU。如果 TensorFlow 可以訪問這些設備，它將通過貪婪的過程自動將計算分配給多個設備。但是，TensorFlow 還允許程序通過名稱范圍放置指定哪些設備將在哪個設備上。要訪問 GPU 設備，必須安裝 GPU 版本的 TensorFlow。要安裝 TensorFlow 的 GPU 版本，請訪問[此鏈接](https://www.tensorflow.org/versions/master/get_started/os_setup.html)。下載，設置并按照特定系統的說明進行操作。請注意，TensorFlow 的 GPU 版本需要 CUDA 才能使用 GPU。在本文中，我們將向您展示各種命令，允許您訪問系統上的各種設備；我們還將演示如何找出 TensorFlow 正在使用的設備。 ## 操作步驟 1. 為了找出 TensorFlow 用于哪些操作的設備，我們需要在會話參數中設置`config`，將`log_device_placement`設置為`True`。當我們從命令行運行腳本時，我們將看到特定的設備放置，如以下輸出所示： ```py import tensorflow as tf sess = tf.Session(config=tf.ConfigProto(log_device_placement=True)) a = tf.constant([1.0, 2.0, 3.0, 4.0, 5.0, 6.0], shape=[2, 3], name='a') b = tf.constant([1.0, 2.0, 3.0, 4.0, 5.0, 6.0], shape=[3, 2], name='b') c = tf.matmul(a, b) # Runs the op. print(sess.run(c)) ``` 1. 從終端，運行以下命令： ```py $python3 using_multiple_devices.py Device mapping: no known devices. I tensorflow/core/common_runtime/direct_session.cc:175] Device mapping: MatMul: /job:localhost/replica:0/task:0/cpu:0 I tensorflow/core/common_runtime/simple_placer.cc:818] MatMul: /job:localhost/replica:0/task:0/cpu:0 b: /job:localhost/replica:0/task:0/cpu:0 I tensorflow/core/common_runtime/simple_placer.cc:818] b: /job:localhost/replica:0/task:0/cpu:0 a: /job:localhost/replica:0/task:0/cpu:0 I tensorflow/core/common_runtime/simple_placer.cc:818] a: /job:localhost/replica:0/task:0/cpu:0 [[ 22\. 28.] [ 49\. 64.]] ``` 1. 默認情況下，TensorFlow 會自動決定如何跨計算設備（CPU 和 GPU）分配計算，有時我們需要了解這些展示位置。這在加載早期的現有模型時非常有用，該模型在我們的計算機具有不同設備時在圖中分配了硬展示位置。我們可以在配置中設置軟放置以解決此問題，如下所示： ```py config = tf.ConfigProto() config.allow_soft_placement = True sess_soft = tf.Session(config=config) ``` 1. 使用 GPU 時，TensorFlow 會自動占用 GPU 內存的很大一部分。雖然通常需要這樣做，但我們可以采取措施更加小心 GPU 內存分配。雖然 TensorFlow 從未發布 GPU 內存，但我們可以通過設置 GPU 內存增長選項，將其分配緩慢增加到最大限制（僅在需要時），如下所示： ```py config.gpu_options.allow_growth = True sess_grow = tf.Session(config=config) ``` 1. 如果我們想對 TensorFlow 使用的 GPU 內存百分比設置硬限制，我們可以使用`config`設置`per_process_gpu_memory_fraction`，如下所示： ```py config.gpu_options.per_process_gpu_memory_fraction = 0.4 sess_limited = tf.Session(config=config) ``` 1. 有時我們可能需要編寫可靠的代碼來確定它是否在 GPU 可用的情況下運行。 TensorFlow 具有內置功能，可以測試 GPU 是否可用。當我們想要編寫在可用時利用 GPU 并為其分配特定操作的代碼時，這很有用。這是通過以下代碼完成的： ```py if tf.test.is_built_with_cuda(): <Run GPU specific code here> ``` 1. 如果我們需要為 GPU 分配特定操作，請輸入以下代碼。這將執行簡單的計算并將操作分配給主 CPU 和兩個輔助 GPU： ```py with tf.device('/cpu:0'): a = tf.constant([1.0, 3.0, 5.0], shape=[1, 3]) b = tf.constant([2.0, 4.0, 6.0], shape=[3, 1]) with tf.device('/gpu:0'): c = tf.matmul(a,b) c = tf.reshape(c, [-1]) with tf.device('/gpu:1'): d = tf.matmul(b,a) flat_d = tf.reshape(d, [-1]) combined = tf.multiply(c, flat_d) print(sess.run(combined)) ``` ## 工作原理當我們想在我們的機器上為 TensorFlow 操作指定特定設備時，我們需要知道 TensorFlow 如何引用這些設備。 TensorFlow 中的設備名稱遵循以下約定： | 設備 | 設備名稱 | | --- | --- | --- | | 主 CPU | `/CPU:0` | | 第二個 CPU | `/CPU:1` | | 主 GPU | `/GPU:0` | | 第二個 GPU | `/GPU:1` | | 第三個 GPU | `/GPU:2` | ## 更多幸運的是，在云中運行 TensorFlow 現在比以往更容易。許多云計算服務提供商都提供 GPU 實例，其中包含主 CPU 和強大的 GPU。 Amazon Web Services（AWS）具有 G 實例和 P2 實例，允許使用功能強大的 GPU，為 TensorFlow 流程提供極快的速度。您甚至可以免費選擇 AWS Machine Images（AMI），它將在安裝了 TensorFlow 的 GPU 實例的情況下啟動選定的實例。 # 并行化 TensorFlow 為了擴展 TensorFlow 并行化的范圍，我們還可以以分布式方式在完全不同的機器上從我們的圖執行單獨的操作。這個秘籍將告訴你如何。 ## 準備在 TensorFlow 發布幾個月后，谷歌發布了分布式 TensorFlow，它是對 TensorFlow 生態系統的一次重大升級，并且允許在不同的工作機器上設置 TensorFlow 集群，并分享訓練和評估的計算任務楷模。使用分布式 TensorFlow 就像為工作器設置參數一樣簡單，然后為不同的工作器分配不同的工作。在這個秘籍中，我們將建立兩個本地工作器并將他們分配到不同的工作。 ## 操作步驟 1. 首先，我們加載 TensorFlow 并使用配置字典文件（端口`2222`和`2223`）定義我們的兩個本地 worker，如下所示： ```py import tensorflow as tf # Cluster for 2 local workers (tasks 0 and 1): cluster = tf.train.ClusterSpec({'local': ['localhost:2222', 'localhost:2223']}) ``` 1. 現在，我們將兩個工作器連接到服務器并使用以下任務編號標記它們： ```py server = tf.train.Server(cluster, job_name="local", task_index=0) server = tf.train.Server(cluster, job_name="local", task_index=1) ``` 1. 現在我們將讓每個工作器完成一項任務。第一個工作器將初始化兩個矩陣（每個矩陣將是 25 乘 25）。第二個工作器將找到所有元素的總和。然后，我們將自動分配兩個總和的總和并打印輸出，如下所示： ```py mat_dim = 25 matrix_list = {} with tf.device('/job:local/task:0'): for i in range(0, 2): m_label = 'm_{}'.format(i) matrix_list[m_label] = tf.random_normal([mat_dim, mat_dim]) # Have each worker calculate the sums sum_outs = {} with tf.device('/job:local/task:1'): for i in range(0, 2): A = matrix_list['m_{}'.format(i)] sum_outs['m_{}'.format(i)] = tf.reduce_sum(A) # Sum all the sums summed_out = tf.add_n(list(sum_outs.values())) with tf.Session(server.target) as sess: result = sess.run(summed_out) print('Summed Values:{}'.format(result)) ``` 1. 輸入上面的代碼后，我們可以在命令提示符下運行以下命令： ```py $ python3 parallelizing_tensorflow.py I tensorflow/core/distributed_runtime/rpc/grpc_channel.cc:197] Initialize GrpcChannelCache for job local -> {0 -> localhost:2222, 1 -> localhost:2223} I tensorflow/core/distributed_runtime/rpc/grpc_server_lib.cc:206] Started server with target: grpc://localhost:2222 I tensorflow/core/distributed_runtime/rpc/grpc_channel.cc:197] Initialize GrpcChannelCache for job local -> {0 -> localhost:2222, 1 -> localhost:2223} I tensorflow/core/distributed_runtime/rpc/grpc_server_lib.cc:206] Started server with target: grpc://localhost:2223 I tensorflow/core/distributed_runtime/master_session.cc:928] Start master session 252bb6f530553002 with config: Summed Values:-21.12611198425293 ``` ## 工作原理使用分布式 TensorFlow 非常簡單。您所要做的就是將工作器 IP 分配給具有名稱的服務器。然后，可以手動或自動為工作器分配操作。 # 將 TensorFlow 投入生產如果我們想在生產環境中使用我們的機器學習腳本，我們首先需要考慮一些要點作為最佳實踐。在本節中，我們將概述其中的一些內容。 ## 準備在本文中，我們想總結并濃縮將 TensorFlow 投入生產的各種技巧。我們將介紹如何最好地保存和加載詞匯表，圖，變量和模型檢查點。我們還將討論如何使用 TensorFlow 的命令行參數解析器并更改 TensorFlow 的日志記錄詳細程度。 ## 操作步驟 1. 運行 TensorFlow 程序時，我們可能需要檢查內存中是否已存在其他圖會話，或者在調試程序后是否清除了圖會話。我們可以使用以下命令行來完成此任務： ```py from tensorflow.python.framework import ops ops.reset_default_graph() ``` 1. 在處理文本（或任何數據管道）時，我們需要確保我們保存處理數據的方式，以便我們可以以相同的方式處理未來的評估數據。例如，如果我們處理文本，我們需要確保我們可以保存并加載詞匯表。以下代碼是如何使用`JSON`庫保存詞匯表字典的示例： ```py import json word_list = ['to', 'be', 'or', 'not', 'to', 'be'] vocab_list = list(set(word_list)) vocab2ix_dict = dict(zip(vocab_list, range(len(vocab_list)))) ix2vocab_dict = {val:key for key,val in vocab2ix_dict.items()} # Save vocabulary import json with open('vocab2ix_dict.json', 'w') as file_conn: json.dump(vocab2ix_dict, file_conn) # Load vocabulary with open('vocab2ix_dict.json', 'r') as file_conn: vocab2ix_dict = json.load(file_conn) ``` > 在這里，我們以`JSON`格式保存了詞匯詞典，但我們也可以將其保存在`text`文件，`csv`甚至二進制格式中。如果詞匯量很大，則首選二進制文件。您還可以考慮使用 Pickle 庫來創建`pkl`二進制文件，但請注意，Pickle 文件在庫和 Python 版本之間不能很好地轉換。 1. 為了保存模型圖和變量，我們創建了一個`Saver()`操作并將其添加到圖中。建議我們在訓練期間定期保存模型。要保存模型，請輸入以下代碼： ```py After model declaration, add a saving operations saver = tf.train.Saver() # Then during training, save every so often, referencing the training generation for i in range(generations): ... if i%save_every == 0: saver.save(sess, 'my_model', global_step=step) # Can also save only specific variables: saver = tf.train.Saver({"my_var": my_variable}) ``` > 請注意，`Saver()`操作也會采用其他參數。如前面的示例所示，它可以使用變量和張量字典來保存特定元素。每隔`n`小時也可以檢查一次，定期執行保存操作。默認情況下，保存操作僅保留最后五個模型保存（出于空間考慮）。可以使用`maximum_to_keep`選項更改此設置。 1. 在保存模型之前，請務必命名模型的重要操作。如果 TensorFlow 沒有名稱，則沒有簡單的方法來加載特定的占位符，操作或變量。 TensorFlow 中的大多數操作和函數都接受`name`參數，如下例所示： ```py conv_weights = tf.Variable(tf.random_normal(), name='conv_weights') loss = tf.reduce_mean(... , name='loss') ``` 1. TensorFlow 還可以使用`tf.apps.flags`庫在命令行上輕松執行參數解析。使用這些函數，我們可以定義字符串，浮點數，整數或布爾值的命令行參數，如下面的代碼片段所示。使用這些標志定義，我們可以運行`tf.app.run()`，它將使用以下標志參數運行`main()`函數： ```py tf.flags.DEFINE_string("worker_locations", "", "List of worker addresses.") tf.flags.DEFINE_float('learning_rate', 0.01, 'Initial learning rate.') tf.flags.DEFINE_integer('generations', 1000, 'Number of training generations.') tf.flags.DEFINE_boolean('run_unit_tests', False, 'If true, run tests.') FLAGS = tf.flags.FLAGS # Need to define a 'main' function for the app to run def main(_): worker_ips = FLAGS.worker_locations.split(",") learning_rate = FLAGS.learning_rate generations = FLAGS.generations run_unit_tests = FLAGS.run_unit_tests # Run the Tensorflow app if __name__ == "__main__": # The following is looking for a "main()" function to run and will pass. tf.app.run() # Can modify this to be more custom: tf.app.run(main=my_main_function(), argv=my_arguments) ``` 1. TensorFlow 具有內置日志記錄，我們可以為其設置級別參數。我們可以設定的水平是`DEBUG`，`INFO`，`WARN`，`ERROR`和`FATAL`。默認為`WARN`，如下所示： ```py tf.logging.set_verbosity(tf.logging.WARN) # WARN is the default value, but to see more information, you can set it to # INFO or DEBUG tf.logging.set_verbosity(tf.logging.DEBUG) ``` ## 工作原理在本節中，我們提供了在 TensorFlow 中創建生產級代碼的提示。我們想介紹應用標志，模型保存和日志記錄等概念，以便用戶可以使用這些工具一致地編寫代碼，并了解在其他代碼中看到這些工具時的含義。還有許多其他方法可以編寫好的生產代碼，但下面的秘籍中將顯示完整的示例。 # 生產環境 TensorFlow 的一個例子生產機器學習模型的一個好方法是將訓練和評估程序分開。在本節中，我們將說明一個評估腳本，該腳本已經擴展到包括單元測試，模型保存和加載以及評估。 ## 準備在本文中，我們將向您展示如何使用上述標準實現評估腳本。代碼實際上包含一個訓練腳本和一個評估腳本，但是對于這個秘籍，我們只會向您展示評估腳本。提醒一下，兩個腳本都可以在[在線 GitHub 倉庫](https://github.com/nfmcclure/tensorflow_cookbook/)和 [Packt 官方倉庫](https://github.com/nfmcclure/tensorflow_cookbook/)中看到。對于即將到來的示例，我們將實現第 9 章，回歸神經網絡中的第一個 RNN 示例，該示例試圖預測文本消息是垃圾郵件還是非垃圾郵件。我們將假設 RNN 模型與詞匯一起被訓練和保存。 ## 操作步驟 1. 首先，我們首先加載必要的庫并聲明 TensorFlow 應用標志，如下所示： ```py import os import re import numpy as np import tensorflow as tf from tensorflow.python.framework import ops ops.reset_default_graph() # Define App Flags tf.flags.DEFINE_string("storage_folder", "temp", "Where to store model and data.") tf.flags.DEFINE_float('learning_rate', 0.0005, 'Initial learning rate.') tf.flags.DEFINE_float('dropout_prob', 0.5, 'Per to keep probability for dropout.') tf.flags.DEFINE_integer('epochs', 20, 'Number of epochs for training.') tf.flags.DEFINE_integer('batch_size', 250, 'Batch Size for training.') tf.flags.DEFINE_integer('rnn_size', 15, 'RNN feature size.') tf.flags.DEFINE_integer('embedding_size', 25, 'Word embedding size.') tf.flags.DEFINE_integer('min_word_frequency', 20, 'Word frequency cutoff.') tf.flags.DEFINE_boolean('run_unit_tests', False, 'If true, run tests.') FLAGS = tf.flags.FLAGS ``` 1. 接下來，我們聲明一個文本清理函數。這與訓練腳本中使用的清潔函數相同，如下所示： ```py def clean_text(text_string): text_string = re.sub(r'([^sw]|_|[0-9])+', '', text_string) text_string = " ".join(text_string.split()) text_string = text_string.lower() return text_string ``` 1. 現在，我們需要加載以下詞匯處理函數： ```py def load_vocab(): vocab_path = os.path.join(FLAGS.storage_folder, "vocab") vocab_processor = tf.contrib.learn.preprocessing.VocabularyProcessor.restore(vocab_path) return vocab_processor ``` 1. 現在我們有了清理文本的方法，并且還有一個詞匯處理器，我們可以將這些函數組合起來為給定的文本創建數據處理管道，如下所示： ```py def process_data(input_data, vocab_processor): input_data = clean_text(input_data) input_data = input_data.split() processed_input = np.array(list(vocab_processor.transform(input_data))) return processed_input ``` 1. 接下來，我們需要一種方法來獲取要評估的數據。為此，我們將要求用戶在屏幕上鍵入文本。然后，我們將處理文本并返回以下處理過的文本： ```py def get_input_data(): input_text = input("Please enter a text message to evaluate: ") vocab_processor = load_vocab() return process_data(input_text, vocab_processor) ``` > 對于此示例，我們通過要求用戶鍵入來創建評估數據。雖然許多應用將通過提供的文件或 API 請求獲取數據，但我們可以相應地更改此輸入數據函數。 1. 對于單元測試，我們需要使用以下代碼確保我們的文本清理函數正常運行： ```py class clean_test(tf.test.TestCase): # Make sure cleaning function behaves correctly def clean_string_test(self): with self.test_session(): test_input = '--Tensorflow's so Great! Dont you think so? ' test_expected = 'tensorflows so great don you think so' test_out = clean_text(test_input) self.assertEqual(test_expected, test_out) ``` 1. 現在我們有了模型和數據，我們可以運行`main`函數。`main`函數將獲取數據，設置圖，加載變量，輸入處理過的數據，然后打印輸出，如下面的代碼片段所示： ```py def main(args): # Get flags storage_folder = FLAGS.storage_folder # Get user input text x_data = get_input_data() # Load model graph = tf.Graph() with graph.as_default(): sess = tf.Session() with sess.as_default(): # Load the saved meta graph and restore variables saver = tf.train.import_meta_graph("{}.meta".format(os.path.join(storage_folder, "model.ckpt"))) saver.restore(sess, os.path.join(storage_folder, "model.ckpt")) # Get the placeholders from the graph by name x_data_ph = graph.get_operation_by_name("x_data_ph").outputs[0] dropout_keep_prob = graph.get_operation_by_name("dropout_keep_prob").outputs[0] probability_outputs = graph.get_operation_by_name("probability_outputs").outputs[0] # Make the prediction eval_feed_dict = {x_data_ph: x_data, dropout_keep_prob: 1.0} probability_prediction = sess.run(tf.reduce_mean(probability_outputs, 0), eval_feed_dict) # Print output (Or save to file or DB connection?) print('Probability of Spam: {:.4}'.format(probability_prediction[1])) ``` 1. 最后，要運行`main()`函數或單元測試，請使用以下代碼： ```py if __name__ == "__main__": if FLAGS.run_unit_tests: # Perform unit tests tf.test.main() else: # Run evaluation tf.app.run() ``` ## 工作原理為了評估模型，我們能夠使用 TensorFlow 的應用標志加載命令行參數，加載模型和詞匯處理器，然后通過模型運行處理過的數據并進行預測。請記住通過命令行運行此腳本，并在創建模型和詞匯表字典之前檢查是否運行了訓練腳本。 # 使用 TensorFlow 服務在本節中，我們將向您展示如何設置 RNN 模型以預測 TensorFlow 上的垃圾郵件或非垃圾郵件文本消息。我們將首先說明如何以 protobuf 格式保存模型，然后將模型加載到本地服務器，監聽端口`9000`以進行輸入。 ## 準備我們通過鼓勵讀者閱讀 [TensorFlow 服務網站](https://www.tensorflow.org/serving/serving_basic)上的官方文檔和簡短教程來開始本節。對于這個例子，我們將在第 9 章，循環神經網絡中重用我們在預測垃圾郵件中使用的大部分 RNN 代碼和 RNNs 秘籍。我們將更改模型保存代碼，以便將 protobuf 模型保存在使用 TensorFlow 服務所需的正確文件夾結構中。 > 請注意，本章中的所有腳本都應該從命令行 bash 提示符執行。有關更新的安裝說明，[請訪問官方安裝站點](https://www.tensorflow.org/serving/setup)。正常安裝就像向 Linux 源添加 gpg-key 并運行以下安裝命令一樣簡單： ```py $ sudo apt install tensorflow-model-server ``` ## 操作步驟 1. 在這里，我們將以與以前相同的方式開始，通過加載必要的庫并設置 TensorFlow 標志，如下所示： ```py import os import re import io import sys import requests import numpy as np import tensorflow as tf from zipfile import ZipFile from tensorflow.python.framework import ops ops.reset_default_graph() # Define App Flags tf.flags.DEFINE_string("storage_folder", "temp", "Where to store model and data.") tf.flags.DEFINE_float('learning_rate', 0.0005, 'Initial learning rate.') tf.flags.DEFINE_float('dropout_prob', 0.5, 'Per to keep probability for dropout.') tf.flags.DEFINE_integer('epochs', 20, 'Number of epochs for training.') tf.flags.DEFINE_integer('batch_size', 250, 'Batch Size for training.') tf.flags.DEFINE_integer('rnn_size', 15, 'RNN feature size.') tf.flags.DEFINE_integer('embedding_size', 25, 'Word embedding size.') tf.flags.DEFINE_integer('min_word_frequency', 20, 'Word frequency cutoff.') tf.flags.DEFINE_boolean('run_unit_tests', False, 'If true, run tests.') FLAGS = tf.flags.FLAGS ``` 1. 我們將以完全相同的方式繼續完成腳本。為簡潔起見，我們只會在訓練腳本中包含差異，這就是我們如何保存 protobuf 模型。這是通過在訓練完成后插入以下代碼來完成的： > 請注意此代碼與教程代碼的相似之處。這里的主要區別在于模型名稱，版本號以及我們正在保存 RNN 而不是 CNN 的事實。 ```py # Save the finished model for TensorFlow Serving (pb file) # Here, it's our storage folder / version number out_path = os.path.join(tf.compat.as_bytes(os.path.join(storage_folder, '1'))) print('Exporting finished model to : {}'.format(out_path)) builder = tf.saved_model.builder.SavedModelBuilder(out_path) # Build the signature_def_map. classification_inputs = tf.saved_model.utils.build_tensor_info(x_data_ph) classification_outputs_classes = tf.saved_model.utils.build_tensor_info(rnn_model_outputs) classification_signature = (tf.saved_model.signature_def_utils.build_signature_def( inputs={tf.saved_model.signature_constants.CLASSIFY_INPUTS: classification_inputs}, outputs={tf.saved_model.signature_constants.CLASSIFY_OUTPUT_CLASSES: classification_outputs_classes}, method_name=tf.saved_model.signature_constants.CLASSIFY_METHOD_NAME)) tensor_info_x = tf.saved_model.utils.build_tensor_info(x_data_ph) tensor_info_y = tf.saved_model.utils.build_tensor_info(y_output_ph) prediction_signature = ( tf.saved_model.signature_def_utils.build_signature_def( inputs={'texts': tensor_info_x}, outputs={'scores': tensor_info_y}, method_name=tf.saved_model.signature_constants.PREDICT_METHOD_NAME)) legacy_init_op = tf.group(tf.tables_initializer(), name='legacy_init_op') builder.add_meta_graph_and_variables( sess, [tf.saved_model.tag_constants.SERVING], signature_def_map={ 'predict_spam': prediction_signature, tf.saved_model.signature_constants.DEFAULT_SERVING_SIGNATURE_DEF_KEY: classification_signature, }, legacy_init_op=legacy_init_op) builder.save() print('Done exporting!') ``` 1. 對我們來說，重要的是要意識到 TensorFlow Serving 需要特定的文件或文件夾結構來加載模型。該腳本將以以下格式安裝文件： ![](https://img.kancloud.cn/d8/ec/d8ecfb33a993023a625a65d75ba6f918_380x136.png) A screenshot of the directory structure that TensorFlow Serving expects. 上面的屏幕截圖顯示了所需的目錄結構。在其中，我們有我們定義的數據目錄`temp`，然后是我們的模型版本號`1`。在版本號目錄中，我們保存我們的 protobuf 模型和一個包含要保存的所需變量的`variables`文件夾。 > 我們應該知道，在我們的數據目錄中，TensorFlow 服務將查找整數文件夾。 TensorFlow 服務將自動啟動并在最大整數下獲取模型。這意味著要部署新模型，我們需要將其標記為版本 2，并將其粘貼在也標記為`2`的新文件夾下。然后，TensorFlow 服務將自動獲取模型。 1. 要啟動我們的服務器，我們使用端口，`model_name`和`model_base_path`參數調用命令`tensorflow_model_server`。然后，TensorFlow Serving 查找版本號文件夾并選擇最大版本編號的模型。然后它將它部署到機器上，命令通過作為參數給出的端口運行。在以下示例中，我們在本地計算機（`0.0.0.0`）上運行，并且接受的默認端口是`9000`： ```py $ tensorflow_model_server --port=9000 --model_name=spam_ham --model_base_path=<directory of our code>/tensorflow_cookbook/10_Taking_TensorFlow_to_Production/06_Using_TensorFlow_Serving/temp/ 2018-08-09 12:05:16.206712: I tensorflow_serving/model_servers/main.cc:153] Building single TensorFlow model file config: model_name: spam_ham model_base_path: .../temp/ 2018-08-09 12:05:16.206874: I tensorflow_serving/model_servers/server_core.cc:459] Adding/updating models. 2018-08-09 12:05:16.206903: I tensorflow_serving/model_servers/server_core.cc:514] (Re-)adding model: spam_ham 2018-08-09 12:05:16.307681: I tensorflow_serving/core/basic_manager.cc:716] Successfully reserved resources to load servable {name: spam_ham version: 1} 2018-08-09 12:05:16.307744: I tensorflow_serving/core/loader_harness.cc:66] Approving load for servable version {name: spam_ham version: 1} 2018-08-09 12:05:16.307773: I tensorflow_serving/core/loader_harness.cc:74] Loading servable version {name: spam_ham version: 1} 2018-08-09 12:05:16.307829: I external/org_tensorflow/tensorflow/contrib/session_bundle/bundle_shim.cc:360] Attempting to load native SavedModelBundle in bundle-shim from: .../temp/1 2018-08-09 12:05:16.307867: I external/org_tensorflow/tensorflow/cc/saved_model/loader.cc:242] Loading SavedModel with tags: { serve }; from: .../temp/1 2018-08-09 12:05:16.313811: I external/org_tensorflow/tensorflow/core/platform/cpu_feature_guard.cc:141] Your CPU supports instructions that this TensorFlow binary was not compiled to use: AVX2 FMA 2018-08-09 12:05:16.325866: I external/org_tensorflow/tensorflow/cc/saved_model/loader.cc:161] Restoring SavedModel bundle. 2018-08-09 12:05:16.329290: I external/org_tensorflow/tensorflow/cc/saved_model/loader.cc:196] Running LegacyInitOp on SavedModel bundle. 2018-08-09 12:05:16.332936: I external/org_tensorflow/tensorflow/cc/saved_model/loader.cc:291] SavedModel load for tags { serve }; Status: success. Took 25074 microseconds. 2018-08-09 12:05:16.332972: I tensorflow_serving/servables/tensorflow/saved_model_warmup.cc:83] No warmup data file found at .../temp/1/assets.extra/tf_serving_warmup_requests 2018-08-09 12:05:16.333335: I tensorflow_serving/core/loader_harness.cc:86] Successfully loaded servable version {name: spam_ham version: 1} 2018-08-09 12:05:16.334678: I tensorflow_serving/model_servers/main.cc:323] Running ModelServer at 0.0.0.0:9000 ... ``` 1. 我們現在可以將二進制數據提交給`<host>:9000`并返回顯示結果的 JSON 響應。我們可以通過任何機器和任何編程語言來完成。不必依賴客戶端擁有 TensorFlow 的本地副本是非常有用的。 ## 工作原理如果我們將早期的生產規模部分與前一部分進行比較，主要區別在于我們在主機上部署了可以響應傳入請求的模型服務器。前面的部分是一個很好的設置示例，用于執行批量結果或在可以加載 TensorFlow 的機器上工作，但秘籍不是很擅長部署可用的模型，可以進行計算，并將結果返回給任何客戶。在本節中，我們將了解如何處理這種架構，如下表所示： | | 第 5 節 - 批量作業 | 第 6 節 - 通過 TensorFlow 服務的作業 | | --- | --- | --- | | 優點 | 不依賴于網絡連接或主機 | 結果與客戶端結構無關，唯一的要求是 Numpy 數組的正確格式化的二進制文件 | | 缺點 | 客戶端必須具有 TensorFlow 和模型文件 | 依靠可用的主機 | | 理想的用途 | 大批量數據 | 生產服務始終可用，通常是小的請求 | 當然，每種方法的優缺點都值得商榷，兩者都能滿足每種情況的要求。還有許多其他可用的架構可以滿足不同的需求，例如 Docker，Kubernetes，Luigi，Django/Flask，Celery，AWS 和 Azure。 ## 更多本章未涉及的架構工具和資源的鏈接如下： * [在 Docker 中使用 TensorFlow 服務](https://www.tensorflow.org/serving/docker) * [在 Kubernetes 中使用 TensorFlow 服務](https://www.tensorflow.org/serving/serving_inception) * [Luigi，批量作業的管道工具](https://github.com/spotify/luigi) * [在 Flask 中使用 TensorFlow](https://guillaumegenthial.github.io/serving.html) * [用于分布式任務排隊的 Python 框架](http://www.celeryproject.org/community/) * [如何在 TensorFlow 模型中使用 AWS lambdas](https://aws.amazon.com/blogs/machine-learning/how-to-deploy-deep-learning-models-with-aws-lambda-and-tensorflow/)