首页 > 解决方案 > 在 Keras LSTM 自动编码器中重构消耗的序列掩码

问题描述

我试图了解 Keras 中不同时间步数的输入的序列到序列自动编码(LSTM)。我对使用这样的网络将可变长度序列编码为固定向量特别感兴趣(不使用 return_sequences=True 用于 LSTM 编码器)。

例如,仅使用遮罩层...

import numpy as np
import tensorflow as tf

test_data = np.array([[[1,2,1],[3,2,1],[0,0,0],[0,0,0]],
                      [[2,1,1],[3,1,2],[1,2,1],[3,3,3]],
                      [[3,2,3],[1,3,2],[1,1,1],[0,0,0]],
                      [[1,1,2],[0,0,0],[0,0,0,],[0,0,0]],
                      [[2,1,3],[4,2,0],[1,1,2],[1,3,2]]],dtype=np.float32)
input_layer = tf.keras.layers.Input((4,3))
masking_layer = tf.keras.layers.Masking(mask_value=0)(input_layer)
# masks correctly to this point

model = tf.keras.models.Model(inputs=input_layer, outputs=masking_layer)

print(model(test_data)._keras_mask)

tf.Tensor(
[[ True  True False False]
 [ True  True  True  True]
 [ True  True  True False]
 [ True False False False]
 [ True  True  True  True]], shape=(5, 4), dtype=bool)

正如预期的那样。添加 LSTM 编码层会消耗此掩码:

encode_layer = tf.keras.layers.LSTM(64)(masking_layer)
model = tf.keras.models.Model(inputs=input_layer, outputs= encode_layer)
print(model(test_data)._keras_mask)

没有产生任何结果(再次,这是预期的)。我想做的是(以某种方式)重构由 masking_layer 生成的掩码并将其应用于我用来馈送 LSTM 解码器的 RepeatVector:

repeated = tf.keras.layers.RepeatVector(5)(encode_layer)
# it's here I'd like to reincarnate the mask, as it should propagate to the decoder

decode_layer = tf.keras.layers.LSTM(5,return_sequences=True)(repeated)

model = tf.keras.models.Model(inputs=input_layer, outputs=decode_layer)

我已经尝试实现一个 Lambda 层来尝试从遮罩层读取遮罩,但我认为我严重误解了这一步

get_mask = lambda x: masking_layer._keras_mask
Lambda_layer = tf.keras.layers.Lambda(lambda x: x, mask=get_mask)(repeated)

这只会生成一个 TypeError: () 接受 1 个位置参数,但给出了 2 个。

我很感激您对复活这样的面具有任何见解。我知道我可以完全重新考虑网络,但我想保留单向量编码表示并避免在编码 LSTM 上 return_sequences=True。

提前谢谢你。

标签: keraslstmtensorflow2.0

解决方案


这似乎对我有用,尽管我很想听听其他解决方案;可能比这更优雅......

我创建了一个自定义层,Reapply_Masking,改编自 Keras 的默认遮罩层。它需要两个输入层: Input[0] 是您要应用遮罩的层(在我的示例中,RepeatVector 层);Input[1] 是原始掩码之前的层(在我的示例中为 input_layer)。

它像 Masking 层一样重新计算掩码,并将 Input[0] 的掩码时间步归零以进行良好测量(Keras 的掩码层也是如此)。

class Reapply_Masking(Layer):
def __init__(self, mask_value, **kwargs):
    super(Reapply_Masking, self).__init__(**kwargs)
    self.supports_masking = True
    self.mask_value = mask_value

def compute_output_shape(self, input_shape_list):
    return input_shape_list[0].shape

def compute_mask(self, input_list, mask=None):
    return K.any(math_ops.not_equal(input_list[1], self.mask_value), axis=-1)

def call(self, input_list):
    to_output = input_list[0]

    boolean_mask = array_ops.squeeze(K.any(
        math_ops.not_equal(input_list[1], self.mask_value), axis=-1, keepdims=True),axis=-1)       

    dim =(input_list[0].shape[-1])
    killer = math_ops.cast(tf.keras.backend.repeat_elements(tf.expand_dims(boolean_mask,axis=2), dim, axis=2),
                           to_output.dtype)

    outputs = to_output * killer
    outputs._keras_mask = boolean_mask

    return outputs

def get_config(self):
    config = {'mask_value': self.mask_value}
    base_config = super(Reapply_Masking, self).get_config()
    return dict(list(base_config.items()) + list(config.items()))

我似乎能够保存/恢复模型,但只能使用 .h5 格式和 custom_objects。


推荐阅读