LSTM - 딥러닝

Long Short-Term Memory (LSTM, 1997)

1순환 신경망의 한계와 LSTM¶

순환 신경망(RNN)은 과거의 정보가 현재에 영향을 줄 수 있도록 설계되어 있어 시퀀스 데이터를 다루는 데 유용한 구조입니다. 그러나 실제 학습 과정에서는 기울기 소실과 기울기 폭발이라는 중요한 한계가 드러납니다.

신경망은 정답과 출력의 차이를 줄이기 위해 역전파를 통해 각 연결의 가중치를 조정하는데, 이 과정은 여러 단계의 미분값을 계속 곱하는 연산으로 이루어집니다. 작은 수를 반복해서 곱하면 값이 점점 작아지고, 큰 수를 반복해서 곱하면 점점 커집니다. 그래서 어떤 경우에는 기울기가 너무 작아져 앞쪽 정보가 거의 전달되지 않고, 반대로 어떤 경우에는 기울기가 지나치게 커져 학습이 불안정해집니다. 결과적으로 RNN은 시퀀스가 길어질수록 먼 과거의 정보를 제대로 학습하지 못합니다.

이 문제를 해결하기 위해 1997년 호흐라이터(Hochreiter)와 슈미트후버(Schmidhuber)는 LSTM(Long Short-Term Memory)을 제안했습니다. 이름 그대로 오래 기억할 정보와 짧게 기억할 정보를 구분해서 다룰 수 있는 구조입니다.

LSTM의 핵심은 셀 상태(cell state)라는 정보를 별도로 흐르게 한다는 점입니다. 셀 상태는 시간의 흐름을 따라 이동하면서 필요에 따라 정보를 추가하거나 제거합니다. 이때 정보를 조절하는 장치가 게이트(gate)이며, LSTM에는 세 가지 게이트가 있습니다.

망각 게이트(forget gate): 지금까지 기억해온 것 중에서 무엇을 잊을지를 결정합니다.
입력 게이트(input gate): 새로 들어온 정보 중에서 어떤 것을 기억할지 정합니다.
출력 게이트(output gate): 현재 상태에서 어떤 정보를 바깥으로 내보낼지를 결정합니다.

세 게이트가 협력하면서 셀 상태를 갱신하고 정보의 흐름을 정교하게 조절합니다. 그 결과 오래된 정보도 잃지 않고 안정적으로 유지하면서 불필요한 정보는 효과적으로 제거할 수 있습니다. 이러한 구조 덕분에 LSTM은 기계 번역, 음성 인식, 감정 분석, 시계열 예측 등 긴 문장이나 긴 시간에 걸친 데이터를 다루어야 하는 문제에서 강한 성능을 보입니다. LSTM은 단순히 정보를 저장하는 것이 아니라, 어떤 정보를 얼마나 오래 기억할지 그리고 언제 지울지를 스스로 판단하는 구조라는 점에서 시계열 데이터 학습의 전환점을 만든 모델입니다.

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import sklearn
import keras

2Keras API¶

import keras
from keras import layers

시퀀스길이, 특성수, 출력수 = 4, 5, 6

inputs = keras.Input(shape=(시퀀스길이, 특성수))
rnn = layers.SimpleRNN(출력수, activation='tanh')
lstm = layers.LSTM(출력수, activation='tanh')

x = rnn(inputs)
print(x.shape)
weights = rnn.get_weights()
W, U, b = weights
print(W.shape, U.shape, b.shape)

x = lstm(inputs)
print(x.shape)
weights = lstm.get_weights()
W, U, b = weights
print(W.shape, U.shape, b.shape)

(None, 6)
(5, 6) (6, 6) (6,)
(None, 6)
(5, 24) (6, 24) (24,)

1949년 1월부터 1960년 12월까지 국제 노선 항공기 이용객수 (단위: 천 명)

출처: DataMarket

data = pd.read_csv(
    '../data/international-airline-passengers.csv', 
    index_col=0, engine='python', skipfooter=3)
data = data.rename(columns=lambda title: title.split(':')[0])
data[:5]

data.plot()

from sklearn.model_selection import train_test_split
from tensorflow.keras.utils import timeseries_dataset_from_array

