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.

Mascaramento e preenchimento com Keras

Veja em TensorFlow.org

Executar no Google Colab

Ver fonte em GitHub

Baixar notebook

Configuração

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

Introdução

O mascaramento (masking) é uma maneira de dizer às camadas de processamento de sequência que determinados intervalos de tempo de uma entrada estão ausentes e, portanto, devem ser ignorados ao processar os dados.

O preenchimento (padding) é uma forma especial de mascaramento em que as etapas mascaradas estão no início ou no final de uma sequência. O preenchimento vem da necessidade de codificar os dados da sequência em lotes contíguos: para que todas as sequências de um lote caibam em um determinado comprimento padrão, é necessário preencher ou truncar algumas sequências.

Vamos dar uma olhada mais de perto.

Preenchimento de dados de sequência

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"],
]

Depois da pesquisa de vocabulário, os dados podem ser vetorizados como 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 comprimentos 3, 5 e 6, respectivamente. Como os dados de entrada para um modelo de aprendizado profundo devem ser um único tensor (de forma, por exemplo, (batch_size, 6, vocab_size) neste caso), as amostras que são mais curtas que o item mais longo precisam ser preenchidas com um valor especial para marcar posição (alternativamente, também pode-se truncar amostras longas antes de preencher amostras curtas).

O Keras fornece uma função utilitária para truncar e preencher listas do Python para terem 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)

Mascaramento

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

Há três maneiras de introduzir máscaras de entrada nos modelos Keras:

• Adicionando uma camada keras.layers.Masking .

• Configurando uma camada keras.layers.Embedding com mask_zero=True .

• Passando um argumento mask manualmente ao chamar camadas que suportam este argumento (por exemplo, camadas RNN).

Camadas geradoras de máscaras: Embedding e Masking

Nos bastidores, essas camadas criarão um tensor de máscara (tensor 2D com forma (batch, sequence_length)) que será anexado à saída do tensor retornada pela camada Masking ou 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 você pode ver no resultado impresso, a máscara é um tensor booleano 2D com formato (batch_size, sequence_length), onde cada entrada False individual indica que o intervalo de tempo correspondente deve ser ignorado durante o processamento.

Propagação de máscaras na API Functional e na API Sequential

Ao utilizar a API Functional ou a API Sequential, uma máscara gerada por uma camada Embedding ou Masking será propagada pela rede para qualquer camada que seja capaz de utilizá-las (por exemplo, camadas RNN). O Keras buscará automaticamente a máscara correspondente a uma entrada e a passará para qualquer camada que saiba como usá-la.

Por exemplo, no modelo Sequential a seguir, a camada LSTM receberá automaticamente uma máscara, o que significa que ignorará os valores preenchidos:

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

Este também é o caso do seguinte modelo da API Functional:

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 as camadas

As camadas que podem processar máscaras (como a camada LSTM) têm um argumento mask em seu método __call__.

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

Assim, você pode passar a saída do método compute_mask() de uma camada produtora de máscaras para o método __call__ de uma camada consumidora de máscaras, dessa 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)

Suporte ao mascaramento em suas camadas personalizadas

Às vezes, pode ser necessário escrever camadas que geram uma máscara (como Embedding) ou camadas que precisem modificar a máscara atual.

Por exemplo, qualquer camada que produza um tensor com uma dimensão de tempo diferente de sua entrada, tal como uma camada Concatenate que concatena na dimensão de tempo, precisará modificar a máscara atual para que as camadas seguintes possam levar considerar corretamente os intervalos de tempo mascarados.

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

Aqui está um exemplo de uma camada TemporalSplit que precisa 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)

Aqui está outro exemplo de uma camada CustomEmbedding 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)

Aceitando a propagação de máscaras em camadas compatíveis

A maioria das camadas não modifica a dimensão de tempo, então não há necessidade de se modificar a máscara atual. No entanto, elas ainda podem querer propagar a máscara atual, inalterada, para a camada seguinte. Este é um comportamento opcional. Por padrão, uma camada personalizada destruirá a máscara atual (já que o framework não tem como saber se a propagação da máscara é segura).

Se você tiver uma camada personalizada que não modifica a dimensão de tempo e quiser que ela seja capaz de propagar a máscara de entrada atual, defina self.supports_masking = True no construtor da camada. Neste caso, o comportamento padrão de compute_mask() será de apenas passar adiante a máscara atual.

Eis aqui um exemplo de uma camada configurada para permitir a propagação 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)

Agora você pode usar essa camada personalizada entre uma camada geradora de máscaras (como Embedding) e uma camada consumidora de máscaras (como LSTM ), e ela passará a máscara para que ela alcance a camada consumidora de 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)

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

Algumas camadas são consumidoras de máscara: elas aceitam um argumento mask na call e o usam para determinar se devem pular determinadas etapas de tempo.

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

Veja a seguir um exemplo simples: uma camada que calcula um softmax sobre a dimensão de tempo (eixo 1) de uma sequência de entrada, enquanto descarta intervalos de tempo 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 o que você precisa saber sobre preenchimento e mascaramento no Keras. Para recapitular:

• "Mascaramento" é como as camadas são capazes de saber quando pular/ignorar determinados intervalos de tempo nas entradas de sequências.

• Algumas camadas são geradoras de máscaras: Embedding pode gerar uma máscara a partir de valores de entrada (se mask_zero=True ), assim como a camada Masking.

• Algumas camadas são consumidoras de máscaras: elas expõem um argumento mask em seu método __call__. É o caso das camadas RNN.

• Nas APIs Functional e Sequential, as informações de máscara são propagadas automaticamente.

• Ao usar camadas de forma standalone, você pode passar os argumentos mask para as 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.