tf.keras的各种自定义和高级用法（待续）

2021 年 2 月 25 日
AI

Keras & Tensorflow 2.0: Custom Layers & Modelswww.kaggle.com

kaggle上的一个很简明的教程。

使用内置方法进行训练和评估 | TensorFlow Coretensorflow.google.cn

这里面的keras的api给了非常多的细致的定义，搭配苏建林的文章，基本够了

tf.keras 各种自定义

metric的自定义是最简单的：

1、使用train_on_batch，个人认为最最灵活的方式了，train_on_batch即训练一个batch size的数据之后，model直接predict得到y_pred，numpy形式，然后sklearn的metric或者自定义基于numpy的metric计算都可以使用，非常的灵活，个人比较推荐train on batch，当然了，直接fit的时候epochs设置为1，然后一个batch 一个batch的读取原始数据也是可以的，这一点上tf.keras是非常灵活的；基于numpy的自定义metric函数可以使用numba来加速大大提高evaluation的效率；

2、通过

import tensorflow.keras.backend as K

def mean_pred(y_true, y_pred):
    return K.mean(y_pred)

model.compile(optimizer='rmsprop',
              loss='binary_crossentropy',
              metrics=['accuracy', mean_pred])

因为在model fit的过程中得到的中间结果是tf的tensor类型，因此无法直接在内部使用numpy函数，可以通过使用backend来写，backend的常规用法和numpy api很类似，基本上可以无缝衔接；

当然，我们也可以将y_true,y_pred通过 ..numpy()的方式转化为numpy，然后使用sklearn的或自定义的基于numpy的评估指标，最后通过convert_to_tensor再转化为tensor也是很灵活的；

loss的自定义也不复杂：

def smape_error(y_true, y_pred):
    return K.mean(K.clip(K.abs(y_pred - y_true),  0.0, 1.0), axis=-1)

和metric的定义类似，我们可以使用tf.keras.backend来定义，但是不能转化为numpy了，因为转化为numpy之后，tf底层无法识别numpy数据类型，无法针对自定义的loss进行autograd，不过其实直接用backend基本够了，自带的函数基本上和常见的numpy函数是一样的。

backend的内置methods

['__builtins__',
 '__cached__',
 '__doc__',
 '__file__',
 '__loader__',
 '__name__',
 '__package__',
 '__path__',
 '__spec__',
 '_sys',
 'abs',
 'all',
 'any',
 'arange',
 'argmax',
 'argmin',
 'backend',
 'batch_dot',
 'batch_flatten',
 'batch_get_value',
 'batch_normalization',
 'batch_set_value',
 'bias_add',
 'binary_crossentropy',
 'cast',
 'cast_to_floatx',
 'categorical_crossentropy',
 'clear_session',
 'clip',
 'concatenate',
 'constant',
 'conv1d',
 'conv2d',
 'conv2d_transpose',
 'conv3d',
 'cos',
 'count_params',
 'ctc_batch_cost',
 'ctc_decode',
 'ctc_label_dense_to_sparse',
 'cumprod',
 'cumsum',
 'depthwise_conv2d',
 'dot',
 'dropout',
 'dtype',
 'elu',
 'epsilon',
 'equal',
 'eval',
 'exp',
 'expand_dims',
 'eye',
 'flatten',
 'floatx',
 'foldl',
 'foldr',
 'function',
 'gather',
 'get_uid',
 'get_value',
 'gradients',
 'greater',
 'greater_equal',
 'hard_sigmoid',
 'image_data_format',
 'in_test_phase',
 'in_top_k',
 'in_train_phase',
 'int_shape',
 'is_keras_tensor',
 'is_sparse',
 'l2_normalize',
 'learning_phase',
 'learning_phase_scope',
 'less',
 'less_equal',
 'local_conv1d',
 'local_conv2d',
 'log',
 'manual_variable_initialization',
 'map_fn',
 'max',
 'maximum',
 'mean',
 'min',
 'minimum',
 'moving_average_update',
 'name_scope',
 'ndim',
 'normalize_batch_in_training',
 'not_equal',
 'one_hot',
 'ones',
 'ones_like',
 'permute_dimensions',
 'placeholder',
 'pool2d',
 'pool3d',
 'pow',
 'print_tensor',
 'prod',
 'random_binomial',
 'random_normal',
 'random_normal_variable',
 'random_uniform',
 'random_uniform_variable',
 'relu',
 'repeat',
 'repeat_elements',
 'reset_uids',
 'reshape',
 'resize_images',
 'resize_volumes',
 'reverse',
 'rnn',
 'round',
 'separable_conv2d',
 'set_epsilon',
 'set_floatx',
 'set_image_data_format',
 'set_learning_phase',
 'set_value',
 'shape',
 'sigmoid',
 'sign',
 'sin',
 'softmax',
 'softplus',
 'softsign',
 'sparse_categorical_crossentropy',
 'spatial_2d_padding',
 'spatial_3d_padding',
 'sqrt',
 'square',
 'squeeze',
 'stack',
 'std',
 'stop_gradient',
 'sum',
 'switch',
 'tanh',
 'temporal_padding',
 'tile',
 'to_dense',
 'transpose',
 'truncated_normal',
 'update',
 'update_add',
 'update_sub',
 'var',
 'variable',
 'zeros',
 'zeros_like']

