Mascheratura e imbottitura con Keras

Visualizza su TensorFlow.org Esegui in Google Colab Visualizza sorgente su GitHub Scarica taccuino

Impostare

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

introduzione

Il mascheramento è un modo per indicare ai livelli di elaborazione della sequenza che mancano alcuni timestep in un input e quindi devono essere ignorati durante l'elaborazione dei dati.

Il riempimento è una forma speciale di mascheramento in cui i passaggi mascherati si trovano all'inizio o alla fine di una sequenza. Il riempimento nasce dalla necessità di codificare i dati della sequenza in batch contigui: per fare in modo che tutte le sequenze di un batch si adattino a una data lunghezza standard, è necessario riempire o troncare alcune sequenze.

Diamo uno sguardo più da vicino.

Dati della sequenza di riempimento

Durante l'elaborazione dei dati di sequenza, è molto comune che i singoli campioni abbiano lunghezze diverse. Considera il seguente esempio (testo tokenizzato come parole):

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

Dopo la ricerca nel vocabolario, i dati potrebbero essere vettorizzati come numeri interi, ad esempio:

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

I dati sono un elenco annidato in cui i singoli campioni hanno rispettivamente lunghezza 3, 5 e 6. Poiché i dati di input per un modello di apprendimento profondo devono essere un singolo tensore (di forma ad es. (batch_size, 6, vocab_size) in questo caso), i campioni più corti dell'elemento più lungo devono essere riempiti con un valore segnaposto (in alternativa, uno potrebbe anche troncare campioni lunghi prima di riempire campioni brevi).

Keras fornisce una funzione di utilità per troncare e tf.keras.preprocessing.sequence.pad_sequences elenchi Python a una lunghezza comune: 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]]

Mascheramento

Ora che tutti i campioni hanno una lunghezza uniforme, il modello deve essere informato che una parte dei dati è effettivamente riempita e deve essere ignorata. Quel meccanismo sta mascherando .

Esistono tre modi per introdurre maschere di input nei modelli Keras:

  • Aggiungi un livello keras.layers.Masking .
  • Configurare un keras.layers.Embedding livello con mask_zero=True .
  • Passa manualmente un argomento mask quando chiami layer che supportano questo argomento (es. Layer RNN).

Livelli di generazione di maschere: Embedding e Masking

Sotto il cofano, questi livelli creeranno un tensore maschera (tensore 2D con forma (batch, sequence_length) ) e lo collegheranno all'output del tensore restituito dal livello 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)
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)

Come puoi vedere dal risultato stampato, la maschera è un tensore booleano 2D con forma (batch_size, sequence_length) , dove ogni singola voce False indica che il (batch_size, sequence_length) corrispondente deve essere ignorato durante l'elaborazione.

Propagazione della maschera nell'API funzionale e nell'API sequenziale

Quando si utilizza l'API funzionale o l'API sequenziale, una maschera generata da un livello di Embedding o Masking verrà propagata attraverso la rete per qualsiasi livello in grado di utilizzarli (ad esempio, i livelli RNN). Keras recupererà automaticamente la maschera corrispondente a un input e la passerà a qualsiasi livello che sappia come usarla.

Ad esempio, nel seguente modello sequenziale, il livello LSTM riceverà automaticamente una maschera, il che significa che ignorerà i valori riempiti:

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

Questo è anche il caso del seguente modello di API funzionale:

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)

Passaggio dei tensori della maschera direttamente ai livelli

I livelli che possono gestire le maschere (come il livello LSTM ) hanno un argomento mask nel loro metodo __call__ .

Nel frattempo, i livelli che producono una maschera (ad esempio Embedding ) espongono un compute_mask(input, previous_mask) che puoi chiamare.

