作者： 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) 的 2D 張量)，並將其附加到 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) 的 2D 布林值張量，其中每個個別的 False 項目都表示在處理期間應忽略對應的時間步。
Functional API 和 Sequential API 中的遮罩傳播
當使用 Functional API 或 Sequential API 時，由 Embedding 或 Masking 層產生的遮罩將透過網路傳播到任何能夠使用它們的層 (例如 RNN 層)。Keras 會自動擷取對應於輸入的遮罩，並將其傳遞給任何知道如何使用它的層。
例如，在下列 Sequential 模型中，LSTM 層將自動接收遮罩，這表示它將忽略填充值
model = keras.Sequential(
[
layers.Embedding(input_dim=5000, output_dim=16, mask_zero=True),
layers.LSTM(32),
]
)
下列 Functional 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 層的情況就是如此。 - 在 Functional API 和 Sequential API 中，遮罩資訊會自動傳播。
- 當以獨立方式使用層時，您可以手動將
mask引數傳遞到層。 - 您可以輕鬆編寫修改目前遮罩、產生新遮罩或取用與輸入相關聯的遮罩的層。