[딥러닝] RNN을 사용한 자연어 처리: 신경망 기계 번역

이전 내용

[딥러닝] RNN을 사용한 자연어 처리: 감성분석

---

신경망 기계 번역(NMT, neural machine translation)

 

신경망 기계 번역(NMT, Neural Machine Translation)은 인공 신경망을 사용하여 한 언어에서 다른 언어로 텍스트를 자동으로 번역하는 기술을 말한다. NMT는 번역의 일관성과 품질을 향상시키기 위해 딥러닝 기술을 사용한다.

- NMT의 주요 개념 및 구성 요소

1. 인코더-디코더 구조:
   • NMT 시스템의 기본 구조는 일반적으로 인코더-디코더(Encoder-Decoder) 아키텍처로 구성된다.
   • 인코더(Encoder): 입력 텍스트를 고정된 길이의 벡터로 변환한다. 주로 RNN(Recurrent Neural Networks), LSTM(Long Short-Term Memory), GRU(Gated Recurrent Units), 또는 최근에는 트랜스포머(Transformer) 모델이 사용된다.
   • 디코더(Decoder): 엔코더에서 생성된 벡터를 텍스트로 다시 변환한다. 디코더도 주로 RNN, LSTM, GRU, 또는 트랜스포머 모델을 사용한다.
2. 어텐션 메커니즘(Attention Mechanism):
   • 어텐션 메커니즘은 입력 시퀀스의 특정 부분에 대해 가중치를 부여하여 디코더가 번역을 더 잘할 수 있도록 도와준다.
   • 이는 특히 긴 문장의 번역에서 중요한 역할을 하며, 디코더가 입력 시퀀스 내 중요한 정보에 집중할 수 있도록 한다.
3. 트랜스포머 모델:
   • 트랜스포머(Transformer)는 최근에 가장 많이 사용되는 모델로서, RNN 구조를 사용하지 않고도 시퀀스를 효과적으로 처리할 수 있다.
   • 셀프 어텐션(Self-Attention) 메커니즘을 통해 문장 내의 단어들 간의 관계를 효과적으로 모델링할 수 있다.
   • 트랜스포머 모델은 Google의 BERT, OpenAI의 GPT 등 여러 모델의 기본 구조로 사용된다.
4. 훈련 및 번역 과정:
   • 훈련: 대규모 병렬 코퍼스(병렬 문장 쌍 데이터셋)를 사용하여 모델을 훈련시킨다. 훈련 과정 중 모델은 입력과 출력 문장 쌍을 통해 번역 규칙을 학습한다.
   • 번역: 훈련된 모델에 새로운 입력 문장을 주면, 모델이 해당 문장을 타겟 언어로 번역한다.

- NMT의 장점

• 문맥 이해: 기존의 통계적 기계 번역(SMT, Statistical Machine Translation)보다 문장의 문맥을 더 잘 이해하며, 더 자연스러운 번역 결과를 제공한다.
• 종단 간 학습(End-to-End Learning): 번역 파이프라인의 여러 단계를 따로 설정할 필요 없이, 통합된 모델로 전체 번역 과정을 학습할 수 있다.
• 확장 가능성: 딥러닝의 특성을 이용해 대규모 데이터셋을 사용하여 모델 성능을 지속적으로 향상시킬 수 있다.

[도움될 자료: 트랜스포머 모델, BERT]

[신경망 모델] 트랜스포머(Transformer) 모델이란?

 

[NLP 모델] BERT(Bidirectional Encoder Representations from Transformers)란?

---

영어 문장을 스페인어로 번역하는 간단한 NMT 모델 만들기

 

- 데이터셋 다운로드

먼저 영어/스페인어 문장 쌍으로 구성된 데이터셋을 다운로드 해야 한다.

from pathlib import Path

url = "https://storage.googleapis.com/download.tensorflow.org/data/spa-eng.zip"
path = tf.keras.utils.get_file("spa-eng.zip", origin=url, cache_dir="datasets",
                               extract=True)
text_path = Path(path).with_name("spa-eng") / "spa.txt"
text = text_path.read_text(encoding='utf-8')

print(text[:500])  # 텍스트 파일의 처음 500자를 출력하여 확인

▶ 각 줄에는 영어 문장과 해당 스페인어 번역이 탭으로 구분되어 있다.

 

- 데이터 정제 및 전처리

