如何从 keras 中的单个自定义损失函数访问所有输出

Question

我正在尝试在 tensorFlow 中重现 this publication 中提出的网络架构。作为一个完全的初学者，我一直使用 this tutorial 作为基础，使用 tensorflow==2.3.2。

为了训练这个网络，他们使用了一个损失，这意味着同时来自网络的两个分支的输出，这让我开始关注 keras 中的自定义损失函数。我知道你可以定义你自己的，只要函数的定义如下所示：

def custom_loss(y_true, y_pred):

我也明白你可以像这样给出其他论点：

def loss_function(margin=0.3):
    def custom_loss(y_true, y_pred):
        # And now you can use margin

然后，您只需在编译模型时调用它们。在使用多个输出时，最常见的方法似乎是建议的here，您可以在其中提供多个损失函数，为每个输出调用一个。 但是，我找不到为损失函数提供多个输出的解决方案，而这正是我需要的。

为了进一步解释，这里是一个最小的工作示例，展示了我尝试过的内容，您可以在this collab 中亲自尝试。

import os
import tensorflow as tf
import keras.backend as K
from tensorflow.keras import datasets, layers, models, applications, losses
from tensorflow.keras.preprocessing import image_dataset_from_directory

_URL = 'https://storage.googleapis.com/mledu-datasets/cats_and_dogs_filtered.zip'
path_to_zip = tf.keras.utils.get_file('cats_and_dogs.zip', origin=_URL, extract=True)
PATH = os.path.join(os.path.dirname(path_to_zip), 'cats_and_dogs_filtered')

train_dir = os.path.join(PATH, 'train')
validation_dir = os.path.join(PATH, 'validation')

BATCH_SIZE = 32
IMG_SIZE = (160, 160)
IMG_SHAPE = IMG_SIZE + (3,)

train_dataset = image_dataset_from_directory(train_dir,
                                             shuffle=True,
                                             batch_size=BATCH_SIZE,
                                             image_size=IMG_SIZE)

validation_dataset = image_dataset_from_directory(validation_dir,
                                                  shuffle=True,
                                                  batch_size=BATCH_SIZE,
                                                  image_size=IMG_SIZE)

data_augmentation = tf.keras.Sequential([
  layers.experimental.preprocessing.RandomFlip('horizontal'),
  layers.experimental.preprocessing.RandomRotation(0.2),
])
preprocess_input = applications.resnet50.preprocess_input
base_model = applications.ResNet50(input_shape=IMG_SHAPE,
                                               include_top=False,
                                               weights='imagenet')
base_model.trainable = True
conv = layers.Conv2D(filters=128, kernel_size=(1,1))
global_pooling = layers.GlobalAveragePooling2D()
horizontal_pooling = layers.AveragePooling2D(pool_size=(1, 5))
reshape = layers.Reshape((-1, 128))

def custom_loss(y_true, y_pred):
    print(y_pred.shape)
    # Do some stuffs involving both outputs
    # Returning something trivial here for correct behavior
    return K.mean(y_pred)

inputs = tf.keras.Input(shape=IMG_SHAPE)
x = data_augmentation(inputs)
x = preprocess_input(x)
x = base_model(x, training=True)

first_branch = global_pooling(x)

second_branch = conv(x)
second_branch = horizontal_pooling(second_branch)
second_branch = reshape(second_branch)

model = tf.keras.Model(inputs, [first_branch, second_branch])
base_learning_rate = 0.0001
model.compile(optimizer=tf.keras.optimizers.Adam(lr=base_learning_rate),
              loss=custom_loss,
              metrics=['accuracy'])
model.summary()

initial_epochs = 10
history = model.fit(train_dataset,
                    epochs=initial_epochs,
                    validation_data=validation_dataset)

这样做时，我认为赋予损失函数的 y_pred 将是一个列表，包含两个输出。但是，在运行它时，我在标准输出中得到的是：

Epoch 1/10
(None, 2048)
(None, 5, 128)

我从中了解到的是，损失函数是对每个输出一个一个地调用的，而不是对所有输出调用一次，这意味着我无法定义一个将同时使用两个输出的损失同时。有什么方法可以实现吗？

