Maskieren und Aufpolstern mit Keras

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

Aufstellen

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

Einführung

Maskierung ist eine Art und Weise sequenzVerarbeitungsSchichten zu sagen , dass bestimmte Zeitschritte in einem Eingang fehlt, und somit übersprungen werden sollte , wenn die Verarbeitung der Daten.

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

Schauen wir genau hin.

Padding-Sequenzdaten

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

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

Nach der Vokabularsuche können die Daten als ganze Zahlen vektorisiert werden, z. B.:

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

Die Daten sind eine verschachtelte Liste, in der einzelne Samples die Länge 3, 5 bzw. 6 haben. Da die Eingangsdaten für ein tiefes Lernmodell ein einzelner Tensor (von Form zB sein muss (batch_size, 6, vocab_size) in diesem Fall), Proben , die kürzer als die längste Element Notwendigkeit sind mit einigen Platzhalter Wert aufgefüllt werden (alternativ ein kann auch lange Samples abschneiden, bevor kurze Samples aufgefüllt werden).

Keras stellt eine Nutzenfunktion trunkieren und Pad 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 nun alle Samples eine einheitliche Länge haben, muss dem Modell mitgeteilt werden, dass ein Teil der Daten tatsächlich aufgefüllt wird und ignoriert werden sollte. Dieser Mechanismus wird maskiert.

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

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

Maskenerzeugenden Schichten: Embedding und Masking

Unter der Haube wird schaffen diese Schichten eine Maske Tensor (2D - Tensor mit Form (batch, sequence_length) ), und befestigen Sie es mit dem Ausgang Tensor durch die zurück Masking oder Embedding Schicht.

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 aus dem gedruckten Ergebnis sehen können, ist der Maske ein 2D boolean Tensor mit Form (batch_size, sequence_length) , wobei jeder einzelner False Eintrag gibt an, dass der entsprechenden Zeitschritt sollte bei der Verarbeitung ignoriert werden.

Maskenpropagation in der Functional API und Sequential API

Wenn die funktionelle API oder Sequential - API, eine Maske von einer erzeugten Embedding oder Masking Schicht für jede Schicht durch das Netzwerk weitergeleitet werden , die sie zur Verwendung geeignet ist (beispielsweise RNN - Schichten). Keras ruft automatisch die einer Eingabe entsprechende Maske ab und übergibt sie an jede Ebene, die weiß, wie sie verwendet wird.

Zum Beispiel in dem folgende Sequential - Modell, die LSTM wird Schicht erhalten automatisch eine Maske, die Mittel wird es gepolsterte Werte ignorieren:

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

Dies gilt auch für das folgende Functional 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 Ebenen übergeben

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

Inzwischen Schichten , die eine Maske (zB produzieren Embedding ) aussetzen einen compute_mask(input, previous_mask) Methode , die Sie aufrufen können.

So können Sie die Ausgabe des Passes compute_mask() Methode einer Maske erzeugenden Schicht auf die __call__ Verfahren einer Maske raubend Schicht, wie folgt aus :

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

Unterstützung der Maskierung in Ihren benutzerdefinierten Ebenen

Manchmal müssen Sie möglicherweise Schreibschichten , die eine Maske (wie erzeugen Embedding ) oder Schichten, die die aktuelle Maske ändern müssen.

Zum Beispiel kann jede Schicht , die einen Tensor mit einer anderen Zeitdimension als seine Eingabe erzeugt, wie beispielsweise eine Concatenate Schicht , die konkateniert auf der Zeitdimension, muß die aktuelle Maske modifizieren , so dass stromabwärts Schichten in der Lage , richtig zu nehmen maskierte Zeitschritte in Konto.

Um dies zu tun, sollten Sie Ihre Schicht , die die Umsetzung layer.compute_mask() Methode, die eine neue Maske erzeugt die Eingabe und die aktuelle Maske gegeben.

Hier ist ein Beispiel für eine TemporalSplit Schicht, 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 einer CustomEmbedding Schicht , die eine Maske zur Erzeugung von aus Eingabewerten geeignet ist:

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)

Aktivieren, um die Ausbreitung auf kompatiblen Ebenen zu maskieren

Die meisten Ebenen ändern die Zeitdimension nicht, müssen also die aktuelle Maske nicht ändern. 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 keine Möglichkeit hat, festzustellen, ob die Weitergabe der Maske sicher ist).

Wenn Sie eine benutzerdefinierte Schicht , die nicht die Zeitdimension nicht verändert, und wenn Sie es wollen in der Lage , die aktuelle Eingabemaske zu propagieren, sollten Sie setzen self.supports_masking = True in der Schicht Konstruktor. In diesem Fall ist das Standardverhalten von compute_mask() ist nur durch die aktuelle Maske passieren.

Hier ist ein Beispiel für einen Layer, der für die Maskenweitergabe auf die Whitelist gesetzt 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 nun diese individuelle Schicht in-zwischen einer Maske erzeugenden Schicht (wie verwenden Embedding ) und einer Maske raubend Schicht (wie LSTM ), und es wird entlang so die Maske passiert , dass sie die Maske raubend Schicht 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.

So schreiben Sie eine solche Schicht, können Sie einfach eine hinzufügen mask=None Argument in Ihrem call Signatur. Die mit den Eingaben verknüpfte Maske wird an Ihren Layer übergeben, wenn sie verfügbar ist.

Hier ist ein einfaches Beispiel unten: ein Layer, der einen Softmax über die Zeitdimension (Achse 1) einer Eingabesequenz berechnet, während maskierte Zeitschritte verworfen werden.

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 Padding & Masking in Keras wissen müssen. Um es zusammenzufassen:

  • "Maskieren" ist, wie Layer wissen, wann bestimmte Zeitschritte in Sequenzeingaben übersprungen / ignoriert werden sollen.
  • Einige Schichten sind masken Generatoren: Embedding eine Maske aus Eingabewerten erzeugen kann (wenn mask_zero=True ), und so kann die Masking Schicht.
  • 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 Schichten in einer eigenständigen Art und Weise verwenden, können Sie den Pass mask Argumente Schichten manuell.
  • Sie können ganz einfach Ebenen schreiben, die die aktuelle Maske ändern, eine neue Maske generieren oder die den Eingaben zugeordnete Maske verbrauchen.