Keras中带有权重的自定义损失函数

Posted 2023-02-16

【中文标题】Keras中带有权重的自定义损失函数

【英文标题】：Custom loss function with weights in Keras

【发布时间】：2020-10-05 03:34:31

【问题描述】：

我是神经网络新手。我想在TensorFlow中做一个自定义的损失函数，但是我需要得到一个权重向量，所以我这样做了：

def my_loss(weights):
  def custom_loss(y, y_pred):
    return weights*(y - y_pred)
  return custom_loss
model.compile(optimizer='adam', loss=my_loss(weights), metrics=['accuracy'])
model.fit(x_train, y_train, batch_size=None,  validation_data=(x_test, y_test), epochs=100)

当我启动它时，我收到此错误：

InvalidArgumentError:  Incompatible shapes: [50000,10] vs. [32,10]

形状是：

print(weights.shape)
print(y_train.shape)

(50000, 10)
(50000, 10)

所以我认为是批次的问题，我对TensorFlow没有很强的背景，所以我尝试使用全局变量以天真的方式解决

batch_index = 0

然后在自定义回调中将其更新到“on_batch_begin”钩子中。但它没有用，这是一个可怕的解决方案。那么，我怎样才能得到具有相应 y 的权重的确切部分？我有办法在自定义损失中获取当前批次索引吗？ 提前感谢您的帮助

【参考方案1】：

Keras 允许您从全局范围获取任何张量。实际上，y_true 和 y_pred 甚至可能不会被使用，as here。

您的模型可以有多个输入（您可以在推理时将此输入设为虚拟，或将权重加载到具有单个输入的模型中）。请注意，您仍然需要它进行验证。

import keras
from keras.layers import *
from keras import backend as K

import numpy as np

inputs_x = Input(shape=(10,))
inputs_w = Input(shape=(10,))

y = Dense(10,kernel_initializer='glorot_uniform' )(inputs_x)

model = keras.Model(inputs=[inputs_x, inputs_w], outputs=[y])

def my_loss(y_true, y_pred):
    return K.abs((y_true-y_pred)*inputs_w)

def my_metrics(y_true, y_pred):
    # just to output something
    return K.mean(inputs_w)



model.compile(optimizer='adam', loss=[my_loss], metrics=[my_metrics])

data = np.random.normal(size=(50000, 10))
labels = np.random.normal(size=(50000, 10))
weights = np.random.normal(size=(50000, 10))


model.fit([data, weights], labels, batch_size=256, validation_data=([data[:100], weights[:100]], labels[:100]), epochs=100)

要在没有权重的情况下进行验证，您需要编译另一个具有不同损失且不使用权重的模型版本。

UPD：还要注意，如果 Keras 返回数组而不是标量，Keras 将汇总您损失的所有元素

---

UPD：Tor tensorflow 2.1.0 似乎变得更加复杂。前进的方向是@marco-cerliani 指出的方向（标签、重量和数据被输入模型，并通过.add_loss() 添加自定义损失张量），但是他的解决方案对我来说开箱即用.第一件事是该模型不想使用无损失，拒绝同时接受输入和输出。所以，我引入了额外的虚拟损失函数。当数据集大小不能被批量大小整除时，就会出现第二个问题。在 keras 和 tf 1.x 中，最后一批问题通常由 steps_per_epoch 和 validation_steps 参数解决，但这里如果在第一批 Epoch 2 上开始失败。所以我需要制作简单的自定义数据生成器。

import tensorflow.keras as keras
from tensorflow.keras.layers import *
from tensorflow.keras import backend as K

import numpy as np

inputs_x = Input(shape=(10,))
inputs_w = Input(shape=(10,))
inputs_l = Input(shape=(10,))


y = Dense(10,kernel_initializer='glorot_uniform' )(inputs_x)

model = keras.Model(inputs=[inputs_x, inputs_w, inputs_l], outputs=[y])

def my_loss(y_true, y_pred):
    return K.abs((y_true-y_pred)*inputs_w)

def my_metrics():
    # just to output something
    return K.mean(inputs_w)

def dummy_loss(y_true, y_pred):
    return 0.


loss = my_loss(y, inputs_l)
metric = my_metrics()

model.add_loss(loss)
model.add_metric(metric, name='my_metric', aggregation='mean')


model.compile(optimizer='adam', loss=dummy_loss)

data = np.random.normal(size=(50000, 10))
labels = np.random.normal(size=(50000, 10))
weights = np.random.normal(size=(50000, 10))

dummy = np.zeros(shape=(50000, 10)) # or in can be labels, no matter now


# looks like it does not like when len(data) % batch_size != 0
# If I set steps_per_epoch, it fails on the second epoch.

