masking_and_padding.ipynb

Kernel: Python 3

Copyright 2020 The TensorFlow Authors.

#@title Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Enmascaramiento y amortiguación con Keras

Ver en TensorFlow.org

Ejecutar en Google Colab

Ver el código fuente en GitHub

Descargar el bloc de notas

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 forma de indicarle a las capas que se encargan del procesamiento de las secuencias que faltan ciertos pasos temporales en una entrada y que, por lo tanto, estos deben omitirse durante el procesamiento de los datos.

La amortiguación es una forma especial de enmascaramiento, donde los pasos enmascarados se encuentran al principio o al final de una secuencia. El amortiguamiento surge por la necesidad de codificar secuencias de datos en lotes contiguos: para que todas las secuencias de un lote se ajusten a una cierta longitud estándar es necesario amortiguar o truncar algunas secuencias.

Veamos cómo funciona esto.

Amortiguación de una secuencia de datos

Cuando se procesan secuencias de datos, es muy común que las muestras individuales tengan diferentes longitudes. Considere el siguiente ejemplo (el texto se convirtió en tokens que funcionan 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 vectorizarse como números enteros, por ejemplo:

[
  [71, 1331, 4231]
  [73, 8, 3215, 55, 927],
  [83, 91, 1, 645, 1253, 927],
]

Los datos consisten en una lista anidada en donde las muestras individuales tienen una longiud de 3, 5, y 6 datos, respectivamente. Dado que los datos de entrada para un modelo de aprendizaje profundo deben ser un tensor único (con forma, por ejemplo, (batch_size, 6, vocab_size) en este caso), las muestras más cortas que el elemento más largo deben amortiguarse con algún valor del marcador de posición (o en su defecto, también es posible truncar las muestras largas antes de amortiguar las muestras cortas).

Keras proporciona una función que es útil para truncar y amortiguar las listas de Python a una longitud habitual: 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)

Enmascaramiento

Ahora que todas las muestras tienen una longitud uniforme, es necesario informarle al modelo que una parte de los datos en realidad es un amortiguado y debe ignorarse. Este mecanismo es el enmascaramiento.

En los modelos de Keras, hay tres formas de establecer máscaras de entrada:

• Agregar una capa keras.layers.Masking.

• Configurar una capa keras.layers.Embedding mediante mask_zero=True.

• Pasar un argumento mask manualmente cuando se llama a las capas que sustentan dicho argumento (por ejemplo, las capas RNN).

Capas que generan el enmascaramiento: Embedding y Masking

En el trasfondo, estas capas crearán una máscara para el tensor (un tensor con forma 2D (batch, sequence_length)), y adjunto a este un tensor de salida devuelto por las capas Masking o 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)

Como puede ver en el resultado impreso, la máscara es un tensor booleano en 2D con forma de (batch_size, sequence_length), donde cada entrada individual False indica que el paso temporal correspondiente debe ignorarse durante el procesamiento.

Propagación de la máscara en la API funcional y en la API secuencial

Cuando se utilizan ya sea la API funcional o la API secuencial, una máscara generada por una capa Embedding o Masking se propagará a través de la red para cualquier capa que sea capaz de utilizarlas (por ejemplo, las capas RNN). Entonces, Keras obtendrá automáticamente la máscara correspondiente a una entrada y la pasará a cualquier capa que sepa utilizarla.

Por ejemplo, en el siguiente modelo secuencial, la capa LSTM recibirá una máscara automáticamente, lo cual significa que ignorará los valores amortiguados:

model = keras.Sequential(
    [layers.Embedding(input_dim=5000, output_dim=16, mask_zero=True), layers.LSTM(32),]
)

Lo mismo ocurre con el siguiente modelo de una 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)

Cómo pasar los tensores de la máscara directamente a las capas

Las capas que pueden soportar máscaras (como la capa LSTM) tienen un argumento de tipo mask en su método __call__.

En cambio, las capas que generan una máscara (por ejemplo, Embedding) exhiben un método compute_mask(input, previous_mask) al que usted puede llamar.

Por lo tanto, puede traspasar la salida del método compute_mask() desde una capa que genera máscaras al método __call__ de una capa que utiliza máscaras, de la siguiente forma:

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)

Cómo brindar asistencia a las máscaras en sus capas personalizadas

Algunas veces, es posible que necesite escribir capas que generen una máscara (como Embedding), o capas que necesiten modificar la máscara actual.

Por ejemplo, cualquier capa que genere un tensor cuya dimensión temporal sea diferente a la de su entrada, como una capa Concatenate a fin de que concatene en la dimensión temporal, necesitará modificar la máscara actual para que las capas posteriores consideren correctamente los pasos temporales enmascarados.

Para ello, su nueva capa debe implementar el método layer.compute_mask(), el cual produce una nueva máscara cuando se proporcionan la entrada y la máscara actual.

A continuación, se muestra el ejemplo de una capa TemporalSplit donde se 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)

El siguiente es otro ejemplo de una capa CustomEmbedding 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)

Cómo elegir la propagación de máscaras en capas que sean compatibles

En la mayoría de las capas no se modifica la dimensión temporal, de modo que no es necesario modificar la máscara actual. Sin embargo, es posible que se desee propagar la máscara actual, sin ningún cambio, a la siguiente capa. Esta es una acción opcional. De forma predeterminada, una capa personalizada destruirá la máscara actual (ya que la estructura no tiene forma de saber si es seguro propagar la máscara).

Si tiene una capa personalizada en la que no se modifique la dimensión temporal, y desea que pueda propagarse la máscara de la entrada actual, debe establecer self.supports_masking = True en el generador de la capa. En este caso, la acción predeterminada de compute_mask() es simplemente traspasar la máscara actual.

A continuación, se muestra el ejemplo de una capa que está 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 usar esta capa personalizada que se encuentra entre una capa generadora de máscaras (como Embedding) y una capa que utiliza máscaras (como LSTM), esto traspasará la máscara para que llegue a la capa que utiliza las máscaras.

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)

Cómo escribir capas que necesitan información sobre la máscara

Algunas capas son consumers de máscaras: aceptan un argumento mask en la cally lo usan para determinar si deben saltarse ciertos pasos temporales.

Para escribir una capa de este tipo, puede simplemente añadir un argumento mask=None en la firma de su call signature. La máscara asociada a las entradas se pasará a su capa siempre que esté disponible.

A continuación, veremos un ejemplo sencillo: una capa que calcula un softmax sobre la dimensión temporal (eje 1) de una secuencia de entrada, mientras descarta los pasos temporales 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

Esto es todo lo que necesita saber sobre la amortiguación y el enmascaramiento en Keras. Para recapitular, recuerde lo siguiente:

• El "enmascaramiento" es la forma en que las capas son capaces de saber cuándo omitir / ignorar ciertos pasos temporales en las entradas de la secuencia.

• Algunas capas son generadoras de máscaras: Embedding puede generar una máscara a partir de los valores de entrada (si mask_zero=True), y lo mismo puede hacer la capa Masking.

• Algunas capas utilizan máscaras: exhiben un argumento mask en su método __call__. Este es el caso de las capas RNN.

• Tanto en la API funcional como en la API secuencial, la información sobre la máscara se propaga automáticamente.

• Cuando las capas se usan de forma independiente, puede traspasar los argumentos mask a las capas de forma manual.

• Puede escribir fácilmente capas que modifiquen la máscara actual, ya sea que generen una nueva máscara o que utilicen la máscara asociada a las entradas.