[딥러닝 with Python] LSTM (Long Short Term Memory)

[본 포스팅은 "만들면서 배우는 생성 AI 2판"을 참조로 작성했습니다]

 

이번에 알아볼 모형은 자기회귀 모델의 대표적인 모형인 LSTM입니다.

 

LSTM은 Long Short Term Memory의 줄임말로 기존의 순환 신경망(RNN)이 시퀀스(Sequence)가 긴 데이터에는 잘 맞지 않는다는 문제를 해결하기 위해 등장한 네트워크 입니다.

 

해당 LSTM은 시계열 예측, 감성 분석, 오디오 분류 등 순차 데이터와 관련된 다양한 문제에 적용되고 있는 여전히 실용성이 높은 모델 중 하나입니다.

 

 

 

LSTM에 대해서 알아보기 전, 텍스트 데이터와 이미지 데이터 간의 차이점에 대해서 알아보도록 하겠습니다.

 

 

텍스트 데이터와 이미지 데이터의 차이점

 

1) 텍스트 데이터는 개별적인 데이터 조각(문자나 단어)으로 구성되어있습니다. 하지만, 이미지의 픽셀의 경우 연속적인 색상 스펙트럼 위의 한 점입니다. 이로 인해 다음과 같은 차이가 생기는데요

 -  이미지의 녹색 픽셀을 파란색 픽셀에 가깝게 변형하기는 쉽지만, 단어 cat을 단어 dog에 가깝게 바꾸는 방법은 명확하게 정의되어 있지 않습니다.

 - 개별 픽셀에 대한 손실 함수의 그레이디언트를 계산하면 손실함수를 를 최소화하기 위해 픽셀의 색이 바뀌어야 할 방향을 알 수 있습니다. 하지만, 이산적인 텍스트 데이터의 경우에는 일반적인 방식으로 역전파를 적용할 수 없기때문에 다른 방법을 찾아야 한다는 문제가 있습니다.

 

2) 텍스트 데이터는 시간 차원이 있지만, 공간 차원은 없습니다. 반면 이미지 데이터는 시간차원이 없고 공간차원이 있습니다. 

 

3) 이미지 데이터에서는 픽셀간의 순서가 매우 중요하지는 않지만, 텍스트 데이터에서는 텍스트 간의 순서가 그 의미 차이를 바꿀 수 있기에 순서가 중요하게 다루어져야 합니다. 

 

4) 이미지 데이터는 픽셀의 일부가 바뀌더라도 전체적인 의미 전달에 있어서 큰 영향을 받지 않을 수 있지만, 텍스트 데이터에서는 단어의 일부가 바뀌게 된다면 그 의미가 완전히 바뀔 수도 있습니다.

 

5) 텍스트 데이터에서는 문법 규칙이 존재하지만, 이미지 데이터에서는 픽셀간의 정해진 규칙이 존재하지는 않습니다.

 

 

이와 같은 텍스트 데이터의 특성을 염두에 두고 텍스트 데이터 처리에 활용되는 LSTM에 대해서 알아보도록 하겠습니다. 

 

 

 

 

LSTM을 활용한 텍스트 처리의 순서는 일반적으로 다음과 같습니다. 

 

 

1. 토큰화(Tokenization) 

- 토큰화는 텍스트를 단어나 문자와 같은 개별 단위로 나누는 작업을 의미합니다. 

- 텍스트를 토큰화 하는 기준을 단어로 할 것인가, 문자로 할 것인가에 따라서 모델의 출력 결과가 영향을 받는데요

 

- 단어를 기준으로 토큰화 한 경우

 *  공백이나 구두점 등을 기준으로 단어를 분리함

 * 예 : 자연어 처리는 재미있다 => ["자연어", "처리", "는", "재미있다"]

 * 장점 : 대부분의 경우 단어는 의미를 갖고 있어서, 단어 단위로 나누면 의미 분석에 유리

 * 단점 : 언어마다 단어 구분 규칙이 다르고, 형태소 분석이 필요한 경우가 있어 복잡할 수 있음

- 문자를 기준으로 토큰화 한 경우

 * 텍스트를 문자 단위로 나눔

 * 모든 문자를 하나으 ㅣ토큰으로 간주

 * 예 : 자연어 처리 => ["자","연","어","처","리"]

 * 장점 : 단어 구분이 필요 없고, 언어 독립적으로 사용할 수 있음

 * 단점 : 단어의 의미를 파악하기 어렵고, 텍스트가 매우 길어질 수 있어 효율성이 떨어질 수 있음

 

2. 임베딩(Embedding) 