def prepare_dataset(data, scaler, sequence_length, batch_size):
    # split data
    train_data, test_data = train_test_split(data, test_size=0.2, shuffle=False)
    # scaling
    train_data = scaler.fit_transform(train_data)
    test_data = scaler.transform(test_data)
    # time series dataset
    delay = sequence_length
    train_dataset = timeseries_dataset_from_array(
        data=train_data[:-delay], targets=train_data[delay:], sequence_length=sequence_length, batch_size=batch_size)
    test_dataset = timeseries_dataset_from_array(
        data=test_data[:-delay], targets=test_data[delay:], sequence_length=sequence_length, batch_size=batch_size)
    return train_dataset, test_dataset

from sklearn.metrics import r2_score

def evaluate(model, train_dataset, test_dataset):
    # get all samples and targets from dataset
    train_samples = np.concatenate([samples.numpy() for samples, targets in train_dataset], axis=0)
    train_targets = np.concatenate([targets.numpy() for samples, targets in train_dataset], axis=0)
    test_samples = np.concatenate([samples.numpy() for samples, targets in test_dataset], axis=0)
    test_targets = np.concatenate([targets.numpy() for samples, targets in test_dataset], axis=0)
    # evaluate
    train_loss = model.evaluate(train_dataset, verbose=False)
    test_loss = model.evaluate(test_dataset, verbose=False)
    # predict
    y_pred = {}
    y_pred['train'] = model.predict(train_dataset)
    y_pred['test'] = model.predict(test_dataset)
    print(f'train loss: {train_loss:.4f}, test loss: {test_loss:.4f}')
    print(f'R2 (train): {r2_score(train_targets, y_pred["train"]):.4f}, R2 (test): {r2_score(test_targets, y_pred["test"]):.4f}')

    y = np.concatenate([train_targets, test_targets], axis=0)
    x_train = np.arange(len(train_samples))
    x_test = np.arange(len(train_samples), len(train_samples) + len(test_samples))
    plt.plot(scaler.inverse_transform(y))
    plt.plot(x_train, scaler.inverse_transform(y_pred['train']))
    plt.plot(x_test, scaler.inverse_transform(y_pred['test']))

from tensorflow.keras import Sequential
from tensorflow.keras.layers import LSTM, Dense

def create_model(input_shape):
    model = Sequential([
        keras.Input(shape=input_shape),
        LSTM(4),
        Dense(1)
    ])
    return model

create_model((None, 1)).summary()

from sklearn.preprocessing import MinMaxScaler

n_features = 1
sequence_length = 1
batch_size = 1

scaler = MinMaxScaler()
train_dataset, test_dataset = prepare_dataset(data, scaler, sequence_length=sequence_length, batch_size=batch_size)
for samples, targets in train_dataset.take(1):
    print(f'X.shape={samples.shape}, y.shape={targets.shape}')
model = create_model(input_shape=(sequence_length, n_features))
model.summary()

model.compile(loss='mean_squared_error', optimizer='adam')
model.fit(train_dataset, epochs=100)

evaluate(model, train_dataset, test_dataset)

3Window 기법¶

x_{t-2}, x_{t-1}, x_{t} \rightarrow x_{t+1} = y

(1)

from sklearn.preprocessing import MinMaxScaler

n_features = 1
sequence_length = 3
batch_size = 1

scaler = MinMaxScaler()
train_dataset, test_dataset = prepare_dataset(data, scaler, sequence_length=sequence_length, batch_size=batch_size)
for samples, targets in train_dataset.take(1):
    print(samples.shape, targets.shape)
model = create_model(input_shape=(sequence_length, n_features))
model.summary()

model.compile(loss='mean_squared_error', optimizer='adam')
model.fit(train_dataset, epochs=100)

evaluate(model, train_dataset, test_dataset)

4GRU (Gated Recurrent Unit)¶

GRU는 LSTM과 같은 문제의식에서 출발했지만 더 가벼운 순환 유닛입니다. LSTM에 비해 연산과 구현이 훨씬 간단한 것이 장점입니다.

GRU의 핵심은 은닉 상태를 조건부로 갱신하는 방식에 있습니다. 입력으로부터 만들어진 새로운 은닉 상태를 얼마나 반영할지는 갱신 게이트(update gate) $z$ 가 결정하고, 이전 은닉 상태를 얼마나 반영할지는 리셋 게이트(reset gate) $r$ 가 결정합니다. 두 게이트만으로 정보의 유지와 갱신을 조절하기 때문에, GRU는 LSTM보다 단순한 구조로도 장기 의존성을 비교적 안정적으로 학습할 수 있습니다.