<jupyter_start><jupyter_text>Chapter 12. custom models and training with tensorflow- Lower-lever Python API- for exact control to write custom    - loss functions    - metrics    - layers    - models    - initializers    - regularizers    - weight constrains    - ...    - fully control training loop, e.g.    - apply special transformations or constraints to the gradients beyond just clippig them    - use multiple optimizers for different parts of the network- automatic graph generation feature  A quick tour of TensorFlow- similar to NumPy, with GPU support- support distributed computing- computation graph optimize computations - model is portable across different platform- autodiff and excellent optimizers  Setup<jupyter_code># Python ≥3.5 is required
import sys
assert sys.version_info >= (3, 5)

# Scikit-Learn ≥0.20 is required
import sklearn
assert sklearn.__version__ >= "0.20"

try:
    # %tensorflow_version only exists in Colab.
    %tensorflow_version 2.x
    !pip install -U tqdm
except Exception:
    pass

# TensorFlow ≥2.0 is required
import tensorflow as tf
from tensorflow import keras
assert tf.__version__ >= "2.0"

# Common imports
import numpy as np
import os

# to make this notebook's output stable across runs
np.random.seed(42)
tf.random.set_seed(42)

# To plot pretty figures
%matplotlib inline
import matplotlib as mpl
import matplotlib.pyplot as plt
mpl.rc('axes', labelsize=14)
mpl.rc('xtick', labelsize=12)
mpl.rc('ytick', labelsize=12)

# Where to save the figures
PROJECT_ROOT_DIR = "."
CHAPTER_ID = "deep"
IMAGES_PATH = os.path.join(PROJECT_ROOT_DIR, "images", CHAPTER_ID)
os.makedirs(IMAGES_PATH, exist_ok=True)

def save_fig(fig_id, tight_layout=True, fig_extension="png", resolution=300):
    path = os.path.join(IMAGES_PATH, fig_id + "." + fig_extension)
    print("Saving figure", fig_id)
    if tight_layout:
        plt.tight_layout()
    plt.savefig(path, format=fig_extension, dpi=resolution)

print("Setup Complete.")<jupyter_output>Setup Complete.<jupyter_text>using tensorflow like numpy  tensors and operations<jupyter_code>tf_const_eg = tf.constant([[1., 2., 3.], 
                           [4., 5., 6.]]) # matrix
tf_const_eg
tf_const_eg.shape
tf_const_eg.dtype
tf_const_num = tf.constant(42) # scalar
tf_const_num
print(tf_const_num.numpy())
print(tf_const_eg.numpy())
print(type(tf_const_eg.numpy()))<jupyter_output><class 'numpy.ndarray'><jupyter_text>indexing tf is similar to that in NumPy<jupyter_code>t = tf_const_eg
t[:, 1:]<jupyter_output><empty_output><jupyter_text>??? **tf.newaxis**<jupyter_code>t[..., 1, tf.newaxis]
# addition by broadcasting
t + 10
# adding via .add() function
tf.add(t, 10)
# element-wise square
tf.square(t)<jupyter_output><empty_output><jupyter_text>The **@** operator in Python is for **matrix multiplication**, equvalent to calling **tf.matmul()** function. **tf.reduce_mean()** = **np.mean()****tf.reduce_sum()** = **np.sum()****tf.reduce_max()** = **np.max()****tf.math.log()** = **np.log()** However,... **tf.transpose(a)** is not the same as **a.T** in numpy, because it creates a new copy of the transposed data<jupyter_code>t = tf.constant([[1., 2., 3.], 
                 [4., 5., 6.]]) # tf matrix

n = np.array([[1., 2., 3.], 
              [4., 5., 6.]]) # np matrix

t
n
print(tf.transpose(t).numpy())
tf.transpose(t)
print(t)
print(tf.transpose(t))
print(t.numpy())
print(n.T)
print(n)<jupyter_output>[[1. 2. 3.]
 [4. 5. 6.]]<jupyter_text>Keras' low-level APIFor code to be **portable** to other Keras implementations, **low level Keras functions** should be used, instead of low level TensofFlow functions.<jupyter_code>from tensorflow import keras