除了使用backend之外，tf.keras目前也支持直接使用tf的内置methods来自定义metric与loss，比如：

def customized_mse(y_true, y_pred):
    return tf.reduce_mean(tf.squre(y_pred - y_true))

然后是比较复杂的自定义优化器：

//github.com/bojone/accum_optimizer_for_keras/blob/master/accum_optimizer.pygithub.com

//kexue.fm/archives/6794kexue.fm

用时间换取效果：Keras梯度累积优化器 – 科学空间|Scientific Spaceskexue.fm

这块的逻辑定义比较复杂，因为用的比较少，暂时不研究了

自定义layer：

1、最简单的，使用lambda层，一行搞定，缺点是无法处理太复杂的逻辑，例如：

from keras.layers import *
from keras import backend as K

x_in = Input(shape=(10,))
x = Lambda(lambda x: x+2)(x_in) # 对输入加上2

Python/tf/keras/layers/Lambda” data-draft-node=”block” data-draft-type=”link-card” class=”LinkCard LinkCard–noImage”>//www.tensorflow.org/api_docs/python/tf/keras/layers/Lambdawww.tensorflow.org

lambda layer的用法和pandas中的lambda的用法基本类似，只不过自定义function的时候要通过tf或者tf.keras.backend的方式进行逻辑代码的撰写；

2、通过继承tf.keras.layer

//www.tensorflow.org/tutorials/customization/custom_layerswww.tensorflow.org

官方文档写的非常详细了：

实现自定义层的最佳方法是继承tf.keras.Layer类并实现：

__init__ ，在其中进行所有与输入无关的变量或常量的初始化
build，在其中知道输入张量的形状，并可以进行其余的初始化
call，在这里进行前向传播逻辑的计算的定义

请注意，不必等到build被调用才创建变量，也可以在__init__中直接定义。但是，在build中创建变量的好处在于，可以根据build的参数 input shape来自动调整变量的shape。而在__init__中创建变量意味着需要显式指定创建变量的形状。

看demo：

class MyDenseLayer(tf.keras.layers.Layer):
  def __init__(self, num_outputs):
    super(MyDenseLayer, self).__init__()
    self.num_outputs = num_outputs

  def build(self, input_shape):
    self.kernel = self.add_weight("kernel",
                                  shape=[int(input_shape[-1]),
                                         self.num_outputs])

  def call(self, input):
    return tf.matmul(input, self.kernel)

layer = MyDenseLayer(10)

主要，这里 call部分的input的shape，会自动关联到 build中的input_shape上；

更加复杂的例子，可见苏剑林大佬的科学空间：

“让Keras更酷一些！”：精巧的层与花式的回调 – 科学空间|Scientific Spaceskexue.fm

