 Voir sur TensorFlow.org |  Exécuter dans Google Colab |  Voir la source sur GitHub |  Télécharger le cahier |
Installer
import numpy as np
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
introduction
Le masquage est un moyen de dire des couches de traitement de séquence que certains pas de temps à une entrée sont manquantes, et devraient donc être ignorés lors du traitement des données.
Padding est une forme particulière de masquage dans lequel les étapes sont masquées au début ou à la fin d'une séquence. Le remplissage vient de la nécessité d'encoder les données de séquence en lots contigus : afin que toutes les séquences d'un lot correspondent à une longueur standard donnée, il est nécessaire de remplir ou de tronquer certaines séquences.
Regardons de près.
Remplissage des données de séquence
Lors du traitement des données de séquence, il est très courant que des échantillons individuels aient des longueurs différentes. Prenons l'exemple suivant (texte symbolisé sous forme de mots) :
[
["Hello", "world", "!"],
["How", "are", "you", "doing", "today"],
["The", "weather", "will", "be", "nice", "tomorrow"],
]
Après la recherche de vocabulaire, les données peuvent être vectorisées sous forme d'entiers, par exemple :
[
[71, 1331, 4231]
[73, 8, 3215, 55, 927],
[83, 91, 1, 645, 1253, 927],
]
Les données sont une liste imbriquée dans laquelle les échantillons individuels ont respectivement une longueur de 3, 5 et 6. Étant donné que les données d'entrée pour un modèle d'apprentissage en profondeur doit être un seul tenseur (de forme , par exemple (batch_size, 6, vocab_size) dans ce cas), les échantillons qui sont plus courtes que le besoin le plus long de l' élément à rembourrées avec une valeur d'espace réservé (alternativement, un peut également tronquer les échantillons longs avant de remplir les échantillons courts).
Keras fournit une fonction d'utilité pour tronquer et pad listes python pour une longueur commune: 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]]
Masquage
Maintenant que tous les échantillons ont une longueur uniforme, le modèle doit être informé qu'une partie des données est en fait du remplissage et doit être ignorée. Ce mécanisme est de masquage.
Il existe trois manières d'introduire des masques de saisie dans les modèles Keras :
- Ajouter une
keras.layers.Maskingcouche. - Configurer une
keras.layers.Embeddingcouche avecmask_zero=True. - Passer un
maskarguments manuellement lors de l' appel des couches prenant en charge cet argument (par exemple les couches RNN).
Les couches génératrices de masque: Embedding et Masking
Sous le capot, ces couches vont créer un tenseur de masque (tenseur 2D avec forme (batch, sequence_length) ), et le fixer à la sortie du tenseur renvoyé par le Masking ou Embedding couche.
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)
Comme vous pouvez le voir sur le résultat imprimé, le masque est un tenseur booléen 2D avec la forme (batch_size, sequence_length) , où chaque individu False entrée indique que le correspondant timestep doit être ignorée au cours du traitement.
Propagation de masque dans l'API fonctionnelle et l'API séquentielle
Lors de l' utilisation de l'API fonctionnelle ou de l'API séquentielle, un masque généré par une Embedding ou Masking couche sera propagé à travers le réseau pour chaque couche qui est capable de les utiliser (par exemple, des couches RNN). Keras récupérera automatiquement le masque correspondant à une entrée et le transmettra à n'importe quel calque qui saura l'utiliser.
Par exemple, dans le modèle séquentiel suivant, la LSTM couche reçoit automatiquement un masque, ce qui signifie qu'il ne tient pas compte des valeurs rembourrées:
model = keras.Sequential(
[layers.Embedding(input_dim=5000, output_dim=16, mask_zero=True), layers.LSTM(32),]
)
C'est également le cas pour le modèle d'API fonctionnelle suivant :
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)
Passer des tenseurs de masque directement aux calques
Les couches qui peuvent manipuler des masques (comme la LSTM couche) ont un mask argument en leur __call__ procédé.
Pendant ce temps, les couches qui produisent un masque (par exemple Embedding ) exposer une compute_mask(input, previous_mask) méthode que vous pouvez appeler.
Ainsi, vous pouvez passer la sortie du compute_mask() méthode d'une couche productrice de masque à la __call__ méthode d'une couche, comme la consommation de masque ceci:
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)>
Prise en charge du masquage dans vos calques personnalisés
Parfois, vous devrez peut - être des couches d'écriture qui génèrent un masque (comme Embedding ), ou des couches qui doivent modifier le masque en cours.
Par exemple, une couche qui produit un tenseur avec une dimension temporelle différente de celle de son entrée, comme une Concatenate couche qui concatène sur la dimension temporelle, devront modifier le masque actuel afin que les couches en aval seront en mesure de prendre correctement pas de temps masqué dans Compte.
Pour ce faire, votre couche doit mettre en œuvre la layer.compute_mask() méthode, qui produit un nouveau masque donné l'entrée et le masque en cours.
Voici un exemple d'une TemporalSplit couche qui doit modifier le masque en cours.
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)
Voici un autre exemple d'un CustomEmbedding couche qui est capable de générer un masque à partir des valeurs d'entrée:
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)
Activer la propagation du masque sur les calques compatibles
La plupart des calques ne modifient pas la dimension temporelle, vous n'avez donc pas besoin de modifier le masque actuel. Cependant, ils peuvent vouloir toujours être en mesure de se propager le masque actuel, sans changement, à la couche suivante. Il s'agit d'un comportement opt-in. Par défaut, un calque personnalisé détruira le masque actuel (puisque le framework n'a aucun moyen de dire si la propagation du masque est sûre).
Si vous avez une couche personnalisée qui ne modifie pas la dimension temporelle, et si vous voulez qu'il soit capable de se propager le masque d'entrée en cours, vous devez définir self.supports_masking = True dans le constructeur de la couche. Dans ce cas, le comportement par défaut de compute_mask() est de passer juste le masque courant à travers.
Voici un exemple de calque mis en liste blanche pour la propagation de masque :
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)
Vous pouvez maintenant utiliser cette couche personnalisée en entre une couche de génération de masque (comme Embedding ) et une couche consommant masque (comme LSTM ), et il passera le masque le long de sorte qu'il atteigne la couche consommant masque.
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')
Écriture de calques nécessitant des informations de masque
Certaines couches sont des consommateurs de masque: ils acceptent un mask argument en call et l' utiliser pour déterminer si pour sauter certaines étapes de temps.
Pour écrire une telle couche, vous pouvez simplement ajouter un mask=None argument en votre call signature. Le masque associé aux entrées sera transmis à votre calque dès qu'il sera disponible.
Voici un exemple simple ci-dessous : une couche qui calcule un softmax sur la dimension temporelle (axe 1) d'une séquence d'entrée, tout en supprimant les pas de temps masqués.
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)))
Résumé
C'est tout ce que vous devez savoir sur le rembourrage et le masquage à Keras. Récapituler:
- Le « masquage » est la façon dont les couches sont capables de savoir quand sauter / ignorer certains pas de temps dans les entrées de séquence.
- Certaines couches sont masque générateurs:
Embeddingpeut générer un masque à partir des valeurs d'entrée (simask_zero=True), et peut donc leMaskingcouche. - Certaines couches sont masque-consommateurs: ils exposent un
maskarguments dans leur__call__méthode. C'est le cas des couches RNN. - Dans l'API fonctionnelle et l'API séquentielle, les informations de masque sont propagées automatiquement.
- Lors de l' utilisation des couches d'une manière autonome, vous pouvez passer les
maskarguments aux couches manuellement. - Vous pouvez facilement écrire des calques qui modifient le masque actuel, qui génèrent un nouveau masque ou qui consomment le masque associé aux entrées.