K = keras.backend
K.square(K.transpose(t)) + 10<jupyter_output><empty_output><jupyter_text>Tensors and NumPy<jupyter_code># create tensor from np array
a = np.array([2., 4., 5.])
tf.constant(a)<jupyter_output><empty_output><jupyter_text>**_.numpy()_** : convert a **tensor** to **numpy array**<jupyter_code># convert tensor to np array
t.numpy()
np.array(t)
# apply tf function on np array
tf.square(a)
# apply np function on tensor
np.square(t)<jupyter_output><empty_output><jupyter_text>type conversions- TF doesn't perform type conversion automatically<jupyter_code>try:
    tf.constant(2.0) + tf.constant(40)
except tf.errors.InvalidArgumentError as ex:
    print(ex)
try:
    tf.constant(2.0) + tf.constant(40., dtype=tf.float64)
except tf.errors.InvalidArgumentError as ex:
    print(ex)<jupyter_output>cannot compute AddV2 as input #1(zero-based) was expected to be a float tensor but is a double tensor [Op:AddV2]<jupyter_text>use **_tf.cast()_** to convert type<jupyter_code>t2 = tf.constant(40., dtype=tf.float64)
tf.constant(2.0) + tf.cast(t2, tf.float32)
tf.constant(b'hello world')
tf.constant('cafe')
u = tf.constant([ord(c) for c in 'cafe'])
u<jupyter_output><empty_output><jupyter_text>variables<jupyter_code>v = tf.Variable([[1., 2., 3.], [4., 5., 6.]])
v
v.assign(2*v)
v[0, 1].assign(42)
v[:, 2].assign([0.1, 1.])
try:
    v[1] = [7., 8., 9.]
except TypeError as ex:
    print(ex)
v.scatter_nd_update(
    indices=[[0, 0], 
             [1, 2]],
    updates=[100., 200.]
)
sparse_delta = tf.IndexedSlices(
    values=[[1., 2., 3.], [4., 5., 6.]], 
    indices=[1, 0]
)
v.scatter_update(sparse_delta)<jupyter_output><empty_output><jupyter_text>other data structures  customizing models & training algorithms  custom loss functionscustomize a **Huber loss** function<jupyter_code>from sklearn.datasets           import fetch_california_housing
from sklearn.model_selection    import train_test_split
from sklearn.preprocessing      import StandardScaler

housing = fetch_california_housing()
X_train_full, X_test, y_train_full, y_test \
= train_test_split(
    housing.data, 
    housing.target.reshape(-1, 1), 
    random_state=42
    )
X_train, X_valid, y_train, y_valid \
= train_test_split(
    X_train_full, 
    y_train_full, 
    random_state=42
    )

scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_valid_scaled = scaler.transform(X_valid)
X_test_scaled = scaler.transform(X_test)
def huber_fn(y_true, y_pred):
    error = y_true - y_pred
    is_small_error  = tf.abs(error) < 1
    squared_loss    = tf.square(error) / 2
    linear_loss     = tf.abs(error) - 0.5
    return tf.where(is_small_error, squared_loss, linear_loss)

plt.figure(figsize=(8, 3.5))
z = np.linspace(-4, 4, 200)
plt.plot(z, huber_fn(0, z), "b-", linewidth=2, label="huber($z$)")
plt.plot(z, z**2 / 2, "b:", linewidth=1, label=r"$\frac{1}{2}z^2$")
plt.plot([-1, -1], [0, huber_fn(0., -1.)], "r--")
plt.plot([1, 1], [0, huber_fn(0., 1.)], "r--")
plt.gca().axhline(y=0, color='k')
plt.gca().axvline(x=0, color='k')
plt.axis([-4, 4, 0, 4])
plt.grid(True)
plt.xlabel("$z$")
plt.legend(fontsize=14)
plt.title("Huber loss", fontsize=14)
plt.show()
input_shape = X_train.shape[1:]

model = keras.models.Sequential([
    keras.layers.Dense(
        30, 
        activation='selu', 
        kernel_initializer='lecun_normal', 
        input_shape=input_shape),
    keras.layers.Dense(1), 
])
model.compile(
    loss=huber_fn, 
    optimizer='nadam', 
    metrics=['mae'])
model.fit(
    X_train_scaled,
    y_train, 
    epochs=10,
    validation_data=(X_valid_scaled, y_valid))<jupyter_output>Epoch 1/10
