How to create a custom callback in Keras

Training jobs often need small side effects that do not belong inside the model definition. A Keras callback runs at defined points in fit(), evaluate(), or predict(), so metric capture, logging, stopping decisions, and sanity checks can stay outside the core training loop.

Custom callbacks subclass keras.callbacks.Callback and override methods such as on_train_begin() or on_epoch_end(). The logs argument carries quantities for the current batch or epoch, and self.model points at the active model when a callback needs to read optimizer state or request training to stop.

The smoke test creates a small epoch recorder around model.fit(). The model and data stay deliberately tiny, and the output proves that three epochs ran, each epoch delivered loss and metric keys, and the callback kept its own recorded state after training finished.

Steps to create a custom Keras callback:

Open a terminal in the Python environment where Keras imports with the backend used by the training project.
```
$ python3 -c "import keras; print(keras.__version__, keras.backend.backend())"
3.15.0 tensorflow
```
The backend name may be tensorflow, jax, or torch. Select it before Python imports keras when the project needs a non-default backend.
Related: How to set the Keras backend

Save the callback smoke test as

custom_callback.py

custom_callback.py

import keras
import numpy as np
 
 
class EpochLogRecorder(keras.callbacks.Callback):
    def on_train_begin(self, logs=None):
        self.seen_epochs = []
        self.log_key_history = []
 
    def on_epoch_end(self, epoch, logs=None):
        logs = logs or {}
        keys = sorted(logs.keys())
        self.seen_epochs.append(epoch + 1)
        self.log_key_history.append(keys)
        print(f"epoch={epoch + 1} log_keys={','.join(keys)}")
 
 
keras.utils.set_random_seed(7)
 
x_train = np.arange(0, 8, dtype="float32").reshape(-1, 1)
y_train = (2 * x_train) + 1
 
model = keras.Sequential(
    [
        keras.layers.Input(shape=(1,)),
        keras.layers.Dense(1),
    ]
)
model.compile(
    optimizer=keras.optimizers.SGD(learning_rate=0.05),
    loss="mse",
    metrics=["mae"],
)
 
callback = EpochLogRecorder()
history = model.fit(
    x_train,
    y_train,
    batch_size=4,
    epochs=3,
    verbose=0,
    callbacks=[callback],
)
 
print(f"epochs recorded: {callback.seen_epochs}")
print(f"log snapshots: {len(callback.log_key_history)}")
print(f"history keys: {sorted(history.history.keys())}")

on_train_begin() initializes callback-owned state before training starts, and on_epoch_end() receives the epoch index plus the metric dictionary after each epoch.

Run the callback smoke test.

$ python3 custom_callback.py
epoch=1 log_keys=loss,mae
epoch=2 log_keys=loss,mae
epoch=3 log_keys=loss,mae
epochs recorded: [1, 2, 3]
log snapshots: 3
history keys: ['loss', 'mae']

One output line per epoch and a final snapshot count of 3 confirm that the callback ran once at each epoch end.

Attach the callback class to the project training call.
```
callback = EpochLogRecorder()
 
history = model.fit(
    train_features,
    train_labels,
    epochs=20,
    validation_data=(val_features, val_labels),
    callbacks=[callback],
)
```
Use on_train_batch_end() when the side effect must run after each training batch. Prefer built-in callbacks such as EarlyStopping or ModelCheckpoint when they already match the project behavior.
Remove the smoke test file after copying the callback pattern into the training code.
```
$ rm custom_callback.py
```