Sehen Sie sich Keynotes, Produktsitzungen, Workshops und mehr in Google I / O an. Siehe Wiedergabeliste

Maskieren und Auffüllen mit Keras

Ansicht auf TensorFlow.org In Google Colab ausführen Quelle auf GitHub anzeigen Notizbuch herunterladen

Einrichten

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

Einführung

Das Maskieren ist eine Möglichkeit, Sequenzverarbeitungsebenen mitzuteilen, dass bestimmte Zeitschritte in einer Eingabe fehlen, und sollte daher bei der Verarbeitung der Daten übersprungen werden.

Das Auffüllen ist eine spezielle Form der Maskierung, bei der sich die maskierten Schritte am Anfang oder am Ende einer Sequenz befinden. Das Auffüllen ergibt sich aus der Notwendigkeit, Sequenzdaten in zusammenhängende Stapel zu codieren: Damit alle Sequenzen in einem Stapel auf eine bestimmte Standardlänge passen, müssen einige Sequenzen aufgefüllt oder abgeschnitten werden.

Schauen wir uns das genauer an.

Auffüllsequenzdaten

Bei der Verarbeitung von Sequenzdaten kommt es häufig vor, dass einzelne Proben unterschiedliche Längen haben. Betrachten Sie das folgende Beispiel (Text als Wörter):

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

Nach der Suche nach Vokabeln können die Daten als Ganzzahlen vektorisiert werden, z.

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

Die Daten sind eine verschachtelte Liste, in der einzelne Stichproben die Länge 3, 5 bzw. 6 haben. Da die Eingabedaten für ein Deep-Learning-Modell ein einzelner Tensor sein müssen (in diesem Fall z. B. (batch_size, 6, vocab_size) ), müssen Stichproben, die kürzer als das längste Element sind, mit einem Platzhalterwert (alternativ einem) aufgefüllt werden kann auch lange Samples abschneiden, bevor kurze Samples aufgefüllt werden).

Keras bietet eine Dienstprogrammfunktion zum tf.keras.preprocessing.sequence.pad_sequences Python-Listen auf eine gemeinsame Länge: 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]]

Maskierung

Da alle Stichproben eine einheitliche Länge haben, muss das Modell darüber informiert werden, dass ein Teil der Daten tatsächlich aufgefüllt wird und ignoriert werden sollte. Dieser Mechanismus maskiert .

Es gibt drei Möglichkeiten, Eingabemasken in Keras-Modelle einzuführen:

  • Fügen Sie eine keras.layers.Masking .
  • Konfigurieren Sie eine keras.layers.Embedding mit mask_zero=True .
  • Übergeben Sie ein mask manuell, wenn Sie Layer aufrufen, die dieses Argument unterstützen (z. B. RNN-Layer).

Maskenerzeugende Ebenen: Embedding und Masking

Unter der Haube erstellen diese Ebenen einen Maskentensor (2D-Tensor mit Form (batch, sequence_length) ) und hängen ihn an die Tensorausgabe an, die von der Masking oder Embedding wird.

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)

Wie Sie dem gedruckten Ergebnis (batch_size, sequence_length) , handelt es sich bei der Maske um einen 2D-Booleschen Tensor mit Form (batch_size, sequence_length) , bei dem jeder einzelne False Eintrag angibt, dass der entsprechende Zeitschritt während der Verarbeitung ignoriert werden sollte.

Maskenausbreitung in der funktionalen API und der sequentiellen API

Bei Verwendung der funktionalen API oder der sequentiellen API wird eine von einer Embedding oder Masking erzeugte Maske für jede Schicht, die sie verwenden kann (z. B. RNN-Schichten), über das Netzwerk weitergegeben. Keras ruft automatisch die Maske ab, die einer Eingabe entspricht, und übergibt sie an jede Ebene, die weiß, wie man sie verwendet.

Im folgenden sequentiellen Modell erhält die LSTM Ebene beispielsweise automatisch eine Maske, was bedeutet, dass aufgefüllte Werte ignoriert werden:

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

Dies gilt auch für das folgende funktionale API-Modell:

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)

Maskentensoren direkt an Schichten übergeben

Schichten , die Masken (wie beispielsweise der Griff LSTM Schicht) eine mask Argument in ihrer __call__ Methode.

In der Zwischenzeit compute_mask(input, previous_mask) Ebenen, die eine Maske erzeugen (z. B. Embedding ), eine Methode compute_mask(input, previous_mask) die Sie aufrufen können.