363/363 [==============================] - 1s 2ms/step - loss: 0.6235 - mae: 0.9953 - val_loss: 0.2862 - val_mae: 0.5866
Epoch 2/10
363/363 [==============================] - 0s 1ms/step - loss: 0.2197 - mae: 0.5177 - val_loss: 0.2382 - val_mae: 0.5281
Epoch 3/10
363/363 [==============================] - 0s 1ms/step - loss: 0.2051 - mae: 0.4975 - val_loss: 0.2196 - val_mae: 0.5040
Epoch 4/10
363/363 [==============================] - 0s 1ms/step - loss: 0.2002 - mae: 0.4908 - val_loss: 0.2069 - val_mae: 0.4917
Epoch 5/10
363/363 [==============================] - 1s 1ms/step - loss: 0.1964 - mae: 0.4843 - val_loss: 0.1940 - val_mae: 0.4751
Epoch 6/10
363/363 [==============================] - 0s 1ms/step - loss: 0.1935 - mae: 0.4795 - val_loss: 0.1893 - val_mae: 0.4705
Epoch 7/10
363/363 [==============================] - 0s 1ms/step - loss: 0.1912 - mae: 0.4761 - val_loss: 0.1851 - val_mae: 0.4627
Epoch 8/10
363/363 [==============================] - 0s 1ms/step - loss: 0.[...]<jupyter_text>save/load models with custom objects<jupyter_code>model.save('my_model_with_a_custom_loss.h5')
model = keras.models.load_model(
    'my_model_with_a_custom_loss.h5', 
    custom_objects={'huber_fn': huber_fn}
)
model.fit(
    X_train_scaled,
    y_train, 
    epochs=2,
    validation_data=(X_valid_scaled, y_valid))
def create_huber(threshold=1.0):
    def huber_fn(y_true, y_pred):
        error = y_true - y_pred
        is_small_error = tf.abs(error) < threshold
        squared_loss = tf.square(error) / 2
        linear_loss = threshold * tf.abs(error) - threshold**2 / 2
        return tf.where(is_small_error, squared_loss, linear_loss)
    return huber_fn
model.compile(
    loss=create_huber(2.0),
    optimizer='nadam', 
    metrics=['mae'])
model.fit(
    X_train_scaled,
    y_train, 
    epochs=2,
    validation_data=(X_valid_scaled, y_valid))
model.save('my_model_with_a_custom_loss_threshold_2.h5')
model = keras.models.load_model(
    'my_model_with_a_custom_loss_threshold_2.h5',
    custom_objects={'huber_fn': create_huber(2.0)})
model.fit(
    X_train_scaled,
    y_train, 
    epochs=2,
    validation_data=(X_valid_scaled, y_valid))
class HuberLoss(keras.losses.Loss):
    def __init__(self, threshold=1.0, **kwargs):
        self.threshold = threshold
        super().__init__(**kwargs)
    def call(self, y_true, y_pred):
        error = y_true - y_pred
        is_small_error = tf.abs(error) < self.threshold
        squared_loss = tf.square(error) / 2
        linear_loss  = self.threshold * tf.abs(error) - self.threshold**2 / 2
        return tf.where(is_small_error, squared_loss, linear_loss)
    def get_config(self):
        base_config = super().get_config()
        return {**base_config, "threshold": self.threshold}
model = keras.models.Sequential([
    keras.layers.Dense(30, activation="selu", kernel_initializer="lecun_normal",
                       input_shape=input_shape),
    keras.layers.Dense(1),
])
model.compile(loss=HuberLoss(2.), optimizer="nadam", metrics=["mae"])
model.fit(X_train_scaled, y_train, epochs=2,
          validation_data=(X_valid_scaled, y_valid))
model.save("my_model_with_a_custom_loss_class.h5")
model = keras.models.load_model(
    "my_model_with_a_custom_loss_class.h5", 
    custom_objects={"HuberLoss": HuberLoss})
model.fit(
    X_train_scaled, 
    y_train, 
    epochs=2,
    validation_data=(X_valid_scaled, y_valid))
model = keras.models.load_model(
    "my_model_with_a_custom_loss_class.h5",
    custom_objects={"HuberLoss": HuberLoss})
model.loss.threshold<jupyter_output><empty_output><jupyter_text>other custom functions<jupyter_code>keras.backend.clear_session()
np.random.seed(42)
tf.random.set_seed(42)
def my_softplus(z): # return value is just tf.nn.softplus(z)
    return tf.math.log(tf.exp(z) + 1.0)

def my_glorot_initializer(shape, dtype=tf.float32):
    stddev = tf.sqrt(2. / (shape[0] + shape[1]))
    return tf.random.normal(shape, stddev=stddev, dtype=dtype)

def my_l1_regularizer(weights):
    return tf.reduce_sum(tf.abs(0.01 * weights))

