 Visualizza su TensorFlow.org |  Esegui in Google Colab |  Visualizza l'origine su GitHub |  Scarica quaderno |
Impostare
import numpy as np
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
introduzione
Masking è un modo per dire strati sequenza di elaborazione che certi Timesteps in un ingresso mancano, e quindi dovrebbe essere ignorato durante l'elaborazione dei dati.
Imbottitura è una forma speciale di mascheratura in cui i passaggi mascherati sono all'inizio o alla fine di una sequenza. Il padding nasce dalla necessità di codificare i dati di sequenza in batch contigui: per far sì che tutte le sequenze di un batch rientrino in una data lunghezza standard, è necessario riempire o troncare alcune sequenze.
Diamo un'occhiata da vicino.
Dati della sequenza di riempimento
Quando si elaborano i dati della 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 del 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 nidificato in cui i singoli campioni hanno rispettivamente lunghezza 3, 5 e 6. Poiché i dati di input per un modello di apprendimento profondo deve essere un unico tensore (di forma ad esempio (batch_size, 6, vocab_size) in questo caso), campioni che sono più brevi le più lunghe necessità l'elemento ad essere imbottito con un valore segnaposto (in alternativa, uno potrebbe anche troncare i campioni lunghi prima di riempire i campioni corti).
Keras fornisce una funzione di utilità per troncare e pad liste Python ad 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]]
Mascheratura
Ora che tutti i campioni hanno una lunghezza uniforme, il modello deve essere informato che una parte dei dati sta effettivamente riempiendo e deve essere ignorata. Tale meccanismo è mascheramento.
Esistono tre modi per introdurre le maschere di input nei modelli Keras:
- Aggiungi un
keras.layers.Maskinglayer. - Configurare un
keras.layers.Embeddinglivello conmask_zero=True. - Passare una
maskargomento manualmente le chiamate a livelli che supportano questo argomento (es strati RNN).
Strati maschera generano: Embedding e Masking
Sotto il cofano, questi strati creerà un tensore maschera (tensore 2D di forma (batch, sequence_length) ), e fissarlo all'uscita tensore restituito dal Masking o Embedding strato.
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 si può vedere dal risultato stampato, la maschera è un tensore booleana 2D di forma (batch_size, sequence_length) , in cui ogni individuo False voce indica che il passo temporale corrispondente dovrebbe essere ignorato durante la lavorazione.
Propagazione della maschera nell'API funzionale e nell'API sequenziale
Quando si utilizza l'API funzionale o l'API sequenziale, una maschera generato da un Embedding o Masking strato sarà propagato attraverso la rete per qualsiasi livello che è in grado di utilizzarle (per esempio, strati 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 LSTM strato riceverà automaticamente una maschera, i quali mezzi ignorerà valori imbottite:
model = keras.Sequential(
[layers.Embedding(input_dim=5000, output_dim=16, mask_zero=True), layers.LSTM(32),]
)
Questo vale anche per il 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)
Passando i tensori della maschera direttamente ai livelli
Strati in grado di gestire le maschere (come il LSTM strato) hanno una mask argomento a loro __call__ metodo.
Nel frattempo, strati che producono una maschera (es Embedding ) espongono un compute_mask(input, previous_mask) metodo che si può chiamare.
Così, è possibile passare l'output del compute_mask() metodo di un livello maschera produttrice al __call__ metodo di uno strato maschera di tempo, 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([[-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)>
Supporto della mascheratura nei livelli personalizzati
A volte, potrebbe essere necessario strati di scrittura che generano una maschera (come Embedding ), o strati che hanno bisogno di modificare la maschera corrente.
Per esempio, ogni strato che produce un tensore con una dimensione temporale diverso rispetto al suo ingresso, ad esempio un Concatenate strato che concatena sulla dimensione tempo, sarà necessario modificare la maschera corrente in modo che gli strati a valle potranno prendere correttamente Timesteps mascherati in account.
Per fare questo, il livello dovrebbe implementare il layer.compute_mask() metodo, che produce una nuova maschera dato l'ingresso e la maschera corrente.
Ecco un esempio di un TemporalSplit strato 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 CustomEmbedding livello che è capace di generare una maschera da valori di ingresso:
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)
Attivazione per mascherare la propagazione sui livelli compatibili
La maggior parte dei livelli non modifica la dimensione temporale, quindi non è necessario modificare la maschera corrente. Tuttavia, essi possono ancora voglia di essere in grado di propagare la maschera corrente, invariato, 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 stabilire se la propagazione della maschera è sicura).
Se si dispone di un layer personalizzato che non modifica la dimensione temporale, e se si vuole che sia in grado di propagare la maschera di input corrente, è necessario impostare self.supports_masking = True nel costruttore layer. In questo caso, il comportamento predefinito di compute_mask() è passare solo la maschera corrente attraverso.
Ecco un esempio di un livello che è nella whitelist 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 strato personalizzato in-tra uno strato di maschera generatrici (come Embedding ) ed uno strato maschera dispendiosa (come LSTM ), e passerà la maschera lungo in modo che raggiunga il livello maschera consumano.
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 necessitano di informazioni sulla maschera
Alcuni strati sono consumatori maschera: accettano un mask argomento nella call e lo usano per determinare se saltare alcuni passaggi di tempo.
Per scrivere un tale strato, si può semplicemente aggiungere una mask=None argomento a vostra call firma. La maschera associata agli input verrà passata al tuo livello ogni volta che sarà disponibile.
Ecco un semplice esempio di seguito: un livello che calcola un softmax sulla dimensione temporale (asse 1) di una sequenza di input, scartando i tempi mascherati.
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)))
Riepilogo
Questo è tutto ciò che devi sapere sull'imbottitura e sulla mascheratura in Keras. Per ricapitolare:
- Il "mascheramento" è il modo in cui i livelli sono in grado di sapere quando saltare/ignorare determinati passaggi temporali negli input di sequenza.
- Alcuni strati sono maschere-generatori:
Embeddingpuò generare una maschera da valori di ingresso (semask_zero=True), e così pure ilMaskingstrato. - Alcuni livelli sono maschera-consumatori: espongono una
maskargomento a loro__call__metodo. Questo è il caso dei livelli RNN. - Nell'API funzionale e nell'API sequenziale, le informazioni sulla maschera vengono propagate automaticamente.
- Quando si utilizza strati in modo indipendente, è possibile passare i
maskargomenti strati manualmente. - Puoi facilmente scrivere livelli che modificano la maschera corrente, che generano una nuova maschera o che consumano la maschera associata agli input.