如何为 LSTM 实现 Keras 自定义损失函数

Question

我有一个多类分类问题，我正在使用 LSTM 来解决它。我一直在使用categorical_crossentropy 训练我的模型。但是在检查模型质量（训练后）时，我必须使用这个自定义指标，其中A 是一个二维惩罚矩阵：

def score(y_true, y_pred):
    S = 0.0
    y_true = y_true.astype(int)
    y_pred = y_pred.astype(int)
    for i in range(0, y_true.shape[0]):
        S -= A[y_true[i], y_pred[i]]
    return S/y_true.shape[0]

这样的自定义指标可以接收y_true 和y_pred 作为Pandas Series 对象的输入，它输出一个负数，越接近零越好。

我想将当前的categorical_crossentropy 损失函数替换为与上述自定义指标具有类似行为的自定义损失函数，即考虑A 惩罚矩阵。

我面临的问题是损失函数的输入是 Tensor 对象，而不是我完全不熟悉的 Pandas Series 对象。不仅如此，当我处理 LSTM 时，我的输入张量的形状是 3D 的：

y_true: Tensor("IteratorGetNext:1", shape=(1, 25131, 12), dtype=uint8)
type(y_true): <class 'tensorflow.python.framework.ops.Tensor'>
y_pred: Tensor("sequential_26/time_distributed_26/Reshape_1:0", shape=(1, 25131, 12), dtype=float32)
type(y_pred): <class 'tensorflow.python.framework.ops.Tensor'>

如果有帮助，这就是我的架构：

callbacks = [EarlyStopping(monitor='val_loss', patience=25)]

model = Sequential()
model.add(Masking(mask_value = 0.))
model.add(Bidirectional(LSTM(64, return_sequences=True, activation = "tanh")))
model.add(Dropout(0.3))
model.add(TimeDistributed(Dense(12, activation='softmax')))
adam = adam_v2.Adam(learning_rate=0.002)

model.compile(optimizer=adam, loss=score, metrics=['accuracy'])

history = model.fit(X_train, y_train, epochs=150, batch_size=1, shuffle=False,
                    validation_data=(X_test, y_test), verbose=2, callbacks=[callbacks])

这些是我对模型的输入数据的形状，我总共有 12 个类：

print(f'{X_train.shape} {X_test.shape} {y_train.shape} {y_test.shape}')
(73, 25131, 29) (25, 23879, 29) (73, 25131, 12) (25, 23879, 12)

这是A惩罚矩阵，大小为12x12，是多类分类问题的类数：

这是我正在为其构建模型的比赛：

https://xeek.ai/challenges/force-well-logs/overview

https://github.com/bolgebrygg/Force-2020-Machine-Learning-competition/tree/master/lithology_competition

Answer 1

当您想要进行模型评估时，您需要一个指标而不是损失。

要将您的函数转换为 tensorflow 中的自定义指标：

1. 将y_true 和y_pred 转换为int 不会得到你想要的，你需要使用tf.argmax 代替，假设y_true 是一次性编码标签，y_pred 是概率。 （因为你用categorical_crossentropy训练，softmax作为输出层的激活函数）

2. 定义自定义指标类

示例代码：

class Score(tf.keras.metrics.Metric):
  def __init__(self, A, name="score", **kwargs):
    super().__init__(name=name, **kwargs)
    self.A = tf.constant(A,dtype=tf.float32)
    self.S = self.add_weight(name="S", initializer="zeros")
    self.num_elems = self.add_weight(name="num_elems", initializer="zeros")

  def update_state(self, y_true, y_pred, sample_weight=None):
    #convert y_true and y_pred to indices and flatten them
    y_true = tf.reshape(tf.argmax(y_true,-1),(-1,))
    y_pred = tf.reshape(tf.argmax(y_pred,-1),(-1,))
    num_elems = y_true.shape[0]
    indices = tf.stack([y_true,y_pred],1)
    #access A using indices and sum
    self.S.assign_add(-tf.reduce_sum(tf.gather_nd(self.A,indices)))
    self.num_elems.assign_add(num_elems)
        
  def result(self):
    return self.S/self.num_elems

此外，model.fit 用于训练模型。如果您只想评估模型，请将Score() 作为指标传递并使用model.evaluate。 示例代码：

model.compile(optimizer=adam, loss=tf.keras.losses.CategoricalCrossentropy(), 
              metrics=['accuracy',Score(A)])
model.evaluate(X_test, y_test, batch_size=1)