Auf diese Weise können Sie die Ausgabe der Methode compute_mask() einer __call__ Ebene wie __call__ an die Methode __call__ einer __call__ Ebene übergeben:

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

Unterstützung der Maskierung in Ihren benutzerdefinierten Ebenen

Manchmal müssen Sie möglicherweise Ebenen schreiben, die eine Maske generieren (z. B. Embedding ), oder Ebenen, die die aktuelle Maske ändern müssen.

Beispielsweise muss jede Ebene, die einen Tensor mit einer anderen Zeitdimension als ihre Eingabe erzeugt, wie z. B. eine Concatenate , die die Zeitdimension verkettet, die aktuelle Maske ändern, damit nachgeschaltete Ebenen maskierte Zeitschritte ordnungsgemäß ausführen können Konto.

Zu diesem layer.compute_mask() sollte Ihre Ebene die Methode layer.compute_mask() implementieren, die unter layer.compute_mask() der Eingabe und der aktuellen Maske eine neue Maske erstellt.

Hier ist ein Beispiel für eine TemporalSplit Ebene, die die aktuelle Maske ändern muss.

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)

Hier ist ein weiteres Beispiel für eine CustomEmbedding Ebene, die aus Eingabewerten eine Maske generieren kann:

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)

Aktivieren Sie diese Option, um die Ausbreitung auf kompatiblen Ebenen zu maskieren

Die meisten Ebenen ändern die Zeitdimension nicht, sodass die aktuelle Maske nicht geändert werden muss. Sie können jedoch nach wie vor in der Lage sein wollen , die aktuelle Maske unverändert an die nächste Schicht zu propagieren. Dies ist ein Opt-In-Verhalten. Standardmäßig zerstört eine benutzerdefinierte Ebene die aktuelle Maske (da das Framework nicht erkennen kann, ob die Weitergabe der Maske sicher ist).

Wenn Sie eine benutzerdefinierte Ebene haben, die die Zeitdimension nicht ändert, und wenn Sie möchten, dass sie die aktuelle Eingabemaske weitergeben kann, sollten Sie im self.supports_masking = True . In diesem Fall besteht das Standardverhalten von compute_mask() darin, nur die aktuelle Maske zu übergeben.

Hier ist ein Beispiel für eine Ebene, die für die Maskenausbreitung in die Whitelist aufgenommen wurde:

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)

Sie können diese benutzerdefinierte Ebene jetzt zwischen einer maskenerzeugenden Ebene (wie Embedding ) und einer maskenverzehrenden Ebene (wie LSTM ) verwenden. Die Maske wird so weitergeleitet, dass sie die maskenverbrauchende Ebene erreicht.

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

Schreiben von Ebenen, die Maskeninformationen benötigen

Einige Schichten sind Maske Verbraucher: sie akzeptieren mask Argument call und es verwenden , um zu bestimmen , ob bestimmte Zeitschritte zu überspringen.

Um eine solche Ebene zu schreiben, können Sie einfach ein Argument mask=None in Ihre call einfügen. Die den Eingaben zugeordnete Maske wird an Ihre Ebene übergeben, sobald sie verfügbar ist.

Im Folgenden finden Sie ein einfaches Beispiel: Eine Ebene, die einen Softmax über die Zeitdimension (Achse 1) einer Eingabesequenz berechnet und dabei maskierte Zeitschritte verwirft.

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

Zusammenfassung

Das ist alles, was Sie über das Auffüllen und Maskieren in Keras wissen müssen. Um es noch einmal zusammenzufassen:

  • Mit "Maskierung" können Ebenen erkennen, wann bestimmte Zeitschritte in Sequenzeingaben übersprungen / ignoriert werden müssen.
  • Einige Ebenen sind Maskengeneratoren: Durch das Embedding kann eine Maske aus Eingabewerten generiert werden (wenn mask_zero=True ), ebenso wie die Masking .
  • Einige Schichten sind masken Verbraucher: sie entlarven mask Argument in ihrer __call__ Methode. Dies ist bei RNN-Schichten der Fall.
  • In der funktionalen API und der sequentiellen API werden Maskeninformationen automatisch weitergegeben.
  • Wenn Sie Ebenen eigenständig verwenden, können Sie die mask manuell an Ebenen übergeben.
  • Sie können problemlos Ebenen schreiben, die die aktuelle Maske ändern, eine neue Maske generieren oder die den Eingaben zugeordnete Maske verwenden.