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

기계 번역 데이터셋

타토에바 : 문장과 번역이 모이는 곳

https://tatoeba.org/ko/

NMT를 위한 벡터 파이프라인

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

스케줄링된 샘플링 Scheduled Sampling

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

모델 훈련

1. 소스 시퀸스와 타겟 시퀀스를 입력받아, 타겟 시퀀스 예측 생성

2. 타겟 시퀀스 예측 레이블을 통해 크로스 엔트로피 손실 계산 크로스 엔트로피 손실은 확률 분포와 예측 분포 사이의 차이를 계산하여, 분류 모델이 예측을 잘 수행하는지를 평가하는 지표입니다.

3. 역전파를 통해 그래디언트를 계산

4. 옵티마이저를 통해 모델 파라미터를 업데이트