[인공지능개론] RNN②

RNN은 Recurrent Neural Network로, sequential data를 처리하기에 적합하다. 각 시점마다 입력을 받아 각 시점에 대한 입력과 hidden state로 출력을 계산하고 이는 현시점의 hidden state가 된다.

RNN을 Pytorch로 구현해보자.

먼저, 'hihello' 예제를 진행해보자.

import torch
import torchvision.datasets as dsets
import torchvision.transforms as transforms
import torch.nn.init

device = 'cuda' if torch.cuda.is_available() else 'cpu'

 

RNN은 torch.nn.Module에서 nn.RNN()만으로 아주 쉽게 구현할 수 있다.

class Net(torch.nn.Module):
  def __init__(self,input_size, hidden_size, output_size):
    super(Net, self).__init__()
    self.rnn = torch.nn.RNN(input_size, hidden_size, batch_first=True)
    self.fc = torch.nn.Linear(hidden_size, output_size, bias = True)

  def forward(self, x): # 구현한 RNN 셀과 출력층을 연결
    x, _status = self.rnn(x)
    x = self.fc(x)
    return x

'hihello' 예제는 char이 순서대로 출력되도록 하는 것이다.

h → i

i → h

h → e

e → l

l → l

l → o

→의 좌측은 input, 우측은 label이 된다. 여기서 각 고유한 char인 e,h,i,l,o를 one hot encoding하면 순서대로 0,1,2,3,4가 할당되어 아래와 같이 나타난다.

e : [1,0,0,0,0]

h : [0,1,0,0,0]

i : [0,0,1,0,0]

l : [0,0,0,1,0]

o : [0,0,0,0,1]

char_vocab은 input_str과 label_str에 포함된 고유한 문자들의 집합을 정렬된 리스트로 만든다. vocab_size는 문자 집합의 크기로, 고유한 문자들의 수를 의미한다. 이 예제에서는 e,h,i,l,o 5가지의 고유한 문자들이 존재하므로 vocab_size는 5이다.

input_str = 'hihell'
label_str = 'ihello'
char_vocab = sorted(list(set(input_str+label_str)))
vocab_size = len(char_vocab)
print('문자 집합의 크기 : {}'.format(vocab_size))

input_size = vocab_size # 입력의 크기 = 문자 집합의 크기
hidden_size = 5
output_size = 5
learning_rate = 0.1

char_to_index = dict((c,i) for i,c in enumerate(char_vocab))
print(char_to_index)

index_to_char = {}
for key, value in char_to_index.items():
  index_to_char[value] = key
print(index_to_char)

char_to_index로 각 char를 고유한 index로 변환하고, index_to_char로 각 index를 고유한 char로 변환하는 dictionary를 정의한다.

input str(x_data)과 label str(y_data)를 위의 dictionary를 이용하여 index로 변환하여 list 형태로 저장한다. 이를 list 안에 넣으면 batch 형태로 변환하여 2차원 list로 만들 수 있다. 이를 x_one_hot으로 각 index를 one hot encoding하여 list로 저장하여 np.eye(vocab_size)는 vocab_size의 단위 행렬을 생성하면 RNN 모델의 입력으로 사용할 수 있다.

x_data = [char_to_index[c] for c in input_str]
y_data = [char_to_index[c] for c in label_str]
print(x_data)
print(y_data)

x_data = [x_data]
y_data = [y_data]
print(x_data)
print(y_data)

x_one_hot = [np.eye(vocab_size)[x] for x in x_data]
print(x_one_hot)

X = torch.FloatTensor(x_one_hot)
Y = torch.LongTensor(y_data)
print('훈련 데이터의 크기 : {}'.format(X.shape))
print('레이블의 크기 : {}'.format(Y.shape))

CNN의 예제에서와 동일하게

loss function은 Cross Entropy, optimizer는 Adam을 사용한다. 

net = Net(input_size,hidden_size,output_size)
outputs = net(X)

criterion = torch.nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(net.parameters(), learning_rate)

위에서 정의해둔 dictionary index_to_char을 이용하여 예측된 결과를 다시 str 형태로 출력할 수 있다.

for i in range(100):
  optimizer.zero_grad()
  outputs = net(X)
  loss = criterion(outputs.view(-1,input_size), Y.view(-1))
  loss.backward()
  optimizer.step()

  result = outputs.data.numpy().argmax(axis=2)
  result_str = ''.join([index_to_char[c] for c in np.squeeze(result)])
  print(i, "loss: ", loss.item(), "prediction: ", result, "true Y: ", y_data, "prediction str: ", result_str)

 

이번에는 긴 문장을 학습시키는 예제를 진행해보자.

0 if you wan > f you want

1 f you want > you want

2 you want > you want t

3 you want t > ou want to

...

168 of the se > of the sea

169 of the sea > f the sea.

길이가 길어졌다는 것을 제외하고는 앞선 'hihello' 예제와 거의 동일하다. char_to_index, index_to_char 대신 char_dic, char_set을 사용한다. 또한, input str(x_data)와 label str(y_data)를 다루는 단계 이전에 x_str과 y_str을 통해 긴 sentence를 적당한 sequence length로 끊어준다. 이를 char_dic을 이용하여 index로 변환하여 저장하는 작업을 하면 앞의 예제와 완벽하게 동일해졌다.

sentence = ("if you want to build a ship, don't drum up people together to "
            "collect wood and don't assign them tasks and work, but rather "
            "teach them to long for the endless immensity of the sea.")

char_set = list(set(sentence))
char_dic = {c: i for i, c in enumerate(char_set)}

print(char_dic)
dic_size = len(char_dic)
print('문자 집합의 크기 : {}'.format(dic_size))

hidden_size = dic_size
sequence_length = 10
learning_rate = 0.1

x_data = []
y_data = []

for i in range(0,len(sentence) - sequence_length):
  x_str = sentence[i:i+sequence_length]
  y_str = sentence[i+1:i+sequence_length+1]
  print(i, x_str,'->',y_str)

  x_data.append([char_dic[c] for c in x_str])
  y_data.append([char_dic[c] for c in y_str])

print(x_data[0])
print(y_data[0])

x_one_hot = [np.eye(dic_size)[x] for x in x_data]
X = torch.FloatTensor(x_one_hot)
Y = torch.LongTensor(y_data)

print('훈련 데이터의 크기 : {}'.format(X.shape))
print('레이블의 크기 : {}'.format(Y.shape))

class Net(torch.nn.Module):
  def __init__(self, input_dim, hidden_dim, layers):
    super(Net, self).__init__()
    self.rnn = torch.nn.RNN(input_dim, hidden_dim, num_layers=layers, batch_first=True)
    self.fc = torch.nn.Linear(hidden_dim, hidden_dim, bias = True)

  def forward(self, x):
    x, _status = self.rnn(x)
    x = self.fc(x)
    return x

net = Net(dic_size, hidden_size, 2)
criterion = torch.nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(net.parameters(), learning_rate)

outputs = net(X)

for i in range(100):
  optimizer.zero_grad()
  outputs = net(X)
  loss = criterion(outputs.view(-1,dic_size), Y.view(-1))
  loss.backward()
  optimizer.step()

  results = outputs.argmax(dim=2)
  predict_str = ''
  for j,result in enumerate(results):
    if j == 0:
      predict_str += ''.join([char_set[t] for t in result])
    else:
      predict_str += char_set[result[-1]]
  print(predict_str)

 

[출처] 한양대학교 장준혁 교수님 인공지능개론 수업