Transformer 번역기 부분에 대해 질문 있습니다.

아래 사이트는 Encoder / Decoder를 하나의 Class로 만들었는데요..강사님도 해당 예제는 많이 보셨을거 같습니다.

https://keras.io/examples/nlp/neural_machine_translation_with_transformer/

아래와 같이 Encoder / Decoder Class를 선언하고..추가로 shape들을 찍어 보았습니다.

(참고로 get_config 함수는 지면상 뺐습니다.)

class TransformerEncoder(keras.layers.Layer):

    def init(self, num_heads, embed_dim, dense_dim, **kwargs):

        super().__init__(**kwargs)

        self.num_heads = num_heads

        self.embed_dim = embed_dim

        self.dense_dim = dense_dim

        self.class_name = 'TransformerEncoder'

        self.attention = keras.layers.MultiHeadAttention(self.num_heads, key_dim=self.embed_dim, value_dim=self.embed_dim)

        self.dense = keras.models.Sequential([keras.layers.Dense(self.dense_dim, activation='relu'),

                                             keras.layers.Dense(self.embed_dim)])

        self.layer1 = keras.layers.LayerNormalization()

        self.layer2 = keras.layers.LayerNormalization()

        self.support_mask = True

    def call(self, inputs, mask=None):

        tf.print('ClassName=TransformerEncoder-------------------------------------------------------')

        if mask is not None:

            #print('ClassName={}, mask.shape={}'.format(self.class_name, mask.shape))

            padding_mask = mask[:, tf.newaxis, :]

            tf.print('mask.shape=', padding_mask.shape)

        else:

            padding_mask = None

            tf.print('mask=None')

        

        attention_out = self.attention(query=inputs, key=inputs, value=inputs, attention_mask=padding_mask)

        tf.print('input.shape=', inputs.shape)

        tf.print('attention_out.shape=', attention_out.shape)

        layer1_out = self.layer1(inputs+attention_out)

        #print('ClassName={}, layer1_out.shape={}'.format(self.class_name, layer1_out.shape))

        dense_out  = self.dense(layer1_out)

        tf.print('dense_out.shape=', dense_out.shape)

        return self.layer2(layer1_out+dense_out)




class TransformerDecoder(keras.layers.Layer):

    def init(self, num_heads, embed_dim, latent_dim, **kwargs):

        super().__init__(**kwargs)

        self.num_heads = num_heads

        self.embed_dim = embed_dim

        self.latent_dim = latent_dim

        self.class_name = 'Decoder'

        self.attention_1 = keras.layers.MultiHeadAttention(num_heads=self.num_heads, key_dim=self.embed_dim)

        self.attention_2 = keras.layers.MultiHeadAttention(num_heads=self.num_heads, key_dim=self.embed_dim)

        self.layer_1 = keras.layers.LayerNormalization()

        self.layer_2 = keras.layers.LayerNormalization()

        self.layer_3 = keras.layers.LayerNormalization()

        self.dense = keras.models.Sequential([keras.layers.Dense(self.latent_dim, activation='relu'),

                                              keras.layers.Dense(self.embed_dim)])

        

        self.supports_masking=True

    def call(self, inputs, encoder_outputs, mask=None):        

        tf.print('ClassName=TransformerDecoder-------------------------------------------------------')

        #inputs.shape=(None, 35, 32), encoder_outputs.shape=(None, 35, 32)     

        #tf.print('aaaaaaaaa=', inputs.shape)

        casual_mask = self.get_casual_attention_mask(inputs)

        tf.print('inputs.shape=', inputs.shape)

        tf.print('casual_mask.shape=', casual_mask.shape)

        

        if mask is not None:

            padding_mask = tf.cast(mask[:, None, :], dtype='int32')

            padding_mask = tf.minimum(padding_mask, casual_mask)

            tf.print('padding_mask.shape=', padding_mask.shape)

        else:

            padding_mask = None

            tf.print('padding_mask = None')

        attention_1_out = self.attention_1(query=inputs, key=inputs, value=inputs, attention_mask=casual_mask)   

        tf.print('attention_1_out.shape=', attention_1_out.shape)

        layer_1_out = self.layer_1(inputs+attention_1_out)

        attention_2_out = self.attention_2(query=layer_1_out, key=encoder_outputs, value=encoder_outputs, attention_mask=padding_mask)

        layer_2_out = self.layer_2(layer_1_out + attention_2_out)

        dense_out = self.dense(layer_2_out)

        tf.print('dense_out.shape=', dense_out.shape)

        return self.layer_3(layer_2_out + dense_out)

    

    def get_casual_attention_mask(self, inputs):

        #input_shape=(2, 35, 32) =(BatchSize, Sequence_len, Embed_Len)

        input_shape = tf.shape(inputs)

        batch_size, sequence_length = input_shape[0], input_shape[1]

        

        #i=(35, 1), j=(35,)

        #i=[[0], [1], [2], ... [35]], j=[0 1 2 ... 34]

        i = tf.range(start=0, limit=sequence_length)[:, tf.newaxis]

        j = tf.range(start=0, limit=sequence_length)

        # tf.cast 함수는 보통 조건에 따른 True, False 의 판단 기준에 따라 True 면 1, False 면 0을 반환한다.

        # mask.shape=(1, 35, 35), mask=[[[1 0 0 .. 0], [1 1 0 0 ...0], [1 1 1 1 ...1 0], [1 1 1 1 ... 1 1]]]

        mask = tf.cast(i>=j, dtype='int32')

        mask = tf.reshape(mask, (1, input_shape[1], input_shape[1]))

        # mult = (2, 1, 1), (BatchSize, 1, 1)

        mult = tf.concat([tf.expand_dims(batch_size, -1), tf.convert_to_tensor([1,1])], axis=0)

        return tf.tile(mask, mult)



