TensorFlow Callbacks 实战：5 个必备回调 + 自定义回调写法

回调函数是 TensorFlow 训练过程中最灵活的钩子——它让你在不修改训练循环代码的情况下，介入训练的每个阶段：每个 epoch 开始前、每个 batch 结束后、训练结束时……几乎所有"想在训练过程中做点什么"的需求，都可以用回调实现。

最常用的 5 个内置回调

不用全记住，先把这 5 个用熟：

1. EarlyStopping —— 训练自动刹车

python
from tensorflow.keras.callbacks import EarlyStopping

early_stop = EarlyStopping(
    monitor="val_loss",
    patience=5,
    restore_best_weights=True,
    mode="min"
)

patience=5 表示连续 5 个 epoch 验证损失没有改善就停。restore_best_weights=True 是关键——不加它，模型停在最后一个 epoch 的权重上，可能已经过拟合了。

常见错误：patience 设太小（2-3），训练还在正常波动就停了。大部分任务 5-10 是合适的起点。

2. ModelCheckpoint —— 自动存档

python
from tensorflow.keras.callbacks import ModelCheckpoint

# 只保存验证集上最好的模型
checkpoint = ModelCheckpoint(
    filepath="best_model.h5",
    monitor="val_loss",
    save_best_only=True,
    mode="min",
    verbose=1
)

# 只保存权重（更省磁盘）
checkpoint = ModelCheckpoint(
    filepath="weights_{epoch:02d}.h5",
    save_weights_only=True,
    save_freq="epoch"
)

save_best_only=True 比 save_freq="epoch" 更实用——前者只在模型刷新最优记录时保存，不会占满磁盘。训练时间长的任务务必加上这个回调，防止中途断线或 OOM 白跑。

3. ReduceLROnPlateau —— 损失停滞时自动降学习率

python
from tensorflow.keras.callbacks import ReduceLROnPlateau

reduce_lr = ReduceLROnPlateau(
    monitor="val_loss",
    factor=0.1,      # 学习率乘以 0.1
    patience=3,       # 连续 3 个 epoch 没改善就降
    min_lr=1e-7,      # 最低不低于这个值
    verbose=1
)

这个回调和 EarlyStopping 配合使用效果最好：先用 ReduceLROnPlateau 降学习率尝试突破瓶颈，如果降了好几次还是没改善，EarlyStopping 再出手停止训练。

4. TensorBoard —— 训练可视化

python
from tensorflow.keras.callbacks import TensorBoard
import datetime

log_dir = "logs/" + datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
tensorboard = TensorBoard(
    log_dir=log_dir,
    histogram_freq=1,
    write_graph=True,
    update_freq="epoch"
)

启动 TensorBoard：tensorboard --logdir=logs/，浏览器打开 localhost:6006。

histogram_freq=1 会记录每层权重的分布变化，对调试梯度消失/爆炸特别有用——如果某层权重分布越来越窄，说明那层基本没在学。

5. CSVLogger —— 训练日志留底

python
from tensorflow.keras.callbacks import CSVLogger

csv_logger = CSVLogger("training_log.csv")

最不起眼但最实用。训练跑完几小时后想回看每个 epoch 的 loss/accuracy 变化，CSV 日志比 TensorBoard 更方便做数据分析和画图。

5 个回调的标准组合

python
callbacks = [
    EarlyStopping(monitor="val_loss", patience=10, restore_best_weights=True),
    ModelCheckpoint("best_model.h5", monitor="val_loss", save_best_only=True),
    ReduceLROnPlateau(monitor="val_loss", factor=0.5, patience=3, min_lr=1e-7),
    TensorBoard(log_dir="logs/" + datetime.datetime.now().strftime("%Y%m%d-%H%M%S")),
    CSVLogger("training_log.csv")
]

model.fit(x_train, y_train,
          epochs=200,
          validation_data=(x_val, y_val),
          callbacks=callbacks)

这套组合覆盖了：自动刹车 + 自动存档 + 自动降学习率 + 可视化 + 日志记录。日常训练够用了。

回调的执行顺序：按列表顺序依次执行。如果你的自定义回调依赖 ModelCheckpoint 的保存结果，确保 ModelCheckpoint 排在前面。

其他内置回调：什么时候才需要

LearningRateScheduler —— 自定义学习率曲线

python
from tensorflow.keras.callbacks import LearningRateScheduler

def lr_schedule(epoch, lr):
    if epoch < 10:
        return 0.001
    elif epoch < 30:
        return 0.0005
    else:
        return 0.0001

lr_callback = LearningRateScheduler(lr_schedule, verbose=1)

和 ReduceLROnPlateau 的区别：LearningRateScheduler 按预定计划降（不看指标），ReduceLROnPlateau 根据指标自适应降。大多数情况 ReduceLROnPlateau 更好用——你不需要提前猜学习率该什么时候降。

BackupAndRestore —— 训练中断恢复

python
from tensorflow.keras.callbacks import BackupAndRestore

backup = BackupAndRestore(backup_dir="backup", save_freq="epoch")

长时间训练（几小时甚至几天）时加上这个，遇到 OOM 或手动中断后可以从上次保存的 epoch 继续。配合 ModelCheckpoint 使用不冲突——BackupAndRestore 只保存训练状态（优化器状态等），ModelCheckpoint 保存模型权重。

LambdaCallback —— 最简自定义

