走近深度学习，认识MoXing：模型定义教程
本文主要讲述MoXing将模型定义在model_fn方法中，并在mox.run时注册该方法。
基本方法：
def model_fn(inputs, mode, **kwargs):
...
return mox.ModelSpec(...)
mox.run(..., model_fn=model_fn, ...)
输入参数：
· inputs: 对应input_fn返回值的输入数据。
· mode: 当前调用model_fn时的运行模式，需要用户在model_fn中做好判断使用相应的模型。mox.ModeKeys中的一个，参考API。如训练态(mox.ModeKeys.TRAIN)和验证态(mox.ModeKeys.EVAL)下的模型是不一样的（如BN层和Dropout层）。
· **kwargs: 扩展参数的预留位置。
返回值：
· mox.ModelSpec的实例
当input_fn返回的输入数据只有一项时，model_fn的输入参数inputs仍然是一个list。
用户的代码可能是如下样例：
def input_fn(mode, **kwargs):
...
return image
def model_fn(inputs, mode, **kwargs):
images = inputs
...
代码看似没什么问题，但是当用户在model_fn中使用images时发现images的shape和预想的不太一样。可能会出现如下错误信息：
ValueError: Input must be of size [batch_size, height, width, C>0]
即使input_fn返回的输入数据只有image，model_fn的输入参数inputs仍然是一个list，为[images]，所以如下代码才是正确的用法：
def input_fn(mode, **kwargs):
...
return image
def model_fn(inputs, mode, **kwargs):
images = inputs[0]
...
model_fn必须返回ModelSpec的实例，根据model_fn中的mode不同，ModelSpec的入参情况为：
· loss: 指定模型的损失值，一个0阶tf.Tensor，或者0阶tf.Tensor的list，多loss案例参考生成对抗模型GAN，当mode==mox.ModeKey.TRAIN时必须提供。
· var_scope: 指定从loss中计算出的梯度需要对应的变量范围，只有在var_scope范围内的tf.Variable的梯度才会被计算和更新。如果loss是一个0阶tf.Tensor，则var_scope为str的list，指定一个或多个variable_scope。当loss是0阶tf.Tensor的list时，var_scope为二阶list，list[i]表示loss[i]的variable_scope，参考生成对抗模型GAN。
· log_info: 一个dict，运行作业时控制台需要打印的指标信息，仅支持0阶tf.Tensor，如{'loss': loss, 'acc': accuracy}，当mode==mox.ModeKey.EVAL时必须提供。
· output_info: 一个dict，运行作业的同时输出tf.Tensor中具体的值到output_fn中，当mode==mox.ModeKey.PREDICT时必须提供，参考利用output_fn做预测。
· export_spec: 一个dict，导出PB模型时指定输入输出节点，必须是一个mox.ExportSpec的实例，当mode==mox.ModeKey.EXPORT时必须提供(注意mox.ModeKey.EXPORT是无法在mox.run中显示指定的，仅当mox.run参数中export_model为有效值时会自动添加该模式)，参考导出PB模型。
· hooks: 一个list, 每个元素都必须是mox.AggregativeSessionRunHook子类的实例，会被tf.Session()执行的hook。参考在model_fn中使用placeholder，训练时打印验证集指标，[使用Early Stopping](使用Early Stopping)
1 使用MoXing模型库的内置模型
目前MoXing集成了一些神经网络模型，用户可以直接使用mox.get_model_fn获取这些模型。以及使用mox.get_model_meta获取这些模型的元信息。
例：训练一个ResNet_v1_50:
import tensorflow as tf
import moxing.tensorflow as mox
slim = tf.contrib.slim
def input_fn(mode, **kwargs):
meta = mox.ImageClassificationRawMetadata(base_dir='/export1/flowers/raw/split/train')
dataset = mox.ImageClassificationRawDataset(meta)
image, label = dataset.get(['image', 'label'])
image = mox.get_data_augmentation_fn(
name='resnet_v1_50',
run_mode=mode,
output_height=224,
output_width=224)(image)
return image, label
def model_fn(inputs, mode, **kwargs):
images, labels = inputs
logits, endpoints = mox.get_model_fn(
name='resnet_v1_50',
run_mode=mode,
num_classes=1000,
weight_decay=0.0001)(images)
loss = tf.losses.softmax_cross_entropy(
logits=logits, onehot_labels=slim.one_hot_encoding(labels, 1000))
return mox.ModelSpec(loss=loss, log_info={'loss': loss})
mox.run(input_fn=input_fn,
model_fn=model_fn,
optimizer_fn=mox.get_optimizer_fn('sgd', learning_rate=0.01),
batch_size=32,
run_mode=mox.ModeKeys.TRAIN,
max_number_of_steps=100)
当用户导出模型时，考虑以下代码导出一个被TF-Serving使用的模型：
可能会遇到如下错误信息：
ValueError: `image_height` and `image_width` should be given to `mox.get_model_fn` when `run_mode` is `mox.ModeKeys.EXPORT (When `export_model` is specified in `mox.run`).
当用户导出模型时，model_fn会以mode=mox.ModeKeys.EXPORT模式调用，当mox.get_model_fn中的run_mode为mode=mox.ModeKeys.EXPORT时，必须指定输入图像的尺寸，修改以下代码段：
logits, endpoints = mox.get_model_fn(
name='resnet_v1_50',
run_mode=mode,
num_classes=1000,
weight_decay=0.0001)(images)
正确的用法为：
model_meta = mox.get_model_meta('resnet_v1_50')
logits, endpoints = mox.get_model_fn(
name='resnet_v1_50',
run_mode=mode,
num_classes=1000,
weight_decay=0.0001,
image_height=model_meta.default_image_size,
image_width=model_meta.default_image_size)(images)
除了使用MoXing内置的神经网络模型，用户可以自定义任何模型，只需要返回值符合规范。MoXing会自动将model_fn中定义的模型使用在多GPU上和分布式上。
在model_fn中调用形如tf.global_variables()或tf.get_collection(tf.GraphKeys.REGULARIZATION_LOSSES)这些方法时，返回值与预期的不符。tf.global_variables()等效于tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES)
当使用单GPU时，这些方法的使用没有问题，但当使用多GPU时，使用mox.get_collection代替tf.get_collection来获取当前GPU上model_fn定义的Collection。
以下为获取模型正则项损失值代码：
def model_fn(inputs, mode, **kwargs):
...
# 错误用法
# reg_losses = tf.get_collection(tf.GraphKeys.REGULARIZATION_LOSSES)
# 正确用法
reg_losses = mox.get_collection(tf.GraphKeys.REGULARIZATION_LOSSES)
2 生成对抗模型GAN
创建并训练一个DCGAN-MNIST模型，由此开源代码转换为MoXing实现方式。
3 利用output_fn做预测
在model_fn中的节点都是以tf.Tensor的形式构建在流图中，MoXing中可以提供output_fn用于获取并输出model_fn中的tf.Tensor的值。
output_fn的基本使用方法：
def input_fn(mode, **kwargs):
...
def model_fn(inputs, mode, **kwargs):
...
predictions = ...
...
return mox.ModelSpec(..., output_dict={'predictions': predictions}, ...)
def output_fn(outputs, **kwargs):
print(outputs)
mox.run(...
output_fn=output_fn,
output_every_n_steps=10,
...)
其中，在model_fn中的output_dict指定输出值对应的tf.Tensor，在mox.run中注册output_fn，当output_every_n_steps为10时，每经过10个step（注意在分布式运行中，这里的step指的是local_step），output_fn就会被调用一次，并且输入参数outputs为一个长度为10的list，每个元素为一个dict: {'predictions': ndarray}。在这里，outputs的值即为：
[{'predictions': ndarray_step_i}, ..., {'predictions': ndarray_step_i+9}]
注意，如果用户使用了多GPU，则outputs每次被调用时的输入参数outputs的长度为GPU数量*output_every_n_steps，分别表示[(step-0,GPU-0), (step-0,GPU-1), (step-1,GPU-0), ..., (step-9,GPU-1)]
案例，用ResNet_v1_50做预测，将max_number_of_steps和output_every_n_steps的值设置一致，也就是说output_fn只会被调用一次，输入参数为所有steps的预测结果prediction。然后将预测的结果输出到DataFrame中并写到文件里。
4 导出PB模型
MoXing在mox.run执行完毕后（训练完成或是验证完成），可以导出模型，基本用法为：
def input_fn(mode, **kwargs):
...
def model_fn(inputs, mode, **kwargs):
...
return mox.ModelSpec(...,
export_spec=mox.ExportSpec(inputs_dict={...}, outputs_dict={...}, ...),
...)
mox.run(...,
export_model=mox.ExportKeys.XXX,
...)
其中，mox.ExportSpec指定了导出模型的输入输出节点，仅能选取model_fn内部定义的tf.Tensor，mox.ExportKeys指定了导出模型的类型。
案例，训练一个ResNet_v1_50模型，在训练结束后导出用于TF-Serving的PB模型：
https://github.com/huaweiyun7759/backup/new/master/MOXING%20CHAPTER%204
当训练完成后，model_fn将以mode=mox.ModeKeys.EXPORT被调用（用户构建导出模型的流图），在此次调用过程中：
1) 当auto_batch为False时，inputs的shape和训练时保持一致，即images.shape=[32, 224, 224, 3], labels.shape=[32]。当auto_batch为True时，inputs中batch_size的维度会被置为None，即images.shape=[None, 224, 224, 3], labels.shape=[None]，所以就会导致输出节点logits的batch_size维度也为None，即logits.shape=[None, 1000]。
2) 导出的模型中计算节点的device信息为空。
DLS服务中预测作业使用的即是mox.ExportKeys.TF_SERVING类型的PB模型。
启动预测作业，如果提示信息类似如下：
tensorflow_serving/sources/storage_path/file_system_storage_path_source.cc:268] No versions of servable resnet_v1_50 found under base path s3://dls-test/log/resnet_v1_50/1/
说明没有找到可以用于TF-Serving的模型文件。导出的模型应有如下目录结构
|- log_dir
|- 1
|- svaed_model.pb
|- variables
|- variables.data-00000-of-00001
|- variables.index
其中1表示模型的版本号，启动预测服务需要指定到目录log_dir这层，在这个案例中就是s3://dls-test/log/resnet_v1_50而不是s3://dls-test/log/resnet_v1_50/1/。
当导出模型的目录下有多个版本号的模型时，如1，2，99，TF-Serving会自动选取数字最大99的模型做预测，当一个作业往该目录下继续输出了模型100，TF-Serving预测服务不需要重启，自动切换到100的模型上。在MoXing中，mox.ExportSpec(..., version=x, ...)，version参数就是用来指定该版本号，缺省值为-1，表示自动自增，即在输出目录下找到最大的版本号并+1，然后保存。
如出现如下错误信息：
InvalidArgumentError (see above for traceback): Default MaxPoolingOp only supports NHWC.
这个错误可能在训练作业、预测作业中遇到。原因是当使用CPU时，模型中的某些算子不支持NHWC数据格式，可能的情况如下：
1）DLS服务中，使用预置模型库训练模型（使用GPU训练），运行参数有data_format=NCHW，训练完成后使用导出的模型启动预测作业（由于目前预测作业仅支持CPU）。预测作业中出现该错误。
2）DLS服务中，使用预置模型库训练模型（使用CPU训练），并且数据格式为NCHW（即运行参数data_format=NCHW。
3）本地MoXing开发，模型中有不支持NCHW的算子，并且使用CPU训练。
如出现如下错误信息：
AssertireplaceString: Export directory already exists. Please specify a different export directory: s3://bucket_name/log/1
导出模型时如果在输出日志路径（train_url或是log_dir）中存在一个1的目录，并且还指定了version=1，则会出现该错误。指定一个不存在的版本号或者将版本号设置为自增(即version=-1)
在model_fn中，如果需要新建变量，建议使用tf.get_variable而不是tf.Variable。
当model_fn中的变量使用了tf.Variable来创建，并且损失值loss的计算中使用到了该变量，可能会出现如下错误信息：
v.name in list_allowed_variable_names_with_port())
AssertireplaceString
这是因为tf.Variable创建的变量无法被MoXing管理，替换为tf.get_variable即可解决。
另外，有一些隐藏调用tf.Variable的地方，如tf.train.AdamOptimizer中创建变量时使用了tf.Variable（仅针对TensorFlow-1.4及以下版本，TensorFlow-1.5及以上版本官方已修复），所以如果使用tf.train.AdamOptimizer遇到了类似的问题，MoXing提供了等价的API： mox.get_optimizer_fn('adam', ...)
5 Hook的使用
MoXing提供了允许在tf.train.MoniteredSession中注册hooks的方法，hooks要求为继承于tf.train.SessionRunHook的子类。MoXing中由于兼容了多GPU和分布式，因此要求用户注册的hooks为mox.AggregativeSessionRunHook的子类。AggregativeSessionRunHook继承于SessionRunHook，用户可以添加由SessionRunHook定义的回调函数begin, after_create_session, before_run, after_run, end。另外，用户还必须额外实现三个返回布尔值方法，support_aggregation，support_sync_workers，run_inter_mode，基本用法如下：
import tensorflow as tf
import moxing.tensorflow as mox
class MyHook(mox.AggregativeSessionRunHook):
def __init__(self, ...):
...
def support_aggregation(self):
return ...
def support_sync_workers(self):
return ...
def run_inter_mode(self):
return ...
...
def input_fn(mode, **kwargs):
...
def model_fn(inputs, mode, **kwargs):
...
hook = MyHook(...)
return mox.ModelSpec(..., hooks=[feed_hook], ...)
mox.run(...)
5.1 在model_fn中使用placeholder
用户可以在model_fn中利用hook创建并填充placeholder，MoXing提供了最基本的实现类FeedSessionRunHook，样例代码如下:
5.2 训练时打印验证集指标
在启动一个训练作业时，通常在训练时要不断观察模型在验证数据集上的各项指标。训练和验证在输入和模型上都不相同，所以至少要构建2个数据流图，分别为训练时的流图和验证时的流图。这就是inter_mode的作用，inter_mode允许在run_mode以外额外创建一个中间模式并在run_mode运行时穿插运行。基本用法：
def input_fn(mode, **kwargs):
...
def model_fn(inputs, mode, **kwargs):
...
mox.run(...,
run_mode=mox.ModeKeys.TRAIN,
inter_mode=mox.ModeKeys.EVAL,
...)
其中input_fn和model_fn都会以mox.ModeKeys.TRAIN和inter_mode=mox.ModeKeys.EVAL这两个模式被调用。
样例，训练一个ResNet_v1_50，使用mox.LogEvaluationMetricHook，每隔一定训练步数在验证数据集上打印loss和accuracy：
https://github.com/huaweiyun7759/backup/blob/master/MOXING%20CHAPTER%204/ResNet_v1_50
控制台输出日志可能会如下：
INFO:tensorflow:step: 0(global step: 0) sample/sec: 12.271 loss: 8.273 accuracy: 0.000
INFO:tensorflow:step: 10(global step: 10) sample/sec: 42.184 loss: 3.977 accuracy: 0.188
INFO:tensorflow:step: 20(global step: 20) sample/sec: 42.211 loss: 2.395 accuracy: 0.156
INFO:tensorflow:step: 30(global step: 30) sample/sec: 42.284 loss: 2.063 accuracy: 0.250
INFO:tensorflow:[VALIDATION METRICS] step: 31 loss: 17737.227 accuracy: 0.000
INFO:tensorflow:step: 40(global step: 40) sample/sec: 42.088 loss: 2.797 accuracy: 0.312
INFO:tensorflow:step: 50(global step: 50) sample/sec: 42.175 loss: 2.335 accuracy: 0.156
INFO:tensorflow:step: 60(global step: 60) sample/sec: 41.986 loss: 4.093 accuracy: 0.156
INFO:tensorflow:[VALIDATION METRICS] step: 63 loss: 99017.656 accuracy: 0.000
INFO:tensorflow:step: 70(global step: 70) sample/sec: 41.681 loss: 2.391 accuracy: 0.375
INFO:tensorflow:step: 80(global step: 80) sample/sec: 41.361 loss: 1.550 accuracy: 0.531
INFO:tensorflow:step: 90(global step: 90) sample/sec: 41.693 loss: 1.992 accuracy: 0.438
INFO:tensorflow:[VALIDATION METRICS] step: 95 loss: 9779.766 accuracy: 0.000
5.3 使用Early Stopping
在Keras-API中提供了tf.keras.callbacks.EarlyStopping的功能，MoXing中也用同样的API，用法和Keras的相似，为mox.EarlyStoppingHook。
Early Stopping是建立在同时提供训练集和验证集的前提上，当训练的模型在验证数据集上的指标(minotor)趋于稳定时，则停止训练。
样例代码，训练一个ResNet_v1_50，每训练一个epoch就在验证数据集上观察评价指标accuracy，当连续3次评价指标accuracy没有上升(第一次无法判断上升还是下降，所以至少评价4次)，则停止训练。
https://github.com/huaweiyun7759/backup/new/master/MOXING%20CHAPTER%204
控制台输出日志可能会如下：
INFO:tensorflow:step: 0(global step: 0) sample/sec: 15.875 loss: 7.753 accuracy: 0.000
INFO:tensorflow:step: 10(global step: 10) sample/sec: 42.087 loss: 3.451 accuracy: 0.312
INFO:tensorflow:[EarlyStopping] step: 19 accuracy: 0.000
INFO:tensorflow:step: 20(global step: 20) sample/sec: 40.802 loss: 4.920 accuracy: 0.250
INFO:tensorflow:step: 30(global step: 30) sample/sec: 41.427 loss: 4.368 accuracy: 0.281
INFO:tensorflow:[EarlyStopping] step: 39 accuracy: 0.000
INFO:tensorflow:step: 40(global step: 40) sample/sec: 41.678 loss: 2.614 accuracy: 0.281
INFO:tensorflow:step: 50(global step: 50) sample/sec: 41.816 loss: 2.788 accuracy: 0.219
INFO:tensorflow:[EarlyStopping] step: 59 accuracy: 0.000
INFO:tensorflow:step: 60(global step: 60) sample/sec: 41.407 loss: 2.861 accuracy: 0.094
INFO:tensorflow:step: 70(global step: 70) sample/sec: 41.929 loss: 2.075 accuracy: 0.469
INFO:tensorflow:[EarlyStopping] step: 79 accuracy: 0.000
Process finished with exit code 0
除了EarlyStopping，MoXing还提供了当检测到Plateau时自动下降学习率，当检测到多次Plateau并且评价指标没有上升或下降是，则停止训练，参考API：mox.PlateauLREarlyStoppingHook
6 利用Keras构建模型
MoXing本身除了支持TensorFlow和TensorFlow-slim的API来构建模型以外，还可以使用Keras-API来构建模型。根据Keras官方教程中的一个案例Multi-input and multi-output models，将其迁移到MoXing框架中，代码如下：
https://github.com/huaweiyun7759/backup/new/master/MOXING%20CHAPTER%204
当运行完成后，将模型以json的形式保存（不包含模型参数值，仅保存数据流图），利用以下代码可以载入该模型并训练（仅载入数据流图，载入模型参数值需要使用checkpoint_path）
import tensorflow as tf
import moxing.tensorflow as mox
from tensorflow.python.keras.losses import binary_crossentropy
from tensorflow.python.keras.models import model_from_json
def input_fn(mode, **kwargs):
main_input = tf.random_uniform(shape=(100,), minval=1, maxval=10000, dtype=tf.int32, name='main_input')
auxiliary_input = tf.random_normal(shape=(5,), name='aux_input')
main_labels = tf.random_uniform(shape=(1,))
auxiliary_labels = tf.random_uniform(shape=(1,))
return main_input, auxiliary_input, main_labels, auxiliary_labels
def model_fn(inputs, mode, **kwargs):
main_input, auxiliary_input, main_labels, auxiliary_labels = inputs
with tf.gfile.Open('/tmp/delete_me/keras_model.json', 'r') as f:
keras_model_json = f.read()
model_croe = model_from_json(keras_model_json)
main_output, auxiliary_output = model_croe([main_input, auxiliary_input])
loss = 1.0 * binary_crossentropy(main_output, main_labels) + \
0.2 * binary_crossentropy(auxiliary_output, auxiliary_labels)
loss = tf.reduce_mean(loss)
return mox.ModelSpec(loss=loss, log_info={'loss': loss})
if __name__ == '__main__':
mox.run(input_fn=input_fn,
model_fn=model_fn,
optimizer_fn=mox.get_optimizer_fn('rmsprop', learning_rate=0.01),
run_mode=mox.ModeKeys.TRAIN,
batch_size=32,
auto_batch=True,
log_dir=None,
max_number_of_steps=1000,
log_every_n_steps=10)
MoXing系列文章下期预告：优化器配置。
- 点赞
- 收藏
- 关注作者
评论（0）