MAX_EMBED=32

MAX_VOCAB=5000

MAX_SEQ = 35

MAX_DENSE=1024

MAX_HEAD=1

encoder_inputs = keras.layers.Input(shape=(MAX_SEQ,), dtype='int64', name='encoder_inputs')

encoder_embed_outs = PositionalEmbedding(MAX_SEQ, MAX_VOCAB, MAX_EMBED)(encoder_inputs)

encoder_transformer_outs = TransformerEncoder(num_heads=1, embed_dim=MAX_EMBED, dense_dim=MAX_DENSE)(encoder_embed_outs)

#encoder_transformer_outs == (None, 80, 256)

decoder_inputs = keras.layers.Input(shape=(MAX_SEQ,), dtype='int64', name='decoder_inputs')

decoder_embed_outs = PositionalEmbedding(MAX_SEQ, MAX_VOCAB, MAX_EMBED)(decoder_inputs)

#decoder_embed_outs == (None, 80, 256)

      

decoder_transformer_outs = TransformerDecoder(1, MAX_EMBED, MAX_DENSE)(decoder_embed_outs, encoder_transformer_outs)

decoder_dropout_outs = keras.layers.Dropout(0.5)(decoder_transformer_outs)

decoder_outputs = keras.layers.Dense(MAX_VOCAB, activation='softmax')(decoder_dropout_outs)

model = keras.Model(inputs=[encoder_inputs, decoder_inputs], outputs=decoder_outputs)

model.summary()

호출(호출은 Log만 찍어보게 BatchSize를 줄였습니다.)

q_train_seqs = q_seqs[:2]

a_train_seqs = a_seqs[:2]

print(q_train_seqs.shape, a_train_seqs.shape)

model.compile(optimizer='adam', loss='sparse_categorical_crossentropy', metrics=['accuracy'])

history = model.fit([q_train_seqs, a_train_seqs[:, :-1]], a_train_seqs[:, 1:], epochs=1)

이렇게 해서 돌리면 결과가 아래와 같이 찍힙니다.

ClassName=TransformerEncoder-------------------------------------------------------
mask=None
input.shape= TensorShape([None, 35, 32])
attention_out.shape= TensorShape([None, 35, 32])
dense_out.shape= TensorShape([None, 35, 32])
ClassName=TransformerDecoder-------------------------------------------------------
inputs.shape= TensorShape([None, 35, 32])
casual_mask.shape= TensorShape([None, 35, 35])
padding_mask = None
attention_1_out.shape= TensorShape([None, 35, 32])
dense_out.shape= TensorShape([None, 35, 32])
1/1 [==============================] - 4s 4s/step - loss: 8.2735 - accuracy: 0.0000e+00

질문은 아래와 같습니다.

1. Encoder 파트에서 mask=None 으로 Mask가 들어있지 않습니다.

   그럼 굳이 mask는 구현할 필요가 없는건지?

    

2. Decoder 파트에서는 첫번째 Self-Attention 부분에서만 제대로된 마스킹 값이 들어가고 Encoder의 Key와 매핑 시키는 부분에서는 마찬가지로 padding_mask가 None 입니다. 이 부분에 대해 조금 더 자세한 설명을 해 주시면 감사하겠습니다.

    

다시 한번 설명 드리면

class TransformerEncoder(keras.layers.Layer):
...

 def call(self, inputs, mask=None):
        tf.print('ClassName=TransformerEncoder-------------------------------------------------------')
        if mask is not None:
            #print('ClassName={}, mask.shape={}'.format(self.class_name, mask.shape))
            padding_mask = mask[:, tf.newaxis, :]
            tf.print('mask.shape=', padding_mask.shape)
        else:
            padding_mask = None
            tf.print('mask=None')

#Encoder 에서 padding_mask는 None으로 찍힘
        attention_out = self.attention(query=inputs, key=inputs, value=inputs, attention_mask=padding_mask)


class TransformerDecoder(keras.layers.Layer):   
    def call(self, inputs, encoder_outputs, mask=None):        
        casual_mask = self.get_casual_attention_mask(inputs)


#mask는 None 으로 들어옴
        if mask is not None:
            padding_mask = tf.cast(mask[:, None, :], dtype='int32')
            padding_mask = tf.minimum(padding_mask, casual_mask)
        else:
            padding_mask = None

# casual_mask는 현재 위치 이후는 0 으로 패딩 된 것 확인
        attention_1_out = self.attention_1(query=inputs, key=inputs, value=inputs, attention_mask=casual_mask)   


        layer_1_out = self.layer_1(inputs+attention_1_out)

# 하지만 padding_mask는 None 임
        attention_2_out = self.attention_2(query=layer_1_out, key=encoder_outputs, value=encoder_outputs, attention_mask=padding_mask)
        layer_2_out = self.layer_2(layer_1_out + attention_2_out)