Deep Learning

Deep Learning with Keras 3

Patterns and best practices based on Deep Learning with Python, 2nd Edition by François Chollet, updated for Keras 3 (Multi-Backend).

Core Workflow

1. Prepare Data: Normalize, split train/val/test, create tf.data.Dataset 2. Build Model: Sequential, Functional, or Subclassing API 3. Compile: model.compile(optimizer, loss, metrics) 4. Train: model.fit(data, epochs, validation_data, callbacks) 5. Evaluate: model.evaluate(test_data)

Model Building APIs

Sequential - Simple stack of layers:

model = keras.Sequential([
    layers.Dense(64, activation="relu"),
    layers.Dense(10, activation="softmax")
])

Functional - Multi-input/output, shared layers, non-linear topologies:

inputs = keras.Input(shape=(64,))
x = layers.Dense(64, activation="relu")(inputs)
outputs = layers.Dense(10, activation="softmax")(x)
model = keras.Model(inputs=inputs, outputs=outputs)

Subclassing - Full flexibility with call() method:

class MyModel(keras.Model):
    def __init__(self):
        super().__init__()
        self.dense1 = layers.Dense(64, activation="relu")
        self.dense2 = layers.Dense(10, activation="softmax")

    def call(self, inputs):
        x = self.dense1(inputs)
        return self.dense2(x)

Quick Reference: Loss & Optimizer Selection

Optimizers: rmsprop (default), adam (popular), sgd (with momentum for fine-tuning)

Domain-Specific Guides

Essential Callbacks

callbacks = [
    keras.callbacks.EarlyStopping(monitor="val_loss", patience=3),
    keras.callbacks.ModelCheckpoint("best.keras", save_best_only=True),
    keras.callbacks.TensorBoard(log_dir="./logs")
]
model.fit(..., callbacks=callbacks)

Utility Scripts

Advanced Computer Vision

Table of Contents

• Image Segmentation
• Modern Architecture Patterns
• Interpreting Convnets
• Object Detection & Beyond

Image Segmentation

Classifying every pixel in an image. Uses encoder-decoder architecture with Conv2DTranspose for upsampling.

def get_segmentation_model(img_size, num_classes):
    inputs = keras.Input(shape=img_size + (3,))
    x = layers.Rescaling(1./255)(inputs)

    # Encoder (downsampling)
    x = layers.Conv2D(64, 3, strides=2, activation="relu", padding="same")(x)
    x = layers.Conv2D(64, 3, activation="relu", padding="same")(x)
    x = layers.Conv2D(128, 3, strides=2, activation="relu", padding="same")(x)
    x = layers.Conv2D(128, 3, activation="relu", padding="same")(x)
    x = layers.Conv2D(256, 3, strides=2, activation="relu", padding="same")(x)
    x = layers.Conv2D(256, 3, activation="relu", padding="same")(x)

    # Decoder (upsampling)
    x = layers.Conv2DTranspose(256, 3, activation="relu", padding="same")(x)
    x = layers.Conv2DTranspose(256, 3, strides=2, activation="relu", padding="same")(x)
    x = layers.Conv2DTranspose(128, 3, activation="relu", padding="same")(x)
    x = layers.Conv2DTranspose(128, 3, strides=2, activation="relu", padding="same")(x)
    x = layers.Conv2DTranspose(64, 3, activation="relu", padding="same")(x)
    x = layers.Conv2DTranspose(64, 3, strides=2, activation="relu", padding="same")(x)

    # Per-pixel classification
    outputs = layers.Conv2D(num_classes, 3, activation="softmax", padding="same")(x)

    return keras.Model(inputs, outputs)

model = get_segmentation_model((200, 200), num_classes=3)
model.compile(optimizer="rmsprop", loss="sparse_categorical_crossentropy")

Modern Architecture Patterns

Residual Connections

Solve vanishing gradients in deep networks by adding skip connections:

def residual_block(x, filters, pooling=False):
    residual = x

    x = layers.Conv2D(filters, 3, activation="relu", padding="same")(x)
    x = layers.Conv2D(filters, 3, activation="relu", padding="same")(x)

    if pooling:
        x = layers.MaxPooling2D(2, padding="same")(x)
        residual = layers.Conv2D(filters, 1, strides=2)(residual)
    elif filters != residual.shape[-1]:
        residual = layers.Conv2D(filters, 1)(residual)

    x = layers.add([x, residual])
    return x

# Usage
x = residual_block(x, filters=32, pooling=True)
x = residual_block(x, filters=64, pooling=True)
x = residual_block(x, filters=128, pooling=False)

Batch Normalization

Normalize activations to stabilize training:

x = layers.Conv2D(32, 3, use_bias=False)(x)  # No bias needed with BN
x = layers.BatchNormalization()(x)
x = layers.Activation("relu")(x)

Depthwise Separable Convolutions

SeparableConv2D is lighter and faster than Conv2D, assuming spatial and channel correlations are independent (Xception architecture):

x = layers.SeparableConv2D(256, 3, padding="same", use_bias=False)(x)

Mini Xception-like Model

inputs = keras.Input(shape=(180, 180, 3))
x = data_augmentation(inputs)
x = layers.Rescaling(1./255)(x)
x = layers.Conv2D(32, 5, use_bias=False)(x)

for size in [32, 64, 128, 256, 512]:
    residual = x

    x = layers.BatchNormalization()(x)
    x = layers.Activation("relu")(x)
    x = layers.SeparableConv2D(size, 3, padding="same", use_bias=False)(x)

    x = layers.BatchNormalization()(x)
    x = layers.Activation("relu")(x)
    x = layers.SeparableConv2D(size, 3, padding="same", use_bias=False)(x)

    x = layers.MaxPooling2D(3, strides=2, padding="same")(x)

    residual = layers.Conv2D(size, 1, strides=2, padding="same", use_bias=False)(residual)
    x = layers.add([x, residual])

x = layers.GlobalAveragePooling2D()(x)
x = layers.Dropout(0.5)(x)
outputs = layers.Dense(1, activation="sigmoid")(x)
model = keras.Model(inputs, outputs)

Interpreting Convnets

Visualizing Intermediate Activations

See what each layer learns:

# Create model that outputs all conv layer activations
layer_outputs = []
layer_names = []
for layer in model.layers:
    if isinstance(layer, (layers.Conv2D, layers.MaxPooling2D)):
        layer_outputs.append(layer.output)
        layer_names.append(layer.name)

activation_model = keras.Model(inputs=model.input, outputs=layer_outputs)

# Get activations for an image
activations = activation_model.predict(img_array)

# Visualize first layer, channel 5
plt.matshow(activations[0][0, :, :, 5], cmap="viridis")

Grad-CAM (Class Activation Maps)

Visualize which image regions contributed most to a prediction:

# 1. Get last conv layer output and predictions
last_conv_layer = model.get_layer("block14_sepconv2_act")
last_conv_layer_model = keras.Model(model.inputs, last_conv_layer.output)

classifier_input = keras.Input(shape=last_conv_layer.output.shape[1:])
x = classifier_input
for layer_name in ["avg_pool", "predictions"]:
    x = model.get_layer(layer_name)(x)
classifier_model = keras.Model(classifier_input, x)

# 2. Compute gradients
with tf.GradientTape() as tape:
    last_conv_output = last_conv_layer_model(img_array)
    tape.watch(last_conv_output)
    preds = classifier_model(last_conv_output)
    top_pred_index = tf.argmax(preds[0])
    top_class_channel = preds[:, top_pred_index]

grads = tape.gradient(top_class_channel, last_conv_output)

# 3. Create heatmap
pooled_grads = tf.reduce_mean(grads, axis=(0, 1, 2)).numpy()
last_conv_output = last_conv_output.numpy()[0]

for i in range(pooled_grads.shape[-1]):
    last_conv_output[:, :, i] *= pooled_grads[i]

heatmap = np.mean(last_conv_output, axis=-1)
heatmap = np.maximum(heatmap, 0)
heatmap /= np.max(heatmap)

# 4. Overlay on image
import matplotlib.cm as cm
jet = cm.get_cmap("jet")
jet_heatmap = jet(np.uint8(255 * heatmap))[:, :, :3]
superimposed = jet_heatmap * 0.4 + original_img / 255.

Visualizing Convnet Filters

What patterns does each filter respond to:

