TensorFlow Keras 自定义组件的实现机制和代码实践

1 .

2 .孔晓泉 谷歌开发技术专家（GDE）（Machine Learning方向） TensorFlow 和 TensorFlow Addons contributor。 TFUG社区主要组织者。Rasa (an open-source conversational AI framework) SuperHero contributor 和 Rasa 中国社区 Founder。曾就职于 Alibaba Group 从事 NLP 算法和框架研发工作。 

3 .TensorFlow Keras 自定义组件 实现机制和代码实践 

4 .• TensorFlow Keras 简介 • Keras 原生扩展机制 • TensorFlow Keras 扩展机制 • 编写你的自定义扩展代码 • 面向代码编程 

5 .TensorFlow Keras 简介 

6 .TensorFlow Keras 是 TensorFlow 对 Keras API 规范的实现。 Keras API 规范不是 https://keras.io/ 的那个 Keras 具体实现。 Keras API 规范定了一组用于构建和训练模型的高级 API。 其具有简单直观、容易学习、写起来代码简洁、扩展方便等特点。 Keras API 目前已经成为多数机器学习框架在 User level API 设计上的重点参考对 象。 

7 .Keras API 规范 涵盖: - 模型的表达 - 损失的表达 - 优化器 - 激活函数 - metric的计算 - 插件机制 