- 임베딩 층은 기본적으로 각 정수 토큰을 embedding 길이의 벡터로 변환하는 룩업 테이블을 말합니다.

- 여기서 말하는 룩업 벡터는 모델에 의해 학습되는 가중치를 말합니다. 

- 임베딩에서 활용하는 룩업 벡터로 이루어진 룩업 테이블의 예시는 아래와 같습니다.

 * 예시 문장 : "자연어 처리는 재미있다"

 * 단어 기준 토큰화 후 => ["자연어", "처리", "는", "재미있다"]   : 총 4개의 토큰

 * 임베딩 차원 : 2로 설정

 * 아래는 이에 따른 룩업 테이블 (각 세부 요소는 룩업 벡터)

* 위에서 보시는 룩업 벡터는 학습 가능한 파라미터(Learnable Parameter)이며, 총 학습 가능한 파라미터의 수(=룩업 벡터의 수)는              4(토큰의 개수) x 2(임베딩 차원) = 8 입니다.

 

 ** 입력 토큰을 원핫 인코딩으로 할 수도 있지만, 임베딩 층이 더 선호되는데요. 그 이유는 임베딩 층은 스스로 학습할 수 있도록 설정되어 있기때문에 성능을 높이기 위해 모델이 토큰의 임베딩 방법을 자유롭게 결정할 수 있기 때문입니다.

 

- 이러한 임베딩을 거치게 된다면, Input 층이 [batch_size, seq_length] 크기의 정수 시퀀스 텐서를 Embedding 층으로 전달하면 이 층은 [batch_size, seq_length, embedding_size] 크기의 텐서를 출력하게 됩니다.

 

 

** LSTM 층에 대해서 알아보기 전 RNN 층에 대해서 알아보겠습니다.

  위 그림에서 보시면 각 t 시간대의 연결된 Input이 하나의 Input Sequence라고 한다면 위와같은 방식으로 순환입력이 진행됩니다. (이전 Sequence들의 모델 출력결과가 그 다음 Sequence 토큰의 결과에 입력으로 들어옴)

  예 : 자연어(t=0), 처리(t=1), 는(t=2), 재미있다(t=3) 

 이와같은 RNN 네트워크의 문제는 시퀀스의 길이가 길어지면서 점차 초기의 입력값에 대한 정보가 소실된다는 것입니다.

 이를 해결하기 위해 나온것이 LSTM이 되겠습니다.

 

3. LSTM

 

- LSTM 셀은 이전 은닉 상태 h_t-1과 현재 단어 임베딩 x_t가 주어졌을 때 새로운 은닉 상태 h_t를 출력하는 네트워크입니다. h_t의 길이는 LSTM에 있는 유닛의 개수와 동일하며, 이는 해당 층을 정해야 하이퍼 파라미터입니다(이는 시퀀스의 길이와 상관 없습니다.)

 

  * 셀 상태(Cell State) : 위 그림에서 셀의 상단부에 직선으로 이어지는 네트워크의 출력 결과(상단 출력 결과)를 의미합니다. 이는 셀의 "기억" 상태를 나타내며, 긴 시퀀스 내에서도 정보를 오랜 기간 동안 유지할 수 있다는 특징이 있습니다. 

   셀 상태는 시간 단계 간에 정보를 전달하여, 중요한 정보가필요할 때까지 유지될 수 있도록 하며, 이는는 직접적인 수정을 받지 않고, 덧셈과 곱셈 연산으로만 업데이트 됩니다.

  셀 상태는 정보가 흐를 때의 그레이디언트 소실문제를 일으키는 sigomoid 또는 tanh 함수의 역할을 최소화 하기 위함입니다.

 

 * 은닉상태(Hidden State) : 위 그림의 하단 출력 결과를 의미합니다. 이는 LSTM 셀이 출력하는 상태로, 다음 LSTM 셀로 전달되며, 동시에 출력 값으로 사용될 수 있습니다.

 은닉 상태는 셀의 현재 출력으로 간주되며, 셀 상태의 일부 정보를 포함합니다. 또한, 다음 시간 단계로 정보를 전달하는 역할도 합니다.

 

 * LSTM 셀의 동작 과정은 아래와 같은 단계로 동작합니다.

 

 

이제 기본적인 자연어 처리 과정 및 LSTM에 대해서 알아본 내용을 바탕으로 실제 데이터를 가지고 구현해보겠습니다. 

 

 

 

이때 사용할 데이터는 레시피 데이터셋입니다. 

 

 

먼저 모델에 활용할 라이브러리를 로드합니다.

import numpy as np

import json

import re

import string

import tensorflow as tf

from tensorflow.keras import layers, models, callbacks, losses

 

 