@tf.function
def gradient_ascent_step(image, filter_index, learning_rate):
    with tf.GradientTape() as tape:
        tape.watch(image)
        activation = feature_extractor(image)
        filter_activation = activation[:, 2:-2, 2:-2, filter_index]
        loss = tf.reduce_mean(filter_activation)
    grads = tape.gradient(loss, image)
    grads = tf.math.l2_normalize(grads)
    image += learning_rate * grads
    return image

def generate_filter_pattern(filter_index):
    image = tf.random.uniform(shape=(1, 200, 200, 3), minval=0.4, maxval=0.6)
    for _ in range(30):
        image = gradient_ascent_step(image, filter_index, learning_rate=10.)
    return image[0].numpy()

Object Detection & Beyond

For tasks like object detection and instance segmentation, consider:

• YOLO (You Only Look Once) - Real-time object detection
• Faster R-CNN - Two-stage detector with region proposals
• Mask R-CNN - Instance segmentation (detection + pixel masks)

These are available in libraries like:

• tensorflow.keras.applications (some detection models)
• TensorFlow Object Detection API
• Detectron2 (Facebook/Meta)

Deep Learning Basics

Based on Deep Learning with Python, 2nd Edition by François Chollet.

Table of Contents

• Tensor Operations
• Data Preparation
• Generalization: Underfitting vs Overfitting
• Common Mistakes
• Quick Recipes

Tensor Operations

Key Attributes

x.ndim    # Number of axes (rank)
x.shape   # Dimensions along each axis
x.dtype   # Data type (float32, int32, etc.)

Common Tensor Shapes

• Vector data: (samples, features) - 2D
• Timeseries/Sequence: (samples, timesteps, features) - 3D
• Images: (samples, height, width, channels) - 4D
• Video: (samples, frames, height, width, channels) - 5D

GradientTape for Automatic Differentiation

x = tf.Variable(3.0)
with tf.GradientTape() as tape:
    y = x ** 2
grad = tape.gradient(y, x)  # dy/dx = 2x = 6

Nested tapes for second-order gradients:

with tf.GradientTape() as outer_tape:
    with tf.GradientTape() as inner_tape:
        y = x ** 3
    first_grad = inner_tape.gradient(y, x)
second_grad = outer_tape.gradient(first_grad, x)

Data Preparation

Normalization (Critical for Training)

# Compute stats on TRAINING data only
mean = train_data.mean(axis=0)
std = train_data.std(axis=0)

# Apply to all splits
train_data = (train_data - mean) / std
val_data = (val_data - mean) / std
test_data = (test_data - mean) / std

Train/Val/Test Split

num_val = int(0.2 * len(data))
num_test = int(0.1 * len(data))

val_data = data[:num_val]
test_data = data[num_val:num_val + num_test]
train_data = data[num_val + num_test:]

K-Fold Cross-Validation

Use when data is limited (< 10,000 samples).

k = 4
num_val_samples = len(data) // k
all_scores = []

for fold in range(k):
    val_data = data[fold * num_val_samples:(fold + 1) * num_val_samples]
    train_data = np.concatenate([
        data[:fold * num_val_samples],
        data[(fold + 1) * num_val_samples:]
    ], axis=0)

    model = build_model()
    model.fit(train_data, train_targets, epochs=100, verbose=0)
    val_score = model.evaluate(val_data, val_targets)
    all_scores.append(val_score)

print(f"Average score: {np.mean(all_scores)}")

Generalization: Underfitting vs Overfitting

Diagnosing from Learning Curves

• Underfitting: Both train and val loss are high
• Overfitting: Train loss decreases, val loss increases
• Good fit: Both losses decrease, then val loss plateaus

Fighting Overfitting (in order of preference)

1. Get more training data - Always the best solution

2. Reduce network capacity:

# Smaller model
model = keras.Sequential([
    layers.Dense(4, activation="relu"),
    layers.Dense(4, activation="relu"),
    layers.Dense(1, activation="sigmoid")
])

3. Add weight regularization:

from tensorflow.keras import regularizers

layers.Dense(16,
    kernel_regularizer=regularizers.l2(0.002),
    activation="relu")

# Options: l1(0.001), l2(0.001), l1_l2(l1=0.001, l2=0.001)

4. Add dropout:

model = keras.Sequential([
    layers.Dense(16, activation="relu"),
    layers.Dropout(0.5),  # Drop 50% of units during training
    layers.Dense(16, activation="relu"),
    layers.Dropout(0.5),
    layers.Dense(1, activation="sigmoid")
])

Common Mistakes

Wrong Learning Rate

# Too high - loss explodes or oscillates
model.compile(optimizer=keras.optimizers.RMSprop(1.0), ...)

# Good starting point
model.compile(optimizer=keras.optimizers.RMSprop(1e-3), ...)

# For fine-tuning pretrained models
model.compile(optimizer=keras.optimizers.RMSprop(1e-5), ...)

Information Bottleneck

Don't use layers smaller than the output dimension in intermediate layers:

# BAD: 4-unit bottleneck loses information for 46-class classification
model = keras.Sequential([
    layers.Dense(64, activation="relu"),
    layers.Dense(4, activation="relu"),   # Bottleneck!
    layers.Dense(46, activation="softmax")
])

# GOOD: Maintain representational capacity
model = keras.Sequential([
    layers.Dense(64, activation="relu"),
    layers.Dense(64, activation="relu"),
    layers.Dense(46, activation="softmax")
])

Quick Recipes

Binary Classification (e.g., Sentiment)

model = keras.Sequential([
    layers.Dense(16, activation="relu"),
    layers.Dense(16, activation="relu"),
    layers.Dense(1, activation="sigmoid")
])
model.compile(optimizer="rmsprop",
              loss="binary_crossentropy",
              metrics=["accuracy"])

Multiclass Classification (e.g., Topic)

model = keras.Sequential([
    layers.Dense(64, activation="relu"),
    layers.Dense(64, activation="relu"),
    layers.Dense(num_classes, activation="softmax")
])
model.compile(optimizer="rmsprop",
              loss="sparse_categorical_crossentropy",  # integer labels
              metrics=["accuracy"])

Regression (e.g., Price Prediction)

model = keras.Sequential([
    layers.Dense(64, activation="relu"),
    layers.Dense(64, activation="relu"),
    layers.Dense(1)  # No activation for regression
])
model.compile(optimizer="rmsprop",
              loss="mse",
              metrics=["mae"])

Deep Learning Best Practices

Table of Contents

• Hyperparameter Optimization
• Model Ensembling
• Mixed Precision Training
• Multi-GPU Training
• TPU Training
• Data Pipeline Optimization
• Production Checklist
• Debugging Tips

Hyperparameter Optimization

KerasTuner

Automated hyperparameter search with various algorithms.

import keras_tuner as kt

def build_model(hp):
    units = hp.Int("units", min_value=16, max_value=64, step=16)
    model = keras.Sequential([
        layers.Dense(units, activation="relu"),
        layers.Dense(10, activation="softmax")
    ])
    optimizer = hp.Choice("optimizer", values=["rmsprop", "adam"])
    model.compile(
        optimizer=optimizer,
        loss="sparse_categorical_crossentropy",
        metrics=["accuracy"]
    )
    return model

HyperModel Class (For Complex Models)

class SimpleMLP(kt.HyperModel):
    def __init__(self, num_classes):
        self.num_classes = num_classes

    def build(self, hp):
        units = hp.Int("units", min_value=16, max_value=64, step=16)
        dropout = hp.Float("dropout", min_value=0.0, max_value=0.5, step=0.1)
        lr = hp.Float("learning_rate", min_value=1e-4, max_value=1e-2, sampling="log")

        model = keras.Sequential([
            layers.Dense(units, activation="relu"),
            layers.Dropout(dropout),
            layers.Dense(self.num_classes, activation="softmax")
        ])
        model.compile(
            optimizer=keras.optimizers.Adam(learning_rate=lr),
            loss="sparse_categorical_crossentropy",
            metrics=["accuracy"]
        )
        return model

Running the Search

tuner = kt.BayesianOptimization(
    build_model,
    objective="val_accuracy",
    max_trials=100,
    executions_per_trial=2,  # Average over multiple runs
    directory="my_tuner",
    project_name="mnist_tuning",
    overwrite=True,
)

tuner.search_space_summary()

tuner.search(
    x_train, y_train,
    batch_size=128,
    epochs=100,
    validation_data=(x_val, y_val),
    callbacks=[keras.callbacks.EarlyStopping(patience=5)],
)

# Get best hyperparameters
best_hps = tuner.get_best_hyperparameters(num_trials=3)
print(best_hps[0].values)

