NLP 공부 (5-3: Transformer)

Transformer 모델은 Seq2Seq와 다르게 RNN을 사용하지 않고 Attention만 사용하기 때문에 단어 입력을 순차적으로 하나하나 받지 않고 한 번에 모든 단어를 입력으로 받는다. 순차적으로 받지 않게되면 단어의 위치 정보를 따로 입력해줄 필요가 있을 것이다. 이를 위해 사용하는 기법이 바로 Positional Encoding이다. Positional Encoding에서는 다음과 같은 함수가 사용된다. 

 

 

Positional Encoding

 

 

위 함수에서는 Sin, Cos 함수를 사용하게 되는데 이는 위치 벡터값이 -1과 1사이의 값을 갖게끔 한다. 코드로는 다음과 같이 작성해볼 수 있다.

 

 

def position(pos, i, dim):
  result = pos / np.power(10000, (2 * i) / np.float32(dim))
  return result

def positionalEncoding(pos, dim):
  result = position(np.arange(pos)[:, np.newaxis], np.arange(dim)[np.newaxis, :], dim)

  result[:, 0::2] = np.sin(result[:, 0::2])
  result[:, 1::2] = np.cos(result[:, 1::2])
  posEncoding = result[np.newaxis, ...]
  return tf.cast(posEncoding, dtype=tf.float32)

 

 

position 함수는 위 Positional Encoding 수식에서 괄호 안의 수식을 만들어준 것이다. positionalEncoding에서는 이 함수를 사용하여 우선 각 단어별로 다른 위치 값을 갖도록 한다. 그 후, 위 수식과 동일하게 짝수 위치의 값들에는 Sin, 홀수 위치의 값들에는 Cos 함수를 적용시킨 후, 차원을 늘려주어 결과값으로 내보낸다. (np.newaxis 는 새로운 차원을 만들어줌) 

 

 

이제, 지금까지 만들어온 함수들을 모두 사용하여 Encoder Decoder를 직접 구성해볼 차례이다. 아래 Transformer Model 구조와 동일하게 구성해보도록 하겠다.

 

 

Transformer

 

class EncoderLayer(tf.keras.layers.Layer):
  def __init__(self, **kwargs):
    super(Encoder, self).__init__()
    self.multihead = MultiHeadAttention(**kwargs)
    self.ffnn = positionwiseFeedforwardNetwork(kwargs["hidDim"], kwargs["outDim"])
    self.normalize = tf.keras.layers.LayerNormalization(epsilon=1e-6)
    self.dropout = tf.keras.layers.Dropout(kwargs["dRate"])
  def call(self, data, mask):
    attention_result, _ = self.mutilhead(data, data, data, mask)
    attention_result = self.dropout(attention_result)
    attention_result = self.normalize(attention_result)
    save_temp = attention_result

    attention_result = self.ffnn(attention_result)
    attention_result = self.dropout(attention_result)
    result = self.normalize(attention_result + save_temp)
    return result

class Encoder(tf.keras.layers.Layer):
  def __init__(self, **kwargs):
    super(Encoder, self).__init__()
    self.dim = kwargs["dim"]
    self.layers = kwargs["layers"]
    self.embedding = tf.keras.layers.Embedding(kwargs["wordDicSize"], self.dim)
    self.positionalEncoding = positionalEncoding(kwargs["maxPositionEncoding"], self.dim)
    self.encoderLayer = [EncoderLayer(**kwargs) for _ in range(self.layers)]
    self.dropout = tf.keras.layers.Dropout(kwargs["dRate"])
  def call(self, data, mask):
    sequenceLength = tf.shape(data)[1]
    data = self.embedding(data)
    data *= tf.math.sqrt(tf.cast(self.dim, tf.float32))
    data += self.positionalEncoding[:, :sequenceLength, :]
    data = self.dropout(data)
    for i in range(self.layers):
      data = self.encoderLayer[i](data, mask)
    return data

 

 

우선 Encoder이다. EncoderLayer 그리고 EncoderLayer들을 여러 번 사용하는 Encoder, 2 개의 Class를 만들어주었다. EncoderLayer에서는 우선 멀티헤드어텐션을 거치고 포지션와이즈피드포워드 네트워크를 거친다. 그 과정에서 Dropout, Normalization, Residual Connection도 반영해주었다. 이 Encoder Layer Class를 활용하여 Encoder에서는 인풋 데이터를 우선 Embedding 해준 후, Positional Encoding 과정을 거쳐 위치 정보를 더해준다. 그리고 Layer 개수로 지정한 만큼 Encoder Layer Class를 거치게 되는 것이다. 이제 Decoder를 작성해보겠다.

 

 