def my_positive_weights(weights): # return value is just tf.nn.relu(weights)
    return tf.where(weights < 0., tf.zeros_like(weights), weights)
layer = keras.layers.Dense(1, activation=my_softplus,
                           kernel_initializer=my_glorot_initializer,
                           kernel_regularizer=my_l1_regularizer,
                           kernel_constraint=my_positive_weights)
keras.backend.clear_session()
np.random.seed(42)
tf.random.set_seed(42)
model = keras.models.Sequential([
    keras.layers.Dense(30, 
                       activation="selu", 
                       kernel_initializer="lecun_normal",
                       input_shape=input_shape),
    keras.layers.Dense(1, 
                       activation=my_softplus,
                       kernel_regularizer=my_l1_regularizer,
                       kernel_constraint=my_positive_weights,
                       kernel_initializer=my_glorot_initializer),
])
model.compile(
    loss="mse", 
    optimizer="nadam", 
    metrics=["mae"])

model.fit(X_train_scaled, y_train, epochs=2,
          validation_data=(X_valid_scaled, y_valid))
model.save("my_model_with_many_custom_parts.h5")
model = keras.models.load_model(
    "my_model_with_many_custom_parts.h5",
    custom_objects={
       "my_l1_regularizer": my_l1_regularizer,
       "my_positive_weights": my_positive_weights,
       "my_glorot_initializer": my_glorot_initializer,
       "my_softplus": my_softplus,
    })
class MyL1Regularizer(keras.regularizers.Regularizer):
    def __init__(self, factor):
        self.factor = factor
    def __call__(self, weights):
        return tf.reduce_sum(tf.abs(self.factor * weights))
    def get_config(self):
        return {"factor": self.factor}
keras.backend.clear_session()
np.random.seed(42)
tf.random.set_seed(42)
model = keras.models.Sequential([
    keras.layers.Dense(30, activation="selu", kernel_initializer="lecun_normal",
                       input_shape=input_shape),
    keras.layers.Dense(1, activation=my_softplus,
                       kernel_regularizer=MyL1Regularizer(0.01),
                       kernel_constraint=my_positive_weights,
                       kernel_initializer=my_glorot_initializer),
])
model.compile(loss="mse", optimizer="nadam", metrics=["mae"])
model.fit(X_train_scaled, y_train, epochs=2,
          validation_data=(X_valid_scaled, y_valid))
model.save("my_model_with_many_custom_parts.h5")
model = keras.models.load_model(
    "my_model_with_many_custom_parts.h5",
    custom_objects={
       "MyL1Regularizer": MyL1Regularizer,
       "my_positive_weights": my_positive_weights,
       "my_glorot_initializer": my_glorot_initializer,
       "my_softplus": my_softplus,
    })<jupyter_output><empty_output><jupyter_text>custom metrics  Losses- e.g. **cross entropy**- used by **Gradient Descent** to **_train_** a model, so they must be **differentiable**- it's ok if they're not easily interpretable by humams. Metrics- e.g. **accuracy**- used to **_evaluate_** a model- can be non-differnetiable- must be easily interpretable by humans.<jupyter_code>keras.backend.clear_session()
np.random.seed(42)
tf.random.set_seed(42)
model = keras.models.Sequential([
    keras.layers.Dense(
        30, 
        activation="selu", 
        kernel_initializer="lecun_normal",
        input_shape=input_shape
    ),
    keras.layers.Dense(1),
])
model.compile(
    loss="mse", 
    optimizer="nadam", 
    metrics=[create_huber(2.0)]
)
model.fit(X_train_scaled, y_train, epochs=2)
model.compile(
    loss=create_huber(2.0), 
    optimizer="nadam", 
    metrics=[create_huber(2.0)]
)
sample_weight = np.random.rand(len(y_train))
history = model.fit(X_train_scaled, y_train, epochs=2, sample_weight=sample_weight)
history.history["loss"][0], history.history["huber_fn"][0] * sample_weight.mean()<jupyter_output><empty_output><jupyter_text>streaming metrics<jupyter_code>precision = keras.metrics.Precision()
precision([0,1,1,1,0,1,0,1], [1,1,0,1,0,1,0,1])
precision([0,1,0,0,1,0,1,1], [1,0,1,1,0,0,0,0])
precision.result()
precision.variables
precision.reset_states()<jupyter_output><empty_output><jupyter_text>create a streaming metric :<jupyter_code>class HuberMetric(keras.metrics.Metric):
    def __init__(self, threshold=1.0, **kwargs):
        super().__init__(**kwargs) # handles base args (e.g., dtype)
        self.threshold = threshold
        #self.huber_fn = create_huber(threshold) # TODO: investigate why this fails
        self.total = self.add_weight("total", initializer="zeros")
        self.count = self.add_weight("count", initializer="zeros")
    def huber_fn(self, y_true, y_pred): # workaround
        error = y_true - y_pred
        is_small_error = tf.abs(error) < self.threshold
        squared_loss = tf.square(error) / 2
        linear_loss  = self.threshold * tf.abs(error) - self.threshold**2 / 2
        return tf.where(is_small_error, squared_loss, linear_loss)
    def update_state(self, y_true, y_pred, sample_weight=None):
        metric = self.huber_fn(y_true, y_pred)
        self.total.assign_add(tf.reduce_sum(metric))
        self.count.assign_add(tf.cast(tf.size(y_true), tf.float32))
    def result(self):
        return self.total / self.count
    def get_config(self):
        base_config = super().get_config()
        return {**base_config, "threshold": self.threshold}