# Get best model
best_model = tuner.get_best_models(num_models=1)[0]

Tuner Types

• kt.RandomSearch - Random sampling
• kt.BayesianOptimization - Bayesian optimization (recommended)
• kt.Hyperband - Adaptive resource allocation

Model Ensembling

Combine predictions from multiple models for better accuracy:

# Train multiple models
models = [get_model() for _ in range(5)]
for model in models:
    model.fit(x_train, y_train, epochs=10, validation_data=(x_val, y_val))

# Ensemble predictions (average)
predictions = np.zeros_like(models[0].predict(x_test))
for model in models:
    predictions += model.predict(x_test)
predictions /= len(models)
final_predictions = np.argmax(predictions, axis=1)

Mixed Precision Training

Use 16-bit floats for faster training on modern GPUs (Volta/Turing or newer):

from tensorflow import keras

# Enable mixed precision globally
keras.mixed_precision.set_global_policy("mixed_float16")

# Build and train model as usual
model = build_model()
model.fit(x_train, y_train, epochs=10)

# Note: Final layer should stay float32 for numerical stability
# (Dense with softmax/sigmoid output is automatically handled)

Benefits:

• 2-3x training speedup on supported GPUs
• Lower memory usage (can use larger batch sizes)
• Minimal accuracy loss

Multi-GPU Training

Single Machine, Multiple GPUs

strategy = tf.distribute.MirroredStrategy()
print(f"Number of devices: {strategy.num_replicas_in_sync}")

with strategy.scope():
    model = build_model()
    model.compile(
        optimizer="rmsprop",
        loss="sparse_categorical_crossentropy",
        metrics=["accuracy"]
    )

# Scale batch size with number of GPUs
global_batch_size = batch_size * strategy.num_replicas_in_sync
model.fit(x_train, y_train, batch_size=global_batch_size, epochs=10)

Multiple Machines (Distributed Training)

strategy = tf.distribute.MultiWorkerMirroredStrategy()

with strategy.scope():
    model = build_model()
    model.compile(...)

TPU Training

Using TPU on Google Colab

# Connect to TPU
resolver = tf.distribute.cluster_resolver.TPUClusterResolver()
tf.config.experimental_connect_to_cluster(resolver)
tf.tpu.experimental.initialize_tpu_system(resolver)

strategy = tf.distribute.TPUStrategy(resolver)
print(f"Number of TPU cores: {strategy.num_replicas_in_sync}")

with strategy.scope():
    model = build_model()
    model.compile(...)

# Use larger batch sizes for TPUs (8 * 128 = 1024 typical)
model.fit(dataset, epochs=10, steps_per_epoch=steps_per_epoch)

Step Fusing for TPU Efficiency

Run multiple training steps per TPU execution to reduce overhead:

# Wrap model with step fusing
@tf.function
def train_multiple_steps(iterator, steps):
    for _ in tf.range(steps):
        x, y = next(iterator)
        with tf.GradientTape() as tape:
            predictions = model(x, training=True)
            loss = loss_fn(y, predictions)
        gradients = tape.gradient(loss, model.trainable_variables)
        optimizer.apply_gradients(zip(gradients, model.trainable_variables))
    return loss

Data Pipeline Optimization

# Optimal data pipeline
dataset = tf.data.Dataset.from_tensor_slices((x, y))
dataset = dataset.shuffle(buffer_size=10000)
dataset = dataset.batch(batch_size)
dataset = dataset.prefetch(tf.data.AUTOTUNE)  # Overlap loading with training
dataset = dataset.cache()  # Cache after first epoch (if fits in memory)

For large datasets:

# Use TFRecord format for efficient I/O
dataset = tf.data.TFRecordDataset(filenames)
dataset = dataset.map(parse_fn, num_parallel_calls=tf.data.AUTOTUNE)
dataset = dataset.batch(batch_size)
dataset = dataset.prefetch(tf.data.AUTOTUNE)

Production Checklist

1. Data

• Normalize inputs (using training set statistics)
• Handle missing values
• Check for data leakage between train/val/test

2. Model

• Start simple, increase complexity as needed
• Use appropriate architecture for data type
• Add regularization (dropout, L2) if overfitting

3. Training

• Use callbacks (EarlyStopping, ModelCheckpoint)
• Monitor both train and val metrics
• Use learning rate scheduling if needed

4. Deployment

• Save model with model.save("model.keras")
• Test inference on new data
• Consider quantization for edge deployment
• Use tf.function for compiled inference

Debugging Tips

# Run eagerly for debugging (slower but allows print statements)
model.compile(..., run_eagerly=True)

# Check for NaN values
tf.debugging.enable_check_numerics()

# Profile training
tensorboard_callback = keras.callbacks.TensorBoard(
    log_dir="./logs",
    profile_batch=(10, 20)  # Profile batches 10-20
)

Deep Learning for Computer Vision

Table of Contents

• Convnets (Convolutional Neural Networks)
• Data Loading
• Data Augmentation
• Transfer Learning
• Available Pretrained Models
• Complete Training Pipeline

Convnets (Convolutional Neural Networks)

Key components: Conv2D, MaxPooling2D, Flatten.

Basic Convnet Structure

inputs = keras.Input(shape=(28, 28, 1))
x = layers.Conv2D(filters=32, kernel_size=3, activation="relu")(inputs)
x = layers.MaxPooling2D(pool_size=2)(x)
x = layers.Conv2D(filters=64, kernel_size=3, activation="relu")(x)
x = layers.MaxPooling2D(pool_size=2)(x)
x = layers.Conv2D(filters=128, kernel_size=3, activation="relu")(x)
x = layers.Flatten()(x)
outputs = layers.Dense(10, activation="softmax")(x)
model = keras.Model(inputs=inputs, outputs=outputs)

Why MaxPooling Matters

Without pooling, the model has too many parameters and limited receptive field:

# BAD: No pooling - 3.5M parameters, small receptive field
inputs = keras.Input(shape=(28, 28, 1))
x = layers.Conv2D(32, 3, activation="relu")(inputs)
x = layers.Conv2D(64, 3, activation="relu")(x)
x = layers.Conv2D(128, 3, activation="relu")(x)
x = layers.Flatten()(x)  # Huge flattened output!

Data Loading

image_dataset_from_directory

from tensorflow.keras.utils import image_dataset_from_directory

train_dataset = image_dataset_from_directory(
    "path/to/train",
    image_size=(180, 180),
    batch_size=32,
    label_mode="binary"  # or "categorical", "int"
)

# Check shapes
for images, labels in train_dataset.take(1):
    print(f"images: {images.shape}")  # (32, 180, 180, 3)
    print(f"labels: {labels.shape}")  # (32,)

Dataset Operations

# Apply transformations
dataset = dataset.map(lambda x: x / 255.)  # Normalize

# Cache and prefetch for performance
dataset = dataset.cache().prefetch(tf.data.AUTOTUNE)

Data Augmentation

Apply random transformations to increase training data diversity:

data_augmentation = keras.Sequential([
    layers.RandomFlip("horizontal"),
    layers.RandomRotation(0.1),   # ±10% rotation
    layers.RandomZoom(0.2),       # ±20% zoom
    layers.RandomTranslation(0.1, 0.1),
    layers.RandomContrast(0.1),
])

# Use in model (only active during training)
inputs = keras.Input(shape=(180, 180, 3))
x = data_augmentation(inputs)
x = layers.Rescaling(1./255)(x)
x = layers.Conv2D(32, 3, activation="relu")(x)
# ...

Transfer Learning

Feature Extraction (Frozen Base)

Use pretrained model as fixed feature extractor:

# Load pretrained base
conv_base = keras.applications.VGG16(
    weights="imagenet",
    include_top=False,
    input_shape=(180, 180, 3)
)
conv_base.trainable = False  # Freeze weights

# Build model
inputs = keras.Input(shape=(180, 180, 3))
x = keras.applications.vgg16.preprocess_input(inputs)
x = conv_base(x)
x = layers.Flatten()(x)
x = layers.Dense(256, activation="relu")(x)
x = layers.Dropout(0.5)(x)
outputs = layers.Dense(1, activation="sigmoid")(x)
model = keras.Model(inputs, outputs)

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

Fast Feature Extraction (No Augmentation)

Extract features once, train classifier on extracted features:

def get_features_and_labels(dataset):
    all_features, all_labels = [], []
    for images, labels in dataset:
        preprocessed = keras.applications.vgg16.preprocess_input(images)
        features = conv_base.predict(preprocessed)
        all_features.append(features)
        all_labels.append(labels)
    return np.concatenate(all_features), np.concatenate(all_labels)

