Masking dan padding dengan Keras

Lihat di TensorFlow.org Jalankan di Google Colab Lihat sumber di GitHub Unduh buku catatan

Mendirikan

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

pengantar

Masking adalah cara untuk memberi tahu lapisan pemrosesan urutan bahwa langkah waktu tertentu dalam sebuah masukan hilang, dan karenanya harus dilewati saat memproses data.

Padding adalah bentuk khusus dari masking di mana langkah-langkah bertopeng berada di awal atau akhir urutan. Pengisi berasal dari kebutuhan untuk menyandikan data urutan ke dalam batch yang berdekatan: agar semua urutan dalam batch sesuai dengan panjang standar tertentu, beberapa urutan perlu di-pad atau dipotong.

Mari kita lihat lebih dekat.

Data urutan padding

Saat memproses data urutan, sangat umum jika sampel individu memiliki panjang yang berbeda. Pertimbangkan contoh berikut (teks yang diberi token sebagai kata-kata):

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

Setelah pencarian kosakata, data mungkin akan di-vektorisasi sebagai bilangan bulat, misalnya:

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

Data adalah daftar bersarang dimana sampel individu memiliki panjang 3, 5, dan 6, masing-masing. Karena data masukan untuk model pembelajaran dalam harus berupa tensor tunggal (dalam hal ini bentuk misalnya (batch_size, 6, vocab_size) ), sampel yang lebih pendek dari item terpanjang perlu diisi dengan beberapa nilai placeholder (sebagai alternatif, satu mungkin juga memotong sampel panjang sebelum mengisi sampel pendek).

Keras menyediakan fungsi utilitas untuk memotong dan memasukkan daftar Python ke panjang yang umum: 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]]

Masking

Sekarang semua sampel memiliki panjang yang seragam, model harus diberi tahu bahwa beberapa bagian dari data sebenarnya padding dan harus diabaikan. Mekanisme itu menutupi .

Ada tiga cara untuk memperkenalkan masker input di model Keras:

  • Tambahkan lapisan keras.layers.Masking .
  • Mengkonfigurasi keras.layers.Embedding layer dengan mask_zero=True .
  • Berikan argumen mask secara manual saat memanggil lapisan yang mendukung argumen ini (misalnya lapisan RNN).

Lapisan penghasil topeng: Embedding dan Masking

Di bawah tenda, lapisan ini akan membuat tensor topeng (tensor 2D dengan bentuk (batch, sequence_length) ), dan melampirkannya ke keluaran tensor yang dikembalikan oleh lapisan Masking atau 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)

Seperti yang Anda lihat dari hasil cetak, mask adalah tensor boolean 2D dengan bentuk (batch_size, sequence_length) , di mana setiap entri False menunjukkan bahwa langkah waktu yang sesuai harus diabaikan selama pemrosesan.

Propagasi mask di Functional API dan Sequential API

Saat menggunakan Functional API atau Sequential API, mask yang dihasilkan oleh lapisan Embedding atau Masking akan disebarkan melalui jaringan untuk setiap lapisan yang mampu menggunakannya (misalnya, lapisan RNN). Keras akan secara otomatis mengambil topeng yang sesuai dengan masukan dan meneruskannya ke lapisan mana pun yang tahu cara menggunakannya.

Misalnya, dalam model Sequential berikut, lapisan LSTM secara otomatis akan menerima mask, yang berarti ia akan mengabaikan nilai yang diberi bantalan:

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

Ini juga kasus untuk model API Fungsional berikut:

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)

Meneruskan tensor topeng langsung ke lapisan

Lapisan yang dapat menangani topeng (seperti lapisan LSTM ) memiliki argumen mask dalam metode __call__ mereka.

Sementara itu, lapisan yang menghasilkan topeng (mis. Embedding ) mengekspos metode compute_mask(input, previous_mask) yang dapat Anda panggil.

Dengan demikian, Anda bisa meneruskan output dari metode compute_mask() dari lapisan penghasil topeng ke metode __call__ dari lapisan yang memakan topeng, seperti ini:

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

