Search
Duplicate

5-4 사전 훈련된 임베딩을 활용한 전이 학습

AG 뉴스 데이터셋

AG 뉴스 데이터셋은 데이터 마이닝과 정보 추출 방법 연구를 목적으로 2005년에 수집한 뉴스 기사 모음입니다. 이번 장의 목표는 텍스트 분류에서 사전 훈련된 단어 임베딩의 효과를 알아보는 것입니다. 기사 제목에 초점을 맞춰 주어진 제목으로 카테고리를 예측하는 다중 분류 작업을 만들 것입니다.
텍스트 전처리는 텍스트를 소문자로 변환 후 쉼표, 마침표, 느낌표 등의 주위에 공백을 추가하고 그 외 구두점 기호는 모두 제거하는 식으로 진행합니다. 데이터셋은 학습, 검증, 테스트 세트로 분할합니다. 다음 코드는 데이터셋의 각 행에서 모델 입력을 나타내는 문자열을 추출하고 Vectorizer를 사용해 벡터로 변환하는 과정을 보여줍니다. 그다음으로 뉴스 카테고리를 나타내는 정수와 쌍을 구성합니다.
다음은 NewsDataset.__getitem__() 메서드 구현입니다.
class NewsDataset(Dataset): @classmethod def load_dataset_and_make_vectorizer(cls, news_csv): # 데이터셋을 로드하고 처음부터 새로운 Vectorizer 만들기 # news_csv(str): 데이터셋의 위치 news_df = pd.read_csv(news_csv) train_news_df = news_df[news_df.split=='train'] # NewsDataset의 인스턴스 return cls(news_df, NewsVectorizer.form_dataframe(train_news_df)) def __getitem__(self, index): # 파이토치 데이터셋의 주요 진입 메서드 # index(int): 데이터 포인트의 인덱스 row = self._target_df.iloc[index] title_vector = self._vectorizer.vectorize(row.title, self._max_seq_length) category_index = self._vectorizer.category_vocab.lookup_token(row.category) return {'x_data': title_vector, # 데이터 포인트의 특성 'y_target': category_index} # 레이블
Python
복사

Vocabulary, Vectorizer, DataLoader

이번 장에서는 Vocabulary 클래스를 상속한 SequenceVocabulary를 만듭니다. 이 클래스에서는 시퀀스 데이터에 사용하는 특수 토큰 4개(UNK 토큰, MASK 토큰, BEGIN-OF-SEQUENCE 토큰, END-OF-SEQUENCE 토큰)가 있는데, 추후에 자세히 설명하겠지만 크게 3가지 용도로 사용됩니다. UNK 토큰은 모델이 드물게 등장하는 단어에 대한 표현을 학습하도록 합니다. 테스트 시에 본 적 없는 단어를 처리할 수 있게 됩니다. MASK 토큰은 Embedding 층의 마스킹 역할을 수행하고 가변 길이의 시퀀스가 있을 시에 손실 계산을 돕습니다. 마지막 2가지 토큰은 시퀀스 경계에 대한 힌트를 신경망에 제공합니다.
텍스트를 벡터의 미니배치로 변환하는 파이프라인의 두 번째 부분은 Vectorizer입니다. 이 클래스는 SequenceVocabulary 객체를 생성하고 캡슐화합니다. 이 예제의 Vectorizer는 단어 빈도를 계산하고 특정 임곗값을 지정하여 Vocabulary에서 사용할 수 있는 전체 단어 집합을 제한합니다. 핵심 목적은 빈도가 낮은 잡음 단어를 제거하여 신호 품질을 개선하고 모델 메모리 사용량 제약을 완화하는 것입니다.
인스턴스 생성 후 Vectorizer의 vectorizer() 메서드는 뉴스 제목 하나를 입력으로 받아 데이터셋에서 가장 긴 제목과 길이가 같은 벡터를 반환합니다. 이 메서드는 2가지 주요 작업을 수행합니다. 첫째, 최대 시퀀스 길이를 사용합니다. 보통 데이터셋이 최대 시퀀스 길이를 관리하고 추론 시 테스트 데이터의 시퀀스 길이를 벡터 길이로 사용하지만, CNN 모델을 사용하므로 추론 시에도 벡터의 크기가 같아야 합니다. 둘째, 단어 시퀀스를 나타내는 0으로 패딩된 정수 벡터를 출력합니다. 이 정수 벡터는 시작 부분에 BEGIN-OF-SEQUENCE 토큰을 추가하고 끝에는 END-OF-SEQUENCE 토큰을 추가함으로써 분류기는 시퀀스 경계를 구분하고 경계 근처의 단어에 중앙에 가까운 단어와는 다르게 반응할 수 있습니다.
다음은 AG 뉴스 데이터 셋을 위한 Vectorizer 구현입니다.
class NewsVectorizer(object): def vectorize(self, title, vector_length = -1): # title(str): 공백으로 나누어진 단어 문자열 # vector_length(int): 인덱스 벡터의 길이 매개변수 indices = [self.title_vocab.begin_seq_index] indices.extend(self.title_vocab.lookup_token(token) for token in title.split(" ")) indices.append(self.title_vocab.end_seq_index) if vector_length < 0: vector_length = len(indices) out_vector = np.zeros(vector_length, dtype=np.int64) out_vector[:len(indices)] = indices out_vector[len(indices):] = self.title_vocab.mask_index # 벡터로 변환된 제목 (넘파이 어레이) return out_vector @classmethod def from_dataframe(cls, news_df, cutoff=25): # 데이터셋 데이터프레임에서 Vectorizer 객체 만들기 # news_df(pandas.DataFrame): 타깃 데이터셋 # cutoff(int): Vocabulary에 포함할 빈도 임곗값 category_vocab = Vocabulary() for category in sorted(set(news_df.category)): category_vocab.add_token(category) word_counts = Counter() for title in news_df.title: for token in title.split(" "): if token not in string.punctuation: word_counts[token] += 1 title_vocab = SequenceVocabulary() for word, word_count in word_counts.items(): if word_count >= cutoff: title_vocab.add_token(word) # NewsVectorizer 객체 return cls(title_vocab, category_vocab)
Python
복사

