本文将介绍如何在多个设备上运行Keras模型,如何保存模型,以及如何使用tensorboard实现回调。
文章目录
1. 如何在多个设备运行Keras模型?
使用TensorFlow作为后端的Keras,在多个GPU上运行单个模型的方法有两种:数据并行和设备并行。大多数情况下使用数据并行。
1.1 数据并行
数据并行即在多个GPU设备上使用相同的模型处理数据的不同部分。
Keras具有内置的实用程序,keras.utils.multi_gpu_model
可以生成任何模型的数据并行版本,并在多达8个GPU上实现准线性加速。
keras.utils.multi_gpu_model(model, gpus=None, cpu_merge=True, cpu_relocation=False)
具体来说,此功能可实现单机多GPU数据并行性。它的工作方式如下:
- 将模型的输入分为多个子批次。
- 在每个子批次上应用模型副本。每个模型副本均在专用GPU上执行。
- 将结果(在CPU上)串联在一起。
如果batch_size
= 64,并设置 gpus=2
,那么该方法将输入分为2个子批次,每个子批次包含32个样本,在一个GPU上处理每个子批次,然后返回整批处理的64个样本。可以在多达8个GPU上实现准线性加速。此功能仅适用于TensorFlow后端。
参数:
- model:Keras模型实例。为了避免OOM错误(OutOfMemory error),可在CPU上构建此模型。(例子在文末)
- gpus:> = 2的整数或整数列表,GPU数量或在其上创建模型副本的GPU ID列表。
- cpu_merge:一个布尔值,用于标识是否在CPU范围内强制合并模型权重。
- cpu_relocation:一个布尔值,用于标识是否在CPU范围内创建模型的权重。如果未在任何先前的设备范围内定义模型,仍然可以通过激活此选项来对其进行恢复。
- 示例1-具有权重的训练模型在CPU上合并
import tensorflow as tf
from keras.applications import Xception
from keras.utils import multi_gpu_model
import numpy as np
num_samples = 1000
height = 224
width = 224
num_classes = 1000
# Instantiate the base model (or "template" model).
# We recommend doing this with under a CPU device scope,
# so that the model's weights are hosted on CPU memory.
# Otherwise they may end up hosted on a GPU, which would
# complicate weight sharing.
with tf.device('/cpu:0'):
model = Xception(weights=None,
input_shape=(height, width, 3),
classes=num_classes)
# Replicates the model on 8 GPUs.
# This assumes that your machine has 8 available GPUs.
parallel_model = multi_gpu_model(model, gpus=8)
parallel_model.compile(loss='categorical_crossentropy',
optimizer='rmsprop')
# Generate dummy data.
x = np.random.random((num_samples, height, width, 3))
y = np.random.random((num_samples, num_classes))
# This `fit` call will be distributed on 8 GPUs.
# Since the batch size is 256, each GPU will process 32 samples.
parallel_model.fit(x, y, epochs=20, batch_size=256)
# Save model via the template model (which shares the same weights):
model.save('model.h5')
- 示例2-使用cpu_relocation在CPU上合并具有权重的训练模型
..
# Not needed to change the device scope for model definition:
model = Xception(weights=None, ..)
try:
parallel_model = multi_gpu_model(model, cpu_relocation=True)
print("Training using multiple GPUs..")
except ValueError:
parallel_model = model
print("Training using single GPU or CPU..")
parallel_model.compile(..)
..
- 示例3-具有权重的训练模型在GPU上合并(英伟达显卡系列可以使用此功能)
..
# Not needed to change the device scope for model definition:
model = Xception(weights=None, ..)
try:
parallel_model = multi_gpu_model(model, cpu_merge=False)
print("Training using multiple GPUs..")
except:
parallel_model = model
print("Training using single GPU or CPU..")
parallel_model.compile(..)
..
1.2 设备并行
设备并行性在于在不同设备上运行同一模型的不同部分。它最适合具有并行架构的模型,例如具有两个分支的模型。一个简单的示例:
# Model where a shared LSTM is used to encode two different sequences in parallel
input_a = keras.Input(shape=(140, 256))
input_b = keras.Input(shape=(140, 256))
shared_lstm = keras.layers.LSTM(64)
# Process the first sequence on one GPU
with tf.device_scope('/gpu:0'):
encoded_a = shared_lstm(tweet_a)
# Process the next sequence on another GPU
with tf.device_scope('/gpu:1'):
encoded_b = shared_lstm(tweet_b)
# Concatenate results on CPU
with tf.device_scope('/cpu:0'):
merged_vector = keras.layers.concatenate([encoded_a, encoded_b],
axis=-1)
2. 如何保存Keras模型?
2.1 保存/加载整个模型(架构+权重+优化器状态)
官方文档:不建议使用pickle或cPickle保存Keras模型。
model.save(filepath)
用来将Keras模型保存到单个HDF5文件中,该文件包含:
- 模型的架构,从而可以重新创建模型
- 模型的权重
- 训练配置(损失,优化器)
- 优化程序的状态,从而可以从您上次中断的地方继续正确地进行训练。
keras.models.load_model(filepath)
用来重新实例化模型。 load_model
还负责使用保存的训练配置来编译模型(除非从未首先编译过模型)。例:
from keras.models import load_model
model.save('my_model.h5') # creates a HDF5 file 'my_model.h5'
del model # deletes the existing model
# returns a compiled model
# identical to the previous one
model = load_model('my_model.h5')
2.2 仅保存/加载模型的架构
如果只需要保存模型的体系结构,而不是模型的权重或训练配置,则可执行以下操作:
# save as JSON
json_string = model.to_json()
# save as YAML
yaml_string = model.to_yaml()
实例:
# convert model architecture to JSON format
architecture = model.to_json()
# save architecture to JSON file
with open('architecture.json', 'wt') as json_file:
json_file.write(architecture)
from keras.models import model_from_json
# load architecture from JSON File
json_file = open('architecture.json', 'rt')
architecture = json_file.read()
json_file.close()
# create model from architecture
model = model_from_json(architecture)
生成的JSON / YAML文件是人类可读的,并且可以根据需要进行手动编辑。
然后,可以根据以下数据构建新的模型:
# model reconstruction from JSON:
from keras.models import model_from_json
model = model_from_json(json_string)
# model reconstruction from YAML:
from keras.models import model_from_yaml
model = model_from_yaml(yaml_string)
2.3 仅保存/加载模型的权重
如果需要保存模型的权重,则可用以下代码进行保存:
model.save_weights('my_model_weights.h5')
若有用于实例化模型的代码,可以将保存的权重加载到具有相同架构的模型中:
model.load_weights('my_model_weights.h5')
如果需要将权重加载到不同的体系结构(共有一些层)中,例如进行微调或转移学习,则可以按层名称加载它们:
model.load_weights('my_model_weights.h5', by_name=True)
例:
"""
Assuming the original model looks like this:
model = Sequential()
model.add(Dense(2, input_dim=3, name='dense_1'))
model.add(Dense(3, name='dense_2'))
...
model.save_weights(fname)
"""
# new model
model = Sequential()
model.add(Dense(2, input_dim=3, name='dense_1')) # will be loaded
model.add(Dense(10, name='new_dense')) # will not be loaded
# load weights from first model; will only affect the first layer, dense_1.
model.load_weights(fname, by_name=True)
2.4 处理已保存模型中的自定义图层(或其他自定义对象)
如果要加载的模型包括自定义图层或其他自定义类或函数,则可用custom_objects
参数指明加载项:
from keras.models import load_model
# Assuming your model includes instance of an "AttentionLayer" class
model = load_model('my_model.h5', custom_objects={'AttentionLayer': AttentionLayer})
以使用自定义对象范围:
from keras.utils import CustomObjectScope
with CustomObjectScope({'AttentionLayer': AttentionLayer}):
model = load_model('my_model.h5')
自定义对象处理的工作方式相同load_model,model_from_json,model_from_yaml:
from keras.models import model_from_json
model = model_from_json(json_string, custom_objects={'AttentionLayer': AttentionLayer})
3. Callbacks
回调是在训练过程的给定阶段应用的一组函数。可以使用回调查看培训期间模型的内部状态和统计信息。可以将回调列表(作为关键字参数回调)传递给序列类或模型类的.fit()方法。然后将在培训的每个阶段调用回调的相关方法。
更多其他方法请参考 官方文档
3.1 tensorboard
我的环境是 TensorFlow 2.1.0 以下Keras文档中的方法已经不适用,因此需要修改:
keras.callbacks.tensorboard_v1.TensorBoard(log_dir=’./logs’, histogram_freq=0, batch_size=32, write_graph=True, write_grads=False,
write_images=False, embeddings_freq=0, embeddings_layer_names=None, embeddings_metadata=None,
embeddings_data=None,update_freq=‘epoch’)
参数:
- log_dir:将TensorBoard解析的日志文件保存到的目录路径。
- histogram_freq:计算模型各层的激活和权重直方图的频率(以历元为单位)。如果设置为0,将不计算直方图。必须为直方图可视化指定验证数据(或拆分)。
- batch_size:要输入到网络以进行直方图计算的输入批次的大小。
- write_graph:是否在TensorBoard中可视化图形。当write_graph设置为True时,日志文件可能会变得很大。
- write_grads:是否在TensorBoard中可视化梯度直方图。 histogram_freq必须大于0。
- write_images:是否编写模型权重以在TensorBoard中可视化为图像。
- embeddings_freq:将保存选定嵌入层的频率(以时间为单位)。如果设置为0,则不会计算嵌入。在TensorBoard的``嵌入’'选项卡中可视化的数据必须以形式传递embeddings_data。
- embeddings_layer_names:需要注意的图层名称列表。如果为None或为空列表,则将监视所有嵌入层。
- embeddings_metadata:将层名称映射到文件名的字典,该嵌入层的元数据保存在该文件名中。如果相同的元数据文件用于所有嵌入层,则可以传递字符串。
- embeddings_data:要在中指定的图层上嵌入的数据 embeddings_layer_names。Numpy数组(如果模型有一个输入)或Numpy数组列表(如果模型有多个输入)。
- update_freq:‘batch’或’epoch’或整数。使用时’batch’,每批之后将损失和指标写入TensorBoard。同样适用于’epoch’。如果使用整数,假设10000,回调将每10000个样本将指标和损失写入TensorBoard。注意,过于频繁地向TensorBoard写入可能会减慢训练速度。
替换为:
tf.keras.callbacks.TensorBoard(
log_dir='logs', histogram_freq=0, write_graph=True, write_images=False,
update_freq='epoch', profile_batch=2, embeddings_freq=0,
embeddings_metadata=None, **kwargs
)
在Keras(TF2.1.0为后端)使用Tensorboard请参考: 在以TensorFlow2.1.0为后端的Keras中使用TensorBoard
3.2 创建一个回调
可以通过扩展基类来创建自定义回调keras.callbacks.Callback。回调可以通过class属性访问其关联的模型self.model。
一个简单的示例,在训练过程中保存了每批次的损失清单:
class LossHistory(keras.callbacks.Callback):
def on_train_begin(self, logs={}):
self.losses = []
def on_batch_end(self, batch, logs={}):
self.losses.append(logs.get('loss'))
示例:记录丢失历史
class LossHistory(keras.callbacks.Callback):
def on_train_begin(self, logs={}):
self.losses = []
def on_batch_end(self, batch, logs={}):
self.losses.append(logs.get('loss'))
model = Sequential()
model.add(Dense(10, input_dim=784, kernel_initializer='uniform'))
model.add(Activation('softmax'))
model.compile(loss='categorical_crossentropy', optimizer='rmsprop')
history = LossHistory()
model.fit(x_train, y_train, batch_size=128, epochs=20, verbose=0, callbacks=[history])
print(history.losses)
# outputs
'''
[0.66047596406559383, 0.3547245744908703, ..., 0.25953155204159617, 0.25901699725311789]
'''
示例:模型检查点
from keras.callbacks import ModelCheckpoint
model = Sequential()
model.add(Dense(10, input_dim=784, kernel_initializer='uniform'))
model.add(Activation('softmax'))
model.compile(loss='categorical_crossentropy', optimizer='rmsprop')
'''
saves the model weights after each epoch if the validation loss decreased
'''
checkpointer = ModelCheckpoint(filepath='/tmp/weights.hdf5', verbose=1, save_best_only=True)
model.fit(x_train, y_train, batch_size=128, epochs=20, verbose=0, validation_data=(X_test, Y_test), callbacks=[checkpointer])