如何在TensorFlow后端的Keras中屏蔽损失函数？

Question

我正试图使用基于Keras和TensorFlow后端的LSTM实现序列到序列任务。 输入是长度可变的英语句子。为了构建一个形状为[batch_number, max_sentence_length]的数据集，我在行尾添加了 EOF并用足够的占位符（例如#）填充每个句子。然后将句子中的每个字符转换为一个one-hot向量，以便数据集具有3-D形状[batch_number, max_sentence_length, character_number]。在LSTM编码器和解码器层之后，计算输出和目标之间的softmax交叉熵。

为了消除模型训练中的填充效应，可以在输入和损失函数上使用掩码。在Keras中，可以使用layers.core.Masking来进行屏蔽输入。在TensorFlow中，可以按如下方式对损失函数进行屏蔽：custom masked loss function in TensorFlow。

然而，由于Keras中用户定义的损失函数只接受参数y_true和y_pred，我找不到一种方法来实现这一点。那么如何将真正的sequence_lengths 输入到损失函数并进行屏蔽？

此外，我发现\keras\engine\training.py中有一个_weighted_masked_objective(fn)函数。它的定义是：

Adds support for masking and sample-weighting to an objective function.

但似乎该函数只能接受fn(y_true, y_pred)。是否有一种使用此函数来解决我的问题的方法？

具体来说，我修改了Yu-Yang的示例。

from keras.models import Model
from keras.layers import Input, Masking, LSTM, Dense, RepeatVector, TimeDistributed, Activation
import numpy as np
from numpy.random import seed as random_seed
random_seed(123)

max_sentence_length = 5
character_number = 3 # valid character 'a, b' and placeholder '#'

input_tensor = Input(shape=(max_sentence_length, character_number))
masked_input = Masking(mask_value=0)(input_tensor)
encoder_output = LSTM(10, return_sequences=False)(masked_input)
repeat_output = RepeatVector(max_sentence_length)(encoder_output)
decoder_output = LSTM(10, return_sequences=True)(repeat_output)
output = Dense(3, activation='softmax')(decoder_output)

model = Model(input_tensor, output)
model.compile(loss='categorical_crossentropy', optimizer='adam')
model.summary()

X = np.array([[[0, 0, 0], [0, 0, 0], [1, 0, 0], [0, 1, 0], [0, 1, 0]],
          [[0, 0, 0], [0, 1, 0], [1, 0, 0], [0, 1, 0], [0, 1, 0]]])
y_true = np.array([[[0, 0, 1], [0, 0, 1], [1, 0, 0], [0, 1, 0], [0, 1, 0]], # the batch is ['##abb','#babb'], padding '#'
          [[0, 0, 1], [0, 1, 0], [1, 0, 0], [0, 1, 0], [0, 1, 0]]])

y_pred = model.predict(X)
print('y_pred:', y_pred)
print('y_true:', y_true)
print('model.evaluate:', model.evaluate(X, y_true))
# See if the loss computed by model.evaluate() is equal to the masked loss
import tensorflow as tf
logits=tf.constant(y_pred, dtype=tf.float32)
target=tf.constant(y_true, dtype=tf.float32)
cross_entropy = tf.reduce_mean(-tf.reduce_sum(target * tf.log(logits),axis=2))
losses = -tf.reduce_sum(target * tf.log(logits),axis=2)
sequence_lengths=tf.constant([3,4])
mask = tf.reverse(tf.sequence_mask(sequence_lengths,maxlen=max_sentence_length),[0,1])
losses = tf.boolean_mask(losses, mask)
masked_loss = tf.reduce_mean(losses)
with tf.Session() as sess:
    c_e = sess.run(cross_entropy)
    m_c_e=sess.run(masked_loss)
    print("tf unmasked_loss:", c_e)
    print("tf masked_loss:", m_c_e)

在Keras和TensorFlow中，输出的比较如下：

如上所示，在某些类型的层之后，掩码被禁用。那么当添加这些层时，如何在Keras中掩盖损失函数呢？

- Shuaaai

---

你需要动态遮罩吗？ - Marcin Możejko

如果“动态遮罩”意味着根据模型的不同输入数据屏蔽损失函数，那么是的，这正是我想要的。 - Shuaaai

Answer 1

如果您的模型中有掩码，它将逐层传播，并最终应用于损失。因此，如果您以正确的方式对序列进行填充和掩码，则填充占位符上的损失将被忽略。

一些详细信息：

解释整个过程有点复杂，因此我将分解为几个步骤：

1. 在compile()中，通过调用compute_mask()来收集掩码并将其应用于损失（为了清晰起见，忽略了不相关的行）。