Mendukung masking di lapisan khusus Anda

Terkadang, Anda mungkin perlu menulis lapisan yang menghasilkan topeng (seperti Embedding ), atau lapisan yang perlu mengubah topeng saat ini.

Misalnya, setiap lapisan yang menghasilkan tensor dengan dimensi waktu yang berbeda dari masukannya, seperti lapisan Concatenate yang digabungkan pada dimensi waktu, perlu memodifikasi topeng saat ini sehingga lapisan hilir akan dapat dengan benar mengambil langkah waktu bertopeng ke Akun.

Untuk melakukan ini, lapisan Anda harus mengimplementasikan metode layer.compute_mask() , yang menghasilkan topeng baru dengan masukan dan topeng saat ini.

Berikut adalah contoh lapisan TemporalSplit yang perlu memodifikasi topeng saat ini.

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)

Berikut adalah contoh lain dari lapisan CustomEmbedding yang mampu menghasilkan mask dari nilai 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)

Memilih untuk menutupi propagasi pada lapisan yang kompatibel

Sebagian besar lapisan tidak mengubah dimensi waktu, jadi tidak perlu mengubah topeng saat ini. Namun, mereka mungkin masih ingin menyebarkan topeng saat ini, tidak berubah, ke lapisan berikutnya. Ini adalah perilaku keikutsertaan. Secara default, lapisan khusus akan menghancurkan topeng saat ini (karena kerangka kerja tidak memiliki cara untuk mengetahui apakah menyebarkan topeng aman untuk dilakukan).

Jika Anda memiliki lapisan khusus yang tidak mengubah dimensi waktu, dan jika Anda ingin lapisan tersebut dapat menyebarkan masker masukan saat ini, Anda harus menyetel self.supports_masking = True di konstruktor lapisan. Dalam kasus ini, perilaku default compute_mask() adalah hanya melewatkan mask saat ini.

Berikut adalah contoh lapisan yang masuk daftar putih untuk penyebaran topeng:

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)

Anda sekarang dapat menggunakan lapisan khusus ini di antara lapisan penghasil topeng (seperti Embedding ) dan lapisan LSTM topeng (seperti LSTM ), dan itu akan meneruskan topeng sehingga mencapai lapisan yang memakan topeng.

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

Menulis lapisan yang membutuhkan informasi topeng

Beberapa lapisan adalah konsumen topeng: mereka menerima argumen mask dalam call dan menggunakannya untuk menentukan apakah akan melewati langkah waktu tertentu.

Untuk menulis lapisan seperti itu, Anda cukup menambahkan argumen mask=None di tanda tangan call Anda. Topeng yang terkait dengan masukan akan diteruskan ke lapisan Anda setiap kali tersedia.

Berikut adalah contoh sederhana di bawah ini: lapisan yang menghitung softmax selama dimensi waktu (sumbu 1) dari urutan masukan, sambil membuang langkah waktu bertopeng.

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

Ringkasan

Itu saja yang perlu Anda ketahui tentang padding & masking di Keras. Ringkasan:

  • "Masking" adalah bagaimana lapisan dapat mengetahui kapan harus melewati / mengabaikan langkah waktu tertentu dalam masukan urutan.
  • Beberapa lapisan adalah generator-topeng: Embedding dapat menghasilkan topeng dari nilai-nilai input (jika mask_zero=True ), dan begitu juga lapisan Masking .
  • Beberapa lapisan adalah konsumen-topeng: mereka mengekspos argumen mask dalam metode __call__ mereka. Ini adalah kasus untuk lapisan RNN.
  • Dalam API Fungsional dan API Sekuensial, informasi topeng disebarkan secara otomatis.
  • Saat menggunakan lapisan secara mandiri, Anda dapat meneruskan argumen mask ke lapisan secara manual.
  • Anda dapat dengan mudah menulis lapisan yang mengubah topeng saat ini, yang menghasilkan topeng baru, atau yang menggunakan topeng yang terkait dengan masukan.