8-5 신경망 기계 번역 실습

class NMTVectorizer(object): """ 어휘 사전을 생성하고 관리합니다 """ def __init__(self, source_vocab, target_vocab, max_source_length, max_target_length): """ 매개변수: source_vocab (SequenceVocabulary): 소스 단어를 정수에 매핑합니다 target_vocab (SequenceVocabulary): 타깃 단어를 정수에 매핑합니다 max_source_length (int): 소스 데이터셋에서 가장 긴 시퀀스 길이 max_target_length (int): 타깃 데이터셋에서 가장 긴 시퀀스 길이 """ self.source_vocab = source_vocab self.target_vocab = target_vocab self.max_source_length = max_source_length self.max_target_length = max_target_length @classmethod def from_dataframe(cls, bitext_df): """ 데이터셋 데이터프레임으로 NMTVectorizer를 초기화합니다 매개변수: bitext_df (pandas.DataFrame): 텍스트 데이터셋 반환값 : NMTVectorizer 객체 """ source_vocab = SequenceVocabulary() target_vocab = SequenceVocabulary() max_source_length = 0 max_target_length = 0 for _, row in bitext_df.iterrows(): source_tokens = row["source_language"].split(" ") if len(source_tokens) > max_source_length: max_source_length = len(source_tokens) for token in source_tokens: source_vocab.add_token(token) target_tokens = row["target_language"].split(" ") if len(target_tokens) > max_target_length: max_target_length = len(target_tokens) for token in target_tokens: target_vocab.add_token(token) return cls(source_vocab, target_vocab, max_source_length, max_target_length)

def _vectorize(self, indices, vector_length=-1, mask_index=0): """인덱스를 벡터로 변환합니다 매개변수: indices (list): 시퀀스를 나타내는 정수 리스트 vector_length (int): 인덱스 벡터의 길이 mask_index (int): 사용할 마스크 인덱스; 거의 항상 0 """ if vector_length < 0: vector_length = len(indices) vector = np.zeros(vector_length, dtype=np.int64) vector[:len(indices)] = indices vector[len(indices):] = mask_index return vector def _get_source_indices(self, text): """ 벡터로 변환된 소스 텍스트를 반환합니다 매개변수: text (str): 소스 텍스트; 토큰은 공백으로 구분되어야 합니다 반환값: indices (list): 텍스트를 표현하는 정수 리스트 """ indices = [self.source_vocab.begin_seq_index] indices.extend(self.source_vocab.lookup_token(token) for token in text.split(" ")) indices.append(self.source_vocab.end_seq_index) return indices def _get_target_indices(self, text): """ 벡터로 변환된 타깃 텍스트를 반환합니다 매개변수: text (str): 타깃 텍스트; 토큰은 공백으로 구분되어야 합니다 반환값: 튜플: (x_indices, y_indices) x_indices (list): 디코더에서 샘플을 나타내는 정수 리스트 y_indices (list): 디코더에서 예측을 나타내는 정수 리스트 """ indices = [self.target_vocab.lookup_token(token) for token in text.split(" ")] x_indices = [self.target_vocab.begin_seq_index] + indices y_indices = indices + [self.target_vocab.end_seq_index] return x_indices, y_indices def vectorize(self, source_text, target_text, use_dataset_max_lengths=True): """ 벡터화된 소스 텍스트와 타깃 텍스트를 반환합니다 벡터화된 소스 텍슽트는 하나의 벡터입니다. 벡터화된 타깃 텍스트는 7장의 성씨 모델링과 비슷한 스타일로 두 개의 벡터로 나뉩니다. 각 타임 스텝에서 첫 번째 벡터가 샘플이고 두 번째 벡터가 타깃이 됩니다. 매개변수: source_text (str): 소스 언어의 텍스트 target_text (str): 타깃 언어의 텍스트 use_dataset_max_lengths (bool): 최대 벡터 길이를 사용할지 여부 반환값: 다음과 같은 키에 벡터화된 데이터를 담은 딕셔너리: source_vector, target_x_vector, target_y_vector, source_length """ source_vector_length = -1 target_vector_length = -1 if use_dataset_max_lengths: source_vector_length = self.max_source_length + 2 target_vector_length = self.max_target_length + 1 source_indices = self._get_source_indices(source_text) source_vector = self._vectorize(source_indices, vector_length=source_vector_length, mask_index=self.source_vocab.mask_index) target_x_indices, target_y_indices = self._get_target_indices(target_text) target_x_vector = self._vectorize(target_x_indices, vector_length=target_vector_length, mask_index=self.target_vocab.mask_index) target_y_vector = self._vectorize(target_y_indices, vector_length=target_vector_length, mask_index=self.target_vocab.mask_index) return {"source_vector": source_vector, "target_x_vector": target_x_vector, "target_y_vector": target_y_vector, "source_length": len(source_indices)}