m = HuberMetric(2.)

# total = 2 * |10 - 2| - 2²/2 = 14
# count = 1
# result = 14 / 1 = 14
m(tf.constant([[2.]]), tf.constant([[10.]]))
# total = total + (|1 - 0|² / 2) + (2 * |9.25 - 5| - 2² / 2) = 14 + 7 = 21
# count = count + 2 = 3
# result = total / count = 21 / 3 = 7
m(tf.constant([[0.], [5.]]), tf.constant([[1.], [9.25]]))

m.result()
m.variables
m.reset_states()
m.variables<jupyter_output><empty_output><jupyter_text>check if the HuberMetric class works well :<jupyter_code>keras.backend.clear_session()
np.random.seed(42)
tf.random.set_seed(42)
model = keras.models.Sequential([
    keras.layers.Dense(30, activation="selu", kernel_initializer="lecun_normal",
                       input_shape=input_shape),
    keras.layers.Dense(1),
])
model.compile(
    loss=create_huber(2.0), 
    optimizer="nadam", 
    metrics=[HuberMetric(2.0)])
model.fit(
    X_train_scaled.astype(np.float32), 
    y_train.astype(np.float32), 
    epochs=2)
model.save("my_model_with_a_custom_metric.h5")
model.fit(X_train_scaled.astype(np.float32), y_train.astype(np.float32), epochs=2)
model.metrics[-1].threshold
class HuberMetric(keras.metrics.Mean):
    def __init__(self, threshold=1.0, name='HuberMetric', dtype=None):
        self.threshold = threshold
        self.huber_fn = create_huber(threshold)
        super().__init__(name=name, dtype=dtype)
    def update_state(self, y_true, y_pred, sample_weight=None):
        metric = self.huber_fn(y_true, y_pred)
        super(HuberMetric, self).update_state(metric, sample_weight)
    def get_config(self):
        base_config = super().get_config()
        return {**base_config, "threshold": self.threshold}
keras.backend.clear_session()
np.random.seed(42)
tf.random.set_seed(42)
model = keras.models.Sequential([
    keras.layers.Dense(30, activation="selu", kernel_initializer="lecun_normal",
                       input_shape=input_shape),
    keras.layers.Dense(1),
])
model.compile(loss=keras.losses.Huber(2.0), optimizer="nadam", weighted_metrics=[HuberMetric(2.0)])
sample_weight = np.random.rand(len(y_train))
history = model.fit(X_train_scaled.astype(np.float32), y_train.astype(np.float32),
                    epochs=2, sample_weight=sample_weight)
history.history["loss"][0], history.history["HuberMetric"][0] * sample_weight.mean()
model.save("my_model_with_a_custom_metric_v2.h5")
model.fit(X_train_scaled.astype(np.float32), y_train.astype(np.float32), epochs=2)
model.metrics[-1].threshold<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.15257688893530363, 'fraction_numerical': 0.03341862043757506, 'mean_word_length': 4.030732860520095, 'pattern_counts': {'":': 14, '<': 74, '<?xml version=': 0, '>': 72, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '18', 'dir': 'jupyter-structured-clean-dedup', 'id': '533184', 'license': 'MIT', 'n_tokens_mistral': 7377, 'n_tokens_neox': 6902, 'n_words': 1296, 'path': 'ch12_tensorflow.ipynb', 'repo_name': 'hopezh/learn_hands_on_ML_v2', 'size': '19151'}