Keras로 마스킹 및 패딩하기

TensorFlow.org에서 보기 Google Colab에서 실행하기 GitHub에서소스 보기 노트북 다운로드하기

설치하기

import numpy as np
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
2022-12-14 22:30:33.575697: W tensorflow/compiler/xla/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libnvinfer.so.7'; dlerror: libnvinfer.so.7: cannot open shared object file: No such file or directory
2022-12-14 22:30:33.575792: W tensorflow/compiler/xla/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libnvinfer_plugin.so.7'; dlerror: libnvinfer_plugin.so.7: cannot open shared object file: No such file or directory
2022-12-14 22:30:33.575802: W tensorflow/compiler/tf2tensorrt/utils/py_utils.cc:38] TF-TRT Warning: Cannot dlopen some TensorRT libraries. If you would like to use Nvidia GPU with TensorRT, please make sure the missing libraries mentioned above are installed properly.

소개

마스킹 은 시퀀스 처리 레이어에 입력의 특정 시간 단계가 누락되어 데이터를 처리 할 때 건너 뛰도록 지시하는 방법입니다.

패딩은 마스킹된 스텝이 시퀀스의 시작 또는 끝에 위치하는 특수한 형태의 마스킹입니다. 패딩이 필요한 이유는 시퀀스 데이터를 연속 배치로 인코딩해야 하는 데 있습니다. 배치의 모든 시퀀스를 지정된 표준 길이에 맞추려면 일부 시퀀스를 패딩 처리하거나 잘라내야 합니다.

자세히 살펴보겠습니다.

패딩 시퀀스 데이터

시퀀스 데이터를 처리 할 때 개별 샘플의 길이가 다른 것이 매우 일반적입니다. 다음 예제를 고려하십시오 (텍스트로 단어로 토큰 화됨).

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

어휘 조회 후 데이터는 다음과 같이 정수로 벡터화 될 수 있습니다.

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

데이터는 개별 샘플의 길이가 각각 3, 5 및 6인 중첩된 목록입니다. 딥 러닝 모델의 입력 데이터는 단일 텐서(이 경우, 예를 들어 (batch_size, 6, vocab_size) 형상의 텐서)여야 하므로 가장 긴 항목보다 짧은 샘플은 일부 자리 표시자 값으로 패딩 처리해야 합니다(또는, 짧은 샘플을 패딩 처리하기 전에 긴 샘플을 잘라낼 수도 있음).

Keras는 Python 목록을 잘라서 공통 길이로 패딩 처리하는 유틸리티 기능을 제공합니다: 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]]

마스킹

이제 모든 샘플의 길이가 균일하므로 데이터의 일부가 실제로 채워져 있다는 사실을 모델에 알려야합니다. 그 메커니즘은 마스킹 입니다.

Keras 모델에서 입력 마스크를 도입하는 세 가지 방법이 있습니다.

  • keras.layers.Masking 레이어를 추가하십시오.
  • mask_zero=Truekeras.layers.Embedding 레이어를 구성하십시오.
  • 이 인수를 지원하는 계층 (예 : RNN 계층)을 호출 할 때 mask 인수를 수동으로 전달하십시오.

마스크 생성 레이어 : EmbeddingMasking

후드 아래에서이 레이어는 마스크 텐서 (모양 (batch, sequence_length) 가진 2D 텐서)를 만들어 Masking 또는 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)

인쇄 된 결과에서 볼 수 있듯이 마스크는 모양이 (batch_size, sequence_length) 인 2D 부울 텐서이며, 각 개별 False 항목은 처리 중에 해당 시간 단계를 무시해야 함을 나타냅니다.

함수형 API 및 순차형 API의 마스크 전파

함수형 API 또는 순차형 API를 사용하는 경우 Embedding 또는 Masking 레이어에서 생성된 마스크는 이를 사용할 수 있는 모든 레이어(예: RNN 레이어)를 대상으로 네트워크를 통해 전파됩니다. Keras는 입력에 해당하는 마스크를 자동으로 가져와서 그 사용 방법을 알고 있는 모든 레이어로 전달합니다.

For instance, in the following Sequential model, the LSTM layer will automatically receive a mask, which means it will ignore padded values:

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

다음과 같은 함수형 API 모델의 경우에도 마찬가지입니다.

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)

마스크 텐서를 레이어로 직접 전달

마스크를 처리 할 수있는 레이어 (예 : LSTM 레이어)는 __call__ 메서드에 mask 인수가 있습니다.

한편 마스크 (예 : Embedding )를 생성하는 레이어는 호출 할 수있는 compute_mask(input, previous_mask) 메소드를 노출합니다.