python
from tensorflow.keras.callbacks import LambdaCallback

# 只想在某个时机做一件简单的事
print_callback = LambdaCallback(
    on_epoch_end=lambda epoch, logs: print(f"Epoch {epoch}: lr={float(model.optimizer.lr):.6f}")
)

一行 lambda 搞定，不需要写完整的 Callback 子类。缺点是不能保存状态，复杂逻辑还是要用类。

自定义回调：真实场景的写法

场景 1：梯度裁剪监控

训练不稳定时，想知道是不是梯度爆炸了：

python
class GradientMonitor(tf.keras.callbacks.Callback):
    def on_batch_end(self, batch, logs=None):
        if batch % 100 != 0:
            return
        grads = self.model.optimizer.get_gradients(
            self.model.total_loss,
            self.model.trainable_weights
        )
        grad_norms = [tf.norm(g).numpy() for g in grads if g is not None]
        if grad_norms:
            max_grad = max(grad_norms)
            if max_grad > 10.0:
                print(f"  Batch {batch}: max gradient norm = {max_grad:.2f} (potential explosion)")

如果频繁打印爆炸警告，说明需要加梯度裁剪：optimizer = Adam(clipnorm=1.0)。

场景 2：验证集上计算自定义指标

TensorFlow 内置的验证指标有限，想算 F1、AUC 或业务指标时：

python
from sklearn.metrics import f1_score

class F1ScoreCallback(tf.keras.callbacks.Callback):
    def __init__(self, validation_data):
        super().__init__()
        self.x_val, self.y_val = validation_data

    def on_epoch_end(self, epoch, logs=None):
        y_pred = self.model.predict(self.x_val, verbose=0)
        y_pred_labels = (y_pred > 0.5).astype(int)
        f1 = f1_score(self.y_val, y_pred_labels, average="macro")
        print(f"  val_f1: {f1:.4f}")
        logs["val_f1"] = f1  # 写入 logs，TensorBoard 和 CSVLogger 会自动记录

把自定义指标写入 logs 字典后，TensorBoard 和 CSVLogger 会自动记录它，不需要额外代码。

场景 3：动态冻结/解冻层

迁移学习中常用：先只训练顶层几轮，再解冻全部层精调。

python
class UnfreezeCallback(tf.keras.callbacks.Callback):
    def __init__(self, unfreeze_at_epoch=5):
        super().__init__()
        self.unfreeze_at_epoch = unfreeze_at_epoch

    def on_epoch_begin(self, epoch, logs=None):
        if epoch == self.unfreeze_at_epoch:
            for layer in self.model.layers:
                layer.trainable = True
            # 重新编译模型以应用更改
            self.model.compile(
                optimizer=self.model.optimizer.__class__(learning_rate=1e-5),
                loss=self.model.loss,
                metrics=["accuracy"]
            )
            print(f"  Unfreezed all layers at epoch {epoch}, lr reduced to 1e-5")

场景 4：训练达到目标精度后自动停止

比 EarlyStopping 更精确的停止条件：

python
class TargetAccuracyCallback(tf.keras.callbacks.Callback):
    def __init__(self, target=0.95):
        super().__init__()
        self.target = target

    def on_epoch_end(self, epoch, logs=None):
        if logs.get("val_accuracy", 0) >= self.target:
            print(f"  Reached {self.target*100}% val accuracy, stopping training")
            self.model.stop_training = True

self.model.stop_training = True 是在回调中中断训练的标准方式，所有回调都能用。

自定义回调的完整生命周期

Callback 基类提供了这些钩子方法，按需重写：

python
class FullLifecycleCallback(tf.keras.callbacks.Callback):
    def on_train_begin(self, logs=None):
        """训练开始前，初始化状态"""

    def on_train_end(self, logs=None):
        """训练结束后，收尾工作"""

    def on_epoch_begin(self, epoch, logs=None):
        """每个 epoch 开始前"""

    def on_epoch_end(self, epoch, logs=None):
        """每个 epoch 结束后，最常用"""

    def on_batch_begin(self, batch, logs=None):
        """每个 batch 开始前"""

    def on_batch_end(self, batch, logs=None):
        """每个 batch 结束后，注意频率别打印太多"""

    def on_predict_begin(self, logs=None):
        """推理开始前"""

    def on_predict_end(self, logs=None):
        """推理结束后"""

on_epoch_end 是用得最多的——大部分监控和决策都在 epoch 级别做。on_batch_end 谨慎使用，如果一个 epoch 有 10000 个 batch，每个 batch 都执行你的回调逻辑，开销不小。

回调使用中的常见问题

多个回调修改学习率会冲突吗？

会。ReduceLROnPlateau 和 LearningRateScheduler 同时使用时，后者会覆盖前者的调整。只用其中一个。

回调里能修改模型结构吗？

不建议。回调里修改模型层（增删层、改激活函数）会导致计算图和优化器状态不一致。但修改 trainable 属性是可以的——只要随后重新编译。

回调里访问训练数据的正确方式

回调的 logs 字典里只有 loss 和 metrics，不包含训练数据。如果回调需要访问数据（如计算自定义指标），在 __init__ 中传入：

python
class MyCallback(tf.keras.callbacks.Callback):
    def __init__(self, validation_data):
        super().__init__()
        self.x_val, self.y_val = validation_data

不要通过 self.model 反向获取训练数据——模型对象里不存这些。