다음은 모델 학습간 활용할 하이퍼 파라미터를 정의해줍니다.

VOCAB_SIZE = 10000

MAX_LEN = 200

EMBEDDING_DIM = 100

N_UNITS = 128

VALIDATION_SPLIT = 0.2

SEED = 42

LOAD_MODEL = False

BATCH_SIZE = 32

EPOCHS = 25

 

 

다음은 학습에 사용할데이터를 다운로드 받아줍니다.

import sys

# 코랩일 경우 노트북에서 celeba 데이터셋을 받습니다.

if 'google.colab' in sys.modules:

    # 캐글-->Setttings-->API-->Create New Token에서

    # kaggle.json 파일을 만들어 코랩에 업로드하세요.

    from google.colab import files

    files.upload()

    !mkdir ~/.kaggle

    !cp kaggle.json ~/.kaggle/

    !chmod 600 ~/.kaggle/kaggle.json

    # celeba 데이터셋을 다운로드하고 압축을 해제합니다.

    !kaggle datasets download -d hugodarwood/epirecipes

    !unzip -q epirecipes.zip

 

이후 데이터 셋을 로드해줍니다.

# 전체 데이터셋 로드

with open("./full_format_recipes.json") as json_data:

    recipe_data = json.load(json_data)

 

 

이제 데이터 셋 중 학습에 활용할 부분 (title, direction) 만을 필터링 해줍니다.

# 데이터셋 필터링

filtered_data = [

    "Recipe for " + x["title"] + " | " + " ".join(x["directions"])

    for x in recipe_data

    if "title" in x

    and x["title"] is not None

    and "directions" in x

    and x["directions"] is not None

]

 

이제 필터링된 레시피의 개수를 확인해봅니다.

# 레시피 개수 확인

n_recipes = len(filtered_data)

print(f"{n_recipes}개 레시피 로드")

 

example = filtered_data[9]

print(example)

 

 

이후 데이터를 단어 단위로토큰화 해줍니다. 

# 구두점을 분리하여 별도의 '단어'로 취급합니다.

def pad_punctuation(s):

    s = re.sub(f"([{string.punctuation}])", r" \1 ", s)

    s = re.sub(" +", " ", s)

    return s

text_data = [pad_punctuation(x) for x in filtered_data]

 

이제 필터링과 전처리된 데이터의 예시를 출력해보면 다음과 같게 나옴을 알 수 있습니다.

 

 

이제 텐서플로에서 활용할 수 있게 데이터를 변환해줍니다

# 텐서플로 데이터셋으로 변환하기

text_ds = (

    tf.data.Dataset.from_tensor_slices(text_data)

    .batch(BATCH_SIZE)

    .shuffle(1000)

)

 

 

이렇게 토큰화 후 텐서플로 형식으로 변환된 데이터를 벡터화 해줍니다.  벡터화 층을 만들어줍니다. 

# 벡터화 층 만들기

vectorize_layer = layers.TextVectorization(

    standardize="lower",

    max_tokens=VOCAB_SIZE,

    output_mode="int",

    output_sequence_length=MAX_LEN + 1,

)

 

Train 세트에 해당 층을 적용하고

# 훈련 세트에 층 적용

vectorize_layer.adapt(text_ds)

vocab = vectorize_layer.get_vocabulary()

 

벡터화된 토큰의 예시를 확인해봅니다.

# 토큰:단어 매핑 샘플 출력하기

for i, word in enumerate(vocab[:10]):

    print(f"{i}: {word}")

 

이제 샘플을 정수로 변화해서 토큰이 문장이라는 시퀀스가 어떻게 벡터화된 토큰의 연속으로 구성되는지 확인해봅니다.

# 동일 샘플을 정수로 변환하여 출력하기

example_tokenised = vectorize_layer(example_data)

print(example_tokenised.numpy())

 

 

이제 훈련세트를 만들어줍니다. 레시피 토큰(입력)과 동일하지만 한 토큰 이동된 벡터(타깃)로 구성된 훈련 세트를 ㅏㄴ듭니다. 

예를 들어, grilled with chicken with boiled 라는 토큰을 주입하면 한 토큰 이동된 단어(potatoes)를 타깃으로 선정하는 것입니다. 

# 레시피와 한 단어 이동한 동일 텍스트로 훈련 세트를 만듭니다.

def prepare_inputs(text):

    text = tf.expand_dims(text, -1)

    tokenized_sentences = vectorize_layer(text)

    x = tokenized_sentences[:, :-1]

    y = tokenized_sentences[:, 1:]

    return x, y

train_ds = text_ds.map(prepare_inputs)

 

 

이제 LSTM 모델을 만들어줍니다.