def generate_nmt_batches(dataset, batch_size, shuffle=True, drop_last=True, device="cpu"): """ 파이토치 DataLoader를 감싸고 있는 제너레이터 함수; NMT 버전 """ dataloader = DataLoader(dataset=dataset, batch_size=batch_size, shuffle=shuffle, drop_last=drop_last) for data_dict in dataloader: lengths = data_dict['x_source_length'].numpy() sorted_length_indices = lengths.argsort()[::-1].tolist() out_data_dict = {} for name, tensor in data_dict.items(): out_data_dict[name] = data_dict[name][sorted_length_indices].to(device) yield out_data_dict

class NMTModel(nn.Module): """ 신경망 기계 번역 모델 """ def __init__(self, source_vocab_size, source_embedding_size, target_vocab_size, target_embedding_size, encoding_size, target_bos_index): """ 매개변수: source_vocab_size (int): 소스 언어의 고유한 단어 개수 source_embedding_size (int): 소스 임베딩 벡터의 크기 target_vocab_size (int): 타깃 언어의 고유한 단어 개수 target_embedding_size (int): 타깃 임베딩 벡터의 크기 encoding_size (int): 인코더 RNN의 크기 target_bos_index (int): BEGIN-OF-SEQUENCE 토큰 인덱스 """ super(NMTModel, self).__init__() self.encoder = NMTEncoder(num_embeddings = source_vocab_size, embedding_size = source_embedding_size, rnn_hidden_size = encoding_size) decoding_size = encoding_size = 2 self.decoder = NMTDecoder(num_embeddings = target_vocab_size, embedding_size = tarrget_embedding_size, rnn_hidden_size = decoding_size, bos_index = target_bos_index) def forward(self, x_source, x_source_lengths, target_sequence): """ 모델의 정방향 계산 매개변수: x_source (torch.Tensor): 소스 텍스트 데이터 센서 x_source.shape는 (batch, vectorizer.max_source_length)입니다. x_source_lengths torch.Tensor): x_source의 시퀀스 길이 target_sequence (torch.Tensor): 타깃 텍스트 데이터 텐서 반환값: decoded_states (torch.Tensor): 각 출력 타임 스텝의 예측 벡터 """ encoder_state, final_hidden_states = self.encoder(x_source, x_source_lengths) decoded_states = self.decoder(encoder_state = encoder_state, initial_hidden_state = final_hidden_states, target_sequence = target_sequence) return decoded_states

class NMTEncoder(nn.Module): def __init__(self, num_embeddings, embedding_size, rnn_hidden_size): """ 매개변수: num_embeddings (int): 임베딩 개수는 소스 어휘 사전의 크기입니다 embedding_size (int): 임베딩 벡터의 크기 rnn_hidden_size (int): RNN 은닉 상태 벡터의 크기 """ super(NMTEncoder, self).__init__() self.source_embedding = nn.Embedding(num_embeddings, embedding_size, padding_idx = 0) self.birnn = nn.GRU(embedding_size, rnn_hidden_size, bidirectional = True, batch_first = True) def forward(self, x_source, x_lengths): x_embedded = self.source_embedding(x_source) # create PackedSequence 생성 x_packed.data.shape = (number_items, # embedding_size) x_lengths = x_lengths.detatch().cpu().numpy() x_packed = pack_padded_sequence(x_embedded, x_lengths, batch_first = True) # x_birnn_h.shape = (num_rnn, batch_size, feature_size) x_birnn_out, x_birnn_h = self.birnn(x_packed) # (batch_size, num_rnn. feature_size)로 변환 x_birnn_h = x_brnn_h.permute(1, 0, 2) # 특성 펼침. (batch_size, num_rnn * feature_size)로 바꾸기 # (참고: -1은 남은 차원에 해당하며, # 두 깨의 RNN 은닉 벡터를 1로 펼칩니다) x_birnn_h = x_birnn_h.contiguous().view(x_birnn_h.size(0), -1) x_unpacked, _ = pad_packed_sequence(x_birnn_out, batch_first = True) return x_unpacked, x_birnn_h