如果我不清楚，或者如果您需要更多详细信息，请告诉我。

Answer 1

我在尝试实现Triplet_Loss 函数时遇到了同样的问题。

我参考了 Keras 对 Siamese Network with Triplet Loss Function 的实现，但有些事情没有解决，我不得不自己实现网络。

def get_siamese_model(input_shape, conv2d_filters):
    # Define the tensors for the input images
    anchor_input = Input(input_shape, name="Anchor_Input")
    positive_input = Input(input_shape, name="Positive_Input")
    negative_input = Input(input_shape, name="Negative_Input")

    body = build_body(input_shape, conv2d_filters)
    # Generate the feature vectors for the images
    encoded_a = body(anchor_input)
    encoded_p = body(positive_input)
    encoded_n = body(negative_input)

    distance = DistanceLayer()(encoded_a, encoded_p, encoded_n)
    # Connect the inputs with the outputs
    siamese_net = Model(inputs=[anchor_input, positive_input, negative_input],
                        outputs=distance)
    return siamese_net

“错误”在DistanceLayer 发布的 Keras 实现中（也在上面的同一链接中）。

class DistanceLayer(tf.keras.layers.Layer):
    """
    This layer is responsible for computing the distance between the anchor
    embedding and the positive embedding, and the anchor embedding and the
    negative embedding.
    """

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    def call(self, anchor, positive, negative):
        ap_distance = tf.math.reduce_sum(tf.math.square(anchor - positive), axis=1, keepdims=True, name='ap_distance')
        an_distance = tf.math.reduce_sum(tf.math.square(anchor - negative), axis=1, keepdims=True, name='an_distance')
        return (ap_distance, an_distance)

当我在训练模型时，损失函数只取了ap_distance 或an_distance 中的一个向量。

最后，修复方法将向量连接在一起（在这种情况下是 axis=1）并在损失函数上将它们分开：

    def call(self, anchor, positive, negative):
        ap_distance = tf.math.reduce_sum(tf.math.square(anchor - positive), axis=1, keepdims=True, name='ap_distance')
        an_distance = tf.math.reduce_sum(tf.math.square(anchor - negative), axis=1, keepdims=True, name='an_distance')
        return tf.concat([ap_distance, an_distance], axis=1)

关于我的自定义损失：

def get_loss(margin=1.0):
    def triplet_loss(y_true, y_pred):
        # The output of the network is NOT A tuple, but a matrix shape (batch_size, 2),
        # containing the distances between the anchor and the positive example,
        # and the anchor and the negative example.
        ap_distance = y_pred[:, 0]
        an_distance = y_pred[:, 1]

        # Computing the Triplet Loss by subtracting both distances and
        # making sure we don't get a negative value.
        loss = tf.math.maximum(ap_distance - an_distance + margin, 0.0)
        # tf.print("\n", ap_distance, an_distance)
        # tf.print(f"\n{loss}\n")
        return loss

    return triplet_loss

Answer 2

好的，这是实现此目的的简单方法。我们可以通过使用loss_weights 参数来实现这一点。我们可以对多个输出进行完全相同的权重，这样我们就可以得到组合的损失结果。所以，对于两个输出我们可以做

loss_weights = 1*output1 + 1*output2

在您的情况下，您的网络有两个输出，名称分别为 reshape 和 global_average_pooling2d。你现在可以这样做

# calculation of loss for one output, i.e. reshape
def reshape_loss(y_true, y_pred):
    # do some math with these two 
    return K.mean(y_pred)

# calculation of loss for another output, i.e. global_average_pooling2d
def gap_loss(y_true, y_pred):
    # do some math with these two 
    return K.mean(y_pred)

现在编译时你需要这样做

model.compile(
    optimizer=tf.keras.optimizers.Adam(lr=base_learning_rate), 
    loss = {
         'reshape':reshape_loss, 
         'global_average_pooling2d':gap_loss
      },
    loss_weights = {
        'reshape':1., 
        'global_average_pooling2d':1.
     }
    )

现在，loss 是 1.*reshape + 1.*global_average_pooling2d 的结果。