weighted_losses = [_weighted_masked_objective(fn) for fn in loss_functions]

# Prepare output masks.
masks = self.compute_mask(self.inputs, mask=None)
if masks is None:
    masks = [None for _ in self.outputs]
if not isinstance(masks, list):
    masks = [masks]

# Compute total loss.
total_loss = None
with K.name_scope('loss'):
    for i in range(len(self.outputs)):
        y_true = self.targets[i]
        y_pred = self.outputs[i]
        weighted_loss = weighted_losses[i]
        sample_weight = sample_weights[i]
        mask = masks[i]
        with K.name_scope(self.output_names[i] + '_loss'):
            output_loss = weighted_loss(y_true, y_pred,
                                        sample_weight, mask)

2. 在Model.compute_mask()内部，调用了run_internal_graph()。
3. 在run_internal_graph()内部，通过迭代调用每一层的Layer.compute_mask()，将模型中的掩码从输入传播到输出。

因此，如果您的模型中使用了Masking层，则不必担心填充占位符上的损失。您可能已经在_weighted_masked_objective()内部看到了这些条目的掩码。

一个小例子：

max_sentence_length = 5
character_number = 2

input_tensor = Input(shape=(max_sentence_length, character_number))
masked_input = Masking(mask_value=0)(input_tensor)
output = LSTM(3, return_sequences=True)(masked_input)
model = Model(input_tensor, output)
model.compile(loss='mae', optimizer='adam')

X = np.array([[[0, 0], [0, 0], [1, 0], [0, 1], [0, 1]],
              [[0, 0], [0, 1], [1, 0], [0, 1], [0, 1]]])
y_true = np.ones((2, max_sentence_length, 3))
y_pred = model.predict(X)
print(y_pred)
[[[ 0.          0.          0.        ]
  [ 0.          0.          0.        ]
  [-0.11980877  0.05803877  0.07880752]
  [-0.00429189  0.13382857  0.19167568]
  [ 0.06817091  0.19093043  0.26219055]]

 [[ 0.          0.          0.        ]
  [ 0.0651961   0.10283815  0.12413475]
  [-0.04420842  0.137494    0.13727818]
  [ 0.04479844  0.17440712  0.24715884]
  [ 0.11117355  0.21645413  0.30220413]]]

# See if the loss computed by model.evaluate() is equal to the masked loss
unmasked_loss = np.abs(1 - y_pred).mean()
masked_loss = np.abs(1 - y_pred[y_pred != 0]).mean()

print(model.evaluate(X, y_true))
0.881977558136

print(masked_loss)
0.881978

print(unmasked_loss)
0.917384

从这个例子可以看出，掩码部分（y_pred中的零元素）的损失被忽略了，model.evaluate()的输出等于masked_loss。

---

编辑：

如果有一个带有return_sequences=False的循环层，则掩码停止传播（即返回的掩码为None）。在RNN.compute_mask()中：

def compute_mask(self, inputs, mask):
    if isinstance(mask, list):
        mask = mask[0]
    output_mask = mask if self.return_sequences else None
    if self.return_state:
        state_mask = [None for _ in self.states]
        return [output_mask] + state_mask
    else:
        return output_mask

如果我理解正确，您希望基于y_true创建一个掩码，每当y_true的值为[0, 0, 1]（“#”的独热编码）时，就希望将损失掩盖。如果是这样，您需要以与Daniel的答案类似的方式掩盖损失值。

主要区别在于最终平均值。平均值应该取决于未经掩盖的值的数量，即K.sum(mask)。此外，y_true可以直接与独热编码向量[0, 0, 1]进行比较。

def get_loss(mask_value):
    mask_value = K.variable(mask_value)
    def masked_categorical_crossentropy(y_true, y_pred):
        # find out which timesteps in `y_true` are not the padding character '#'
        mask = K.all(K.equal(y_true, mask_value), axis=-1)
        mask = 1 - K.cast(mask, K.floatx())

        # multiply categorical_crossentropy with the mask
        loss = K.categorical_crossentropy(y_true, y_pred) * mask

        # take average w.r.t. the number of unmasked entries
        return K.sum(loss) / K.sum(mask)
    return masked_categorical_crossentropy

masked_categorical_crossentropy = get_loss(np.array([0, 0, 1]))
model = Model(input_tensor, output)
model.compile(loss=masked_categorical_crossentropy, optimizer='adam')

上面代码的输出结果表明，损失仅计算未被掩盖的值：

model.evaluate: 1.08339476585
tf unmasked_loss: 1.08989
tf masked_loss: 1.08339

由于我在tf.reverse中将axis参数从[0,1]更改为[1]，所以该值与您的不同。