train_features, train_labels = get_features_and_labels(train_dataset)

# Train lightweight classifier
classifier = keras.Sequential([
    keras.Input(shape=(5, 5, 512)),
    layers.Flatten(),
    layers.Dense(256, activation="relu"),
    layers.Dropout(0.5),
    layers.Dense(1, activation="sigmoid")
])
classifier.fit(train_features, train_labels, epochs=20)

Fine-Tuning

Unfreeze top layers of base after initial training:

# After training frozen model...
conv_base.trainable = True

# Freeze all layers except last 4
for layer in conv_base.layers[:-4]:
    layer.trainable = False

# Recompile with lower learning rate
model.compile(
    optimizer=keras.optimizers.RMSprop(learning_rate=1e-5),
    loss="binary_crossentropy",
    metrics=["accuracy"]
)

# Continue training
model.fit(train_dataset, epochs=30, validation_data=val_dataset)

Available Pretrained Models

# Common architectures (all from keras.applications)
keras.applications.VGG16(weights="imagenet", include_top=False)
keras.applications.ResNet50(weights="imagenet", include_top=False)
keras.applications.Xception(weights="imagenet", include_top=False)
keras.applications.EfficientNetB0(weights="imagenet", include_top=False)
keras.applications.MobileNetV2(weights="imagenet", include_top=False)

# Each has its own preprocess_input function
from keras.applications.resnet50 import preprocess_input

Complete Training Pipeline

# 1. Load data
train_dataset = image_dataset_from_directory("train", image_size=(180, 180), batch_size=32)
val_dataset = image_dataset_from_directory("val", image_size=(180, 180), batch_size=32)

# 2. Define augmentation
data_augmentation = keras.Sequential([
    layers.RandomFlip("horizontal"),
    layers.RandomRotation(0.1),
    layers.RandomZoom(0.2),
])

# 3. Build model with augmentation
inputs = keras.Input(shape=(180, 180, 3))
x = data_augmentation(inputs)
x = layers.Rescaling(1./255)(x)
x = layers.Conv2D(32, 3, activation="relu")(x)
x = layers.MaxPooling2D(2)(x)
x = layers.Conv2D(64, 3, activation="relu")(x)
x = layers.MaxPooling2D(2)(x)
x = layers.Conv2D(128, 3, activation="relu")(x)
x = layers.MaxPooling2D(2)(x)
x = layers.Conv2D(256, 3, activation="relu")(x)
x = layers.Flatten()(x)
x = layers.Dropout(0.5)(x)
outputs = layers.Dense(1, activation="sigmoid")(x)
model = keras.Model(inputs, outputs)

# 4. Compile and train
model.compile(optimizer="rmsprop", loss="binary_crossentropy", metrics=["accuracy"])

callbacks = [
    keras.callbacks.ModelCheckpoint("best.keras", save_best_only=True),
    keras.callbacks.EarlyStopping(patience=5, restore_best_weights=True)
]

model.fit(train_dataset, epochs=100, validation_data=val_dataset, callbacks=callbacks)

Generative Deep Learning

Table of Contents

• Text Generation
• Variational Autoencoders (VAEs)
• Generative Adversarial Networks (GANs)
• DeepDream
• Neural Style Transfer

Text Generation

Temperature Sampling

Control randomness/creativity when sampling from probability distributions:

def sample_next(predictions, temperature=1.0):
    predictions = np.asarray(predictions).astype("float64")
    predictions = np.log(predictions) / temperature
    exp_preds = np.exp(predictions)
    predictions = exp_preds / np.sum(exp_preds)
    probas = np.random.multinomial(1, predictions, 1)
    return np.argmax(probas)

# temperature=0.2: More deterministic, repetitive
# temperature=1.0: Balanced
# temperature=1.5: More random, creative (may be nonsensical)

Text Generation Callback

class TextGenerator(keras.callbacks.Callback):
    def __init__(self, prompt, generate_length, temperatures=(0.2, 0.5, 1.0)):
        self.prompt = prompt
        self.generate_length = generate_length
        self.temperatures = temperatures

    def on_epoch_end(self, epoch, logs=None):
        for temperature in self.temperatures:
            print(f"== Temperature {temperature} ==")
            sentence = self.prompt
            for i in range(self.generate_length):
                tokenized = text_vectorization([sentence])
                predictions = self.model(tokenized)
                next_token = sample_next(predictions[0, i, :], temperature)
                sampled_token = tokens_index[next_token]
                sentence += " " + sampled_token
            print(sentence)

Transformer-based Language Model

inputs = keras.Input(shape=(None,), dtype="int64")
x = PositionalEmbedding(sequence_length, vocab_size, embed_dim)(inputs)
x = TransformerDecoder(embed_dim, latent_dim, num_heads)(x, x)
outputs = layers.Dense(vocab_size, activation="softmax")(x)
model = keras.Model(inputs, outputs)
model.compile(loss="sparse_categorical_crossentropy", optimizer="rmsprop")

Variational Autoencoders (VAEs)

Project data into a continuous, structured latent space for generation.

Sampler Layer (Reparameterization Trick)

class Sampler(layers.Layer):
    def call(self, z_mean, z_log_var):
        batch_size = tf.shape(z_mean)[0]
        z_size = tf.shape(z_mean)[1]
        epsilon = tf.random.normal(shape=(batch_size, z_size))
        return z_mean + tf.exp(0.5 * z_log_var) * epsilon

VAE Encoder

latent_dim = 2

encoder_inputs = keras.Input(shape=(28, 28, 1))
x = layers.Conv2D(32, 3, activation="relu", strides=2, padding="same")(encoder_inputs)
x = layers.Conv2D(64, 3, activation="relu", strides=2, padding="same")(x)
x = layers.Flatten()(x)
x = layers.Dense(16, activation="relu")(x)
z_mean = layers.Dense(latent_dim, name="z_mean")(x)
z_log_var = layers.Dense(latent_dim, name="z_log_var")(x)
encoder = keras.Model(encoder_inputs, [z_mean, z_log_var], name="encoder")

VAE Decoder

latent_inputs = keras.Input(shape=(latent_dim,))
x = layers.Dense(7 * 7 * 64, activation="relu")(latent_inputs)
x = layers.Reshape((7, 7, 64))(x)
x = layers.Conv2DTranspose(64, 3, activation="relu", strides=2, padding="same")(x)
x = layers.Conv2DTranspose(32, 3, activation="relu", strides=2, padding="same")(x)
decoder_outputs = layers.Conv2D(1, 3, activation="sigmoid", padding="same")(x)
decoder = keras.Model(latent_inputs, decoder_outputs, name="decoder")

Complete VAE with Custom Training

class VAE(keras.Model):
    def __init__(self, encoder, decoder, **kwargs):
        super().__init__(**kwargs)
        self.encoder = encoder
        self.decoder = decoder
        self.sampler = Sampler()
        self.total_loss_tracker = keras.metrics.Mean(name="total_loss")
        self.reconstruction_loss_tracker = keras.metrics.Mean(name="reconstruction_loss")
        self.kl_loss_tracker = keras.metrics.Mean(name="kl_loss")

    @property
    def metrics(self):
        return [self.total_loss_tracker, self.reconstruction_loss_tracker, self.kl_loss_tracker]

    def train_step(self, data):
        with tf.GradientTape() as tape:
            z_mean, z_log_var = self.encoder(data)
            z = self.sampler(z_mean, z_log_var)
            reconstruction = self.decoder(z)

            # Reconstruction loss
            reconstruction_loss = tf.reduce_mean(
                tf.reduce_sum(keras.losses.binary_crossentropy(data, reconstruction), axis=(1, 2))
            )

            # KL divergence loss
            kl_loss = -0.5 * (1 + z_log_var - tf.square(z_mean) - tf.exp(z_log_var))
            kl_loss = tf.reduce_mean(tf.reduce_sum(kl_loss, axis=1))

            total_loss = reconstruction_loss + kl_loss

        grads = tape.gradient(total_loss, self.trainable_weights)
        self.optimizer.apply_gradients(zip(grads, self.trainable_weights))

        self.total_loss_tracker.update_state(total_loss)
        self.reconstruction_loss_tracker.update_state(reconstruction_loss)
        self.kl_loss_tracker.update_state(kl_loss)

        return {m.name: m.result() for m in self.metrics}

