实现自定义损失函数

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

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

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

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

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

使用Python函数的简单损失函数

创建自定义损失最直接的方法是定义一个Python函数，该函数接受两个参数：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`实现高级损失函数

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

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

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

重写__init__方法以接受任何超参数并调用父构造函数（super().__init__(...)）。您可以在此指定默认的reduction策略（例如，SUM_OVER_BATCH_SIZE、SUM、NONE），并为损失提供一个name。
重写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$ 是一个权重因子，常用于平衡正/负样本。对于类别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}
# )