 Ver en TensorFlow.org |  Ejecutar en Google Colab |  Ver fuente en GitHub |  Descargar cuaderno |
Configuración
import numpy as np
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
Introducción
El enmascaramiento es una manera de contar capas de secuencia de procesamiento que ciertos timesteps en una entrada faltan, y por lo tanto debe ser saltado al procesar los datos.
El relleno es una forma especial de enmascaramiento, donde los pasos son enmascarados al principio o al final de una secuencia. El relleno proviene de la necesidad de codificar datos de secuencias en lotes contiguos: para que todas las secuencias de un lote se ajusten a una longitud estándar determinada, es necesario rellenar o truncar algunas secuencias.
Echemos un vistazo de cerca.
Datos de secuencia de relleno
Al procesar datos de secuencias, es muy común que las muestras individuales tengan diferentes longitudes. Considere el siguiente ejemplo (texto tokenizado como palabras):
[
["Hello", "world", "!"],
["How", "are", "you", "doing", "today"],
["The", "weather", "will", "be", "nice", "tomorrow"],
]
Después de la búsqueda de vocabulario, los datos pueden ser vectorizados como números enteros, por ejemplo:
[
[71, 1331, 4231]
[73, 8, 3215, 55, 927],
[83, 91, 1, 645, 1253, 927],
]
Los datos son una lista anidada donde las muestras individuales tienen una longitud de 3, 5 y 6, respectivamente. Puesto que los datos de entrada para un modelo de aprendizaje profundo debe ser una sola tensor (de por ejemplo, forma (batch_size, 6, vocab_size) en este caso), las muestras que son más cortos que el más largo necesidad elemento a ser rellenados con un valor de marcador de posición (alternativamente, uno también podría truncar muestras largas antes de rellenar muestras cortas).
Keras proporciona una función de utilidad para truncar y cojín listas de Python a una longitud común: tf.keras.preprocessing.sequence.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 "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.preprocessing.sequence.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]]
Enmascaramiento
Ahora que todas las muestras tienen una longitud uniforme, se debe informar al modelo de que una parte de los datos se está rellenando y se debe ignorar. Ese mecanismo está enmascarando.
Hay tres formas de introducir máscaras de entrada en los modelos de Keras:
- Añadir un
keras.layers.Maskingcapa. - Configurar un
keras.layers.Embeddingcapa conmask_zero=True. - Pasar una
maskargumento manualmente cuando se llama a capas que apoyan este argumento (por ejemplo, capas RNN).
Capas generadoras de Máscara: Embedding y Masking
Bajo el capó, estas capas crearán un tensor máscara (tensor 2D con forma (batch, sequence_length) ), y adjuntarlo a la salida tensor devuelto por el Masking o Embedding capa.
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)
Como se puede ver a partir del resultado impreso, la máscara es un tensor boolean 2D con forma (batch_size, sequence_length) , donde cada individuo False entrada indica que el paso de tiempo correspondiente debe ser ignorada durante el procesamiento.
Propagación de máscaras en la API funcional y la API secuencial
Al utilizar la API funcional o la API secuencial, una máscara generada por una Embedding o Masking capa se propagará a través de la red para cualquier capa que es capaz de utilizar ellos (por ejemplo, capas de RNN). Keras buscará automáticamente la máscara correspondiente a una entrada y la pasará a cualquier capa que sepa cómo usarla.
Por ejemplo, en el siguiente modelo secuencial, el LSTM capa recibirá automáticamente una máscara, lo que significa que ignorar los valores acolchados:
model = keras.Sequential(
[layers.Embedding(input_dim=5000, output_dim=16, mask_zero=True), layers.LSTM(32),]
)
Este es también el caso del siguiente modelo de API funcional:
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)
Pasar tensores de máscara directamente a capas
Capas que pueden manejar máscaras (como el LSTM capa) tienen una mask argumento en su __call__ método.
Mientras tanto, las capas que producen una máscara (por ejemplo Embedding ) exponen un compute_mask(input, previous_mask) método que se puede llamar.
Por lo tanto, se puede pasar la salida del compute_mask() método de una capa de máscara que producen a la __call__ método de una capa de máscara que consume, como este:
class MyLayer(layers.Layer):
def __init__(self, **kwargs):
super(MyLayer, self).__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([[-3.6287602e-04, 8.8942451e-03, -4.5623952e-03, ...,
3.6509466e-04, -4.3871473e-03, -1.7532009e-03],
[ 2.6261162e-03, -2.5420082e-03, 7.6517118e-03, ...,
5.8210879e-03, -1.5617531e-03, -1.7562184e-03],
[ 6.8687932e-03, 1.2330032e-03, -1.2028826e-02, ...,
2.0486799e-03, 5.7172528e-03, 2.6641595e-03],
...,
[-3.4327951e-04, 1.3967649e-03, -1.2102776e-02, ...,
3.8406218e-03, -2.3374180e-03, -4.9669710e-03],
[-2.3023323e-03, 1.8474255e-03, 2.7329330e-05, ...,
6.1798934e-03, 4.2709545e-04, 3.9026213e-03],
[ 7.4090287e-03, 1.9879336e-03, -2.0261200e-03, ...,
8.2100276e-03, 8.7051848e-03, 9.9167246e-03]], dtype=float32)>
Compatibilidad con el enmascaramiento en sus capas personalizadas
A veces, es posible que necesite capas de escritura que generan una máscara (como Embedding ), o capas que necesitan para modificar la máscara actual.
Por ejemplo, cualquier capa que produce un tensor con una dimensión de tiempo diferente de su entrada, tal como un Concatenate capa que concatena en la dimensión de tiempo, tendrá que modificar la máscara actual, de modo que las capas de aguas abajo serán capaces de tomar correctamente timesteps enmascarados en cuenta.
Para ello, la capa debe implementar el layer.compute_mask() método, que produce una nueva máscara dada la entrada y la máscara actual.
Aquí está un ejemplo de una TemporalSplit capa que necesita modificar la máscara actual.
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)
Este es otro ejemplo de un CustomEmbedding capa que es capaz de generar una máscara a partir de los valores de entrada:
class CustomEmbedding(keras.layers.Layer):
def __init__(self, input_dim, output_dim, mask_zero=False, **kwargs):
super(CustomEmbedding, self).__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 True False True True True True] [ True True True True True True True True False True]], shape=(3, 10), dtype=bool)
Optar por enmascarar la propagación en capas compatibles
La mayoría de las capas no modifican la dimensión del tiempo, por lo que no es necesario modificar la máscara actual. Sin embargo, es posible que todavía quieren ser capaces de propagar la máscara actual, sin cambios, a la siguiente capa. Este es un comportamiento de aceptación. De forma predeterminada, una capa personalizada destruirá la máscara actual (ya que el marco no tiene forma de saber si es seguro propagar la máscara).
Si usted tiene una capa personalizada que no modifica la dimensión de tiempo, y si usted quiere que sea capaz de propagar la máscara de entrada de corriente, se debe configurar self.supports_masking = True en el constructor capa. En este caso, el comportamiento predeterminado de compute_mask() es que sólo tiene que pasar a través de la máscara actual.
Este es un ejemplo de una capa incluida en la lista blanca para la propagación de máscaras:
class MyActivation(keras.layers.Layer):
def __init__(self, **kwargs):
super(MyActivation, self).__init__(**kwargs)
# Signal that the layer is safe for mask propagation
self.supports_masking = True
def call(self, inputs):
return tf.nn.relu(inputs)
Ahora puede utilizar esta capa personalizada en el medio de una capa de generación de máscaras (como Embedding ) y una capa de máscara que consume (como LSTM ), y se pasa a lo largo de la máscara para que llegue a la capa de máscara que consume.
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')
Capas de escritura que necesitan información de máscara
Algunas capas son consumidores de máscara: aceptan una mask argumento a call y lo utilizan para determinar si se debe omitir ciertos intervalos de tiempo.
Para escribir una capa de este tipo, simplemente puede añadir una mask=None de argumentos en la call firma. La máscara asociada con las entradas se pasará a su capa siempre que esté disponible.
Aquí hay un ejemplo simple a continuación: una capa que calcula un softmax sobre la dimensión de tiempo (eje 1) de una secuencia de entrada, mientras descarta los pasos de tiempo enmascarados.
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)))
Resumen
Eso es todo lo que necesita saber sobre relleno y enmascaramiento en Keras. Recordar:
- "Enmascaramiento" es cómo las capas pueden saber cuándo omitir/ignorar ciertos pasos de tiempo en las entradas de secuencia.
- Algunas capas son máscara-generadores:
Embeddingpuede generar una máscara a partir de los valores de entrada (simask_zero=True), y así puede elMaskingcapa. - Algunas capas son máscaras-consumidores: se exponen una
maskargumento en su__call__método. Este es el caso de las capas RNN. - En la API funcional y la API secuencial, la información de la máscara se propaga automáticamente.
- Cuando el uso de capas de una manera independiente, puede pasar las
maskargumentos para capas de forma manual. - Puede escribir fácilmente capas que modifiquen la máscara actual, que generen una máscara nueva o que consuman la máscara asociada con las entradas.