TextVectorization 층이 처리하지 못하는 스페인어 문자들(¡, ¿)을 제거한 다음, 문장 쌍을 파싱하고 섞는다. 그리고 언어별로 하나씩 두 개의 리스트로 나눈다.

import numpy as np

text = text.replace("¡", "").replace("¿", "")
pairs = [line.split("\t") for line in text.splitlines()]
np.random.seed(42)
np.random.shuffle(pairs)
sentences_en, sentences_es = zip(*pairs)  # 쌍을 2개의 리스트로 분리

 

처음 몇 개의 문장을 확인해본다.

for i in range(3):
    print(sentences_en[i], '=>', sentences_es[i])

 

- TextVectorization 층 생성

그리고 언어마다 하나씩 두 개의 TextVectorization 층을 만들어 텍스트에 적용한다.

Layers = tf.keras.layers

vocab_size = 1000
max_length = 50
text_vec_layer_en = Layers.TextVectorization(
    vocab_size, output_sequence_length=max_length)
text_vec_layer_es = Layers.TextVectorization(
    vocab_size, output_sequence_length=max_length)
text_vec_layer_en.adapt(sentences_en)
text_vec_layer_es.adapt([f'startofseq {s} endofseq' for s in sentences_es])

 

두 어휘 사전에 있는 처음 10개의 토큰을 살펴본다.

print(text_vec_layer_en.get_vocabulary()[:10])
print(text_vec_layer_es.get_vocabulary()[:10])

 

- 훈련 세트와 검증 세트 생성

처음 10만 개의 문장 쌍은 훈련에 사용하고 나머지는 검증에 사용한다.

디코더의 입력은 스페인어 문장과 SOS 토큰이고, 타깃은 스페인어 문장과 EOS 토큰이다.

X_train = tf.constant(sentences_en[:100_000])
X_valid = tf.constant(sentences_en[100_000:])
X_train_dec = tf.constant([f'startofseq {s}' for s in sentences_es[:100_000]])
X_valid_dec = tf.constant([f'startofseq {s}' for s in sentences_es[100_000:]])
Y_train = text_vec_layer_es([f'endofseq {s}' for s in sentences_es[:100_000]])
Y_valid = text_vec_layer_es([f'endofseq {s}' for s in sentences_es[100_000:]])

 

모델 구성이 순차적이지 않으므로 함수형 API를 사용한다. 인코더와 디코더에 각각 하나씩 두 개의 텍스트 입력이 필요하므로 이것부터 만든다.

Layers = tf.keras.layers

encoder_inputs = Layers.Input(shape=[], dtype=tf.string)
decoder_inputs = Layers.Input(shape=[], dtype=tf.string)

 

- 문장 인코딩, 마스킹

앞서 준비한 TextVectorization 층을 사용하여 문장을 인코딩한다. 그다음 마스킹이 자동으로 처리되도록 각 언어에 대한 Embedding 층을 mask_zero=True로 설정한다. 임베딩 크기는 튜닝할 수 있는 하이퍼파라미터이다.

embed_size = 128
encoder_inputs_ids = text_vec_layer_es(encoder_inputs)
decoder_inputs_ids = text_vec_layer_es(decoder_inputs)
encoder_embedding_layer = Layers.Embedding(vocab_size, embed_size, mask_zero=True)
decoder_embedding_layer = Layers.Embedding(vocab_size, embed_size, mask_zero=True)
encoder_embeddings = encoder_embedding_layer(encoder_inputs_ids)
decoder_embeddings = decoder_embedding_layer(decoder_inputs_ids)

 

- 인코더 생성 및 임베딩된 입력 전달

encoder = Layers.LSTM(512, return_state=True)
encoder_outputs, *encoder_state = encoder(encoder_embeddings)

 

- 위의 상태를 디코더의 초기 상태로 사용

decoder = Layers.LSTM(512, return_state=True)
decoder_outputs = decoder(decoder_embeddings, initial_state=encoder_state)

 

- 디코더의 출력을 소프트맥스 활성화 함수가 있는 Dense 층에 전달

디코더의 출력을 소프트맥스 활성화 함수가 있는 Dense 층에 전달 하여 각 스텝에 대한 단어 확률을 얻을 수 있다.

output_layer = tf.keras.layers.Dense(vocab_size, activation="softmax")
Y_proba = output_layer(decoder_outputs)

 

- 케라스 Model 객체를 만들고 컴파일하고 훈련한다.

model = tf.keras.Model(inputs=[encoder_inputs, decoder_inputs],
                       outputs=[Y_proba])
