본 예제에서는 RNN으로 성씨를 생성하는 간단한 시퀀스 예측 task를 만듭니다. 각 타임 스텝에 대해 성씨에 포함될 수 있는 문자 집합에 대한 확률 분포를 계산하여, 예측을 향상시키거나 새로운 성씨를 생성해낼 수 있습니다.
이번 실습 예제에서는 조건이 없는 SurnameGenerationModel과 RNN 초기 은닉 상태에 특정 국적을 임베딩하여 활용해 편향성을 준, 조건이 있는 SurnameGenerationModel을 만들어 적용해보겠습니다.
1. SurnameDataset 클래스
이번에 활용할 데이터와 Dataset 클래스는 이전에 수행한 실습인 6-2 RNN 실습 : 성씨 국적 분류 (1) 에서 활용한 Dataset 클래스와 매우 유사합니다. 기존의 Dataset과 달라진 점은 __getitem__()에서 vector 정보(surname_vector, nationality_index, vector_length)가 아닌 예측 타겟에 대한 정수 시퀀스, 출력 정수 시퀀스(from_vector, to_vector)를 출력한다는 점입니다.
Dataset 소스코드
2. 벡터 변환 클래스
이전과 마찬가지로 3가지 주요 클래스인 Vocabulary, Vectorizer, DataLoader를 사용합니다.
1.
SequenceVocabulary에서는 개별 문자 토큰들을 정수로 매핑하는 작업을 수행합니다.
2.
SurnameVectorizer에서는 위에서 매핑한 숫자값으로 벡터화를 진행합니다.
3.
DataLoader에서는 Vectorizer에서 만들어진 벡터들을 미니배치로 만들어줍니다.
SequenceVocabulary와 DataLoader는 앞서 6-2 RNN 실습 : 성씨 국적 분류 (1) 에서 다룬 예제와 동일한 소스코드를 활용하므로, 소스코드만 첨부해두겠습니다.
SequenceVocabulary 소스코드
DataLoader 소스코드
SurnameVectorizer
시퀀스 예측 task에서는 각 타임 스텝(입력 차례)마다 토큰 샘플과 토큰 타깃에 대한 정수 시퀀스 2개를 입력으로 받습니다. 주로 하나의 토큰 시퀀스에 대해 토큰을 하나씩 엇갈리게 하여 샘플과 타깃을 구성합니다. 이 과정은 다음과 같습니다.
1.
SequenceVocabulary에서 토큰을 적절한 인덱스에 매핑하기
2.
BEGIN-OF-SEQUENCE, END-OF-SEQUENCE 토큰에 해당하는 인덱스 번호를 시퀀스 앞 뒤에 추가하기
(이제 모든 데이터는 첫번째와 마지막 인덱스가 동일한 시퀀스가 됩니다)
3.
토큰 시퀀스의 마지막을 제외한 모든 시퀀스 토큰을 포함하도록 잘라 입력 시퀀스 만들기
입력 시퀀스에 포함되지 않고 제외된 부분은 mask 토큰의 인덱스로 채웁니다
4.
토큰 시퀀스의 첫번째를 제외환 모든 시퀀스 토큰을 포함하도록 잘라 출력 시퀀스 만들기
출력 시퀀스에 포함되지 않고 제외된 부분은 mask 토큰의 인덱스로 채웁니다
위 과정을 SurnameVectorizer의 vectorizer() 메서드에서 수행하게 됩니다.
def vectorize(self, surname, vector_length=-1):
""" 성씨를 샘플과 타깃 벡터로 변환합니다
성씨 벡터를 두 개의 벡터 surname[:-1]와 surname[1:]로 나누어 출력합니다.
각 타임스텝에서 첫 번째 벡터가 샘플이고 두 번째 벡터가 타깃입니다.
매개변수:
surname (str): 벡터로 변경할 성씨
vector_length (int): 인덱스 벡터의 길이를 맞추기 위한 매개변수
반환값:
튜플: (from_vector, to_vector)
from_vector (numpy.ndarray): 샘플 벡터
to_vector (numpy.ndarray): 타깃 벡터 vector
"""
# 시퀀스의 앞 뒤에 BEGIN index와 END index를 붙여줍니다
# Vocabulary에서 토큰에 매핑된 index를 찾아 넣어줍니다
indices = [self.char_vocab.begin_seq_index]
indices.extend(self.char_vocab.lookup_token(token) for token in surname)
indices.append(self.char_vocab.end_seq_index)
if vector_length < 0:
vector_length = len(indices) - 1
# 마지막 시퀀스를 제외한 모든 시퀀스를 포함하도록 잘라 입력 시퀀스를 생성합니다
# 포함되지 않은 시퀀스는 mask 토큰의 index로 가려집니다
from_vector = np.empty(vector_length, dtype=np.int64)
from_indices = indices[:-1]
from_vector[:len(from_indices)] = from_indices
from_vector[len(from_indices):] = self.char_vocab.mask_index
# 첫번째 시퀀스를 제외한 모든 시퀀스를 포함하도록 잘라 출력 시퀀스를 생성합니다
# 포함되지 않은 시퀀스는 mask 토큰의 index로 가려집니다
to_vector = np.empty(vector_length, dtype=np.int64)
to_indices = indices[1:]
to_vector[:len(to_indices)] = to_indices
to_vector[len(to_indices):] = self.char_vocab.mask_index
# 생성한 시퀀스 쌍 - 입력 시퀀스와 출력 시퀀스 - 를 반환합니다
return from_vector, to_vector
Python
복사
SurnameVectorizer 소스코드
3. 모델 1) 조건 없는 SurnameGenerationModel
뒤에서 다룰 모델 2는 은닉 상태(은닉 벡터)에 미리 값을 저장해 편향된 계산을 하도록 유도하지만, 본 모델에서는 초기값을 0으로 설정해 은닉 상태의 영향력을 없애고 시작합니다.
여기에서 7-2 게이팅 : LSTM, GRU 에서 다뤘던 GRU를 self.rnn에 넣어주며 모델 구성에 활용하는 것을 아래 코드에서 확인할 수 있습니다. 구성 요소를 바꾸는 작업은 이처럼 크게 어렵지 않으며, LSTM 역시 비슷한 방식으로 모델의 구성요소로 넣어줄 수 있습니다.
__init__ 함수에서는 본 모델의 임베딩 층, GRU, Linear층을 초기화해줍니다. 임베딩 층은 정수를 3차원의 텐서로 변환시켜주고, 이 텐서가 GRU를 통과하며 상태 벡터가 연산되게 됩니다.
아래 코드의 forward 함수에서 볼 수 있듯이, 문자 시퀀스를 받아와 임베딩하여 rnn(GRU)를 통해 상태를 순차적으로 계산합니다. 이후, linear층(fc)에서 예측 확률을 계산하게 됩니다.
class SurnameGenerationModel(nn.Module):
def __init__(self, char_embedding_size, char_vocab_size, rnn_hidden_size,
batch_first=True, padding_idx=0, dropout_p=0.5):
"""
매개변수:
char_embedding_size (int): 문자 임베딩 크기
char_vocab_size (int): 임베딩될 문자 개수
rnn_hidden_size (int): RNN의 은닉 상태 크기
batch_first (bool): 0번째 차원이 배치인지 시퀀스인지 나타내는 플래그
padding_idx (int): 텐서 패딩을 위한 인덱스;
torch.nn.Embedding를 참고하세요
dropout_p (float): 드롭아웃으로 활성화 출력을 0으로 만들 확률
"""
super(SurnameGenerationModel, self).__init__()
self.char_emb = nn.Embedding(num_embeddings=char_vocab_size,
embedding_dim=char_embedding_size,
padding_idx=padding_idx)
self.rnn = nn.GRU(input_size=char_embedding_size,
hidden_size=rnn_hidden_size,
batch_first=batch_first)
self.fc = nn.Linear(in_features=rnn_hidden_size,
out_features=char_vocab_size)
self._dropout_p = dropout_p
def forward(self, x_in, apply_softmax=False):
"""모델의 정방향 계산
매개변수:
x_in (torch.Tensor): 입력 데이터 텐서
x_in.shape는 (batch, input_dim)입니다.
apply_softmax (bool): 소프트맥스 활성화를 위한 플래그로 훈련시에는 False가 되어야 합니다.
반환값:
결과 텐서. tensor.shape는 (batch, char_vocab_size)입니다.
"""
x_embedded = self.char_emb(x_in)
y_out, _ = self.rnn(x_embedded)
batch_size, seq_size, feat_size = y_out.shape
y_out = y_out.contiguous().view(batch_size * seq_size, feat_size)
y_out = self.fc(F.dropout(y_out, p=self._dropout_p))
if apply_softmax:
y_out = F.softmax(y_out, dim=1)
new_feat_size = y_out.shape[-1]
y_out = y_out.view(batch_size, seq_size, new_feat_size)
return y_out
Python
복사
4. 모델 2) 조건 있는 SurnameGenerationModel
이번에는 성씨를 생성하는 과정에서 국적을 고려하도록 모델을 형성합니다. 즉, 은닉 상태에 국적을 임베딩하여 RNN의 초기 은닉 상태를 만들어주어, 성씨와 국적 사이의 규칙에 조금 더 민감하게 반응하도록 만들어줍니다.
아래의 코드와 위 조건 없는 모델의 코드의 다른 점은 국적 인덱스를 매핑하는 임베딩 층 nation_emb이 추가되었다는 점입니다. 국적 인덱스를 RNN의 은닉 층과 같은 크기의 벡터로 매핑하고, forward 과정에서 RNN의 초기 은닉 상태로 전달됩니다. 아래 소스코드에서는 들여쓰기 없는 주석으로 추가된 부분을 표시해두었습니다.
class SurnameGenerationModel(nn.Module):
def __init__(self, char_embedding_size, char_vocab_size, num_nationalities,
rnn_hidden_size, batch_first=True, padding_idx=0, dropout_p=0.5):
"""
매개변수:
char_embedding_size (int): 문자 임베딩 크기
char_vocab_size (int): 임베딩될 문자 개수
rnn_hidden_size (int): RNN의 은닉 상태 크기
batch_first (bool): 0번째 차원이 배치인지 시퀀스인지 나타내는 플래그
padding_idx (int): 텐서 패딩을 위한 인덱스;
torch.nn.Embedding를 참고하세요
dropout_p (float): 드롭아웃으로 활성화 출력을 0으로 만들 확률
"""
super(SurnameGenerationModel, self).__init__()
self.char_emb = nn.Embedding(num_embeddings=char_vocab_size,
embedding_dim=char_embedding_size,
padding_idx=padding_idx)
# 국적 임베딩 층 추가
self.nation_emb = nn.Embedding(num_embeddings=num_nationalities,
embedding_dim=rnn_hidden_size)
self.rnn = nn.GRU(input_size=char_embedding_size,
hidden_size=rnn_hidden_size,
batch_first=batch_first)
self.fc = nn.Linear(in_features=rnn_hidden_size,
out_features=char_vocab_size)
self._dropout_p = dropout_p
def forward(self, x_in, nationality_index, apply_softmax=False):
"""모델의 정방향 계산
매개변수:
x_in (torch.Tensor): 입력 데이터 텐서
x_in.shape는 (batch, max_seq_size)입니다.
nationality_index (torch.Tensor): 각 데이터 포인트를 위한 국적 인덱스
RNN의 은닉 상태를 초기화하는데 사용합니다.
apply_softmax (bool): 소프트맥스 활성화를 위한 플래그로 훈련시에는 False가 되어야 합니다.
반환값:
결과 텐서. tensor.shape는 (batch, char_vocab_size)입니다.
"""
x_embedded = self.char_emb(x_in)
# 국적 임베딩 층 추가
# hidden_size: (num_layers * num_directions, batch_size, rnn_hidden_size)
nationality_embedded = self.nation_emb(nationality_index).unsqueeze(0)
y_out, _ = self.rnn(x_embedded, nationality_embedded)
batch_size, seq_size, feat_size = y_out.shape
y_out = y_out.contiguous().view(batch_size * seq_size, feat_size)
y_out = self.fc(F.dropout(y_out, p=self._dropout_p))
if apply_softmax:
y_out = F.softmax(y_out, dim=1)
new_feat_size = y_out.shape[-1]
y_out = y_out.view(batch_size, seq_size, new_feat_size)
return y_out
Python
복사
모델 훈련과 결과
시퀀스의 타임 스텝마다 예측을 만들기 때문에 손실을 계산하기 위해서는 이전 예제에서 두 가지 변경해야할 것들이 있습니다. 첫째로 계산을 위해 3차원 텐서를 2차원 텐서인 행렬로 변환시켜야 합니다. 두번째로 가변 길이 시퀀스를 위해 마스킹 인덱스를 준비해야 합니다. 마스킹된 위치에서는 손실을 계산하지 않습니다.
아래 코드를 이용해 3차원 텐서와 가변 길이 시퀀스 이슈를 다룹니다. 예측과 타깃을 손실 함수가 기대하는 크기(예측 2차원, 타깃 1차원)로 정규화하면, 각 행은 하나의 샘플, 즉 시퀀스에 있는 하나의 타임 텝을 나타내게됩니다. 다음으로 ignore_index를 mask_index로 지정하여 크로스 엔트로피 손실을 사용합니다. 이는 손실함수가 타깃에서 마스킹된 인덱스의 위치를 무시하도록 합니다.
def normalize_sizes(y_pred, y_true):
"""텐서 크기 정규화
매개변수:
y_pred (torch.Tensor): 모델의 출력
3차원 텐서이면 행렬로 변환합니다.
y_true (torch.Tensor): 타깃 예측
행렬이면 벡터로 변환합니다.
"""
if len(y_pred.size()) == 3:
y_pred = y_pred.contiguous().view(-1, y_pred.size(2))
if len(y_true.size()) == 2:
y_true = y_true.contiguous().view(-1)
return y_pred, y_true
Python
복사
def sequence_loss(y_pred, y_true, mask_index):
y_pred, y_true = normalize_sizes(y_pred, y_true)
return F.cross_entropy(y_pred, y_true, ignore_index=mask_index)
Python
복사
모델 하이퍼파라미터는 대부분 문자 어휘 사전 크기에 따라 결정됩니다. 이때 크기는 모델 입력에 나타나는 이산적인 토큰의 개수이고 타임 스텝마다 출력에 나타나는 클래스 개수입니다. 그 외 모델에 사용되는 하이퍼파라미터는 문자 임베딩 크기와 RNN 은닉 상태 크기입니다. 다음 코드에서 하이퍼파라미터와 훈련 설정을 보겠습니다.
args = Namespace(
# 날짜와 경로 정보
surname_csv="data/surnames/surnames_with_splits.csv",
vectorizer_file="vectorizer.json",
model_state_file="model.pth",
save_dir="model_storage/ch7/model2_conditioned_surname_generation",
# 모델 하이퍼파라미터
char_embedding_size=32,
rnn_hidden_size=32,
# 훈련 하이퍼파라미터
seed=1337,
learning_rate=0.001,
batch_size=128,
num_epochs=100,
early_stopping_criteria=5,
# 실행 옵션
catch_keyboard_interrupt=True,
cuda=True,
expand_filepaths_to_save_dir=True,
reload_from_files=False,
)
Python
복사
다음 코드에서는 forward() 메서드의 단계를 수정해서 새로운 반복문을 만듭니다. 여기에서 타임 스텝마다 예측을 계산한 뒤 다음 타임 스텝의 입력으로 사용합니다. 이는 모델이 어떤 성씨를 생성했는지 조사하여 질적으로 평가하기 위함입니다. 모델은 타임 스텝마다 소프트맥스 함수를 사용해 확률 분포로 변환된 예측 벡터를 출력합니다. 확률 분포를 사용하면 torch.multinomial() 샘플링 함수를 이용할 수 있습니다. 이 함수는 인덱스 확률에 비례하여 인덱스를 선택하며, 샘플링은 매번 다른 출력을 만드는 랜덤한 과정입니다.
def sample_from_model(model, vectorizer, num_samples=1, sample_size=20,
temperature=1.0):
"""모델이 만든 인덱스 시퀀스를 샘플링합니다.
매개변수:
model (SurnameGenerationModel): 훈련 모델
vectorizer (SurnameVectorizer): SurnameVectorizer 객체
nationalities (list): 국적을 나타내는 정수 리스트
sample_size (int): 샘플의 최대 길이
temperature (float): 무작위성 정도
0.0 < temperature < 1.0 이면 최대 값을 선택할 가능성이 높습니다
temperature > 1.0 이면 균등 분포에 가깝습니다
반환값:
indices (torch.Tensor): 인덱스 행렬
shape = (num_samples, sample_size)
"""
begin_seq_index = [vectorizer.char_vocab.begin_seq_index
for _ in range(num_samples)]
begin_seq_index = torch.tensor(begin_seq_index,
dtype=torch.int64).unsqueeze(dim=1)
indices = [begin_seq_index]
for time_step in range(sample_size):
x_t = indices[time_step]
x_emb_t = model.char_emb(x_t)
rnn_out_t, h_t = model.rnn(x_emb_t, h_t)
prediction_vector = model.fc(rnn_out_t.squeeze(dim=1))
probability_vector = F.softmax(prediction_vector / temperature, dim=1)
indices.append(torch.multinomial(probability_vector, num_samples=1))
indices = torch.stack(indices).squeeze().permute(1, 0)
return indices
Python
복사
다음 코드에서는 sample_from_model() 함수에서 얻은 샘플링 인덱스를 사람이 읽을 수 있는 문자열로 바꾸기 위해서 성씨를 벡터화하는 SequenceVocabulary를 사용합니다. 문자열을 만들 때는 END-OF-SEQUENCE 인덱스까지만 인덱스를 사용합니다. 모델이 성씨를 종료할 때를 학습했다고 가정하기 떄문입니다.
def decode_samples(sampled_indices, vectorizer):
"""인덱스를 성씨 문자열로 변환합니다
매개변수:
sampled_indices (torch.Tensor): `sample_from_model` 함수에서 얻은 인덱스
vectorizer (SurnameVectorizer): SurnameVectorizer 객체
"""
decoded_surnames = []
vocab = vectorizer.char_vocab
for sample_index in range(sampled_indices.shape[0]):
surname = ""
for time_step in range(sampled_indices.shape[1]):
sample_item = sampled_indices[sample_index, time_step].item()
if sample_item == vocab.begin_seq_index:
continue
elif sample_item == vocab.end_seq_index:
break
else:
surname += vocab.lookup_index(sample_item)
decoded_surnames.append(surname)
return decoded_surnames
Python
복사
다음으로 조건이 있는 SurnameGenerationModel을 위해 sample_from_model() 함수를 수정하여 샘플 개수 대신에 국적 인덱스의 리스트를 받습니다. 이 함수는 국적 인덱스를 임베딩으로 바꾸어 GRU의 초기 은닉 상태로 사용하게 됩니다.
def sample_from_model(model, vectorizer, nationalities, sample_size=20,
temperature=1.0):
"""모델이 만든 인덱스 시퀀스를 샘플링합니다.
매개변수:
model (SurnameGenerationModel): 훈련 모델
vectorizer (SurnameVectorizer): SurnameVectorizer 객체
nationalities (list): 국적을 나타내는 정수 리스트
sample_size (int): 샘플의 최대 길이
temperature (float): 무작위성 정도
0.0 < temperature < 1.0 이면 최대 값을 선택할 가능성이 높습니다
temperature > 1.0 이면 균등 분포에 가깝습니다
반환값:
indices (torch.Tensor): 인덱스 행렬
shape = (num_samples, sample_size)
"""
num_samples = len(nationalities)
begin_seq_index = [vectorizer.char_vocab.begin_seq_index
for _ in range(num_samples)]
begin_seq_index = torch.tensor(begin_seq_index,
dtype=torch.int64).unsqueeze(dim=1)
indices = [begin_seq_index]
nationality_indices = torch.tensor(nationalities, dtype=torch.int64).unsqueeze(dim=0)
h_t = model.nation_emb(nationality_indices)
for time_step in range(sample_size):
x_t = indices[time_step]
x_emb_t = model.char_emb(x_t)
rnn_out_t, h_t = model.rnn(x_emb_t, h_t)
prediction_vector = model.fc(rnn_out_t.squeeze(dim=1))
probability_vector = F.softmax(prediction_vector / temperature, dim=1)
indices.append(torch.multinomial(probability_vector, num_samples=1))
indices = torch.stack(indices).squeeze().permute(1, 0)
return indices
Python
복사
이제 국적 인덱스를 순회하면서 각 국적에서 샘플링을 수행합니다. 출력을 보면 모델이 성씨 철자에 있는 어떤 패턴을 따름을 알 수 있습니다.
model = model.cpu()
for index in range(len(vectorizer.nationality_vocab)):
nationality = vectorizer.nationality_vocab.lookup_index(index)
print("{} 샘플: ".format(nationality))
sampled_indices = sample_from_model(model, vectorizer,
nationalities=[index] * 3,
temperature=0.7)
for sampled_surname in decode_samples(sampled_indices, vectorizer):
print("- " + sampled_surname)
Python
복사
Arabic 샘플:
- Bakin
- Heran
- Soib
Chinese 샘플:
- Luag
- Rur
- Dao
Czech 샘플:
- Ponnoir
- Stonaj
- Teutche
Dutch 샘플:
- Fmitzim
- Fablelb
- Ulskomov
English 샘플:
- Cintee
- Hillen
- Vannid
.....
Python
복사
이전 글 읽기