8 .TensorFlow Keras VS Keras Keras (https://keras.io/) 的多个 backend 中就有 TensorFlow. 那么原生的 Keras 和 TensorFlow Keras 有什么区别呢？ TensorFlow Keras 其中包括对 TensorFlow 特定功能的支持， 例如 Eager Execution，tf.data pipeline 和 Estimator。 tf.keras 使 TensorFlow 易于使用，而不会牺牲灵活性和性能。 

9 .François Chollet 

10 .Keras 原生扩展机制 

11 .自定义 Layer 

12 .自定义 metric 

13 .自定义 Callback 

14 .TensorFlow Keras 扩展机制 

15 .TensorFlow Keras： • 兼容 Karas • 为 Keras (https://keras.io/) 编写的自定义组件无需修改即可 直接在 TensorFlow Keras 调用 • 增加了自己的实现方案 • 采用了更加现代更加灵活的自定义实现方式 • 对独有的 Eager Execution 的支持 • tf.function 自动生成图 

16 .自定义 Layer class Sparsemax(tf.keras.layers.Layer): def __init__(self, axis=-1, **kwargs): super(Sparsemax, self).__init__(**kwargs) self.supports_masking = True self.axis = axis def call(self, inputs): return sparsemax(inputs, axis=self.axis) def get_config(self): config = {'axis': self.axis} base_config = super(Sparsemax, self).get_config() return dict(list(base_config.items()) + list(config.items())) def compute_output_shape(self, input_shape): return input_shape 

17 .自定义 Loss class ContrastiveLoss(tf.keras.losses.Loss): def __init__(self, margin=1.0, reduction=tf.keras.losses.Reduction.SUM_OVER_BATCH_SIZE, name="contrasitve_loss"): super(ContrastiveLoss, self).__init__(reduction=reduction, name=name) self.margin = margin def call(self, y_true, y_pred): return contrastive_loss(y_true, y_pred, self.margin) def get_config(self): config = { "margin": self.margin, } base_config = super(ContrastiveLoss, self).get_config() return dict(list(base_config.items()) + list(config.items())) 

18 .自定义 Metric class RSquare(tf.keras.metrics.Metric): def __init__(self, name='r_square', dtype=tf.float32): super(RSquare, self).__init__(name=name, dtype=dtype) self.squared_sum = self.add_weight("squared_sum", initializer="zeros") self.sum = self.add_weight("sum", initializer="zeros") self.res = self.add_weight("residual", initializer="zeros") self.count = self.add_weight("count", initializer="zeros") def update_state(self, y_true, y_pred): y_true = tf.convert_to_tensor(y_true, tf.float32) y_pred = tf.convert_to_tensor(y_pred, tf.float32) self.squared_sum.assign_add(tf.reduce_sum(y_true**2)) self.sum.assign_add(tf.reduce_sum(y_true)) self.res.assign_add( tf.reduce_sum(tf.square(tf.subtract(y_true, y_pred)))) self.count.assign_add(tf.cast(tf.shape(y_true)[0], tf.float32)) def result(self): mean = self.sum / self.count total = self.squared_sum - 2 * self.sum * mean + self.count * mean**2 return 1 - (self.res / total) def reset_states(self): # The state of the metric will be reset at the start of each epoch. self.squared_sum.assign(0.0) self.sum.assign(0.0) self.res.assign(0.0) self.count.assign(0.0) 

19 .编写你的自定义扩展代码 

20 .自定义 Layer class <LayerNmae>(tf.keras.layers.Layer): def __init__(self): // your code here def build(self, input_shape): // your code here def call(self, inputs): // your code here def get_config(self): // your code here def compute_output_shape(self, input_shape): // your code here 

21 .自定义 Layer > __init__ class <LayerNmae>(tf.keras.layers.Layer): use_XXX=True, def __init__(self): XXX_initializer='glorot_uniform’, // your code here XXX_regularizer=None, XXX_constraint=None, def build(self, input_shape): // your code here def call(self, inputs): // your code here def get_config(self): // your code here def compute_output_shape(self, input_shape): // your code here 

22 .自定义 Layer > build class <LayerNmae>(tf.keras.layers.Layer): if self.use_XXX: def __init__(self): self.XXX = self.add_weight( // your code here 'XXX', shape=[self.XXX,], def build(self, input_shape): initializer=self.XXX_initializer, // your code here regularizer=self.XXX_regularizer, constraint=self.XXX_constraint, def call(self, inputs): dtype=self.dtype, // your code here trainable=True) else: def get_config(self): self.XXX = None // your code here self.built = True def compute_output_shape(self, input_shape): // your code here 

23 .自定义 Layer > call class <LayerNmae>(tf.keras.layers.Layer): return tensor_shape.TensorShape(input_shape) def __init__(self): // your code here def build(self, input_shape): // your code here def call(self, inputs): // your code here def get_config(self): // your code here def compute_output_shape(self, input_shape): // your code here 

24 .自定义 Layer > call class <LayerNmae>(tf.keras.layers.Layer): outputs = tf.math.mat_mul(inputs, self.XXX) def __init__(self): return outputs // your code here def build(self, input_shape): // your code here def call(self, inputs): // your code here def get_config(self): // your code here def compute_output_shape(self, input_shape): // your code here 

25 .自定义 Layer > call config = { ‘XXX’: self.XXX, class <LayerNmae>(tf.keras.layers.Layer): ‘activation’: def __init__(self): activations.serialize(self.activation), // your code here ‘use_XXX’: self.use_XXX, ‘XXX_initializer’: def build(self, input_shape): initializers.serialize(self.kernel_XXX), // your code here ‘XXX_regularizer’: regularizers.serialize(self.XXX_regularizer), def call(self, inputs): ‘XXX_constraint’: // your code here constraints.serialize(self.XXX_constraint) } def get_config(self): // your code here base_config = super(LayerNmae,self).get_config() def compute_output_shape(self, input_shape): return dict(list(base_config.items()) + // your code here list(config.items())) 

26 .自定义 Loss class <LossName>(tf.keras.losses.Loss): def __init__(self, // your code here def call(self, y_true, y_pred): // your code here def get_config(self): // your code here 

27 .自定义 Loss class <LossName>(tf.keras.losses.Loss): __init__ 和 get_config 的作用和 def __init__(self, Layer 层的同名方法相似 // your code here def call(self, y_true, y_pred): call() 函数的内容需要实现者自 // your code here 行实现，返回一个实数，这样优 化器才能利用梯度优化网络权重 def get_config(self): // your code here 

28 .自定义 Metric class MetricName(tf.keras.metrics.Metric): def __init__(self, name=‘<metric_name>’): // your code here def update_state(self, y_true, y_pred): // your code here def result(self): // your code here def reset_states(self): # The state of the metric will be reset at the start of each epoch. // your code here def get_config(self): // your code here 

29 .自定义 Metric class MetricName(tf.keras.metrics.Metric): __init__ 与 get_config 和前面的 def __init__(self, name=‘<metric_name>’): Layer 和 Loss 类的同名方法是 // your code here 一样的 def update_state(self, y_true, y_pred): // your code here 同时，由于 Metric 是有状态的， 所以在 __init__ 里使用 def result(self): self.add_weight() 增加一个状态 // your code here 变量 def reset_states(self): # The state of the metric will be reset at the start of each epoch. // your code here def get_config(self): // your code here