model.compile(loss="sparse_categorical_crossentropy", optimizer="nadam",
              metrics=["accuracy"])
model.fit((X_train, X_train_dec), Y_train, epochs=10,
          validation_data=((X_valid, X_valid_dec), Y_valid))

▶ 정확도, 검증 정확도가 100%로 나오는거보니, 무언가 잘못됐다.

문장 인코딩, 마스킹 부분부터 차례대로 코드를 수정하고 훈련을 재수행해봤다.

 

- 사용자 정의 메모리 셀 생성을 위한 유틸리티 함수 생성

훈련이 끝나면 모델을 사욯아여 새로운 영어 문장을 스페인어로 번역할 수 있으나, model.predict()를 호출하는 것처럼 간단하지 만은 않다. 왜냐하면 디코더는 이전 타임 스텝에서 예측한 단어를 입력으로 기대하기 때문이다.

이를 수행하는 한 가지 방법은 이전 출력을 기록하여 다음 타임 스텝에서 인코더에게 주입하는 사용자 정의 메모리 셀을 만드는 것인데, 간단하게 하려면 모델을 여러 번 호출하고 매번 단어를 하나씩 늘려서 예측하면 된다. 이를 위한 간단한 유틸리티 함수를 만든다.

def translate(sentence_en):
    translation= ""
    for word_idx in range(max_length):
        X = np.array([sentence_en]) # 인코더 입력
        X_dec = np.array(["starttofseq " + translation]) # 디코더 입력
        y_proba = model.predict((X, X_dec))[0, word_idx] # 마지막 토큰 확률
        predicted_word_id = np.argmax(y_proba)
        predicted_word = text_vec_layer_es.get_vocabulary()[predicted_word_id]
        if predicted_word == "endofseq":
            break
        translation += " " + predicted_word
    return translation.strip()

 

이 함수는 한 번에 한 단어씩 계속 예측하여 점차적으로 번역을 완료하고 EOS 토큰에 도달하면 중지한다.

짧은 문장, 긴 문장 하나씩 수행해본다.

print(translate('Thank you so much, my friend'))
print(translate('I like soccer and also going to the mountain'))

▶ 어떤 과정에서 잘못된건지, 에러도 안 뜨고, 출력도 안 뜬다... (다시 해봤음에도 여전히.)

 

다음 날 세 번째 시도를 해보니, 다행히도 모델 컴파일 및 훈련도 잘 되고, translate 함수도 잘 출력되었다.

파파고에 번역을 해보니,

해당 스페인어가 영어로 위 내용처럼 번역되는데, [UNK]는 Unknown을 뜻한다. 즉, 아래 문장은 제대로 번역이 안 되었다는 의미. 심지어, 마지막 단어는 mountain이 아닌 music으로 번역하였다.

 

---

양방향 RNN

 

일반적인 단방향 RNN은 시퀀스를 한 방향 (보통 앞에서 뒤로)으로만 처리한다. 이는 입력 시퀀스의 입력 데이터가 과거의 정보만 반영된다는 단점이 있다. 양방향 RNN은 이를 보완하기 위해 시퀀스를 두 방향 (앞에서 뒤로, 뒤에서 앞으로) 모두 처리하여, 각 시점에서 양방향의 맥락 정보를 반영할 수 있게 만든다.

• 순방향 RNN: 입력 시퀀스를 왼쪽에서 오른쪽으로 읽는다.
• 역방향 RNN: 입력 시퀀스를 오른쪽에서 왼쪽으로 읽는다.
• 양방향 RNN: 순방향과 역방향 RNN을 동시에 처리하여 각 시점의 출력이 양쪽 맥락 정보를 모두 포함하도록 한다.

양방향 RNN의 주요 장점은 다음과 같다

• 모든 시점에서 시퀀스의 전체 맥락을 반영할 수 있다.
• 특히 후반부 단어들이 중요한 정보일 때 유용하다.

케라스에서 양방향 순환 층을 구현하려면 tf.keras.layers.Bidirectional로 순환 층을 감싼다.

tf.random.set_seed(42)
encoder = tf.keras.layers.Bidirectional(
    tf.keras.layers.LSTM(256, return_state=True))

 

LSTM은 두 개의 상태(단기 상태, 장기 상태)만 기대하므로, 두 개의 단기 상태와 두 개의 장기 상태를 연결한다.

