作者：Scott Zhu，Francois Chollet
 在 TensorFlow.org 上查看
|
 在 Google Colab 中运行
|
 在 GitHub 上查看源代码
|
 在 keras.io 上查看
|
设置
import numpy as np
import tensorflow as tf
import keras
from keras import layers
简介
掩码是一种告诉序列处理层输入中的某些时间步长丢失，因此在处理数据时应跳过这些时间步长的方法。
填充是掩码的一种特殊形式，其中掩码的步长位于序列的开头或结尾。填充源于将序列数据编码为连续批次的需要：为了使批次中的所有序列都适合给定的标准长度，有必要填充或截断一些序列。
让我们仔细看看。
填充序列数据
在处理序列数据时，单个样本通常具有不同的长度。考虑以下示例（文本按单词标记化）
[
["Hello", "world", "!"],
["How", "are", "you", "doing", "today"],
["The", "weather", "will", "be", "nice", "tomorrow"],
]
在词汇查找后，数据可能被矢量化为整数，例如
[
[71, 1331, 4231]
[73, 8, 3215, 55, 927],
[83, 91, 1, 645, 1253, 927],
]
数据是一个嵌套列表，其中单个样本的长度分别为 3、5 和 6。由于深度学习模型的输入数据必须是一个单一张量（形状例如 (batch_size, 6, vocab_size) 在这种情况下），短于最长项目的样本需要用一些占位符值填充（或者，也可以在填充短样本之前截断长样本）。
Keras 提供了一个实用函数来截断和填充 Python 列表以达到共同的长度：tf.keras.utils.pad_sequences.
raw_inputs = [
[711, 632, 71],
[73, 8, 3215, 55, 927],
[83, 91, 1, 645, 1253, 927],
]
# By default, this will pad using 0s; it is configurable via the
# "value" parameter.
# Note that you could use "pre" padding (at the beginning) or
# "post" padding (at the end).
# We recommend using "post" padding when working with RNN layers
# (in order to be able to use the
# CuDNN implementation of the layers).
padded_inputs = tf.keras.utils.pad_sequences(raw_inputs, padding="post")
print(padded_inputs)
[[ 711 632 71 0 0 0] [ 73 8 3215 55 927 0] [ 83 91 1 645 1253 927]]
掩码
现在所有样本都具有统一的长度，模型必须被告知数据的一部分实际上是填充，应该被忽略。这种机制就是掩码。
在 Keras 模型中引入输入掩码有三种方法
- 添加一个
keras.layers.Masking层。 - 使用
mask_zero=True配置一个keras.layers.Embedding层。 - 在调用支持此参数的层（例如 RNN 层）时，手动传递
mask参数。
掩码生成层：Embedding 和 Masking
在幕后，这些层将创建一个掩码张量（形状为 (batch, sequence_length) 的二维张量），并将其附加到 Masking 或 Embedding 层返回的张量输出。
embedding = layers.Embedding(input_dim=5000, output_dim=16, mask_zero=True)
masked_output = embedding(padded_inputs)
print(masked_output._keras_mask)
masking_layer = layers.Masking()
# Simulate the embedding lookup by expanding the 2D input to 3D,
# with embedding dimension of 10.
unmasked_embedding = tf.cast(
tf.tile(tf.expand_dims(padded_inputs, axis=-1), [1, 1, 10]), tf.float32
)
masked_embedding = masking_layer(unmasked_embedding)
print(masked_embedding._keras_mask)
tf.Tensor( [[ True True True False False False] [ True True True True True False] [ True True True True True True]], shape=(3, 6), dtype=bool) tf.Tensor( [[ True True True False False False] [ True True True True True False] [ True True True True True True]], shape=(3, 6), dtype=bool)
从打印结果可以看出，掩码是一个形状为 (batch_size, sequence_length) 的二维布尔张量，其中每个单独的 False 项表示在处理过程中应忽略相应的时步。
功能 API 和顺序 API 中的掩码传播
在使用功能 API 或顺序 API 时，由 Embedding 或 Masking 层生成的掩码将通过网络传播到任何能够使用它们的层（例如，RNN 层）。Keras 将自动获取与输入相对应的掩码，并将其传递给任何知道如何使用它的层。
例如，在以下顺序模型中，LSTM 层将自动接收掩码，这意味着它将忽略填充值
model = keras.Sequential(
[
layers.Embedding(input_dim=5000, output_dim=16, mask_zero=True),
layers.LSTM(32),
]
)
以下功能 API 模型也是如此
inputs = keras.Input(shape=(None,), dtype="int32")
x = layers.Embedding(input_dim=5000, output_dim=16, mask_zero=True)(inputs)
outputs = layers.LSTM(32)(x)
model = keras.Model(inputs, outputs)
将掩码张量直接传递给层
可以处理掩码的层（例如 LSTM 层）在其 __call__ 方法中有一个 mask 参数。
同时，生成掩码的层（例如 Embedding）公开了一个 compute_mask(input, previous_mask) 方法，您可以调用该方法。
因此，您可以将掩码生成层的 compute_mask() 方法的输出传递给掩码使用层的 __call__ 方法，如下所示
class MyLayer(layers.Layer):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.embedding = layers.Embedding(input_dim=5000, output_dim=16, mask_zero=True)
self.lstm = layers.LSTM(32)
def call(self, inputs):
x = self.embedding(inputs)
# Note that you could also prepare a `mask` tensor manually.
# It only needs to be a boolean tensor
# with the right shape, i.e. (batch_size, timesteps).
mask = self.embedding.compute_mask(inputs)
output = self.lstm(x, mask=mask) # The layer will ignore the masked values
return output
layer = MyLayer()
x = np.random.random((32, 10)) * 100
x = x.astype("int32")
layer(x)
<tf.Tensor: shape=(32, 32), dtype=float32, numpy=
array([[ 1.1063378e-04, -5.7033719e-03, 3.0645048e-03, ...,
3.6328615e-04, -2.8766368e-03, -1.3289017e-03],
[-9.2790304e-03, -1.5139847e-02, 5.7660388e-03, ...,
3.5337124e-03, 4.0699611e-03, -3.9524431e-04],
[-3.4190060e-03, 7.9529232e-04, 3.7830453e-03, ...,
-6.8300538e-04, 4.7965860e-03, 4.4357078e-03],
...,
[-4.3796434e-04, 3.5149506e-03, 5.0854073e-03, ...,
6.3023632e-03, -4.6664057e-03, -2.1111544e-03],
[ 1.2171637e-03, -1.8671650e-03, 8.6708134e-03, ...,
-2.6730294e-03, -1.6238958e-03, 5.9354519e-03],
[-7.1832030e-03, -6.0863695e-03, 4.3814078e-05, ...,
3.8765911e-03, -1.7828923e-03, -2.3530782e-03]], dtype=float32)>
在自定义层中支持掩码
有时，您可能需要编写生成掩码的层（如 Embedding），或需要修改当前掩码的层。
例如，任何生成时间维度与其输入不同的张量的层，例如在时间维度上连接的 Concatenate 层，都需要修改当前掩码，以便下游层能够正确地考虑掩码的时步。
为此，您的层应实现 layer.compute_mask() 方法，该方法根据输入和当前掩码生成新的掩码。
以下是一个需要修改当前掩码的 TemporalSplit 层的示例。
class TemporalSplit(keras.layers.Layer):
"""Split the input tensor into 2 tensors along the time dimension."""
def call(self, inputs):
# Expect the input to be 3D and mask to be 2D, split the input tensor into 2
# subtensors along the time axis (axis 1).
return tf.split(inputs, 2, axis=1)
def compute_mask(self, inputs, mask=None):
# Also split the mask into 2 if it presents.
if mask is None:
return None
return tf.split(mask, 2, axis=1)
first_half, second_half = TemporalSplit()(masked_embedding)
print(first_half._keras_mask)
print(second_half._keras_mask)
tf.Tensor( [[ True True True] [ True True True] [ True True True]], shape=(3, 3), dtype=bool) tf.Tensor( [[False False False] [ True True False] [ True True True]], shape=(3, 3), dtype=bool)
以下是一个能够从输入值生成掩码的 CustomEmbedding 层的另一个示例
class CustomEmbedding(keras.layers.Layer):
def __init__(self, input_dim, output_dim, mask_zero=False, **kwargs):
super().__init__(**kwargs)
self.input_dim = input_dim
self.output_dim = output_dim
self.mask_zero = mask_zero
def build(self, input_shape):
self.embeddings = self.add_weight(
shape=(self.input_dim, self.output_dim),
initializer="random_normal",
dtype="float32",
)
def call(self, inputs):
return tf.nn.embedding_lookup(self.embeddings, inputs)
def compute_mask(self, inputs, mask=None):
if not self.mask_zero:
return None
return tf.not_equal(inputs, 0)
layer = CustomEmbedding(10, 32, mask_zero=True)
x = np.random.random((3, 10)) * 9
x = x.astype("int32")
y = layer(x)
mask = layer.compute_mask(x)
print(mask)
tf.Tensor( [[ True True True True True True True True True True] [ True True True True False True False True True True] [ True False True False True True True True True True]], shape=(3, 10), dtype=bool)
在兼容层上选择加入掩码传播
大多数层不会修改时间维度，因此不需要修改当前掩码。但是，它们可能仍然希望能够传播当前掩码（未更改）到下一层。这是可选行为。默认情况下，自定义层将销毁当前掩码（因为框架无法判断传播掩码是否安全）。
如果您有一个不修改时间维度的自定义层，并且希望它能够传播当前输入掩码，您应该在层构造函数中设置 self.supports_masking = True。在这种情况下，compute_mask() 的默认行为是只将当前掩码传递过去。
以下是一个被列入白名单以进行掩码传播的层的示例
@keras.saving.register_keras_serializable()
class MyActivation(keras.layers.Layer):
def __init__(self, **kwargs):
super().__init__(**kwargs)
# Signal that the layer is safe for mask propagation
self.supports_masking = True
def call(self, inputs):
return tf.nn.relu(inputs)
您现在可以在掩码生成层（如 Embedding）和掩码使用层（如 LSTM）之间使用此自定义层，它将传递掩码，以便它到达掩码使用层。
inputs = keras.Input(shape=(None,), dtype="int32")
x = layers.Embedding(input_dim=5000, output_dim=16, mask_zero=True)(inputs)
x = MyActivation()(x) # Will pass the mask along
print("Mask found:", x._keras_mask)
outputs = layers.LSTM(32)(x) # Will receive the mask
model = keras.Model(inputs, outputs)
Mask found: KerasTensor(type_spec=TensorSpec(shape=(None, None), dtype=tf.bool, name=None), name='Placeholder_1:0')
编写需要掩码信息的层
有些层是掩码使用者：它们在 call 中接受一个 mask 参数，并使用它来确定是否跳过某些时间步。
要编写这样的层，您只需在 call 签名中添加一个 mask=None 参数。只要可用，与输入相关的掩码就会传递给您的层。
以下是一个简单的示例：一个在输入序列的时间维度（轴 1）上计算 softmax 的层，同时丢弃掩码的时步。
@keras.saving.register_keras_serializable()
class TemporalSoftmax(keras.layers.Layer):
def call(self, inputs, mask=None):
broadcast_float_mask = tf.expand_dims(tf.cast(mask, "float32"), -1)
inputs_exp = tf.exp(inputs) * broadcast_float_mask
inputs_sum = tf.reduce_sum(
inputs_exp * broadcast_float_mask, axis=-1, keepdims=True
)
return inputs_exp / inputs_sum
inputs = keras.Input(shape=(None,), dtype="int32")
x = layers.Embedding(input_dim=10, output_dim=32, mask_zero=True)(inputs)
x = layers.Dense(1)(x)
outputs = TemporalSoftmax()(x)
model = keras.Model(inputs, outputs)
y = model(np.random.randint(0, 10, size=(32, 100)), np.random.random((32, 100, 1)))
总结
这就是您需要了解的关于 Keras 中的填充和掩码的所有内容。概括一下
- "掩码"是层如何知道何时跳过/忽略序列输入中的某些时间步。
- 有些层是掩码生成器：
Embedding可以从输入值生成掩码（如果mask_zero=True），Masking层也可以。 - 有些层是掩码使用者：它们在其
__call__方法中公开了一个mask参数。RNN 层就是这种情况。 - 在功能 API 和顺序 API 中，掩码信息会自动传播。
- 在以独立方式使用层时，您可以手动将
mask参数传递给层。 - 您可以轻松编写修改当前掩码、生成新掩码或使用与输入相关的掩码的层。