Esta página foi traduzida pela API Cloud Translation.
Switch to English

Mascaramento e preenchimento com Keras

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

Configuração

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

Introdução

O mascaramento é uma maneira de informar às camadas de processamento de sequência que certos passos de tempo em uma entrada estão ausentes e, portanto, devem ser ignorados ao processar os dados.

Preenchimento é uma forma especial de mascaramento em que as etapas mascaradas estão no início ou no início de uma sequência. O preenchimento vem da necessidade de codificar os dados da 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 mais 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 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 as amostras individuais têm comprimento 3, 5 e 6, respectivamente. Uma vez que 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 do que o item mais longo precisam ser preenchidas com algum valor de espaço reservado (alternativamente, um também pode truncar amostras longas antes de preencher as amostras curtas).

Keras fornece uma função de utilitário para truncar e preencher listas Python com 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 está realmente preenchendo e deve ser ignorada. Esse mecanismo está mascarando .

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

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

Camadas geradoras de máscara: Embedding e Masking

Sob o capô, essas camadas criarão um tensor de máscara (tensor 2D com forma (batch, sequence_length) ) e o anexarão à saída do tensor retornada pela camada de 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)
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 no resultado impresso, a máscara é um tensor booleano 2D com forma (batch_size, sequence_length) , onde cada entrada individual False indica que o tempo correspondente deve ser ignorado durante o processamento.

Propagação de máscara na API Funcional e API Sequencial

Ao usar a API Funcional ou API Sequencial, uma máscara gerada por uma camada de Embedding ou Masking será propagada pela rede para qualquer camada que seja capaz de usá-las (por exemplo, camadas 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 a seguir, a camada LSTM receberá automaticamente uma máscara, o que significa que irá ignorar os valores preenchidos:

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

Esse também é o caso para o 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 as camadas

As camadas que podem manipular máscaras (como a camada LSTM ) têm um argumento de 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 de produção de máscara para o método __call__ de uma camada de consumo de 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([[ 0.00251055, -0.00290168, -0.00297247, ...,  0.00444937,
         0.0038675 , -0.00085511],
       [-0.00323313, -0.00033371,  0.0051134 , ...,  0.00064341,
        -0.00038007,  0.00942041],
       [ 0.00260642, -0.00067008,  0.00469904, ..., -0.00506479,
        -0.0024022 , -0.00024282],
       ...,
       [ 0.0014047 ,  0.00593606, -0.00268462, ...,  0.00080882,
        -0.00059643,  0.00060116],
       [ 0.01004083,  0.00241335,  0.00406337, ...,  0.00781425,
        -0.00170158,  0.00497941],
       [ 0.00763601,  0.00389311,  0.00035602, ..., -0.00125764,
        -0.00150181,  0.00356462]], dtype=float32)>

Suportando o mascaramento em suas camadas personalizadas

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

Por exemplo, qualquer camada que produza um tensor com uma dimensão de tempo diferente de sua entrada, como uma camada Concatenate que concatena na dimensão de tempo, precisará modificar a máscara atual para que as camadas posteriores sejam capazes de levar passos de tempo mascarados para dentro conta.

Para fazer isso, 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)
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 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)
tf.Tensor(
[[ True  True  True  True  True False  True  True  True  True]
 [False  True  True  True  True  True  True  True  True  True]
 [ True False False  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 do tempo, portanto, não é necessário modificar a máscara atual. No entanto, eles ainda podem querer propagar a máscara atual, inalterada, para a próxima camada. Este é um comportamento opcional. Por padrão, uma camada personalizada destruirá a máscara atual (uma vez que a estrutura não tem como dizer se a propagação da máscara é segura).

Se você tem uma camada personalizada que não modifica a dimensão do tempo e deseja que ela propague a máscara de entrada atual, deve definir self.supports_masking = True no construtor de camada. Nesse caso, o comportamento padrão de compute_mask() é apenas passar a máscara atual.

Aqui está um exemplo de 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 esta camada personalizada entre uma camada de geração de máscara (como Embedding ) e uma camada de consumo de máscara (como LSTM ), e ela passará a máscara de modo que alcance a camada de consumo de 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: Tensor("embedding_4/NotEqual:0", shape=(None, None), dtype=bool)

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

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

Para escrever tal camada, você pode simplesmente adicionar um argumento mask=None em sua assinatura de call . 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 do tempo (eixo 1) de uma sequência de entrada, enquanto descarta os passos 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 * 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 máscara no Keras. Para recapitular:

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