实现自定义损失函数

虽然Keras在tf.keras.losses中提供了一套完整的标准损失函数 (loss function)，适用于许多常见的机器学习 (machine learning)任务，但您不可避免地会遇到它们不足以满足要求的情况。您可能需要：

• 实现研究论文中描述的损失函数。
• 创建针对特定业务逻辑或数据特点量身定制的损失函数。
• 将多个损失分量与特定权重 (weight)组合。
• 通过专门的损失公式处理类不平衡等问题。

TensorFlow提供了直接的方法来定义您自己的自定义损失函数，并将其集成到标准的Keras训练流程中（model.compile()、model.fit()）。本节将展示如何创建这些自定义损失，使您能够精确控制模型优化的目标。

定义自定义损失主要有两种方式：

1. 作为简单的Python函数： 适用于无状态的损失计算。
2. 作为tf.keras.losses.Loss的子类： 对于需要超参数 (parameter) (hyperparameter)、内部状态或自定义序列化逻辑的损失函数，此方法更受青睐。

使用Python函数的简单损失函数 (loss function)

创建自定义损失最直接的方法是定义一个Python函数，该函数接受两个参数 (parameter)：y_true（真实标签）和y_pred（模型的预测）。此函数应返回一个张量，其中包含批处理中每个样本的损失值。Keras会根据您稍后可能指定或默认的reduction参数，处理这些每样本损失的聚合（归约）。

该函数必须完全使用TensorFlow操作，以确保它可以通过tf.function追踪到图中，并在GPU或TPU等加速器上高效运行。

我们来实现Huber损失，它对异常值的敏感度低于均方误差（MSE）。对于小误差，它呈二次方行为；对于大误差，它呈线性行为。公式如下：

$$L_{\delta}(y, f(x)) = \begin{cases} \frac{1}{2}(y - f(x))^2 & \text{当 } |y - f(x)| \le \delta \\ \delta (|y - f(x)| - \frac{1}{2}\delta) & \text{否则} \end{cases}$$

这里，$y$ 代表 y_true，$f(x)$ 代表 y_pred，而 $\delta$ 是一个阈值参数。

import tensorflow as tf

def huber_loss(y_true, y_pred, delta=1.0):
  """
  计算y_true和y_pred之间的Huber损失。

  参数：
    y_true: 真实值。形状 = [batch_size, d0, .. dN]
    y_pred: 预测值。形状 = [batch_size, d0, .. dN]
    delta: 损失从二次方变为线性的点。

  返回：
    一个tf.Tensor，包含每个样本的Huber损失值。
    形状 = [batch_size, d0, .. dN-1]
  """
  y_true = tf.cast(y_true, dtype=y_pred.dtype) # 确保类型匹配
  error = y_true - y_pred
  abs_error = tf.abs(error)
  quadratic = tf.minimum(abs_error, delta)
  linear = abs_error - quadratic
  return 0.5 * tf.square(quadratic) + delta * linear

# --- 使用示例 ---
# 假设您有一个已编译的Keras模型：
# model.compile(optimizer='adam', loss=huber_loss)

# 您也可以通过functools.partial或lambda传递配置
# from functools import partial
# model.compile(optimizer='adam', loss=partial(huber_loss, delta=0.8))
# model.compile(optimizer='adam', loss=lambda yt, yp: huber_loss(yt, yp, delta=0.8))

# Keras会自动处理归约（通常是对批处理进行平均）

这种基于函数的方法对于直接的、无状态的计算来说是清晰且简单的。然而，如果您的损失函数需要可配置参数（如上面的delta，尽管我们硬编码了默认值或使用了包装器）或者需要在批处理之间保持状态（对于损失函数来说不常见），那么子类化tf.keras.losses.Loss会更有效。使用lambda或partial等包装器也可能阻碍模型序列化。

通过子类化tf.keras.losses.Loss实现高级损失函数 (loss function)

为了获得更大的灵活性、可维护性，并更清晰地处理超参数 (parameter) (hyperparameter)，您可以通过子类化tf.keras.losses.Loss来创建自定义损失。这种面向对象的方法使您能够：

• 在初始化期间（__init__）将配置参数（超参数）传递给损失函数。
• 可能管理内部状态（尽管对于典型损失函数来说不太常见）。
• 通过get_config实现自定义序列化。

要子类化tf.keras.losses.Loss，您主要需要：

1. 重写__init__方法以接受任何超参数并调用父构造函数（super().__init__(...)）。您可以在此指定默认的reduction策略（例如，SUM_OVER_BATCH_SIZE、SUM、NONE），并为损失提供一个name。
2. 重写call(self, y_true, y_pred)方法。这是核心损失计算逻辑所在之处，类似于基于函数的方法。它接收y_true和y_pred，并应返回每样本损失张量。

我们来实现Focal Loss，它常用于具有极端类不平衡的对象检测或分类任务。它会降低易分类样本的贡献，让模型侧重于难以分类的样本。

用于二分类的Focal Loss公式是：

$$FL(p_t) = - \alpha_t (1 - p_t)^\gamma \log(p_t)$$

其中：