inputs = layers.Input(shape=(None,), dtype="int32")

x = layers.Embedding(VOCAB_SIZE, EMBEDDING_DIM)(inputs)

x = layers.LSTM(N_UNITS, return_sequences=True)(x)

outputs = layers.Dense(VOCAB_SIZE, activation="softmax")(x)

lstm = models.Model(inputs, outputs)

lstm.summary()

if LOAD_MODEL:

    # model.load_weights('./models/model')

    lstm = models.load_model("./models/lstm", compile=False)

 

 

이제 해당 모델의 loss와 옵티마이저를 설정해줍니다. 이때 loss sparse categorical cross entropy를 활용하며, 이는 레이블이 원핫 인코딩 된 벡터가 아닌 정수일 때 사용합니다. 

 

이후 각 훈련 에폭이 끝날 때 텍스트를 생성하는 콜백 함수를 만들어줍니다.

# TextGenerator 체크포인트 만들기

class TextGenerator(callbacks.Callback):

    def __init__(self, index_to_word, top_k=10):

        self.index_to_word = index_to_word

        self.word_to_index = {

            word: index for index, word in enumerate(index_to_word)

        }

    def sample_from(self, probs, temperature):

        probs = probs ** (1 / temperature)

        probs = probs / np.sum(probs)

        return np.random.choice(len(probs), p=probs), probs

    def generate(self, start_prompt, max_tokens, temperature):

        start_tokens = [

            self.word_to_index.get(x, 1) for x in start_prompt.split()

        ]

        sample_token = None

        info = []

        while len(start_tokens) < max_tokens and sample_token != 0:

            x = np.array([start_tokens])

            y = self.model.predict(x, verbose=0)

            sample_token, probs = self.sample_from(y[0][-1], temperature)

            info.append({"prompt": start_prompt, "word_probs": probs})

            start_tokens.append(sample_token)

            start_prompt = start_prompt + " " + self.index_to_word[sample_token]

        print(f"\n생성된 텍스트:\n{start_prompt}\n")

        return info

    def on_epoch_end(self, epoch, logs=None):

        self.generate("recipe for", max_tokens=100, temperature=1.0)

 

그리고 모델을 저장할 체크포인트 파일을 만들어 줍니다.

# 모델 저장 체크포인트 만들기

model_checkpoint_callback = callbacks.ModelCheckpoint(

    filepath="./checkpoint/checkpoint.ckpt",

    save_weights_only=True,

    save_freq="epoch",

    verbose=0,

)

tensorboard_callback = callbacks.TensorBoard(log_dir="./logs")

# 시작 프롬프트 토큰화

text_generator = TextGenerator(vocab)

 

이후 학습을 진행해줍니다.

lstm.fit(

    train_ds,

    epochs=EPOCHS,

    callbacks=[model_checkpoint_callback, tensorboard_callback, text_generator],

)

 

이제 학습된 모델을 저장해줍니다.

# 최종 모델 저장

lstm.save("./models/lstm")

 

 

 

이제 학습된 모델을 사용해 텍스트를 생성해봅니다.

def print_probs(info, vocab, top_k=5):

    for i in info:

        print(f"\n프롬프트: {i['prompt']}")

        word_probs = i["word_probs"]

        p_sorted = np.sort(word_probs)[::-1][:top_k]

        i_sorted = np.argsort(word_probs)[::-1][:top_k]

        for p, i in zip(p_sorted, i_sorted):

            print(f"{vocab[i]}:   \t{np.round(100*p,2)}%")

        print("--------\n")

 

아래와 같은 recipe for roasted vegetables l chop 1 이라는 입력을 넣어주고 나오는 한개의 단어를 예측해봅니다.

 

이때 temperature는 샘플링 과정을 얼마나 결정적으로 만들어줄지 결정해주는 변수로 0에 가까울소록 샘플링이 더 결정적으로, 1에 가까울수록  모델에서 출력되는 확률로 모델을 선택하게 됩니다.

info = text_generator.generate(

    "recipe for roasted vegetables | chop 1 /", max_tokens=10, temperature=1.0

)

모델에 의해 생성된 결과는 "of" 입니다.

 

 

프롬프트에 따른 결과를 보면 

print_probs(info, vocab)

 

각 프롬프트에 따라서 아래와 같이 각 벡터들이 나올 확률들이 제공됩니다.

 

이번에는 조금더 결정적이게 만들어보기 위해 temperature를 0.2로 선정해서 결과를 만들어보겠습니다.

info = text_generator.generate(

    "recipe for roasted vegetables | chop 1 /", max_tokens=10, temperature=0.2

)

print_probs(info, vocab)