# Training
vae = VAE(encoder, decoder)
vae.compile(optimizer=keras.optimizers.Adam())
vae.fit(mnist_digits, epochs=30, batch_size=128)

Sampling from Latent Space

n = 30
figure = np.zeros((28 * n, 28 * n))

grid_x = np.linspace(-1, 1, n)
grid_y = np.linspace(-1, 1, n)[::-1]

for i, yi in enumerate(grid_y):
    for j, xi in enumerate(grid_x):
        z_sample = np.array([[xi, yi]])
        x_decoded = vae.decoder.predict(z_sample)
        digit = x_decoded[0].reshape(28, 28)
        figure[i * 28:(i + 1) * 28, j * 28:(j + 1) * 28] = digit

plt.imshow(figure, cmap="Greys_r")

Generative Adversarial Networks (GANs)

Two networks competing: Generator creates fakes, Discriminator distinguishes real from fake.

Discriminator

discriminator = keras.Sequential([
    keras.Input(shape=(64, 64, 3)),
    layers.Conv2D(64, kernel_size=4, strides=2, padding="same"),
    layers.LeakyReLU(alpha=0.2),
    layers.Conv2D(128, kernel_size=4, strides=2, padding="same"),
    layers.LeakyReLU(alpha=0.2),
    layers.Conv2D(128, kernel_size=4, strides=2, padding="same"),
    layers.LeakyReLU(alpha=0.2),
    layers.Flatten(),
    layers.Dropout(0.2),
    layers.Dense(1, activation="sigmoid"),
], name="discriminator")

Generator

latent_dim = 128

generator = keras.Sequential([
    keras.Input(shape=(latent_dim,)),
    layers.Dense(8 * 8 * 128),
    layers.Reshape((8, 8, 128)),
    layers.Conv2DTranspose(128, kernel_size=4, strides=2, padding="same"),
    layers.LeakyReLU(alpha=0.2),
    layers.Conv2DTranspose(256, kernel_size=4, strides=2, padding="same"),
    layers.LeakyReLU(alpha=0.2),
    layers.Conv2DTranspose(512, kernel_size=4, strides=2, padding="same"),
    layers.LeakyReLU(alpha=0.2),
    layers.Conv2D(3, kernel_size=5, padding="same", activation="sigmoid"),
], name="generator")

Complete GAN with Adversarial Training

class GAN(keras.Model):
    def __init__(self, discriminator, generator, latent_dim):
        super().__init__()
        self.discriminator = discriminator
        self.generator = generator
        self.latent_dim = latent_dim
        self.d_loss_metric = keras.metrics.Mean(name="d_loss")
        self.g_loss_metric = keras.metrics.Mean(name="g_loss")

    def compile(self, d_optimizer, g_optimizer, loss_fn):
        super().compile()
        self.d_optimizer = d_optimizer
        self.g_optimizer = g_optimizer
        self.loss_fn = loss_fn

    @property
    def metrics(self):
        return [self.d_loss_metric, self.g_loss_metric]

    def train_step(self, real_images):
        batch_size = tf.shape(real_images)[0]

        # Generate fake images
        random_latent_vectors = tf.random.normal(shape=(batch_size, self.latent_dim))
        generated_images = self.generator(random_latent_vectors)

        # Combine with real images
        combined_images = tf.concat([generated_images, real_images], axis=0)

        # Labels: 1 for fake, 0 for real (with noise for stability)
        labels = tf.concat([tf.ones((batch_size, 1)), tf.zeros((batch_size, 1))], axis=0)
        labels += 0.05 * tf.random.uniform(tf.shape(labels))

        # Train discriminator
        with tf.GradientTape() as tape:
            predictions = self.discriminator(combined_images)
            d_loss = self.loss_fn(labels, predictions)
        grads = tape.gradient(d_loss, self.discriminator.trainable_weights)
        self.d_optimizer.apply_gradients(zip(grads, self.discriminator.trainable_weights))

        # Train generator (fool discriminator)
        random_latent_vectors = tf.random.normal(shape=(batch_size, self.latent_dim))
        misleading_labels = tf.zeros((batch_size, 1))  # Claim fakes are real

        with tf.GradientTape() as tape:
            predictions = self.discriminator(self.generator(random_latent_vectors))
            g_loss = self.loss_fn(misleading_labels, predictions)
        grads = tape.gradient(g_loss, self.generator.trainable_weights)
        self.g_optimizer.apply_gradients(zip(grads, self.generator.trainable_weights))

        self.d_loss_metric.update_state(d_loss)
        self.g_loss_metric.update_state(g_loss)
        return {"d_loss": self.d_loss_metric.result(), "g_loss": self.g_loss_metric.result()}

# Training
gan = GAN(discriminator, generator, latent_dim)
gan.compile(
    d_optimizer=keras.optimizers.Adam(learning_rate=0.0001),
    g_optimizer=keras.optimizers.Adam(learning_rate=0.0001),
    loss_fn=keras.losses.BinaryCrossentropy(),
)
gan.fit(dataset, epochs=100)

GAN Training Tips

