使用 TensorFlow NumPy 撰寫 Keras 模型
作者： lukewood
建立日期 2021/08/28
上次修改日期 2021/08/28
描述： 概述如何使用 TensorFlow NumPy API 撰寫 Keras 模型。
簡介
NumPy 是一個非常成功的 Python 線性代數函式庫。
TensorFlow 最近推出了 tf_numpy，這是 NumPy API 的一個大型子集的 TensorFlow 實作。由於 tf_numpy，您可以使用 NumPy 風格撰寫 Keras 層或模型！
TensorFlow NumPy API 與 TensorFlow 生態系統完全整合。支援自動微分、TensorBoard、Keras 模型回呼、TPU 分散式和模型匯出等功能。
讓我們來看幾個範例。
設定
import os
os.environ["KERAS_BACKEND"] = "tensorflow"
import tensorflow as tf
import tensorflow.experimental.numpy as tnp
import keras
from keras import layers
為了測試我們的模型，我們將使用波士頓房價迴歸資料集。
(x_train, y_train), (x_test, y_test) = keras.datasets.boston_housing.load_data(
path="boston_housing.npz", test_split=0.2, seed=113
)
input_dim = x_train.shape[1]
def evaluate_model(model: keras.Model):
loss, percent_error = model.evaluate(x_test, y_test, verbose=0)
print("Mean absolute percent error before training: ", percent_error)
model.fit(x_train, y_train, epochs=200, verbose=0)
loss, percent_error = model.evaluate(x_test, y_test, verbose=0)
print("Mean absolute percent error after training:", percent_error)
Downloading data from https://storage.googleapis.com/tensorflow/tf-keras-datasets/california_housing.npz
743530/743530 ━━━━━━━━━━━━━━━━━━━━ 0s 0us/step
使用 TNP 子類化 keras.Model
使用 Keras API 的最靈活方式是子類化 [keras.Model](/api/models/model#model-class) 類別。子類化 Model 類別可讓您完全自訂訓練迴圈中發生的情況。這使得子類化 Model 成為研究人員的熱門選擇。
在此範例中，我們將實作一個 Model 子類別，該子類別使用 TNP API 對波士頓房屋資料集執行迴歸。請注意，當將 TNP API 與 keras 一起使用時，微分和梯度下降會自動處理。
首先，讓我們定義一個簡單的 TNPForwardFeedRegressionNetwork 類別。
class TNPForwardFeedRegressionNetwork(keras.Model):
def __init__(self, blocks=None, **kwargs):
super().__init__(**kwargs)
if not isinstance(blocks, list):
raise ValueError(f"blocks must be a list, got blocks={blocks}")
self.blocks = blocks
self.block_weights = None
self.biases = None
def build(self, input_shape):
current_shape = input_shape[1]
self.block_weights = []
self.biases = []
for i, block in enumerate(self.blocks):
self.block_weights.append(
self.add_weight(
shape=(current_shape, block),
trainable=True,
name=f"block-{i}",
initializer="glorot_normal",
)
)
self.biases.append(
self.add_weight(
shape=(block,),
trainable=True,
name=f"bias-{i}",
initializer="zeros",
)
)
current_shape = block
self.linear_layer = self.add_weight(
shape=(current_shape, 1),
name="linear_projector",
trainable=True,
initializer="glorot_normal",
)
def call(self, inputs):
activations = inputs
for w, b in zip(self.block_weights, self.biases):
activations = tnp.matmul(activations, w) + b
# ReLu activation function
activations = tnp.maximum(activations, 0.0)
return tnp.matmul(activations, self.linear_layer)
就像任何其他 Keras 模型一樣，我們可以使用任何支援的優化器、損失、指標或我們想要的回呼。
讓我們看看模型的表現如何！
model = TNPForwardFeedRegressionNetwork(blocks=[3, 3])
model.compile(
optimizer="adam",
loss="mean_squared_error",
metrics=[keras.metrics.MeanAbsolutePercentageError()],
)
evaluate_model(model)
WARNING: All log messages before absl::InitializeLog() is called are written to STDERR
I0000 00:00:1699909864.025985 48611 device_compiler.h:187] Compiled cluster using XLA! This line is logged at most once for the lifetime of the process.
Mean absolute percent error before training: 99.96772766113281
Mean absolute percent error after training: 40.94866180419922
太棒了！我們的模型似乎正在有效地學習解決手邊的問題。
我們也可以使用 TNP 撰寫自己的自訂損失函數。
def tnp_mse(y_true, y_pred):
return tnp.mean(tnp.square(y_true - y_pred), axis=0)
keras.backend.clear_session()
model = TNPForwardFeedRegressionNetwork(blocks=[3, 3])
model.compile(
optimizer="adam",
loss=tnp_mse,
metrics=[keras.metrics.MeanAbsolutePercentageError()],
)
evaluate_model(model)
Mean absolute percent error before training: 99.99896240234375
Mean absolute percent error after training: 52.533199310302734
使用 TNP 實作基於 Keras 層的模型
如果需要，TNP 也可以用於面向層的 Keras 程式碼結構。讓我們實作相同的模型，但使用分層方法！
def tnp_relu(x):
return tnp.maximum(x, 0)
class TNPDense(keras.layers.Layer):
def __init__(self, units, activation=None):
super().__init__()
self.units = units
self.activation = activation
def build(self, input_shape):
self.w = self.add_weight(
name="weights",
shape=(input_shape[1], self.units),
initializer="random_normal",
trainable=True,
)
self.bias = self.add_weight(
name="bias",
shape=(self.units,),
initializer="zeros",
trainable=True,
)
def call(self, inputs):
outputs = tnp.matmul(inputs, self.w) + self.bias
if self.activation:
return self.activation(outputs)
return outputs
def create_layered_tnp_model():
return keras.Sequential(
[
TNPDense(3, activation=tnp_relu),
TNPDense(3, activation=tnp_relu),
TNPDense(1),
]
)
model = create_layered_tnp_model()
model.compile(
optimizer="adam",
loss="mean_squared_error",
metrics=[keras.metrics.MeanAbsolutePercentageError()],
)
model.build((None, input_dim))
model.summary()
evaluate_model(model)
Model: "sequential"
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓ ┃ Layer (type) ┃ Output Shape ┃ Param # ┃ ┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩ │ tnp_dense (TNPDense) │ (None, 3) │ 27 │ ├─────────────────────────────────┼───────────────────────────┼────────────┤ │ tnp_dense_1 (TNPDense) │ (None, 3) │ 12 │ ├─────────────────────────────────┼───────────────────────────┼────────────┤ │ tnp_dense_2 (TNPDense) │ (None, 1) │ 4 │ └─────────────────────────────────┴───────────────────────────┴────────────┘
Total params: 43 (172.00 B)
Trainable params: 43 (172.00 B)
Non-trainable params: 0 (0.00 B)
Mean absolute percent error before training: 100.00006866455078
Mean absolute percent error after training: 43.57806396484375
您也可以在 TNP 層和原生 Keras 層之間無縫切換！
def create_mixed_model():
return keras.Sequential(
[
TNPDense(3, activation=tnp_relu),
# The model will have no issue using a normal Dense layer
layers.Dense(3, activation="relu"),
# ... or switching back to tnp layers!
TNPDense(1),
]
)
model = create_mixed_model()
model.compile(
optimizer="adam",
loss="mean_squared_error",
metrics=[keras.metrics.MeanAbsolutePercentageError()],
)
model.build((None, input_dim))
model.summary()
evaluate_model(model)
Model: "sequential_1"
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓ ┃ Layer (type) ┃ Output Shape ┃ Param # ┃ ┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩ │ tnp_dense_3 (TNPDense) │ (None, 3) │ 27 │ ├─────────────────────────────────┼───────────────────────────┼────────────┤ │ dense (Dense) │ (None, 3) │ 12 │ ├─────────────────────────────────┼───────────────────────────┼────────────┤ │ tnp_dense_4 (TNPDense) │ (None, 1) │ 4 │ └─────────────────────────────────┴───────────────────────────┴────────────┘
Total params: 43 (172.00 B)
Trainable params: 43 (172.00 B)
Non-trainable params: 0 (0.00 B)
Mean absolute percent error before training: 100.0
Mean absolute percent error after training: 55.646610260009766
Keras API 提供各種各樣的層。在專案中，能夠將它們與 NumPy 程式碼一起使用可以節省大量的時間。
分佈策略
TensorFlow NumPy 和 Keras 與 TensorFlow 分佈策略整合。這使得跨多個 GPU，甚至是整個 TPU Pod 執行分散式訓練變得簡單。
gpus = tf.config.list_logical_devices("GPU")
if gpus:
strategy = tf.distribute.MirroredStrategy(gpus)
else:
# We can fallback to a no-op CPU strategy.
strategy = tf.distribute.get_strategy()
print("Running with strategy:", str(strategy.__class__.__name__))
with strategy.scope():
model = create_layered_tnp_model()
model.compile(
optimizer="adam",
loss="mean_squared_error",
metrics=[keras.metrics.MeanAbsolutePercentageError()],
)
model.build((None, input_dim))
model.summary()
evaluate_model(model)
INFO:tensorflow:Using MirroredStrategy with devices ('/job:localhost/replica:0/task:0/device:GPU:0',)
Running with strategy: MirroredStrategy
Model: "sequential_2"
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓ ┃ Layer (type) ┃ Output Shape ┃ Param # ┃ ┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩ │ tnp_dense_5 (TNPDense) │ (None, 3) │ 27 │ ├─────────────────────────────────┼───────────────────────────┼────────────┤ │ tnp_dense_6 (TNPDense) │ (None, 3) │ 12 │ ├─────────────────────────────────┼───────────────────────────┼────────────┤ │ tnp_dense_7 (TNPDense) │ (None, 1) │ 4 │ └─────────────────────────────────┴───────────────────────────┴────────────┘
Total params: 43 (172.00 B)
Trainable params: 43 (172.00 B)
Non-trainable params: 0 (0.00 B)
Mean absolute percent error before training: 100.0
Mean absolute percent error after training: 44.573463439941406
TensorBoard 整合
使用 Keras API 的眾多好處之一是能夠透過 TensorBoard 監控訓練。將 TensorFlow NumPy API 與 Keras 一起使用可讓您輕鬆利用 TensorBoard。
keras.backend.clear_session()
若要從 Jupyter 筆記本載入 TensorBoard，您可以執行以下指令
%load_ext tensorboard
models = [
(
TNPForwardFeedRegressionNetwork(blocks=[3, 3]),
"TNPForwardFeedRegressionNetwork",
),
(create_layered_tnp_model(), "layered_tnp_model"),
(create_mixed_model(), "mixed_model"),
]
for model, model_name in models:
model.compile(
optimizer="adam",
loss="mean_squared_error",
metrics=[keras.metrics.MeanAbsolutePercentageError()],
)
model.fit(
x_train,
y_train,
epochs=200,
verbose=0,
callbacks=[keras.callbacks.TensorBoard(log_dir=f"logs/{model_name}")],
)
/opt/conda/envs/keras-tensorflow/lib/python3.10/site-packages/keras/src/callbacks/tensorboard.py:676: UserWarning: Model failed to serialize as JSON. Ignoring... Invalid format specifier
warnings.warn(f"Model failed to serialize as JSON. Ignoring... {exc}")
若要從 Jupyter 筆記本載入 TensorBoard，您可以使用 %tensorboard 指令
%tensorboard --logdir logs
TensorBoard 會監控指標並檢查訓練曲線。
TensorBoard 也允許您探索模型中使用的計算圖。
在除錯期間，能夠內省模型可能會很有價值。
結論
使用 tensorflow_numpy API 將現有的 NumPy 程式碼移植到 Keras 模型非常容易！透過與 Keras 整合，您可以獲得使用現有的 Keras 回呼、指標和優化器、輕鬆分發訓練和使用 Tensorboard 的能力。
將更複雜的模型（例如 ResNet）遷移到 TensorFlow NumPy API 將是一個很好的後續學習練習。
網路上有幾個開源的 NumPy ResNet 實作。