class Dense_with_Center_loss(Layer):

    def __init__(self, output_dim, **kwargs):
        self.output_dim = output_dim
        super(Dense_with_Center_loss, self).__init__(**kwargs)

    def build(self, input_shape):
        # 添加可训练参数
        self.kernel = self.add_weight(name='kernel',
                                      shape=(input_shape[1], self.output_dim),
                                      initializer='glorot_normal',
                                      trainable=True)
        self.bias = self.add_weight(name='bias',
                                    shape=(self.output_dim,),
                                    initializer='zeros',
                                    trainable=True)
        self.centers = self.add_weight(name='centers',
                                       shape=(self.output_dim, input_shape[1]),
                                       initializer='glorot_normal',
                                       trainable=True)

    def call(self, inputs):
        # 对于center loss来说，返回结果还是跟Dense的返回结果一致
        # 所以还是普通的矩阵乘法加上偏置
        self.inputs = inputs
        return K.dot(inputs, self.kernel) + self.bias

    def compute_output_shape(self, input_shape):
        return (input_shape[0], self.output_dim)

    def loss(self, y_true, y_pred, lamb=0.5):
        # 定义完整的loss
        y_true = K.cast(y_true, 'int32') # 保证y_true的dtype为int32
        crossentropy = K.sparse_categorical_crossentropy(y_true, y_pred, from_logits=True)
        centers = K.gather(self.centers, y_true[:, 0]) # 取出样本中心
        center_loss = K.sum(K.square(centers - self.inputs), axis=1) # 计算center loss
        return crossentropy + lamb * center_loss


f_size = 2

x_in = Input(shape=(784,))
f = Dense(f_size)(x_in)

dense_center = Dense_with_Center_loss(10)
output = dense_center(f)

model = Model(x_in, output)
model.compile(loss=dense_center.loss,
              optimizer='adam',
              metrics=['sparse_categorical_accuracy'])

# 这里是y_train是类别的整数id，不用转为one hot
model.fit(x_train, y_train, epochs=10)

这里是完全使用keras.backend的方式来定义的，当然也可以使用tf的函数来定义，这里给了一种非常灵活的方式，loss和层的权重有关系，将loss直接定义在层中。

//www.tensorflow.org/guide/keras/custom_layers_and_models?hl=zh-cnwww.tensorflow.org

//keras.io/examples/keras_recipes/antirectifier/keras.io

Keras documentation: Simple custom layer example: Antirectifier

Keras documentation: Simple custom layer example: Antirectifierkeras.io

另外就是序列化存储model的时候需要对自定义layer指定get config的方法，具体可见上。不过其实如果要序列化保存模型，个人觉得保存weight和biase就行，再用的时候定义一样的结构然后load weight，这样可以避免很多麻烦的问题。

自定义model：

自定义model其实很简单了，本质上就是对常规的keras的model搭建做了一层封装：

//www.tensorflow.org/tutorials/customization/custom_layerswww.tensorflow.org

class ResnetIdentityBlock(tf.keras.Model):
  def __init__(self, kernel_size, filters):
    super(ResnetIdentityBlock, self).__init__(name='')
    filters1, filters2, filters3 = filters

    self.conv2a = tf.keras.layers.Conv2D(filters1, (1, 1))
    self.bn2a = tf.keras.layers.BatchNormalization()

    self.conv2b = tf.keras.layers.Conv2D(filters2, kernel_size, padding='same')
    self.bn2b = tf.keras.layers.BatchNormalization()

    self.conv2c = tf.keras.layers.Conv2D(filters3, (1, 1))
    self.bn2c = tf.keras.layers.BatchNormalization()

  def call(self, input_tensor, training=False):
    x = self.conv2a(input_tensor)
    x = self.bn2a(x, training=training)
    x = tf.nn.relu(x)

    x = self.conv2b(x)
    x = self.bn2b(x, training=training)
    x = tf.nn.relu(x)

    x = self.conv2c(x)
    x = self.bn2c(x, training=training)

    x += input_tensor
    return tf.nn.relu(x)


block = ResnetIdentityBlock(1, [1, 2, 3])