Ayuda a proteger la Gran Barrera de Coral con TensorFlow en Kaggle Únete Challenge

Enmascaramiento y acolchado con Keras

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 secuencia 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 más de cerca.

Datos de secuencia de relleno

Al procesar datos de secuencia, 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 se pueden vectorizar 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 puede 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 en realidad 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.Masking capa.
  • Configurar un keras.layers.Embedding capa con mask_zero=True .
  • Pasar una mask argumento 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áscara 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 también es 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 las 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)>

Apoyando 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 de 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 suscripción voluntaria. 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.

A continuación, se muestra un ejemplo de una capa que está 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')

Escribir capas 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 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 el acolchado y el enmascaramiento en Keras. Recordar:

  • El "enmascaramiento" es la forma en que las capas pueden saber cuándo omitir / ignorar ciertos pasos de tiempo en las entradas de secuencia.
  • Algunas capas son máscara-generadores: Embedding puede generar una máscara a partir de los valores de entrada (si mask_zero=True ), y así puede el Masking capa.
  • Algunas capas son máscaras-consumidores: se exponen una mask argumento 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 mask argumentos para capas de forma manual.
  • Puede escribir fácilmente capas que modifiquen la máscara actual, que generen una nueva máscara o que consuman la máscara asociada con las entradas.