따라서 마스크 생성 레이어의 compute_mask() 메서드 출력을 다음과 같이 마스크 소비 레이어의 __call__ 메서드로 전달할 수 있습니다.

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([[-2.4582020e-03,  4.7728089e-03,  5.7384128e-05, ...,
         4.0099793e-03,  5.1399190e-03, -1.6484645e-03],
       [-2.8507239e-03,  4.7343396e-04, -6.7692003e-03, ...,
         4.8456267e-03,  5.1636202e-03, -7.9966765e-03],
       [-1.0827024e-02,  5.4355534e-03,  8.1839506e-03, ...,
        -7.6012569e-03,  7.3094978e-03, -1.0679217e-02],
       ...,
       [ 7.9554797e-04,  2.1807807e-03,  5.9871853e-04, ...,
        -4.9506580e-03, -6.6169836e-03,  9.8764733e-04],
       [-2.5283429e-03,  4.3892669e-03, -4.1671372e-03, ...,
         8.9579745e-04,  2.5786126e-03,  1.5738398e-03],
       [ 4.1612322e-03, -3.5828012e-03,  3.0661416e-03, ...,
         5.2880589e-03,  3.1845085e-03, -3.1711904e-03]], dtype=float32)>

사용자 정의 레이어에서 마스킹 지원

Sometimes, you may need to write layers that generate a mask (like Embedding), or layers that need to modify the current mask.

For instance, any layer that produces a tensor with a different time dimension than its input, such as a Concatenate layer that concatenates on the time dimension, will need to modify the current mask so that downstream layers will be able to properly take masked timesteps into account.

이렇게 하려면 입력과 현재 마스크가 지정되었을 때 새 마스크를 생성하는 layer.compute_mask() 메서드를 레이어에서 구현해야 합니다.

다음은 현재 마스크를 수정해야하는 TemporalSplit 레이어의 예제입니다.

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)

입력 값에서 마스크를 생성 할 수있는 CustomEmbedding 레이어의 다른 예는 다음과 같습니다.

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 False  True]
 [ True  True  True  True  True  True  True  True  True  True]], shape=(3, 10), dtype=bool)

호환 가능한 레이어에서 전파를 마스크하도록 선택

대부분의 레이어는 시간 차원을 수정하지 않으므로 현재 마스크를 수정할 필요가 없습니다. 그러나 그들은 여전히 현재 마스크를 변경하지 않고 다음 레이어로 전파 할 수 있기를 원할 수 있습니다. 이것은 옵트 인 동작입니다. 기본적으로 사용자 정의 레이어는 현재 마스크를 제거합니다 (프레임 워크에서 마스크 전파가 안전한지 여부를 알 수있는 방법이 없기 때문에).

시간 차원을 수정하지 않는 사용자 정의 레이어가 있고 현재 입력 마스크를 전파 할 수 있으려면 레이어 생성자에서 self.supports_masking = True 를 설정해야합니다. 이 경우 compute_mask() 의 기본 동작은 현재 마스크를 통과하는 것입니다.

마스크 전파가 허용 된 레이어의 예는 다음과 같습니다.

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)

이제 마스크를 생성하는 레이어(예: Embedding)와 마스크를 소비하는 레이어(예: LSTM) 사이에서 이 사용자 정의 레이어를 사용하여 마스크 소비 레이어까지 마스크를 전달할 수 있습니다.

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

마스크 정보가 필요한 레이어 작성

일부 레이어는 마스크 소비자입니다. 이러한 레이어는 call에서 mask 인수를 허용하고 이를 사용하여 특정 타임스텝을 건너뛸지 여부를 결정합니다.

이러한 계층을 작성하려면 call 서명에 mask=None 인수를 추가하면됩니다. 입력과 관련된 마스크는 가능할 때마다 레이어로 전달됩니다.

다음은 간단한 예입니다. 마스크 된 시간 단계를 버리고 입력 시퀀스의 시간 차원 (축 1)에 대한 소프트 맥스를 계산하는 레이어입니다.

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

요약

이것이 Keras의 패딩 및 마스킹에 대해 알아야 할 전부입니다. 요약하자면:

  • "Masking"은 레이어가 시퀀스 입력에서 특정 시간 단계를 건너 뛰거나 무시할 때를 알 수있는 방법입니다.
  • 일부 레이어는 마스크 생성기입니다. Embedding 하면 입력 값에서 마스크를 생성 할 수 있으며 ( mask_zero=True ) Masking 레이어도 생성 할 수 있습니다.
  • 일부 레이어는 마스크 소비자입니다. __call__ 메서드에 mask 인수를 표시합니다. 이것은 RNN 계층의 경우입니다.
  • 함수형 API 및 Sequential API에서 마스크 정보는 자동으로 전파됩니다.
  • 독립형 방식으로 레이어를 사용하는 경우 mask 인수를 레이어에 수동으로 전달할 수 있습니다.
  • 현재 마스크를 수정하거나 새 마스크를 생성하거나 입력과 관련된 마스크를 사용하는 레이어를 쉽게 작성할 수 있습니다.