Mascaramento e preenchimento com Keras

Mantenha tudo organizado com as coleções Salve e categorize o conteúdo com base nas suas preferências.

Veja no TensorFlow.org Executar no Google Colab Ver fonte no GitHub Baixar caderno

Configurar

import numpy as np
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers

Introdução

Mascaramento é uma maneira de dizer camadas de processamento de sequências que certos Timesteps em uma entrada estão em falta, e, portanto, devem ser ignoradas ao processar os dados.

Estofamento é uma forma especial de mascaramento em que os passos são mascarados no início ou no fim de uma sequência. O preenchimento vem da necessidade de codificar dados de sequência em lotes contíguos: para fazer com que todas as sequências em um lote se ajustem a um determinado comprimento padrão, é necessário preencher ou truncar algumas sequências.

Vamos dar uma olhada de perto.

Dados de sequência de preenchimento

Ao processar dados de sequência, é muito comum que amostras individuais tenham comprimentos diferentes. Considere o seguinte exemplo (texto tokenizado como palavras):

[
  ["Hello", "world", "!"],
  ["How", "are", "you", "doing", "today"],
  ["The", "weather", "will", "be", "nice", "tomorrow"],
]

Após a pesquisa de vocabulário, os dados podem ser vetorizados como números inteiros, por exemplo:

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

Os dados são uma lista aninhada em que amostras individuais têm comprimento 3, 5 e 6, respectivamente. Uma vez que os dados de entrada para um modelo de aprendizagem profunda deve ser um único tensor (de forma eg (batch_size, 6, vocab_size) , neste caso), as amostras que são mais curtos do que a necessidade item de mais tempo para ser preenchido com algum valor de espaço reservado (em alternativa, um também pode truncar amostras longas antes de preencher amostras curtas).

Keras fornece uma função de utilidade para truncar e almofada listas de Python para um comprimento comum: 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]]

Mascaramento

Agora que todas as amostras têm um comprimento uniforme, o modelo deve ser informado de que alguma parte dos dados é na verdade preenchimento e deve ser ignorada. Esse mecanismo está mascarando.

Existem três maneiras de introduzir máscaras de entrada em modelos Keras:

  • Adicionar uma keras.layers.Masking camada.
  • Configurar uma keras.layers.Embedding camada com mask_zero=True .
  • Passe uma mask argumento manualmente ao chamar camadas que suportam este argumento (por exemplo, camadas RNN).

As camadas de máscara de geração: Embedding e Masking

Sob o capô, estas camadas irá criar um tensor máscara (tensor 2D com a forma (batch, sequence_length) ), e coloque-o a saída do tensor retornado pelo Masking ou Embedding camada.

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 você pode ver a partir do resultado impresso, a máscara é um boolean tensor 2D com forma (batch_size, sequence_length) , onde cada indivíduo False entrada indica que a iteração correspondente deve ser ignorado durante o processamento.

Propagação de máscara na API funcional e na API sequencial

Quando usando a API funcional ou a API sequencial, de uma máscara produzida por um Embedding ou Masking camada irá ser propagado através da rede para qualquer camada que é capaz de utilizá-las (por exemplo, camadas de RNN). Keras irá buscar automaticamente a máscara correspondente a uma entrada e passá-la para qualquer camada que saiba como usá-la.

Por exemplo, no modelo sequencial seguinte, o LSTM camada receberá automaticamente uma máscara, o que significa que irá ignorar valores acolchoados:

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

Esse também é o caso do seguinte 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)

Passando tensores de máscara diretamente para camadas

As camadas que podem lidar com máscaras (tais como o LSTM camada) tem uma mask argumento no seu __call__ método.

Enquanto isso, as camadas que produzem uma máscara (por exemplo Embedding ) expor um compute_mask(input, previous_mask) método que você pode chamar.

Assim, você pode passar a saída do compute_mask() método de uma camada produtora de máscara à __call__ método de uma camada que consome máscara, 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)>

Suportando mascaramento em suas camadas personalizadas

Às vezes, você pode precisar de camadas de gravação que geram uma máscara (como Embedding ), ou camadas que precisam modificar a máscara atual.

Por exemplo, qualquer camada que produz um tensor com uma dimensão de tempo diferente do que a sua entrada, tais como um Concatenate camada que encadeia sobre a dimensão de tempo, será necessário modificar a máscara de corrente de modo que as camadas a jusante irá ser capaz de tomar adequadamente Timesteps mascarados em conta.

Para fazer isso, sua camada deve implementar a layer.compute_mask() método, que produz uma nova máscara dada a entrada e a máscara atual.

Aqui está um exemplo de uma TemporalSplit camada que precisa para modificar a máscara atual.

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)

Aqui é outro exemplo de um CustomEmbedding camada que é capaz de gerar uma máscara a partir de 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 mascarar a propagação em camadas compatíveis

A maioria das camadas não modifica a dimensão de tempo, portanto, não é necessário modificar a máscara atual. No entanto, eles ainda podem querer ser capaz de propagar a máscara atual, inalterada, para a próxima camada. Este é um comportamento opt-in. Por padrão, uma camada personalizada destruirá a máscara atual (já que a estrutura não tem como saber se a propagação da máscara é segura).

Se você tem uma camada personalizada que não modifica a dimensão de tempo, e se você quer que ele seja capaz de propagar a máscara de entrada atual, você deve definir self.supports_masking = True no construtor camada. Neste caso, o comportamento padrão do compute_mask() é apenas para passar a máscara de corrente através.

Aqui está um exemplo de uma camada que está na lista de permissões para propagação de máscara:

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)

Agora você pode usar essa camada personalizado in-entre uma camada de geração de máscara (como Embedding ) e uma camada que consome máscara (como LSTM ), e passará a máscara ao longo de modo que atinge a camada que consome máscara.

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')

Escrevendo camadas que precisam de informações de máscara

Algumas camadas são consumidores máscara: eles aceitam uma mask argumento na call e usá-lo para determinar se deve ignorar determinados intervalos de tempo.

Para escrever camada tal, você pode simplesmente adicionar uma mask=None argumento em sua call assinatura. A máscara associada às entradas será passada para sua camada sempre que estiver disponível.

Aqui está um exemplo simples abaixo: uma camada que calcula um softmax sobre a dimensão de tempo (eixo 1) de uma sequência de entrada, enquanto descarta timesteps mascarados.

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)))

Resumo

Isso é tudo que você precisa saber sobre preenchimento e mascaramento em Keras. Para recapitular:

  • "Mascarar" é como as camadas são capazes de saber quando pular/ignorar certos passos de tempo em entradas de sequência.
  • Algumas camadas são máscara-geradores: Embedding pode gerar uma máscara a partir de valores de entrada (se mask_zero=True ), e por isso pode o Masking camada.
  • Algumas camadas são Mask-consumidores: eles expõem uma mask argumento em sua __call__ método. Este é o caso das camadas RNN.
  • Na API Funcional e na API Sequencial, as informações de máscara são propagadas automaticamente.
  • Ao usar camadas de forma independente, você pode passar as mask argumentos para camadas manualmente.
  • Você pode facilmente escrever camadas que modificam a máscara atual, que geram uma nova máscara ou que consomem a máscara associada às entradas.