Quindi, puoi passare l'output del metodo compute_mask() di un livello che produce maschera al metodo __call__ di un livello che consuma maschera, in questo modo:

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.0071368 ,  0.00202324,  0.00393163, ..., -0.00365972,
        -0.00194294, -0.00275828],
       [ 0.00865301, -0.00411554, -0.00328279, ...,  0.00395685,
         0.01023738, -0.0013066 ],
       [ 0.0115475 , -0.00367757, -0.0049072 , ...,  0.00312295,
         0.00557074,  0.00681297],
       ...,
       [ 0.00537544, -0.00517081,  0.00668133, ...,  0.00428408,
         0.00251086, -0.00211114],
       [ 0.00286667, -0.00301991, -0.0095289 , ...,  0.00381294,
         0.00675705, -0.00599195],
       [-0.0045211 ,  0.0019338 , -0.00031986, ...,  0.00275819,
        -0.00126366, -0.00347176]], dtype=float32)>

Supporto del mascheramento nei livelli personalizzati

A volte, potrebbe essere necessario scrivere livelli che generano una maschera (come Embedding ) o livelli che devono modificare la maschera corrente.

Ad esempio, qualsiasi livello che produce un tensore con una dimensione temporale diversa dal suo input, come un livello Concatenate che si concatena sulla dimensione temporale, dovrà modificare la maschera corrente in modo che i livelli a valle siano in grado di prendere correttamente i timestep account.

Per fare ciò, il tuo livello dovrebbe implementare il metodo layer.compute_mask() , che produce una nuova maschera dato l'input e la maschera corrente.

Ecco un esempio di un livello TemporalSplit che deve modificare la maschera corrente.

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)

Ecco un altro esempio di un livello CustomEmbedding grado di generare una maschera dai valori di input:

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 False  True]
 [ True  True  True  True  True  True  True  True  True  True]
 [ True  True  True False  True  True  True  True  True  True]], shape=(3, 10), dtype=bool)

Attivazione per mascherare la propagazione su livelli compatibili

La maggior parte dei livelli non modifica la dimensione temporale, quindi non è necessario modificare la maschera corrente. Tuttavia, potrebbero comunque voler essere in grado di propagare la maschera corrente, invariata, al livello successivo. Questo è un comportamento di partecipazione. Per impostazione predefinita, un livello personalizzato distruggerà la maschera corrente (poiché il framework non ha modo di dire se la propagazione della maschera è sicura).

Se si dispone di un livello personalizzato che non modifica la dimensione temporale e si desidera che sia in grado di propagare la maschera di input corrente, è necessario impostare self.supports_masking = True nel costruttore del livello. In questo caso, il comportamento predefinito di compute_mask() è semplicemente passare la maschera corrente.

Ecco un esempio di un livello che è autorizzato per la propagazione della maschera:

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)

È ora possibile utilizzare questo livello personalizzato tra un livello di generazione di maschere (come Embedding ) e uno che consuma maschera (come LSTM ) e farà passare la maschera in modo che raggiunga il livello che consuma la maschera.

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

Scrittura di livelli che richiedono informazioni sulla maschera

Alcuni livelli sono consumatori maschera: accettano un argomento mask nella call e lo utilizzano per determinare se saltare determinati passaggi temporali.

Per scrivere un tale livello, puoi semplicemente aggiungere un argomento mask=None nella firma della call . La maschera associata agli input verrà passata al tuo livello ogni volta che è disponibile.

Di seguito è riportato un semplice esempio: un livello che calcola un softmax sulla dimensione temporale (asse 1) di una sequenza di input, scartando le fasi temporali mascherate.

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

Sommario

Questo è tutto ciò che devi sapere su imbottitura e mascheratura in Keras. Ricapitolando:

  • Il "mascheramento" è il modo in cui i livelli sono in grado di sapere quando saltare / ignorare determinati timestep negli input di sequenza.
  • Alcuni livelli sono generatori di maschere: l' Embedding può generare una maschera dai valori di input (se mask_zero=True ), così come il livello di Masking .
  • Alcuni livelli sono consumatori di maschere: espongono un argomento mask nel loro metodo __call__ . Questo è il caso dei livelli RNN.
  • Nell'API funzionale e nell'API sequenziale, le informazioni sulla maschera vengono propagate automaticamente.
  • Quando si utilizzano i livelli in modo autonomo, è possibile passare manualmente gli argomenti della mask ai livelli.
  • È possibile scrivere facilmente livelli che modificano la maschera corrente, che generano una nuova maschera o che consumano la maschera associata agli input.