在具有条件的 keras 中实现自定义损失函数

Posted 2023-02-23

【中文标题】在具有条件的 keras 中实现自定义损失函数

【英文标题】：Implementing custom loss function in keras with condition

【发布时间】：2018-12-08 13:49:59

【问题描述】：

我需要一些有关 keras 损失函数的帮助。我一直在使用 Tensorflow 后端在 keras 上实现自定义损失函数。

我已经在 numpy 中实现了自定义损失函数，但如果它可以翻译成 keras 损失函数那就太好了。损失函数采用数据框和一系列用户 ID。如果 user_id 不同，则相同 user_id 的欧几里得距离为正数和负数。该函数返回数据帧的总标量距离。

def custom_loss_numpy (encodings, user_id):
# user_id: a pandas series of users
# encodings: a pandas dataframe of encodings

    batch_dist = 0

    for i in range(len(user_id)):
         first_row = encodings.iloc[i,:].values
         first_user = user_id[i]

         for j in range(i+1, len(user_id)):
              second_user = user_id[j]
              second_row = encodings.iloc[j,:].values

        # compute distance: if the users are same then Euclidean distance is positive otherwise negative.
            if first_user == second_user:
                tmp_dist = np.linalg.norm(first_row - second_row)
            else:
                tmp_dist = -np.linalg.norm(first_row - second_row)

            batch_dist += tmp_dist

    return batch_dist

我已经尝试实现 keras 损失函数。我从 y_true 和 y_pred 张量对象中提取了 numpy 数组。

def custom_loss_keras(y_true, y_pred):
    # session of my program
    sess = tf_session.TF_Session().get()

    with sess.as_default():
        array_pred = y_pred.eval()
        print(array_pred)

但我收到以下错误。

tensorflow.python.framework.errors_impl.InvalidArgumentError: You must feed a value for placeholder tensor 'dense_1_input' with dtype float and shape [?,102]
 [[Node: dense_1_input = Placeholder[dtype=DT_FLOAT, shape=[?,102], _device="/job:localhost/replica:0/task:0/device:CPU:0"]()]]

我们将不胜感激任何形式的帮助。

【参考方案1】：

首先，在 Keras 损失函数中无法“从y_true 和y_pred 中提取numpy 数组”。您必须使用 Keras 后端函数（或 TF 函数）操作张量来计算损失。

换句话说，最好考虑一种“矢量化”的方式来计算损失，而不使用 if-else 和循环。

您的损失函数可以通过以下步骤计算：

在encodings 中的所有向量对之间生成成对欧几里得距离矩阵。 生成一个矩阵I，如果user_i == user_j，其元素I_ij为1，如果user_i != user_j，则为-1。 将两个矩阵逐元素相乘，然后将元素相加得到最终损失。

这是一个实现：

def custom_loss_keras(user_id, encodings):
    # calculate pairwise Euclidean distance matrix
    pairwise_diff = K.expand_dims(encodings, 0) - K.expand_dims(encodings, 1)
    pairwise_squared_distance = K.sum(K.square(pairwise_diff), axis=-1)

    # add a small number before taking K.sqrt for numerical safety
    # (K.sqrt(0) sometimes becomes nan)
    pairwise_distance = K.sqrt(pairwise_squared_distance + K.epsilon())

    # this will be a pairwise matrix of True and False, with shape (batch_size, batch_size)
    pairwise_equal = K.equal(K.expand_dims(user_id, 0), K.expand_dims(user_id, 1))

    # convert True and False to 1 and -1
    pos_neg = K.cast(pairwise_equal, K.floatx()) * 2 - 1

    # divide by 2 to match the output of `custom_loss_numpy`, but it's not really necessary
    return K.sum(pairwise_distance * pos_neg, axis=-1) / 2

我假设user_id 是上面代码中的整数。这里的技巧是使用K.expand_dims 来实现成对操作。乍一看可能有点难以理解，但还是很有用的。

它应该给出与custom_loss_numpy 大致相同的损失值（由于K.epsilon() 会有一点差异）：

encodings = np.random.rand(32, 10)
user_id = np.random.randint(10, size=32)

print(K.eval(custom_loss_keras(K.variable(user_id), K.variable(encodings))).sum())
-478.4245

print(custom_loss_numpy(pd.DataFrame(encodings), pd.Series(user_id)))
-478.42953553795815

---

我在损失函数中犯了一个错误。

在训练中使用该函数时，由于 Keras 自动将 y_true 更改为至少 2D，因此参数 user_id 不再是 1D 张量。它的形状是(batch_size, 1)。

要使用此功能，必须移除多余的轴：

def custom_loss_keras(user_id, encodings):
    pairwise_diff = K.expand_dims(encodings, 0) - K.expand_dims(encodings, 1)
    pairwise_squared_distance = K.sum(K.square(pairwise_diff), axis=-1)
    pairwise_distance = K.sqrt(pairwise_squared_distance + K.epsilon())

    user_id = K.squeeze(user_id, axis=1)  # remove the axis added by Keras
    pairwise_equal = K.equal(K.expand_dims(user_id, 0), K.expand_dims(user_id, 1))

    pos_neg = K.cast(pairwise_equal, K.floatx()) * 2 - 1
    return K.sum(pairwise_distance * pos_neg, axis=-1) / 2

【参考方案2】：

在 Keras 中实现参数化自定义损失函数有两个步骤。首先，为系数/度量编写一个方法。其次，编写一个包装函数来按照 Keras 需要的方式格式化。

对于像 DICE 这样的简单自定义损失函数，直接使用 Keras 后端而不是 tensorflow 实际上要干净得多。这是以这种方式实现的系数示例：

import keras.backend as K
def dice_coef(y_true, y_pred, smooth, thresh):
    y_pred = y_pred > thresh
    y_true_f = K.flatten(y_true)
    y_pred_f = K.flatten(y_pred)
    intersection = K.sum(y_true_f * y_pred_f)
    return (2. * intersection + smooth) / (K.sum(y_true_f) + K.sum(y_pred_f) + smooth)

现在是棘手的部分。 Keras 损失函数只能以(y_true, y_pred) 作为参数。所以我们需要一个单独的函数来返回另一个函数：

def dice_loss(smooth, thresh):
    def dice(y_true, y_pred)
        return -dice_coef(y_true, y_pred, smooth, thresh)
    return dice

最后你可以在Keras中使用如下compile：

# build model 
model = my_model()
# get the loss function
model_dice = dice_loss(smooth=1e-5, thresh=0.5)
# compile model
model.compile(loss=model_dice)

【讨论】：

@PyMatFlow 感谢您的帮助。但我不认为你的建议可以帮助我解决我的问题。我不想在 Keras 中实现参数化的自定义损失函数。我想从 y_pred 中提取 numpy 数组并检查其中的元素。