NewsClassifier 모델

단어 임베딩을 초기 임베딩 행렬로 사용하려면 먼저 디스크에서 임베딩을 로드한 다음 실제 데이터에 있는 단어에 해당하는 임베딩의 일부를 선택합니다. 마지막으로 Embedding 층의 가중치 행렬을 선택한 임베딩으로 지정합니다. 첫 번째와 두 번째 단계를 다음 코드에서 설명하겠습니다. 어휘 사전에 기반하여 단어 임베딩의 부분 집합을 선택합니다.
def load_glove_from_file(glove_filepath): # glove_filepath (str): 임베딩 파일 경로 word_to_index = {} embeddings = [] with open(glove_filepath, "r") as fp: for index, line in enumerate(fp): line = line.split(" ") # each line: word num1 num2 ... word_to_index[line[0]] = index # word = line[0] embedding_i = np.array([float(val) for val in line[1:]]) embeddings.append(embedding_i) # word_to_index (dict), embeddings (numpy.ndarary) return word_to_index, np.stack(embeddings) def make_embedding_matrix(glove_filepath, words): # 특정 단어 집합에 대한 임베딩 행렬 만들기 # glove_filepath (str): 임베딩 파일 경로 # words (list): 단어 리스트 word_to_idx, glove_embeddings = load_glove_from_file(glove_filepath) embedding_size = glove_embeddings.shape[1] final_embeddings = np.zeros((len(words), embedding_size)) for i, word in enumerate(words): if word in word_to_idx: final_embeddings[i, :] = glove_embeddings[word_to_idx[word]] else: embedding_i = torch.ones(1, embedding_size) torch.nn.init.xavier_uniform_(embedding_i) final_embeddings[i, :] = embedding_i # final_embeddings (numpu.ndarray): 임베딩 행렬 return final_embeddings
Python
복사
입력 토큰 인덱스를 벡터 표현으로 매핑하는 Embedding층을 사용합니다. 다음 코드에서는 Embedding 층의 가중치를 사전 훈련된 임베딩으로 바꾸게 됩니다. forward() 메서드에서 이 임베딩을 사용해 인덱스를 벡터로 매핑합니다.
class NewsClassifier(nn.Module): def __init__(self, embedding_size, num_embeddings, num_channels, hidden_dim, num_classes, dropout_p, pretrained_embeddings=None, padding_idx=0): """ 매개변수: embedding_size (int): 임베딩 벡터의 크기 num_embeddings (int): 임베딩 벡터의 개수 num_channels (int): 합성곱 커널 개수 hidden_dim (int): 은닉 차원 크기 num_classes (int): 클래스 개수 dropout_p (float): 드롭아웃 확률 pretrained_embeddings (numpy.array): 사전에 훈련된 단어 임베딩 기본값은 None padding_idx (int): 패딩 인덱스 """ super(NewsClassifier, self).__init__() if pretrained_embeddings is None: self.emb = nn.Embedding(embedding_dim=embedding_size, num_embeddings=num_embeddings, padding_idx=padding_idx) else: pretrained_embeddings = torch.from_numpy(pretrained_embeddings).float() self.emb = nn.Embedding(embedding_dim=embedding_size, num_embeddings=num_embeddings, padding_idx=padding_idx, _weight=pretrained_embeddings) self.convnet = nn.Sequential( nn.Conv1d(in_channels=embedding_size, out_channels=num_channels, kernel_size=3), nn.ELU(), nn.Conv1d(in_channels=num_channels, out_channels=num_channels, kernel_size=3, stride=2), nn.ELU(), nn.Conv1d(in_channels=num_channels, out_channels=num_channels, kernel_size=3, stride=2), nn.ELU(), nn.Conv1d(in_channels=num_channels, out_channels=num_channels, kernel_size=3), nn.ELU() ) self._dropout_p = dropout_p self.fc1 = nn.Linear(num_channels, hidden_dim) self.fc2 = nn.Linear(hidden_dim, num_classes) def forward(self, x_in, apply_softmax=False): """분류기의 정방향 계산 매개변수: x_in (torch.Tensor): 입력 데이터 텐서 x_in.shape는 (batch, dataset._max_seq_length)입니다. apply_softmax (bool): 소프트맥스 활성화 함수를 위한 플래그 크로스-엔트로피 손실을 사용하려면 False로 지정합니다 반환값: 결과 텐서. tensor.shape은 (batch, num_classes)입니다. """ # 임베딩을 적용하고 특성과 채널 차원을 바꿉니다 x_embedded = self.emb(x_in).permute(0, 2, 1) features = self.convnet(x_embedded) # 평균 값을 계산하여 부가적인 차원을 제거합니다 remaining_size = features.size(dim=2) features = F.avg_pool1d(features, remaining_size).squeeze(dim=2) features = F.dropout(features, p=self._dropout_p) # MLP 분류기 intermediate_vector = F.relu(F.dropout(self.fc1(features), p=self._dropout_p)) prediction_vector = self.fc2(intermediate_vector) if apply_softmax: prediction_vector = F.softmax(prediction_vector, dim=1) return prediction_vector
Python
복사