class DecoderLayer(tf.keras.layers.Layer):
  def __init__(self, **kwargs):
    super(DecoderLayer, self).__init__()
    self.multihead = MultiHeadAttention(**kwargs)
    self.ffnn = positionwiseFeedforwardNetwork(kwargs["hidDim"], kwargs["outDim"])
    self.normalize = tf.keras.layers.LayerNormalization(epsilon=1e-6)
    self.dropout = tf.keras.layers.Dropout(kwargs["dRate"])
  def call(self, data, encoderResult, lookaheadMask, paddingMask):
    attentionResult, attentions = self.multihead(data, data, data, lookaheadMask)
    attentionResult = self.dropout(attentionResult)
    res1 = self.normalize(attentionResult+data)
    attentionResult2, attentions2 = self.multihead(encoderResult, encoderResult, res1, paddingMask)
    attentionResult2 = self.dropout(attentionResult2)
    res2 = self.normalize(attentionResult2 + res1)
    res3 = self.ffnn(res2)
    res3 = self.dropout(res3)
    res4 = self.normalize(res3 + res2)
    return res4, attentions, attentions2

class Decoder(tf.keras.layers.Layer):
  def __init__(self, **kwargs):
    super(Decoder, self).__init__()
    self.dim = kwargs["dim"]
    self.layers = kwargs["layers"]
    self.embedding = tf.keras.layers.Embedding(kwargs["targetDicSize"], self.dim)
    self.positionalEncoding = positionalEncoding(kwargs["maxPositionEncoding"], self.dim)
    self.decoderLayer = [DecoderLayer(**kwargs) for _ in range(self.layers)]
    self.dropout = tf.keras.layers.Dropout(kwargs["rate"])
  def call(self, data, encoderResult, lookaheadMask, paddingMask):
    sequenceLength = tf.shape(data)[1]
    attentionWeights = {}

    data = self.embedding(data)
    data *= tf.math.sqrt(tf.cast(self.dim, tf.float32))
    data += self.positionalEncoding[:, :sequenceLength, :]
    data = self.dropout(data)
    for i in range(self.layers):
      data, block1, block2 = self.decoderLayer[i](data, encoderResult, lookaheadMask, paddingMask)
      attentionWeights["Decoder_Layer{}_Block1".format(i+1)] = block1
      attentionWeights["Decoder_Layer{}_Block2".format(i+1)] = block2
    return data, attentionWeights

 

 

Encoder과 상당히 유사하게 작성되었다. 차이점으로는 Decoder Self-Attention 연산에 순방향 마스크가 입력값으로 추가되고 Attention 연산 시 Encoder Result 벡터가 입력으로 사용되어 Decoder의 정보 벡터와 Attention 연산을 한다는 것이다. 마지막으로 이 둘 다 사용하여 Transformer Class를 작성한다.

 

 

class Transformer(tf.keras.Model):
  def __init__(self, **kwargs):
    super(Transformer, self).__init__()
    self.endTokenIdx = kwargs["endTokenIdx"]
    self.encoder = Encoder(**kwargs)
    self.decoder = Decoder(**kwargs)
    self.lastLayer = tf.keras.layers.Dense(kwargs["targetDicSize"])
  def call(self, data):
    input, target = data
    encoderPaddingMask ,lookaheadMask ,decoderPaddingMask = makeMasks(input, target)
    encoderResult = self.encoder(input, encoderPaddingMask)
    decoderResult, _ = self.decoder(target, encoderResult, lookaheadMask, decoderPaddingMask)
    result = self.lastLayer(decoderResult)
    return result
  def predict(self, data):
    input = data
    target = tf.expand_dims([SOS_INDEX], 0)
    encoderPaddingMask, lookaheadMask, decoderPaddingMask = makeMasks(input, target)
    encoderResult = self.encoder(input, encoderPaddingMask)
    predictions = list()
    for token in range(MAX_SEQUENCE):
      decoderResult, _ = self.decoder(target, encoderResult, lookaheadMask, decoderPaddingMask)
      result = self.lastLayer(docoderResult)
      result2 = tf.argmax(result, -1).numpy()
      prediction = result2[0][-1]
      if prediction == self.endTokenIdx:
        break
      predictions.append(prediction)
      target = tf.expand_dims([SOS_INDEX]+predictions, 0)
      _, lookaheadMask, decoderPaddingMask = makeMasks(input, target)
    return predictions

 

 

최종적으로 작성한 Transformer 모델은 Encoder, Decoder class와 마지막으로 거칠 Dense(FFNN) 한 층으로 구성되어 있다. 위 모델을 사용하여 학습을 진행하고 Transformer class의 predict 함수로 예측까지 진행해보았다.

 

 

predict

 

총 100번의 Epoch을 돌렸는데 좋은 성과를 보이지는 못했다. 더 많은 데이터, 혹은 이후에 다룰 BERT, GPT와 같은 사전학습 모델을 사용하여 성능을 높여야할 것으로 생각된다.