[NLP] Sequence to Sequence(시퀀스 투 시퀀스) 코드

https://acdongpgm.tistory.com/216

[NLP] Sequence to Sequence (시퀀스 투 시퀀스), Attention(어텐션) 개념

시퀀스 투 시퀀스의 모델 구조와 어떻게 동작하는지에 대해 코드를 통해 이해해 보고자 한다.

 

예제에서 사용한 데이터 셋은 연애 질문에 대한 Q&A 데이터이고 시퀀스 투 시퀀스로 간단한 챗봇을 만들고자 한다.

출처 : 텐서 플로 2와 머신러닝으로 시작하는 자연어 처리

위 책에 있는 예제를 그대로 따라 해보고 이해해보자.

 

전처리가 가장 중요하다는 사실은 알고있지만, 코드가 길어지고 모델의 핵심구조 파악이 더 우선이기 때문에

전처리 과정은 코드블록으로 작성하지 않고 요약함.

---

전처리 과정

1. 토큰 생성

 - PAD : 패딩 토큰 자릿수를 맞춰주기 위해 빈 공간에 들어가는 토큰

 - STD : 아웃풋 시퀀스의 시작을 알려주는 토큰

 - END : 아웃풋 시퀀스의 끝을 알려주는 토큰

 - UNK : 사전에 없는 단어 Unknown 토큰

 

2. Vocab 생성

 - 토큰을 포함한 데이터에 사용된 모든 단어를 사전으로 정리

 - 단어 -> 숫자 , 숫자 -> 단어

 

3. 정수 인코딩

 - vocab 을 가지고 학습/테스트 데이터의 단어들을 정수로 변환

 

인풋 데이터 : 

'3박4일 정도 놀러 가고 싶다'      ---->      [9031, 6756, 553, 5199, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]

 

아웃풋 데이터 : 