# So, I proceded with data generator

class DataGenerator(keras.utils.Sequence):
    'Generates data for Keras'
    def __init__(self, x, w, y, y2, batch_size, shuffle=True):
        'Initialization'
        self.x = x
        self.w = w
        self.y = y
        self.y2 = y2
        self.indices = list(range(len(self.x)))
        self.shuffle = shuffle
        self.batch_size = batch_size
        self.on_epoch_end()

    def __len__(self):
        'Denotes the number of batches per epoch'
        return len(self.indices) // self.batch_size

    def __getitem__(self, index):
        'Generate one batch of data'
        # Generate indexes of the batch

        ids = self.indices[index*self.batch_size:(index+1)*self.batch_size]

        # the last None to remove weird warning
        # https://***.com/questions/59317919
        return [self.x[ids], self.w[ids], self.y[ids]], self.y2[ids], [None]

    def on_epoch_end(self):
        'Updates indexes after each epoch'
        if self.shuffle == True:
            np.random.shuffle(self.indices)

batch_size = 256

train_generator = DataGenerator(data,weights,labels, dummy, batch_size=batch_size, shuffle=True)

val_generator = DataGenerator(data[:2*batch_size],weights[:2*batch_size],labels[:2*batch_size], dummy[:2*batch_size], batch_size=batch_size, shuffle=True)

model.fit(x=train_generator, validation_data=val_generator,epochs=100)

【讨论】：

我必须将其应用于 CNN，但它不起作用。该解决方案似乎正是我所需要的，但我花了几个小时试图让它发挥作用，但没有成功。我还在此笔记本link 中制作了一个玩具问题，只是为了有个想法。非常感谢@MarcoCerliani，我不想滥用你的耐心

我也找到了@Slowpoke 的答案，它“似乎正是我需要的”但不是。请参阅下面的适用于 TF 2.7.0 的答案。请注意，您显然可以使用 run_eagerly=True（反之亦然？）使 TF 不那么挑剔（但更慢？​​），这与图形模式和磁带有关。

【参考方案2】：

这是一种将附加参数传递给自定义损失函数的解决方法，在您的情况下是权重数组。诀窍在于使用虚假输入，这些输入有助于以正确的方式构建和使用损失。不要忘记 keras 处理固定的批量维度

我在回归问题中提供了一个虚拟示例

def mse(y_true, y_pred, weights):
    error = y_true-y_pred
    return K.mean(K.square(error) + K.sqrt(weights))

X = np.random.uniform(0,1, (1000,10))
y = np.random.uniform(0,1, 1000)
w = np.random.uniform(0,1, 1000)

inp = Input((10,))
true = Input((1,))
weights = Input((1,))
x = Dense(32, activation='relu')(inp)
out = Dense(1)(x)

m = Model([inp,true,weights], out)
m.add_loss( mse( true, out, weights ) )
m.compile(loss=None, optimizer='adam')
m.fit(x=[X, y, w], y=None, epochs=3)

## final fitted model to compute predictions (remove W if not needed)
final_m = Model(inp, out)

【参考方案3】：

和@Michael Moretti 一样，我在这一切方面也是新手（深度学习、Python、TensorFlow、Keras ......）。这个问题是在 19 个月前提出的，在“TF 年”中事情进展很快。

显然，在某些时候，您可以编写一个带有参数(y_true, y_pred) 的 Python 函数并将其传递给您对model.compile() 的调用，一切都很好。现在这似乎在一些简单的情况下有效，但不是一般情况下。在试图理解为什么它对我不起作用时，我发现了这个 SO 问题和其他相关问题。正是@M.Innat 对this question 的回答让我走上了正轨。但实际上他的相关最终示例CustomMSE 抄袭自Keras Guide section on Custom Losses。这个例子展示了如何编写与 TensorFlow 版本：2.7.0 完全兼容的自定义损失，以及如何通过基于keras.losses.Loss 的类的构造函数向其传递额外的参数。致电model.compile():

class CustomMSE(keras.losses.Loss):
    def __init__(self, regularization_factor=0.1, name="custom_mse"):
        super().__init__(name=name)
        self.regularization_factor = regularization_factor

    def call(self, y_true, y_pred):
        mse = tf.math.reduce_mean(tf.square(y_true - y_pred))
        reg = tf.math.reduce_mean(tf.square(0.5 - y_pred))
        return mse + reg * self.regularization_factor

model.compile(optimizer=keras.optimizers.Adam(), loss=CustomMSE())

为获得最佳结果，请确保自定义损失函数（即自定义损失类的 call() 方法）内的所有计算都使用 TensorFlow 运算符完成，并且所有输入和输出数据都表示为 TF 张量.