class NMTDecoder(nn.Module): def __init__(self, num_embeddings, embedding_size, rnn_hidden_size, bos_index): """ 매개변수: num_embeddings (int): 임베딩 개수는 타깃 어휘 사전의 고유한 단어의 개수입니다 embeddin_size (int): 임베딩 벡터 크기 rnn_hidden_size (int): RNN 은닉 상태 크기 bos_index (int): begin-of-sequence 인덱스 """ super(NMTDecoder, self).__init__() self._rnn_hidden_size = rnn_hidden_size self.target_embedding = nn.Embedding(num_embeddings = num_embeddings, embedding_dim = embedding_size, padding_ide = 0) self.gru_cell = nn.GRUCell(embedding_size + rnn_hidden_size, rnn_hidden_size) self.hidden_map = nn.Linear(rnn_hidden_size, rnn_hidden_size) self.classifie = nn.Linear(rnn_hidden_size * 2, num_embeddings) self.bos_index = bos_index def _init_indices(self, batch_size): """ BEGIN-OF-SEQUENCE 인덱스 벡터를 반환합니다 """ return torch.ones(batch_size, dtype = torch.int64) * self.bos_index def _init_context_vectors(self, batch_size): """ 문맥 벡터를 초기화하기 위한 0 벡터를 반환합니다 """ return torch.zeros(batch_size, self._rnn_hidden_size) def forward(self, encoder_state, initial_hidden_state, target_sequence): """ """ # 가정: 첫 번째 차원은 배치 차원입니다 # 즉 입력은 (Batch, Seq) # 시퀀스에 대해 반복해야 하므로 (Seq, Batch)로 차원을 바꿉니다 target_sequence = target_sequence.permute(1,0) # 주어진 인코더의 은닉 상태를 초기 은닉 상태로 사용합니다 h_t = self.hidden_map(initial_hidden_state) batch_size = encoder_state_size(0) # 문맥 벡터를 0으로 초기화합니다 context_vectors - self._init_context_vectors(batch_size) # 처 단어 y_t를 BOS로 초기화합니다 y_t_index = self._init_indeices(batch_size) h_t = h_t.to(encoder_state.device) y_index = y_t_index.to(encoder_state.device) context_vectors = context_vectors.to(encoder_state.device) output_vectors = [] #분석을 위해 GPU에서 캐싱된 모든 텐서를 가져와 저장합니다 self._cached_p_attn = [] self._cached_ht = [] self._cached_decoder_state = encoder_state.cpu().detatch().numpy() output_sequence_size = target_sequence_size(0) for i in range(output_sequence_size): # 1단계: 단어를 임베딩하고 이전 문맥과 연결합니다 y_input_vector = self.target_embedding(target_sequence[i]) rnn_input = torch.cat([y_input_vector, context_vectors], dim = 1) # 2단계: GRU를 적용하고 새로운 은닉 벡터를 얻습니다 h_t = self.gru_cell(rnn_input, h_t) self._cached_ht.append(h_t.cpu().data.numpy()) # 3단계: 현재 은닉 상태를 사용해 인코더의 상태를 주목합니다 context_vectors, p_attn, _ = \ verbose_attention(encoder_state_vectors = encoder_state, query_vector = h_t) # 부가 작업: 시각화를 위해 어텐션 확률을 저장합니다 self._cached_p_attn.append(p_attn.cpu().detatch().numpy()) # 4단계: 현재 은닉 상태와 문맥 벡터를 사용해 다음 단어를 예측합니다 prediction_vector = torch.cat((context_vectors, h_t), dim = 1) score_for_y_t_index = self.classifier(prediction_vector) # 부가 작업: 예측 성능 점수를 기록합니다 output_vectors.append(score_for_y_t_index)

def verbose_attention(encoder_state_vectors, query_vector): """ 원소별 연산을 사용하는 어텐션 메커니즘 버전 매개변수: encoder_state_vectors (torch.Tensor): 인코더의 양방향 GRU에서 출력된 3차원 텐서 query_vector (torch.Tensor): 디코더 GRU의 은닉 상태 """ batch_size, num_vectors, vector_size = encoder_state_vectors.size() vector_scores = torch.sum(encoder_state_vectors * query_vector.view(batch_size, 1, vector_size), dim=2) # 쿼리 값과 각 인코더 RNN의 은닉 벡터들의 곱 vector_probabilities = F.softmax(vector_scores, dim=1) # 값들을 0~1의 값으로 확률 분포화로 가중치 구하기 weighted_vectors = encoder_state_vectors * vector_probabilities.view(batch_size, num_vectors, 1) # 은닉 벡터값과 가중치의 곱 구하기 context_vectors = torch.sum(weighted_vectors, dim=1) # 이 값들의 합으로 컨텍스트 벡터 또는 어텐션 값 구하기 return context_vectors, vector_probabilities, vector_scores def terse_attention(encoder_state_vectors, query_vector): """ 점곱을 사용하는 어텐션 메커니즘 버전 매개변수: encoder_state_vectors (torch.Tensor): 인코더의 양방향 GRU에서 출력된 3차원 텐서 query_vector (torch.Tensor): 디코더 GRU의 은닉 상태 """ vector_scores = torch.matmul(encoder_state_vectors, query_vector.unsqueeze(dim=2)).squeeze() # matmul 함수를 이용해 쿼리*키 값 구하기 vector_probabilities = F.softmax(vector_scores, dim=-1) # 가중치 구하기 context_vectors = torch.matmul(encoder_state_vectors.transpose(-2, -1), vector_probabilities.unsqueeze(dim=2)).squeeze() # 컨텍스트 벡터 값 구하기 return context_vectors, vector_probabilities