모델 훈련

훈련 과정은 데이터셋 초기화, 모델 초기화, 손실 함수 초기화, 옵티마이저 초기화, 훈련 세트에 대한 반복, 모델 파라미터 업데이트, 검증 세트에 대한 반복과 성능 측정을 한 뒤에 특정 횟수 동안 이 데이터셋을 반복합니다. 다음 코드는 하이퍼파라미터를 포함한 예제의 훈련 매개변수입니다.
args = Namespace( # 날짜와 경로 정보 news_csv="data/ag_news/news_with_splits.csv", vectorizer_file="vectorizer.json", model_state_file="model.pth", save_dir="model_storage/ch5/document_classification", # 모델 하이퍼파라미터 glove_filepath='data/glove/glove.6B.100d.txt', use_glove=False, embedding_size=100, hidden_dim=100, num_channels=100, # 훈련 하이퍼파라미터 seed=1337, learning_rate=0.001, dropout_p=0.1, batch_size=128, num_epochs=100, early_stopping_criteria=5, # 실행 옵션 cuda=True, catch_keyboard_interrupt=True, reload_from_files=False, expand_filepaths_to_save_dir=True )
Python
복사

모델 평가와 예측

모델이 작업을 잘 수행하는지 평가하는 방법은 두 가지로, 테스트 세트를 사용하여 정량적으로 평가하거나 분류 결과를 개별적으로 조사하여 질적으로 평가하는 방법이 있습니다.

테스트 데이터로 평가하기

classifier.eval() 메서드를 사용해 모델을 평가 모드로 설정하여 드롭아웃과 역전파를 끈 후 훈련 세트 및 검증 세트와 같은 방식으로 테스트 세트를 반복합니다. 전체 훈련 과정에서 테스트 세트는 딱 한 번만 사용해야 합니다.

새로운 뉴스 제목의 카테고리 예측하기

훈련의 목적은 실전에 배치하여 처음 접하는 데이터에 대해 추론 혹은 예측을 수행하기 위함입니다. 새로운 뉴스 제목의 카테고리를 예측하기 위해서는 먼저 훈련할 때 데이터를 전처리한 방식으로 텍스트를 전처리해야 합니다. 전처리된 문자열은 훈련에 사용한 Vectorizer를 사용해 벡터로 바꾸고 파이토치 텐서로 변환합니다. 그다음으로 이 텐서에 분류기를 적용합니다. 예측 벡터에서 최댓값을 찾아 카테고리 이름을 조회하는데, 이 과정을 코드로 살펴보겠습니다.
def predict_category(title, classifier, vectorizer, max_length): """뉴스 제목을 기반으로 카테고리를 예측합니다 매개변수: title (str): 원시 제목 문자열 classifier (NewsClassifier): 훈련된 분류기 객체 vectorizer (NewsVectorizer): 해당 Vectorizer max_length (int): 최대 시퀀스 길이 """ title = preprocess_text(title) vectorized_title = \ torch.tensor(vectorizer.vectorize(title, vector_length=max_length)) result = classifier(vectorized_title.unsqueeze(0), apply_softmax=True) probability_values, indices = result.max(dim=1) predicted_category = vectorizer.category_vocab.lookup_index(indices.item()) return {'category': predicted_category, 'probability': probability_values.item()}
Python
복사
이전 글 읽기