RNN(Recurrent Neural Network) 정리

RNN 기본 개념

• RNN은 입력과 출력을 시퀀스 단위로 처리하는 시퀀스(Sequence) 모델이다.
  
  • 시퀀스(Sequence): 배열. 순서를 갖는 값들의 모임을 의미한다.
  • ex) 번역기 - 입력: 번역하고자 하는 단어의 시퀀스인 문장, 출력: 번역된 단어의 시퀀스인 문장
• MLP, CNN 등의 신경망들은 전부 은닉층에서 활성화 함수를 지난 값은 오직 출력층으로만 향하는 피드 포워드 신경망(Feed Forward Neural Network)이다.
• RNN(Recurrent Neural Network)는 은닉층의 노드에서 활성화 함수를 통해 나온 결과값을 출력층 방향으로도 보내면서, 다시 은닉층 노드의 다음 계산의 입력으로 보내므로 순환(Recurrent)의 의미를 담고 있다.

출처: 위키독스(wikidocs)

 

• RNN에서 은닉층에서 활성화 함수를 통해 결과를 내보내는 역할을 하는 노드를 셀(cell)이라고 한다.
• 이 셀은 이전의 값을 기억하려고 하는 일종의 메모리 역할을 수행하므로 이를 메모리 셀 또는 RNN 셀이라고 표현한다.
• 메모리 셀이 출력층 방향 또는 다음 시점인 t+1의 자신에게 보내는 값을 은닉 상태(hidden state)라고 한다. 
• RNN은 입력과 출력의 길이를 다르게 설계할 수 있으므로 일 대 다(one-to-many), 다 대 일(many-to-one), 다 대 다(many-to-many) 등 다양한 형태로 사용할 수 있다.

출처: 위키독스(wikidocs)

 

• 1) 일 대 다(one-to-many): 하나의 이미지 입력에 대해서 사진의 제목을 출력하는 이미지 캡셔닝(image captioning)
• 2) 다 대 일(many-to-one): 입력 문서가 긍정적인지 부정적인지를 판별하는 감성 분류(sentiment classification), 스팸 메일 분류(spam detection)
• 3) 다 대 다(many-to-many): 사용자가 문장을 입력하면 대답 문장을 출력하는 챗봇, 입력 문장으로부터 번역된 문장을 출력하는 번역기, 개체명 인식, 품사 태깅

출처: 위키독스(wikidocs)

 

RNN 연산 과정

출처: 위키독스(wikidocs)

• 은닉층

 

• 출력층 (f: 활성화 함수)

 

RNN 필기 내용 정리

 

RNN Pytorch 구현

import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
import torch
import torch.nn as nn
import torch.optim as optim
import matplotlib.pyplot as plt
 
# 데이터 불러오기
df = pd.read_csv("kospi.csv")
 
# 데이터 스케일링
scaler = MinMaxScaler()
df[['Open', 'High', 'Low', 'Volume']] = scaler.fit_transform(df[['Open', 'High', 'Low', 'Close', 'Volume']])
 
# 독립변수, 종속변수 지정
X = df[['Open', 'High', 'Low', 'Volume']].values
y = df['Close'].values
 
# 독립변수와 종속변수를 입력받으면 시퀀스 길이만큼 잘라서 x_seq에 저장, 예측할 값은 y_seq에 저장
def seq_data(x, y, sequence_length):
    x_seq = []
    y_seq = []
    for i in range(len(x)-sequence_length):
        x_seq.append(x[i:i+sequence_length])
        y_seq.append(y[i+sequence_length])  
     
    return torch.FloatTensor(x_seq).to(device), torch.FloatTensor(y_seq).to(device).view([-1, 1])
 
split = 200
sequence_length = 5
x_seq, y_seq = seq_data(X, y, sequence_length)
 
x_train_seq = x_seq[:split]
y_train_seq = y_seq[:split]
 
x_test_seq = x_seq[split:]
y_test_seq = y_seq[split:]
 
train = torch.utils.data.TensorDataset(x_train_seq, y_train_seq)
test = torch.utils.data.TensorDataset(x_test_seq, y_test_seq)
 
# 배치 사이즈를 20으로 지정하여 저장
batch_size = 20
train_loader = torch.utils.data.DataLoader(dataset=train, batch_size=batch_size, shuffle=True)
test_loader = torch.utils.data.DataLoader(dataset=test, batch_size=batch_size, shuffle=False)
 
# RNN 하이퍼파라미터 설정
input_size = x_seq.size(2)                  # size = (426, 5, 4) = (데이터 개수, 시퀀스 길이, 피처 개수)
# 순환 레이어의 개수
num_layers = 2
# 은닉층 피처 개수
hidden_size = 8
 
# 모델 설정
class VanillaRNN(nn.Module):
     
    def __init__(self, input_size, hidden_size, sequence_length, num_layers, device):
        super(VanillaRNN, self).__init__()
        self.device =device
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        self.rnn = nn.RNN(input_size, hidden_size, num_layers, batch_first=True)     # batch 사이즈를 처음으로 가게 만듦.
        self.fc = nn.Sequential(nn.Linear(hidden_size*sequence_length, 1), nn.Sigmoid())
         
    def forward(self, x):
        h0 = torch.zeros(self.num_layers, x.size()[0], self.hidden_size).to(self.device)      # 초기 hidden state 설정
        out, _ = self.rnn(x, h0)                                              # out: RNN의 마지막 레이어로 부터 나온 output feature 반환, hn: hidden state 반환
        out = out.reshape(out.shape[0], -1)                                   # many to many 전략. out을 일렬로 피는 과정
        out = self.fc(out)                                                    # output layer의 값을 linear에 통과시키고 sigmoid 함수를 적용.
        return out
 
# 모델 선언
model = VanillaRNN(input_size=input_size,
                   hidden_size=hidden_size,
                   sequence_length=sequence_length,
                   num_layers=num_layers,
                   device=device).to(device)
 
# 비용함수, 학습률, 학습 횟수, 옵티마이저 설정
criterion = nn.MSELoss()    # 실제 target 값과 예측한 모델의 거리 제곱의 합(mean squared error)
lr = 1e-3
num_epochs = 200
optimizer = optim.Adam(model.parameters(), lr=lr)
 
# 학습 진행
loss_graph = []
n = len(train_loader)
 
for epoch in range(num_epochs):
    running_loss = 0.0
     
    for data in train_loader:
         
        seq, target = data   # 배치 데이터
        out = model(seq)
        loss = criterion(out, target)
         
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        running_loss += loss.item()
     
    loss_graph.append(running_loss/n)
    if epoch % 100 == 0:
        print('[epoch: %d] loss: %.4f' %(epoch, running_loss/n))