class NMTDecoder(nn.Module): def __init__(self, num_embeddings, embedding_size, rnn_hidden_size, bos_index): super(NMTDecoder, self).__init__() # 앞서 작성한 코드와 동일한 정의 # 생략 def forward(self, encoder_state, initial_hidden_state, target_sequence, sample_probability=0.0): """ 모델의 정방향 계산 매개변수: encoder_state (torch.Tensor): NMTEncoder의 출력 initial_hidden_state (torch.Tensor): NMTEncoder의 마지막 은닉 상태 target_sequence (torch.Tensor): 타깃 텍스트 데이터 텐서 sample_probability (float): 스케줄링된 샘플링 파라미터 디코더 타임 스텝마다 모델 예측에 사용할 확률 반환값: output_vectors (torch.Tensor): 각 타임 스텝의 예측 벡터 """ # 샘플 확률을 지정해줍니다 # 0과 1 사이의 적당한 값을 지정해줍니다 # 0 : 예측 시퀀스만을 사용 # 1 : 타겟 시퀀스만을 사용 if target_sequence is None: sample_probability = 0.5 else: # 가정: 첫 번째 차원은 배치 차원입니다 # 즉 입력은 (Batch, Seq) # 시퀀스에 대해 반복해야 하므로 (Seq, Batch)로 차원을 바꿉니다 target_sequence = target_sequence.permute(1, 0) output_sequence_size = target_sequence.size(0) # 주어진 인코더의 은닉 상태를 초기 은닉 상태로 사용합니다 h_t = self.hidden_map(initial_hidden_state) batch_size = encoder_state.size(0) # 문맥 벡터를 0으로 초기화합니다 context_vectors = self._init_context_vectors(batch_size) # 첫 단어 y_t를 BOS로 초기화합니다 y_t_index = self._init_indices(batch_size) h_t = h_t.to(encoder_state.device) y_t_index = y_t_index.to(encoder_state.device) context_vectors = context_vectors.to(encoder_state.device) output_vectors = [] self._cached_p_attn = [] self._cached_ht = [] self._cached_decoder_state = encoder_state.cpu().detach().numpy() # 반복문 이전까지는 기존의 Decoder 코드와 동일합니다 for i in range(output_sequence_size): # 스케줄링된 샘플링 사용 여부 # 생성한 난수와 확률 값을 비교해 샘플링 사용 유무를 선택합니다 use_sample = np.random.random() < sample_probability if not use_sample: y_t_index = target_sequence[i] # 단계 1: 단어를 임베딩하고 이전 문맥과 연결합니다 y_input_vector = self.target_embedding(y_t_index) rnn_input = torch.cat([y_input_vector, context_vectors], dim=1) # 단계 2: GRU를 적용하고 새로운 은닉 벡터를 얻습니다 h_t = self.gru_cell(rnn_input, h_t) self._cached_ht.append(h_t.cpu().detach().numpy()) # 단계 3: 현재 은닉 상태를 사용해 인코더의 상태를 주목합니다 context_vectors, p_attn, _ = verbose_attention(encoder_state_vectors=encoder_state, query_vector=h_t) # 부가 작업: 시각화를 위해 어텐션 확률을 저장합니다 self._cached_p_attn.append(p_attn.cpu().detach().numpy()) # 단게 4: 현재 은닉 상태와 문맥 벡터를 사용해 다음 단어를 예측합니다 prediction_vector = torch.cat((context_vectors, h_t), dim=1) score_for_y_t_index = self.classifier(F.dropout(prediction_vector, 0.3)) if use_sample: p_y_t_index = F.softmax(score_for_y_t_index * self._sampling_temperature, dim=1) # _, y_t_index = torch.max(p_y_t_index, 1) y_t_index = torch.multinomial(p_y_t_index, 1).squeeze() # 부가 작업: 예측 성능 점수를 기록합니다 output_vectors.append(score_for_y_t_index) output_vectors = torch.stack(output_vectors).permute(1, 0, 2) return output_vectors