Given the answer provided by @NikoNyrh is correct, I just want to add how you can make a custom pre-processing layer with tensorflow that you can integrate in your model.
Assume you want to build a min-max scaler:
import tensorflow as tf
from tensorflow.keras.layers.experimental.preprocessing import PreprocessingLayer
from tensorflow.keras.layers import Input
class MinMaxScaler(PreprocessingLayer):
def __init__(self, **kwargs):
super().__init__(**kwargs)
# assume normalization should occur on last axis
self.axis = 1
# weights: to be set in build()
self.min_value: tf.Variable = None
self.max_value: tf.Variable = None
def build(self, input_shape):
super().build(input_shape)
input_shape = tf.TensorShape(input_shape).as_list()
shape = (input_shape[self.axis],)
self.min_value = self.add_weight(
name="min_value", shape=shape, dtype=self.compute_dtype,
initializer='zeros', trainable=False)
self.max_value = self.add_weight(
name="max_value", shape=shape, dtype=self.compute_dtype,
initializer='zeros', trainable=False)
def call(self, inputs):
# Here the normalization occurs on the model' inputs
inputs = tf.convert_to_tensor(inputs, dtype=self.compute_dtype)
_range = self.max_value - self.min_value + tf.keras.backend.epsilon()
return (inputs - self.min_value) / _range
def update_state(self, data):
# Normalization logic here
data = tf.convert_to_tensor(data, dtype=self.compute_dtype)
# find min-max boundaries
batch_min = tf.reduce_min(data, axis=0)
batch_max = tf.reduce_max(data, axis=0)
# update the weights
self.min_value.assign(tf.minimum(self.min_value, batch_min))
self.max_value.assign(tf.maximum(self.max_value, batch_max))
def reset_state(self):
self.min_value.assign(tf.zeros_like(self.min_value))
self.max_value.assign(tf.zeros_like(self.max_value))
You can use the MinMaxScaler layer within a keras model easily:
# build some model
inp = Input(input_shape=your_input_shape)
scaler = MinMaxScaler()
norm_inp = scaler(inp)
# ... hidden layers
out = ...
model = tf.keras.Model(inp, out)
model.compile(...)
# before training the model, you should call adapt() to find the normalization weights
scaler.adapt(some_data)
print(scaler.min_value, scaler.max_value)
# finally you can train the model
model.fit(...)
# save everything
model.save_weights() # for example
Basically, the adapt() method computes the weights/parameters that are used to normalize the data when calling model(x), for example during inference. The custom layer can be included in a tf.keras.Model having all the saving and loading benefits, i.e., you don't have to keep track of the normalization statistics by yourself. Last advantage, is that the normalization/scaling logic can run on CPU/GPU and TPU as well.
Indeed, this is a minimum working example so in real code you want to add more functionalities and checks.