'여행은 언제나 좋죠' -------> [ 9142, 11845, 5855, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
 *2는 <END> 토큰

---

파라미터 정의 : 

MODEL_NAME = 'seq2seq_kor'
BATCH_SIZE = 2
MAX_SEQUENCE = 25
EPOCH = 30
UNITS = 1024
EMBEDDING_DIM = 256
VALIDATION_SPLIT = 0.1

인코더 레이어:

class Encoder(tf.keras.layers.Layer):
    def __init__(self, vocab_size, embedding_dim, enc_units, batch_sz):
        super(Encoder, self).__init__()
        self.batch_sz = batch_sz
        self.enc_units = enc_units
        self.vocab_size = vocab_size 
        self.embedding_dim = embedding_dim          
        
        self.embedding = tf.keras.layers.Embedding(self.vocab_size, self.embedding_dim)
        self.gru = tf.keras.layers.GRU(self.enc_units,
                                       return_sequences=True,
                                       return_state=True,
                                       recurrent_initializer='glorot_uniform')

    def call(self, x, hidden):
        x = self.embedding(x)
        output, state = self.gru(x, initial_state = hidden)
        return output, state

    def initialize_hidden_state(self, inp):
        return tf.zeros((tf.shape(inp)[0], self.enc_units))

 

인코더 레이어는 간단하게 임베딩을 하고 RNN 계열의 모델인 GRU 모델을 거쳐서 인풋 정보를 요약한다.

여기서 기존 seq2seq 과 어텐션 기법을 추가한 seq2 seq의 차이점이 살짝 들어가는데.

GRU 파라미터에 return_sequences = True 로 설정할 경우 time step 별 hidden state를 모두 출력하게 된다.

* 마지막 hidden state 만 output 으로 사용하는 것이 아니라 각각의 hidden state를 전부 사용한다. Attention을 하기 위함.

 

return_state = True 를 한 경우에는 마지막 time step에서의 output(hidden state), hidden state

와 cell state 가 출력된다. 즉 마지막 output값이 2번 출력이 되고 cell state가 나온다.

 

어텐션 레이어:

class BahdanauAttention(tf.keras.layers.Layer):
    def __init__(self, units):
        super(BahdanauAttention, self).__init__()
        self.W1 = tf.keras.layers.Dense(units)
        self.W2 = tf.keras.layers.Dense(units)
        self.V = tf.keras.layers.Dense(1)

    def call(self, query, values):
        hidden_with_time_axis = tf.expand_dims(query, 1)

        score = self.V(tf.nn.tanh(
            self.W1(values) + self.W2(hidden_with_time_axis)))

        attention_weights = tf.nn.softmax(score, axis=1)

        context_vector = attention_weights * values
        context_vector = tf.reduce_sum(context_vector, axis=1)

        return context_vector, attention_weights

BahadanauAttention 모델은 뉴럴 네트워크를 통해 score를 구하는 방식이다.

 

그래서 Dense 레이어를 2개 거치고 하나로 모아서 Softmax를 거쳐 attention_weights를 계산한다.

attention_weight를 value(현재 hidden state)와 곱해서 context_vector를 구한다.

 

디코더 레이어 : 

class Decoder(tf.keras.layers.Layer):
    def __init__(self, vocab_size, embedding_dim, dec_units, batch_sz):
        super(Decoder, self).__init__()
        
        self.batch_sz = batch_sz
        # GRU layer number
        self.dec_units = dec_units
        
        # Embedding
        self.vocab_size = vocab_size 
        self.embedding_dim = embedding_dim  
        
        self.embedding = tf.keras.layers.Embedding(self.vocab_size, self.embedding_dim)
        self.gru = tf.keras.layers.GRU(self.dec_units,
                                       return_sequences=True,
                                       return_state=True,
                                       recurrent_initializer='glorot_uniform')
        self.fc = tf.keras.layers.Dense(self.vocab_size)

        self.attention = BahdanauAttention(self.dec_units)
        
    def call(self, x, hidden, enc_output):
        context_vector, attention_weights = self.attention(hidden, enc_output)

        x = self.embedding(x)

        x = tf.concat([tf.expand_dims(context_vector, 1), x], axis=-1)

        output, state = self.gru(x)
        output = tf.reshape(output, (-1, output.shape[2]))
            
        x = self.fc(output)
        
        return x, state, attention_weights

디코더 부분에 Attention을 진행한다는 것을 볼 수 있고

 

현재 hidden_state 와 현재 hidden_state와 attent_weight를 곱한 context_vector를 tf.concat 하는

부분이 존재한다. 이것을 다시 인풋으로 Dense Layer를 거쳐서 모든 단어들에 대한 Softmax 계산을 해서

가장 스코어(확률)이 높은 단어를 찾아낸다. <END> 토큰이 나올 때까지

---

시퀀스 투 시퀀스 모델 :

class seq2seq(tf.keras.Model):
    def __init__(self, vocab_size, embedding_dim, enc_units, dec_units, batch_sz, end_token_idx=2):    
        super(seq2seq, self).__init__()
        self.end_token_idx = end_token_idx
        self.encoder = Encoder(vocab_size, embedding_dim, enc_units, batch_sz) 
        self.decoder = Decoder(vocab_size, embedding_dim, dec_units, batch_sz) 

    def call(self, x):
        inp, tar = x
        
        enc_hidden = self.encoder.initialize_hidden_state(inp)
        enc_output, enc_hidden = self.encoder(inp, enc_hidden)

        dec_hidden = enc_hidden

        predict_tokens = list()
        for t in range(0, tar.shape[1]):
            dec_input = tf.dtypes.cast(tf.expand_dims(tar[:, t], 1), tf.float32) 
            predictions, dec_hidden, _ = self.decoder(dec_input, dec_hidden, enc_output)
            predict_tokens.append(tf.dtypes.cast(predictions, tf.float32))   
        return tf.stack(predict_tokens, axis=1)
    
    def inference(self, x):
        inp  = x

        enc_hidden = self.encoder.initialize_hidden_state(inp)
        enc_output, enc_hidden = self.encoder(inp, enc_hidden)

        dec_hidden = enc_hidden
        
        dec_input = tf.expand_dims([char2idx[std_index]], 1)
        
        predict_tokens = list()
        for t in range(0, MAX_SEQUENCE):
            predictions, dec_hidden, _ = self.decoder(dec_input, dec_hidden, enc_output)
            predict_token = tf.argmax(predictions[0])
            
            if predict_token == self.end_token_idx:
                break
            
            predict_tokens.append(predict_token)
            dec_input = tf.dtypes.cast(tf.expand_dims([predict_token], 0), tf.float32)   
            
        return tf.stack(predict_tokens, axis=0).numpy()

실제 predict 하는 과정은 inference를 통해 이루어진다.

 

모델 정의 : 

model = seq2seq(vocab_size, EMBEDDING_DIM, UNITS, UNITS, BATCH_SIZE, char2idx[end_index])
model.compile(loss=loss, optimizer=tf.keras.optimizers.Adam(1e-3), metrics=[accuracy])
#model.run_eagerly = True

 

모델 학습 : 

PATH = DATA_OUT_PATH + MODEL_NAME
if not(os.path.isdir(PATH)):
        os.makedirs(os.path.join(PATH))
        
checkpoint_path = DATA_OUT_PATH + MODEL_NAME + '/weights.h5'
    
cp_callback = ModelCheckpoint(
    checkpoint_path, monitor='val_accuracy', verbose=1, save_best_only=True, save_weights_only=True)

earlystop_callback = EarlyStopping(monitor='val_accuracy', min_delta=0.0001, patience=10)

history = model.fit([index_inputs, index_outputs], index_targets,
                    batch_size=BATCH_SIZE, epochs=EPOCH,
                    validation_split=VALIDATION_SPLIT, callbacks=[earlystop_callback, cp_callback])

 

결과 테스트 :

query = '남자친구 승진 선물로 뭐가 좋을까?'

test_index_inputs , _ = enc_processing([query],char2idx)
predict_tokens = model.inference(test_index_inputs)

print(' '.join([idx2char[str(t)] for t in predict_tokens]))

 

시퀀스 투 시퀀스는 결과가 다소 어색하다. 위 결과가 그나마 가장 매끄러운 부분이다.

many-to-many 모델이기 때문에  <end> 토큰이 나오지 않고 계속해서 같은 단어를 반복하는 경우도 있었다.

다음에는 이보다 더 성능이 향상된 모델인 transformer 모델을 가지고 구현해 볼 생각이다.

 

그리고 시퀀스 투 시퀀스 같은 모델은 확실히 subclass 모델로 구현하는 게 이해하기 쉽고

편한 것 같다. 앞으로 모든 모델은 subclass로 구현해서 익숙해질 필요가 있을 것 같다.