encoder_outputs, *encoder_state = encoder(encoder_embeddings)
encoder_state = [tf.concat(encoder_state[::2], axis=-1),  # 단기 상태 (0 & 2)
                 tf.concat(encoder_state[1::2], axis=-1)]  # 장기 상태 (1 & 3)

 

모델 컴파일 및 훈련

decoder = tf.keras.layers.LSTM(512, return_sequences=True)
decoder_outputs = decoder(decoder_embeddings, initial_state=encoder_state)
output_layer = tf.keras.layers.Dense(vocab_size, activation="softmax")
Y_proba = output_layer(decoder_outputs)
model = tf.keras.Model(inputs=[encoder_inputs, decoder_inputs],
                       outputs=[Y_proba])
model.compile(loss="sparse_categorical_crossentropy", optimizer="nadam",
              metrics=["accuracy"])
model.fit((X_train, X_train_dec), Y_train, epochs=10,
          validation_data=((X_valid, X_valid_dec), Y_valid))

translate 함수 출력

print(translate('Thank you so much, my friend'))

이전보다 번역이 더 안 좋다.

---

빔 서치

 

빔 서치는 시퀀스 예측 문제에서 가능한 출력 후보들을 탐색하는 데 사용되는 효율적인 기법이다. 기존의 탐욕적 탐색(greedy search) 방식은 가장 높은 확률의 출력을 선택해 나가지만, 이는 최적의 결과를 보장하지 않는다. 빔 서치는 여러 후보 경로를 동시에 탐색하여 더 나은 출력 시퀀스를 찾을 가능성을 높인다.

• 빔 너비 (Beam Width): 동시에 유지할 후보 경로의 수를 의미한다. 빔 너비가 크면 더 많은 후보를 탐색하지만 계산 비용도 증가한다.
• 탐색 과정: 각 단계에서 현재까지의 후보 경로들을 확장하여 다음 단어에 대한 확률을 계산한다. 이 중에서 높은 확률의 후보를 유지하며 반복적으로 진행한다.

빔 서치의 주요 장점은 다음과 같다

• 높은 품질의 출력을 얻을 확률이 높다.
• 단순한 탐욕적 탐색보다 더 최적의 솔루션을 찾을 가능성이 높다.

k개의 가능성 있는 문장의 리스트를 유지하고 디코더 단계마다 이 문장의 단어를 하나씩 생성하여 가능성 있는 k개의 문장을 만든다. 파라미터 k를 빔 너비(beam width)라고 부른다.

def beam_search(sentence_en, beam_width, verbose=False):
    X = np.array([sentence_en])  # 인코더 입력
    X_dec = np.array(["startofseq"])  # 디코더 입력
    y_proba = model.predict((X, X_dec))[0, 0]  # 첫 번째 토큰의 확률
    top_k = tf.math.top_k(y_proba, k=beam_width)
    top_translations = [  # 촤상의 (log_proba, translation) 리스트 
        (np.log(word_proba), text_vec_layer_es.get_vocabulary()[word_id])
        for word_proba, word_id in zip(top_k.values, top_k.indices)
    ]
    
    # verbose 모드에서 상위 첫 단어 표시.
    if verbose:
        print("상위 첫 단어:", top_translations)

    for idx in range(1, max_length):
        candidates = []
        for log_proba, translation in top_translations:
            if translation.endswith("endofseq"):
                candidates.append((log_proba, translation))
                continue  # 번역이 완료되었으므로 번역을 이어가지 않음.
            X = np.array([sentence_en])  # 인코더 입력
            X_dec = np.array(["startofseq " + translation])  # 디코더 입력
            y_proba = model.predict((X, X_dec))[0, idx]  # 마지막 토큰의 확률
            for word_id, word_proba in enumerate(y_proba):
                word = text_vec_layer_es.get_vocabulary()[word_id]
                candidates.append((log_proba + np.log(word_proba),
                                   f"{translation} {word}"))
        top_translations = sorted(candidates, reverse=True)[:beam_width]

        # verbose 모드의 경우 지금까지의 최상의 번역을 출력.
        if verbose:
            print("지금까지 최상의 번역:", top_translations)

        if all([tr.endswith("endofseq") for _, tr in top_translations]):
            return top_translations[0][1].replace("endofseq", "").strip()

 

sentence_en = "I love cats and dogs"
translate(sentence_en)

beam_search(sentence_en, beam_width=3, verbose=True)

 

---

다음 내용

[딥러닝] RNN 자연어 처리: 어텐션 매커니즘, 트랜스포머

---

[출처]

핸즈 온 머신러닝