• Use LeakyReLU instead of ReLU
• Add noise to discriminator labels
• Use separate optimizers for generator and discriminator
• Start with lower learning rates (1e-4)
• Use BatchNormalization in generator (but not in discriminator's first layer)

DeepDream

Maximize activations of specific layers using gradient ascent on input image:

def compute_loss(image, layers_and_weights):
    features = feature_extractor(image)
    loss = tf.zeros(shape=())
    for name, coeff in layers_and_weights.items():
        activation = features[name]
        loss += coeff * tf.reduce_mean(tf.square(activation[:, 2:-2, 2:-2, :]))
    return loss

@tf.function
def gradient_ascent_step(image, learning_rate):
    with tf.GradientTape() as tape:
        tape.watch(image)
        loss = compute_loss(image, layers_and_weights)
    grads = tape.gradient(loss, image)
    grads = tf.math.l2_normalize(grads)
    image += learning_rate * grads
    return loss, image

Neural Style Transfer

Apply style of one image to content of another:

Losses:

• Content loss: MSE between feature maps of content and generated image
• Style loss: MSE between Gram matrices of style and generated image
• Total variation loss: Encourages spatial smoothness

def gram_matrix(x):
    x = tf.transpose(x, (2, 0, 1))
    features = tf.reshape(x, (tf.shape(x)[0], -1))
    gram = tf.matmul(features, tf.transpose(features))
    return gram

def style_loss(style, combination):
    S = gram_matrix(style)
    C = gram_matrix(combination)
    return tf.reduce_sum(tf.square(S - C))

def content_loss(base, combination):
    return tf.reduce_sum(tf.square(combination - base))

Working with Keras

Table of Contents

• Model Building APIs
• Custom Metrics
• Callbacks
• Custom Training Loops
• Performance Optimization
• Model Serialization
• TensorBoard Visualization

Model Building APIs

1. Sequential API

Best for simple stacks where each layer has one input/output.

import keras
from keras import layers, ops

model = keras.Sequential([
    layers.Dense(64, activation="relu"),
    layers.Dense(10, activation="softmax")
])

# Or incrementally
model = keras.Sequential()
model.add(layers.Dense(64, activation="relu"))
model.add(layers.Dense(10, activation="softmax"))

2. Functional API

For multi-input/output, shared layers, non-linear topologies.

Multi-Input/Output Example:

title = keras.Input(shape=(vocab_size,), name="title")
text_body = keras.Input(shape=(vocab_size,), name="text_body")
tags = keras.Input(shape=(num_tags,), name="tags")

features = layers.Concatenate()([title, text_body, tags])
features = layers.Dense(64, activation="relu")(features)

priority = layers.Dense(1, activation="sigmoid", name="priority")(features)
department = layers.Dense(num_depts, activation="softmax", name="dept")(features)

model = keras.Model(
    inputs=[title, text_body, tags],
    outputs=[priority, department]
)

# Compile with per-output losses
model.compile(
    optimizer="rmsprop",
    loss={"priority": "mse", "dept": "categorical_crossentropy"},
    metrics={"priority": ["mae"], "dept": ["accuracy"]}
)

Accessing Layer Connectivity:

model.layers[3].input   # Get layer's input tensor
model.layers[3].output  # Get layer's output tensor

# Create new model reusing intermediate outputs
features = model.layers[4].output
new_output = layers.Dense(3, activation="softmax")(features)
new_model = keras.Model(inputs=model.inputs, outputs=[*model.outputs, new_output])

3. Subclassing API

Full flexibility with loops, conditionals, dynamic behavior.

class CustomerTicketModel(keras.Model):
    def __init__(self, num_departments):
        super().__init__()
        self.concat_layer = layers.Concatenate()
        self.mixing_layer = layers.Dense(64, activation="relu")
        self.priority_scorer = layers.Dense(1, activation="sigmoid")
        self.department_classifier = layers.Dense(num_departments, activation="softmax")

    def call(self, inputs):
        title = inputs["title"]
        text_body = inputs["text_body"]
        tags = inputs["tags"]
        features = self.concat_layer([title, text_body, tags])
        features = self.mixing_layer(features)
        priority = self.priority_scorer(features)
        department = self.department_classifier(features)
        return priority, department

Note: Subclassed models can't be inspected/plotted like Functional models.

Custom Metrics

Using keras.ops makes your metric work with TensorFlow, JAX, and PyTorch backends.

class RootMeanSquaredError(keras.metrics.Metric):
    def __init__(self, name="rmse", **kwargs):
        super().__init__(name=name, **kwargs)
        self.mse_sum = self.add_weight(name="mse_sum", initializer="zeros")
        self.total_samples = self.add_weight(name="total_samples",
                                              initializer="zeros", dtype="int32")

    def update_state(self, y_true, y_pred, sample_weight=None):
        # Use keras.ops for backend-agnostic math
        y_true = ops.one_hot(y_true, num_classes=ops.shape(y_pred)[1])
        mse = ops.sum(ops.square(y_true - y_pred))
        self.mse_sum.assign_add(mse)
        self.total_samples.assign_add(ops.shape(y_pred)[0])

    def result(self):
        return ops.sqrt(self.mse_sum / ops.cast(self.total_samples, "float32"))

    def reset_state(self):
        self.mse_sum.assign(0.)
        self.total_samples.assign(0)

# Usage
model.compile(..., metrics=["accuracy", RootMeanSquaredError()])

Callbacks

Built-in Callbacks

callbacks = [
    keras.callbacks.EarlyStopping(
        monitor="val_accuracy",
        patience=2,
        restore_best_weights=True
    ),
    keras.callbacks.ModelCheckpoint(
        filepath="best_model.keras",
        monitor="val_loss",
        save_best_only=True
    ),
    keras.callbacks.ReduceLROnPlateau(
        monitor="val_loss",
        factor=0.5,
        patience=3
    ),
    keras.callbacks.TensorBoard(log_dir="./logs")
]
model.fit(..., callbacks=callbacks)

Custom Callbacks

import matplotlib.pyplot as plt

class LossHistory(keras.callbacks.Callback):
    def on_train_begin(self, logs):
        self.per_batch_losses = []

    def on_batch_end(self, batch, logs):
        self.per_batch_losses.append(logs.get("loss"))

    def on_epoch_end(self, epoch, logs):
        plt.plot(self.per_batch_losses)
        plt.savefig(f"loss_epoch_{epoch}.png")
        self.per_batch_losses = []

Custom Training Loops

Basic Training Step (TensorFlow Backend)

import tensorflow as tf # Required for GradientTape

model = get_model()
loss_fn = keras.losses.SparseCategoricalCrossentropy()
optimizer = keras.optimizers.RMSprop()
metrics = [keras.metrics.SparseCategoricalAccuracy()]

def train_step(inputs, targets):
    with tf.GradientTape() as tape:
        predictions = model(inputs, training=True)
        loss = loss_fn(targets, predictions)
    gradients = tape.gradient(loss, model.trainable_weights)
    optimizer.apply_gradients(zip(gradients, model.trainable_weights))

    for metric in metrics:
        metric.update_state(targets, predictions)
    return loss

# Training loop
for epoch in range(epochs):
    for metric in metrics:
        metric.reset_state()
    for inputs_batch, targets_batch in dataset:
        loss = train_step(inputs_batch, targets_batch)
    print(f"Epoch {epoch}: {metrics[0].result():.4f}")

Custom train_step with fit()

Override train_step to customize while keeping fit() benefits:

class CustomModel(keras.Model):
    def train_step(self, data):
        inputs, targets = data
        with tf.GradientTape() as tape:
            predictions = self(inputs, training=True)
            loss = self.compiled_loss(targets, predictions)
        gradients = tape.gradient(loss, self.trainable_weights)
        self.optimizer.apply_gradients(zip(gradients, self.trainable_weights))
        self.compiled_metrics.update_state(targets, predictions)
        return {m.name: m.result() for m in self.metrics}

# Use like normal model
model = CustomModel(inputs, outputs)
model.compile(optimizer="rmsprop", loss="mse", metrics=["mae"])
model.fit(train_data, epochs=10)

Performance Optimization

tf.function Decorator

Compiles Python code to TensorFlow graph for 2-10x speedup:

@tf.function
def train_step(inputs, targets):
    with tf.GradientTape() as tape:
        predictions = model(inputs, training=True)
        loss = loss_fn(targets, predictions)
    gradients = tape.gradient(loss, model.trainable_weights)
    optimizer.apply_gradients(zip(gradients, model.trainable_weights))
    return loss

Dataset Optimization

dataset = tf.data.Dataset.from_tensor_slices((x, y))
dataset = dataset.shuffle(buffer_size=1024)
dataset = dataset.batch(32)
dataset = dataset.prefetch(tf.data.AUTOTUNE)  # Overlap data loading with training
dataset = dataset.cache()  # Cache in memory after first epoch

Model Serialization

Saving/Loading with Custom Objects

# When saving models with custom layers
class TransformerEncoder(layers.Layer):
    def get_config(self):
        config = super().get_config()
        config.update({
            "embed_dim": self.embed_dim,
            "dense_dim": self.dense_dim,
            "num_heads": self.num_heads,
        })
        return config

# Loading
model = keras.models.load_model(
    "model.keras",
    custom_objects={"TransformerEncoder": TransformerEncoder}
)

TensorBoard Visualization

# During training
tensorboard = keras.callbacks.TensorBoard(
    log_dir="./logs",
    histogram_freq=1,  # Log weight histograms
    write_graph=True,
    write_images=True
)
model.fit(..., callbacks=[tensorboard])

# View in terminal
# tensorboard --logdir ./logs

Keras 3 & Multi-Backend Guide

Keras 3 is a multi-backend deep learning framework that can run on top of JAX, TensorFlow, or PyTorch.

Core Concepts

1. Setup & Backend Selection

Always import keras directly. Set the backend via environment variable before importing keras.

import os
os.environ["KERAS_BACKEND"] = "jax"  # Options: "jax", "tensorflow", "torch"

import keras

2. Backend-Agnostic Operations (keras.ops)

To write layers/metrics that work on ANY backend, use keras.ops (which mimics the NumPy API) instead of tf.* or torch.*.

from keras import ops

# Instead of tf.reduce_sum or torch.sum
x = ops.ones((2, 2))
y = ops.sum(x, axis=1)

3. Cross-Framework Compatibility

• Data Pipelines: tf.data is recommended for all backends. It produces data efficiently.
• Saving: Always use the .keras extension. It saves the model architecture and weights in a backend-agnostic format (zip archive).

  model.save("my_model.keras")

4. Custom Components (Layers/Models)

Implement call() using keras.ops.

class MyLayer(keras.layers.Layer):
    def call(self, inputs):
        return ops.square(inputs)

Migration from Keras 2 (TensorFlow)

Debugging

• JAX: Fast, but compilation overhead. Good for research.
• TensorFlow: Robust, good deployment ecosystem.
• PyTorch: Eager execution by default, easy to debug.

To debug, you can temporarily switch backend to tensorflow or torch if you are familiar with their eager execution modes, or use run_eagerly=True in compile().

Deep Learning for Text & Transformers

Table of Contents

• Text Preprocessing
• Bag-of-Words Approaches
• Word Embeddings
• Transformer Architecture
• Sequence-to-Sequence Learning
• When to Use Which Approach

Text Preprocessing

TextVectorization Layer

from tensorflow.keras.layers import TextVectorization

text_vectorization = TextVectorization(
    max_tokens=20000,           # Vocabulary size
    output_mode="int",          # Output integer indices
    output_sequence_length=600, # Pad/truncate to this length
)

# Fit on training data
text_only_train_ds = train_ds.map(lambda x, y: x)
text_vectorization.adapt(text_only_train_ds)

# Apply to datasets
int_train_ds = train_ds.map(lambda x, y: (text_vectorization(x), y))

Output Modes

# Integer sequences (for RNNs, Transformers)
TextVectorization(output_mode="int", output_sequence_length=600)

# Binary bag-of-words
TextVectorization(output_mode="multi_hot", max_tokens=20000)

# TF-IDF weighted
TextVectorization(output_mode="tf_idf", max_tokens=20000)

# N-grams (bigrams)
TextVectorization(ngrams=2, output_mode="multi_hot", max_tokens=20000)

Custom Standardization

import re, string

def custom_standardization(input_string):
    lowercase = tf.strings.lower(input_string)
    return tf.strings.regex_replace(
        lowercase, f"[{re.escape(string.punctuation)}]", "")

text_vectorization = TextVectorization(
    standardize=custom_standardization,
    split="whitespace",
)

Bag-of-Words Approaches

Simple but effective for many text classification tasks:

# Binary unigram model
text_vectorization = TextVectorization(max_tokens=20000, output_mode="multi_hot")
text_vectorization.adapt(text_only_train_ds)

inputs = keras.Input(shape=(20000,))
x = layers.Dense(16, activation="relu")(inputs)
x = layers.Dropout(0.5)(x)
outputs = layers.Dense(1, activation="sigmoid")(x)
model = keras.Model(inputs, outputs)
model.compile(optimizer="rmsprop", loss="binary_crossentropy", metrics=["accuracy"])

Word Embeddings

Learning Embeddings from Scratch

inputs = keras.Input(shape=(None,), dtype="int64")
embedded = layers.Embedding(input_dim=max_tokens, output_dim=256)(inputs)
x = layers.Bidirectional(layers.LSTM(32))(embedded)
x = layers.Dropout(0.5)(x)
outputs = layers.Dense(1, activation="sigmoid")(x)
model = keras.Model(inputs, outputs)

Masking Padded Values

# mask_zero=True tells downstream layers to ignore padding (index 0)
embedded = layers.Embedding(
    input_dim=max_tokens,
    output_dim=256,
    mask_zero=True
)(inputs)

Using Pretrained Embeddings (GloVe)

# Load GloVe embeddings
embeddings_index = {}
with open("glove.6B.100d.txt") as f:
    for line in f:
        word, coefs = line.split(maxsplit=1)
        coefs = np.fromstring(coefs, "f", sep=" ")
        embeddings_index[word] = coefs

# Create embedding matrix
embedding_dim = 100
vocabulary = text_vectorization.get_vocabulary()
word_index = dict(zip(vocabulary, range(len(vocabulary))))

embedding_matrix = np.zeros((max_tokens, embedding_dim))
for word, i in word_index.items():
    if i < max_tokens:
        embedding_vector = embeddings_index.get(word)
        if embedding_vector is not None:
            embedding_matrix[i] = embedding_vector

# Create frozen embedding layer
embedding_layer = layers.Embedding(
    max_tokens,
    embedding_dim,
    embeddings_initializer=keras.initializers.Constant(embedding_matrix),
    trainable=False,  # Freeze pretrained weights
    mask_zero=True,
)

Transformer Architecture

Positional Embedding

Transformers are order-agnostic, so we inject position information:

class PositionalEmbedding(layers.Layer):
    def __init__(self, sequence_length, input_dim, output_dim, **kwargs):
        super().__init__(**kwargs)
        self.token_embeddings = layers.Embedding(input_dim, output_dim)
        self.position_embeddings = layers.Embedding(sequence_length, output_dim)
        self.sequence_length = sequence_length
        self.input_dim = input_dim
        self.output_dim = output_dim

    def call(self, inputs):
        length = tf.shape(inputs)[-1]
        positions = tf.range(start=0, limit=length, delta=1)
        embedded_tokens = self.token_embeddings(inputs)
        embedded_positions = self.position_embeddings(positions)
        return embedded_tokens + embedded_positions

    def compute_mask(self, inputs, mask=None):
        return tf.math.not_equal(inputs, 0)

    def get_config(self):
        config = super().get_config()
        config.update({
            "sequence_length": self.sequence_length,
            "input_dim": self.input_dim,
            "output_dim": self.output_dim,
        })
        return config

Transformer Encoder

For text classification, sequence encoding:

class TransformerEncoder(layers.Layer):
    def __init__(self, embed_dim, dense_dim, num_heads, **kwargs):
        super().__init__(**kwargs)
        self.embed_dim = embed_dim
        self.dense_dim = dense_dim
        self.num_heads = num_heads
        self.attention = layers.MultiHeadAttention(
            num_heads=num_heads, key_dim=embed_dim)
        self.dense_proj = keras.Sequential([
            layers.Dense(dense_dim, activation="relu"),
            layers.Dense(embed_dim),
        ])
        self.layernorm_1 = layers.LayerNormalization()
        self.layernorm_2 = layers.LayerNormalization()

    def call(self, inputs, mask=None):
        if mask is not None:
            mask = mask[:, tf.newaxis, :]
        attention_output = self.attention(inputs, inputs, attention_mask=mask)
        proj_input = self.layernorm_1(inputs + attention_output)
        proj_output = self.dense_proj(proj_input)
        return self.layernorm_2(proj_input + proj_output)

    def get_config(self):
        config = super().get_config()
        config.update({
            "embed_dim": self.embed_dim,
            "dense_dim": self.dense_dim,
            "num_heads": self.num_heads,
        })
        return config

Text Classification with Transformer

vocab_size = 20000
sequence_length = 600
embed_dim = 256
num_heads = 2
dense_dim = 32

inputs = keras.Input(shape=(None,), dtype="int64")
x = PositionalEmbedding(sequence_length, vocab_size, embed_dim)(inputs)
x = TransformerEncoder(embed_dim, dense_dim, num_heads)(x)
x = layers.GlobalMaxPooling1D()(x)
x = layers.Dropout(0.5)(x)
outputs = layers.Dense(1, activation="sigmoid")(x)
model = keras.Model(inputs, outputs)

Sequence-to-Sequence Learning

Transformer Decoder

For generation tasks (translation, text generation):

class TransformerDecoder(layers.Layer):
    def __init__(self, embed_dim, dense_dim, num_heads, **kwargs):
        super().__init__(**kwargs)
        self.embed_dim = embed_dim
        self.dense_dim = dense_dim
        self.num_heads = num_heads
        self.attention_1 = layers.MultiHeadAttention(num_heads=num_heads, key_dim=embed_dim)
        self.attention_2 = layers.MultiHeadAttention(num_heads=num_heads, key_dim=embed_dim)
        self.dense_proj = keras.Sequential([
            layers.Dense(dense_dim, activation="relu"),
            layers.Dense(embed_dim),
        ])
        self.layernorm_1 = layers.LayerNormalization()
        self.layernorm_2 = layers.LayerNormalization()
        self.layernorm_3 = layers.LayerNormalization()
        self.supports_masking = True

    def get_causal_attention_mask(self, inputs):
        input_shape = tf.shape(inputs)
        batch_size, sequence_length = input_shape[0], input_shape[1]
        i = tf.range(sequence_length)[:, tf.newaxis]
        j = tf.range(sequence_length)
        mask = tf.cast(i >= j, dtype="int32")
        mask = tf.reshape(mask, (1, sequence_length, sequence_length))
        mult = tf.concat([tf.expand_dims(batch_size, -1), tf.constant([1, 1], dtype=tf.int32)], axis=0)
        return tf.tile(mask, mult)

    def call(self, inputs, encoder_outputs, mask=None):
        causal_mask = self.get_causal_attention_mask(inputs)
        if mask is not None:
            padding_mask = tf.cast(mask[:, tf.newaxis, :], dtype="int32")
            padding_mask = tf.minimum(padding_mask, causal_mask)

        # Self-attention (masked)
        attention_output_1 = self.attention_1(
            query=inputs, value=inputs, key=inputs, attention_mask=causal_mask)
        attention_output_1 = self.layernorm_1(inputs + attention_output_1)

        # Cross-attention with encoder outputs
        attention_output_2 = self.attention_2(
            query=attention_output_1,
            value=encoder_outputs,
            key=encoder_outputs,
            attention_mask=padding_mask)
        attention_output_2 = self.layernorm_2(attention_output_1 + attention_output_2)

        proj_output = self.dense_proj(attention_output_2)
        return self.layernorm_3(attention_output_2 + proj_output)

    def get_config(self):
        config = super().get_config()
        config.update({
            "embed_dim": self.embed_dim,
            "dense_dim": self.dense_dim,
            "num_heads": self.num_heads,
        })
        return config

End-to-End Transformer (Machine Translation)

embed_dim = 256
dense_dim = 2048
num_heads = 8

# Encoder
encoder_inputs = keras.Input(shape=(None,), dtype="int64", name="english")
x = PositionalEmbedding(sequence_length, vocab_size, embed_dim)(encoder_inputs)
encoder_outputs = TransformerEncoder(embed_dim, dense_dim, num_heads)(x)

# Decoder
decoder_inputs = keras.Input(shape=(None,), dtype="int64", name="spanish")
x = PositionalEmbedding(sequence_length, vocab_size, embed_dim)(decoder_inputs)
x = TransformerDecoder(embed_dim, dense_dim, num_heads)(x, encoder_outputs)
x = layers.Dropout(0.5)(x)
decoder_outputs = layers.Dense(vocab_size, activation="softmax")(x)

transformer = keras.Model([encoder_inputs, decoder_inputs], decoder_outputs)
transformer.compile(optimizer="rmsprop", loss="sparse_categorical_crossentropy", metrics=["accuracy"])

Inference (Decoding)

def decode_sequence(input_sentence):
    tokenized_input = source_vectorization([input_sentence])
    decoded_sentence = "[start]"

    for i in range(max_decoded_length):
        tokenized_target = target_vectorization([decoded_sentence])[:, :-1]
        predictions = transformer([tokenized_input, tokenized_target])
        sampled_token_index = np.argmax(predictions[0, i, :])
        sampled_token = spa_index_lookup[sampled_token_index]
        decoded_sentence += " " + sampled_token
        if sampled_token == "[end]":
            break

    return decoded_sentence

When to Use Which Approach

Sequence models capture word order and are better when order matters (sentiment, QA). Bag-of-words is faster and works well for topic classification.

Deep Learning for Timeseries

Table of Contents

• Data Preparation
• Baseline: Common-Sense Approach
• Architectures
• Recurrent Dropout
• Practical Tips
• Multi-Step Forecasting

Data Preparation

timeseries_dataset_from_array

Efficiently create sliding window datasets:

sampling_rate = 6       # Sample every 6 timesteps (1 hour if 10-min data)
sequence_length = 120   # Look back 120 samples
delay = sampling_rate * (sequence_length + 24 - 1)  # Predict 24 steps ahead
batch_size = 256

train_dataset = keras.utils.timeseries_dataset_from_array(
    data=raw_data[:-delay],         # Input sequences
    targets=temperature[delay:],     # Target values
    sampling_rate=sampling_rate,
    sequence_length=sequence_length,
    shuffle=True,
    batch_size=batch_size,
    start_index=0,
    end_index=num_train_samples
)

val_dataset = keras.utils.timeseries_dataset_from_array(
    data=raw_data[:-delay],
    targets=temperature[delay:],
    sampling_rate=sampling_rate,
    sequence_length=sequence_length,
    shuffle=True,
    batch_size=batch_size,
    start_index=num_train_samples,
    end_index=num_train_samples + num_val_samples
)

# Check shapes
for samples, targets in train_dataset.take(1):
    print(f"samples: {samples.shape}")  # (batch, sequence_length, features)
    print(f"targets: {targets.shape}")  # (batch,)

Normalization

Always normalize timeseries data:

mean = raw_data[:num_train_samples].mean(axis=0)
std = raw_data[:num_train_samples].std(axis=0)
raw_data -= mean
raw_data /= std

Baseline: Common-Sense Approach

Always establish a baseline before deep learning:

def evaluate_naive_method(dataset):
    """Predict last observed value."""
    total_abs_err = 0.
    samples_seen = 0
    for samples, targets in dataset:
        preds = samples[:, -1, 1] * std[1] + mean[1]  # Last value, denormalized
        total_abs_err += np.sum(np.abs(preds - targets))
        samples_seen += samples.shape[0]
    return total_abs_err / samples_seen

print(f"Baseline MAE: {evaluate_naive_method(val_dataset):.2f}")

Architectures

1. Dense Baseline

Often surprisingly competitive:

inputs = keras.Input(shape=(sequence_length, num_features))
x = layers.Flatten()(inputs)
x = layers.Dense(16, activation="relu")(x)
outputs = layers.Dense(1)(x)
model = keras.Model(inputs, outputs)
model.compile(optimizer="rmsprop", loss="mse", metrics=["mae"])

2. 1D Convnets

Fast and good for capturing local patterns:

inputs = keras.Input(shape=(sequence_length, num_features))
x = layers.Conv1D(8, 24, activation="relu")(inputs)
x = layers.MaxPooling1D(2)(x)
x = layers.Conv1D(8, 12, activation="relu")(x)
x = layers.MaxPooling1D(2)(x)
x = layers.Conv1D(8, 6, activation="relu")(x)
x = layers.GlobalAveragePooling1D()(x)
outputs = layers.Dense(1)(x)
model = keras.Model(inputs, outputs)

3. Simple RNN

inputs = keras.Input(shape=(sequence_length, num_features))
x = layers.SimpleRNN(16)(inputs)
outputs = layers.Dense(1)(x)
model = keras.Model(inputs, outputs)

4. LSTM (Long Short-Term Memory)

Better at long-term dependencies:

inputs = keras.Input(shape=(sequence_length, num_features))
x = layers.LSTM(32, recurrent_dropout=0.25)(inputs)
x = layers.Dropout(0.5)(x)
outputs = layers.Dense(1)(x)
model = keras.Model(inputs, outputs)

5. GRU (Gated Recurrent Unit)

Simpler/faster than LSTM, often similar performance:

inputs = keras.Input(shape=(sequence_length, num_features))
x = layers.GRU(32, recurrent_dropout=0.25)(inputs)
x = layers.Dropout(0.5)(x)
outputs = layers.Dense(1)(x)
model = keras.Model(inputs, outputs)

6. Stacked RNNs

Increase capacity with multiple layers:

inputs = keras.Input(shape=(sequence_length, num_features))
x = layers.GRU(32, recurrent_dropout=0.5, return_sequences=True)(inputs)  # Must return sequences
x = layers.GRU(32, recurrent_dropout=0.5)(x)
x = layers.Dropout(0.5)(x)
outputs = layers.Dense(1)(x)
model = keras.Model(inputs, outputs)

7. Bidirectional RNNs

Process sequence in both directions. Good for NLP, usually not for forecasting:

inputs = keras.Input(shape=(sequence_length, num_features))
x = layers.Bidirectional(layers.LSTM(16))(inputs)
outputs = layers.Dense(1)(x)
model = keras.Model(inputs, outputs)

When to use: When context from "future" is available (NLP, not real-time forecasting).

Recurrent Dropout

Use recurrent_dropout to regularize RNNs:

# Dropout on recurrent connections (not input/output)
layers.LSTM(32, recurrent_dropout=0.25)
layers.GRU(32, recurrent_dropout=0.5)

# Note: recurrent_dropout prevents cuDNN optimization
# For faster training with recurrent_dropout, use unroll=True
layers.LSTM(32, recurrent_dropout=0.2, unroll=True)

Practical Tips

Key Settings

• return_sequences=True for stacking RNN layers
• recurrent_dropout for RNN regularization
• Dropout after RNN layer for additional regularization

When to Use What

Training Recipe

callbacks = [
    keras.callbacks.ModelCheckpoint("best_model.keras", save_best_only=True),
    keras.callbacks.EarlyStopping(patience=5, restore_best_weights=True)
]

model.compile(optimizer="rmsprop", loss="mse", metrics=["mae"])
history = model.fit(
    train_dataset,
    epochs=50,
    validation_data=val_dataset,
    callbacks=callbacks
)

Multi-Step Forecasting

For predicting multiple future steps:

# Option 1: Multi-output model
inputs = keras.Input(shape=(sequence_length, num_features))
x = layers.LSTM(32)(inputs)
outputs = layers.Dense(24)(x)  # Predict 24 future steps
model = keras.Model(inputs, outputs)
model.compile(optimizer="rmsprop", loss="mse")

# Option 2: Autoregressive (feed predictions back as input)
def forecast_autoregressive(model, input_seq, steps):
    predictions = []
    current_seq = input_seq.copy()
    for _ in range(steps):
        pred = model.predict(current_seq[np.newaxis, :, :])
        predictions.append(pred[0, 0])
        current_seq = np.roll(current_seq, -1, axis=0)
        current_seq[-1, 0] = pred[0, 0]  # Add prediction to sequence
    return np.array(predictions)