• $p_t = p$ 如果 $y=1$ (真实标签为正)
• $p_t = 1-p$ 如果 $y=0$ (真实标签为负)
• $p$ 是模型对标签为 $y=1$ 的类的估计概率。
• $\gamma \ge 0$ 是聚焦参数。增加 $\gamma$ 会减少分类良好的样本（$p_t$ 接近1）的损失贡献。$\gamma=0$ 恢复为标准交叉熵。
• $\alpha_t$ 是一个权重 (weight)因子，常用于平衡正/负样本。对于类别1，$\alpha_t = \alpha$；对于类别0，$\alpha_t = 1-\alpha$。

import tensorflow as tf
import tensorflow.keras.backend as K

class FocalLoss(tf.keras.losses.Loss):
  """
  实现用于二分类的Focal Loss函数。

  参数：
    alpha: 用于平衡正/负类的权重因子。
           浮点数，范围 [0, 1]。默认值为 0.25。
    gamma: 聚焦参数。非负浮点数。默认值为 2.0。
    reduction: 要应用的tf.keras.losses.Reduction类型。
                 默认为 SUM_OVER_BATCH_SIZE。
    name: 损失实例的可选名称。默认为 'focal_loss'。
  """
  def __init__(self, alpha=0.25, gamma=2.0,
               reduction=tf.keras.losses.Reduction.SUM_OVER_BATCH_SIZE,
               name='focal_loss'):
    super().__init__(reduction=reduction, name=name)
    self.alpha = alpha
    self.gamma = gamma

  def call(self, y_true, y_pred):
    """
    计算focal损失。

    参数：
      y_true: 真实标签（二进制0或1）。形状 [batch_size, 1] 或 [batch_size]。
      y_pred: 预测概率。形状 [batch_size, 1] 或 [batch_size]。

    返回：
      与归约策略兼容的损失张量。
    """
    y_true = tf.cast(y_true, dtype=y_pred.dtype)
    # 确保预测是概率（例如，如果输入是logits，则应用sigmoid）
    # 为了稳定性，裁剪预测以避免log(0)
    epsilon = K.epsilon()
    y_pred = tf.clip_by_value(y_pred, epsilon, 1. - epsilon)

    # 计算 p_t
    p_t = tf.where(tf.equal(y_true, 1), y_pred, 1. - y_pred)

    # 计算 alpha_t
    alpha_t = tf.where(tf.equal(y_true, 1), self.alpha, 1. - self.alpha)

    # 计算focal损失分量
    cross_entropy = -tf.math.log(p_t)
    weight = alpha_t * tf.pow(1. - p_t, self.gamma)

    # 计算最终的focal损失
    loss = weight * cross_entropy
    return loss # 归约由基类Loss处理

  def get_config(self):
    """返回可序列化的配置字典。"""
    base_config = super().get_config()
    return {**base_config, "alpha": self.alpha, "gamma": self.gamma}

# --- 使用示例 ---
# model.compile(optimizer='adam', loss=FocalLoss(alpha=0.3, gamma=1.5))

# 由于get_config的存在，模型保存将正常工作
# loaded_model = tf.keras.models.load_model(
#     'my_model.keras',
#     custom_objects={'FocalLoss': FocalLoss}
# )

实现自定义损失函数 (loss function)时的注意事项

• TensorFlow操作： 无论您的损失函数是一个简单函数还是Loss子类，请始终在其中使用tf操作（tf.math、tf.where、tf.cast等）。这可以确保与图执行、自动微分和硬件加速的兼容性。避免使用TensorFlow无法追踪的NumPy或纯Python操作。
• 形状兼容性： 密切关注y_true和y_pred的形状。它们通常需要兼容才能进行元素级操作。如果在调试时遇到错误，请使用tf.debugging.assert_shapes或打印形状。
• 数值稳定性： 注意可能导致NaN或Inf值的操作，例如log(0)或除以零。如FocalLoss示例所示，适当使用tf.clip_by_value将输入限制在安全范围内，或添加一个小的epsilon值（tf.keras.backend.epsilon()）。
• 归约： 当使用简单函数时，Keras会应用默认的归约（SUM_OVER_BATCH_SIZE）或在model.compile(..., loss=custom_loss(reduction=...))中指定的归约（如果损失函数被设计为接受该参数 (parameter)，但这不太常见）。当子类化tf.keras.losses.Loss时，归约策略会根据传递给父__init__的reduction参数自动处理。call方法应返回每样本损失。
• 序列化： 如果您计划使用model.save()保存模型并在以后加载，那么子类化tf.keras.losses.Loss并实现get_config是非常重要的。加载模型时，您需要将自定义损失类传递给tf.keras.models.load_model的custom_objects参数。用作损失的简单Python函数通常不会随模型架构一起序列化，这要求您在加载时重新提供该函数。
• 与model.fit集成： 自定义函数损失和Loss子类都适用于标准的Keras model.fit训练循环。Keras会自动处理损失的应用、通过tf.GradientTape计算梯度以及更新模型权重 (weight)。

这里是一个简单的可视化图，比较了均方误差（MSE）与Huber损失（$\delta=1.0$）。请注意，Huber损失对于较大误差呈线性增长，这使得它与MSE的二次方增长相比，受异常值的影响较小。

均方误差（MSE）与Huber损失（$\delta=1.0$）作为预测误差的函数比较。

通过实现自定义损失函数，您获得了一个工具，可以精确地根据当前任务调整模型的训练目标，解决机器学习 (machine learning)项目中的特定难题和要求。