- Yu-Yang

---

@Shuaaai 啊，通过seq2seq，我以为你是指像这个例子中的模型。我已经更新了答案。请看看是否符合你的要求。 - Yu-Yang

首先，非常感谢你。是的，我想要基于y_true的掩码。我运行了你更新后的代码，但是它会引发一个错误：“ValueError: Dimensions must be equal, but are 5 and 3 for 'Equal' (op: 'Equal') with input shapes: [2,5,3], [3,1]。”这是由于版本不同还是其他原因引起的？ - Shuaaai

我的Keras版本是2.0.4，TensorFlow版本是1.1.0。我刚刚从"https://github.com/Shuaaai/Mask-loss-function"拉取了我的代码和错误截图。你能帮我检查一下吗？谢谢。 - Shuaaai

很抱歉，我没有使用这个模型的经验。初步看来，我认为您可能不应该在此问题中屏蔽“填充空格”。如果答案包含填充空格，模型应该学会预测空格。考虑例子“12 + 34 = 46”和“12 + 34 = 468”，后者显然是错误的。模型应该在输入“12+34”时输出4、6和一个填充空格。 - Yu-Yang

换句话说，如果模型预测得相当准确，填充位置不应该产生太多损失。那么，遮蔽填充空间与否并不是很重要。 - Yu-Yang

Answer 2

如果你没有像Yu-Yang的回答中所示使用掩码，你可以尝试以下方法。

如果你有目标数据Y并且已经使用掩码值填充，则可以：

import keras.backend as K
def custom_loss(yTrue,yPred):

    #find which values in yTrue (target) are the mask value
    isMask = K.equal(yTrue, maskValue) #true for all mask values

    #since y is shaped as (batch, length, features), we need all features to be mask values
    isMask = K.all(isMask, axis=-1) #the entire output vector must be true
        #this second line is only necessary if the output features are more than 1

    #transform to float (0 or 1) and invert
    isMask = K.cast(isMask, dtype=K.floatx())
    isMask = 1 - isMask #now mask values are zero, and others are 1

    #multiply this by the inputs:
       #maybe you might need K.expand_dims(isMask) to add the extra dimension removed by K.all
     yTrue = yTrue * isMask   
     yPred = yPred * isMask

     return someLossFunction(yTrue,yPred)

如果只对输入数据进行填充，或者Y的长度为0，则可以在函数外使用自己的掩码：

masks = [
   [1,1,1,1,1,1,0,0,0],
   [1,1,1,1,0,0,0,0,0],
   [1,1,1,1,1,1,1,1,0]
]
 #shape (samples, length). If it fails, make it (samples, length, 1). 

import keras.backend as K

masks = K.constant(masks)

由于掩码取决于您的输入数据，因此您可以使用掩码值来确定放置零的位置，例如：

masks = np.array((X_train == maskValue).all(), dtype='float64')    
masks = 1 - masks

#here too, if you have a problem with dimensions in the multiplications below
#expand masks dimensions by adding a last dimension = 1.

并使您的函数从外部获取掩码（如果更改输入数据，则必须重新创建损失函数）：

def customLoss(yTrue,yPred):

    yTrue = masks*yTrue
    yPred = masks*yPred

    return someLossFunction(yTrue,yPred)

---

有人知道Keras是否自动屏蔽损失函数吗？因为它提供了一个Masking层，对输出没有任何说明，也许它会自动屏蔽？

- Daniel Möller

---

Daniel - 这个答案非常差。长度掩码是动态分配给 y_true 和 y_pred 的，因此您无法在外部定义它们，因为掩码是会变的。如果按照您提供的方式这样做，那么最终会得到一个常量掩码，而这不是 OP 所期望的东西。 - Marcin Możejko

@MarcinMożejko，非常感谢您。我的答案确实是一个糟糕的回答。 - Daniel Möller

还不如余洋的好，但如果他们不使用遮罩层，这可能适用。 - Daniel Möller

如果您在模型函数中定义自定义损失，仍然可以访问掩码张量。因此，这个答案是有效的。 - jonperl

@DanielMöller 在你的 customLoss 代码片段中：如果掩码将某些 yTrue 和 yPred 值设置为零，那么这是否意味着 yTrue=yPred 并且损失人为地增加了？ - Helen

@Helen，人为地降低了损失，但真正预期的效果是这些值的损失应该是“恒定”的，这意味着它们永远不会影响训练。但是，为了平衡事物，向其他元素添加一些权重以弥补零可能是一个好主意。 - Daniel Möller

Answer 3

我采用了两个回答，为多个时间步长、单个缺失目标值、使用return_sequences=True的LSTM（或其他递归NN）制定了一种方法。

由于 isMask = K.all(isMask, axis=-1) ，Daniels的回答无法满足多个目标的要求。 移除这个聚合使得函数不可微分，可能是因为我从未运行过纯函数，无法确定其是否能够拟合模型。

我将Yu-Yangs和Daniels的回答合并在一起，并且它有效地工作了。

from tensorflow.keras.layers import Layer, Input, LSTM, Dense, TimeDistributed
from tensorflow.keras import Model, Sequential
import tensorflow.keras.backend as K
import numpy as np


mask_Value = -2
def get_loss(mask_value):
    mask_value = K.variable(mask_value)
    def masked_loss(yTrue,yPred):
        
        #find which values in yTrue (target) are the mask value
        isMask = K.equal(yTrue, mask_Value) #true for all mask values
    
        #transform to float (0 or 1) and invert
        isMask = K.cast(isMask, dtype=K.floatx())
        isMask = 1 - isMask #now mask values are zero, and others are 1
        isMask
        
        #multiply this by the inputs:
        #maybe you might need K.expand_dims(isMask) to add the extra dimension removed by K.all
        yTrue = yTrue * isMask   
        yPred = yPred * isMask
        
        # perform a root mean square error, whereas the mean is in respect to the mask
        mean_loss = K.sum(K.square(yPred - yTrue))/K.sum(isMask)
        loss = K.sqrt(mean_loss)
    
        return loss
        #RootMeanSquaredError()(yTrue,yPred)
        
    return masked_loss

# define timeseries data
n_sample = 10
timesteps = 5
feat_inp = 2
feat_out = 2

X = np.random.uniform(0,1, (n_sample, timesteps, feat_inp))
y = np.random.uniform(0,1, (n_sample,timesteps, feat_out))

# define model
model = Sequential()
model.add(LSTM(50, activation='relu',return_sequences=True, input_shape=(timesteps, feat_inp)))
model.add(Dense(feat_out))
model.compile(optimizer='adam', loss=get_loss(mask_Value))
model.summary()

# %%
model.fit(X, y, epochs=50, verbose=0)

- Alexander Vocaet

Answer 4

请注意，Yu-Yang的答案似乎在Tensorflow Keras 2.7.0上无法使用。

令人惊讶的是，model.evaluate不会计算masked_loss或unmasked_loss。相反，它假定所有被屏蔽的输入步骤的损失为零（但仍将这些步骤包括在mean()计算中）。这意味着每个被屏蔽的时间步实际上都会减少计算出的误差！

#%% Yu-yang's example
# https://dev59.com/2VYN5IYBdhLWcg3wsZ1R#47060797
import tensorflow as tf
import tensorflow.keras as keras
import numpy as np
# Fix the random seed for repeatable results
np.random.seed(5)
tf.random.set_seed(5)

max_sentence_length = 5
character_number = 2

input_tensor = keras.Input(shape=(max_sentence_length, character_number))
masked_input = keras.layers.Masking(mask_value=0)(input_tensor)
output = keras.layers.LSTM(3, return_sequences=True)(masked_input)
model = keras.Model(input_tensor, output)
model.compile(loss='mae', optimizer='adam')

X = np.array([[[0, 0], [0, 0], [1, 0], [0, 1], [0, 1]],
              [[0, 0], [0, 1], [1, 0], [0, 1], [0, 1]]])
y_true = np.ones((2, max_sentence_length, 3))
y_pred = model.predict(X)
print(y_pred)

# See if the loss computed by model.evaluate() is equal to the masked loss
unmasked_loss = np.abs(1 - y_pred).mean()
masked_loss = np.abs(1 - y_pred[y_pred != 0]).mean()

print(f"model.evaluate= {model.evaluate(X, y_true)}")
print(f"masked loss= {masked_loss}")
print(f"unmasked loss= {unmasked_loss}") 

输出：

[[[ 0.          0.          0.        ]
  [ 0.          0.          0.        ]
  [ 0.05340272 -0.06415359 -0.11803789]
  [ 0.08775083  0.00600774 -0.10454659]
  [ 0.11212641  0.07632366 -0.04133942]]

 [[ 0.          0.          0.        ]
  [ 0.05394626  0.08956442  0.03843312]
  [ 0.09092357 -0.02743799 -0.10386454]
  [ 0.10791279  0.04083341 -0.08820333]
  [ 0.12459432  0.09971555 -0.02882453]]]
1/1 [==============================] - 1s 658ms/step - loss: 0.6865
model.evaluate= 0.6864957213401794
masked loss= 0.9807082414627075
unmasked loss= 0.986495852470398 

（这是一条评论，而非答案。）

- Dean_me