[프로젝트] HR_면접자 정보 맞추기 프로젝트 (3)

https://dangingsu.tistory.com/47

[프로젝트] HR_면접자 정보 맞추기 프로젝트 (2)

지난 2편에 이어서 진행되겠습니다!

1편에서는 전처리, EDA 등 주로 데이터에 대해 알아보았습니다.

2편에서는 모델 파인튜닝에 대해서 알아보았습니다.

3편에서는 추론 결과를 보면서 어떻게 성능을 더 올릴 수 있을지에 대해 논의해보겠습니다.

 

1. Inference 결과

2편에서 학습시켰던 모델을 받아서 추론을 실행시켜보겠습니다.

코드는 아래와 같습니다.

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch

## 모델과 토크나이저 로드
model = AutoModelForSequenceClassification.from_pretrained('Junhoee/Kobigbird_HR').to(device)
tokenizer = AutoTokenizer.from_pretrained("Junhoee/Kobigbird_HR")

## test_df의 answer_text에 대해 추론 수행
def predict_combined_label(test_df):
    all_predictions = []
    
    ## 배치 단위로 데이터를 처리할 수 있도록 설정 (메모리 절약을 위해)
    batch_size = 16
    for i in range(0, len(test_df), batch_size):
        ## answer_text 추출 및 토크나이징
        batch_texts = test_df['answer_text'].iloc[i:i+batch_size].tolist()
        inputs = tokenizer(batch_texts, padding=True, truncation=True, return_tensors="pt", max_length=512)

        ## 입력 데이터를 GPU로 이동
        inputs = {key: val.to(device) for key, val in inputs.items()}

        ## 추론 수행
        with torch.no_grad():
            outputs = model(**inputs)

        ## 로짓(logits) 출력
        logits = outputs.logits

        ## 예측 클래스 추출
        predicted_class = torch.argmax(logits, dim=1).cpu().numpy()
        all_predictions.extend(predicted_class)

    return all_predictions

## 예측 수행
test_df['predicted_combined_label_encoded'] = predict_combined_label(test_df)

• 학습시켰던 모델 받아와주었습니다
• train_df 가 아닌, test_df의 answer_text column에 대해 예측을 수행하는 방식으로 코드를 작성했습니다

 

 

사진을 보면 모든 답변을 클래스 0인 "35세~44세의 경력직 여성"으로 예측한 것을 볼 수 있습니다. 데이터가 과적합된듯 하네요.. 이를 해결하기 위해서는 몇 가지 아이디어들이 떠오르는데 이번 프로젝트에서는 '텍스트 데이터 증강법'과 '클래스 불균형에 따른 가중치 조정'을 수행하도록 하겠습니다.

---

2. 텍스트 데이터 증강법 적용

K-TACC에서 개발한 Bert Augmentation 방법을 사용하기로 했습니다.

https://github.com/kyle-bong/K-TACC

GitHub - kyle-bong/K-TACC: 문맥을 고려한 한국어 텍스트 데이터 증강

 

1) 증강 방법 소개

해당 증강 방법을 조금 소개드리자면 랜덤 토큰 대체, 랜덤 토큰 삽입, 랜덤 부사 삽입 등 여러 증강 방법들이 있습니다.

위 url과 아래 증강 예시, 점수를 참고해서 기존 base 보다 성능이 향상된 Adverb augmentation 기법과 Random Masking Insertion 기법만 적용하도록 하겠습니다.

 

2) 증강 코드

코드가 매우 많이 엄청 상당히 길지만 Bert augmentation의 코드를 한 번에 담다보니까 길어졌고 중간 중간에 잘 보시면 클래스를 불러오고 함수 정의하고 데이터셋에 적용하는 부분이 나뉘어져 있으니 잘 확인해주시면 감사하겠습니다! 그래서 아래 코드를 적용하면 기존 약 45,000개의 데이터셋에서 3배인 135,000개의 데이터셋으로 증강이 완료되겠습니다.

!pip install selenium
!pip install kiwipiepy
!pip install quickspacer
!pip install -q sentence_transformers

import transformers
import re
import random
import numpy as np
from bs4 import BeautifulSoup
from selenium import webdriver
import random
import requests
from kiwipiepy import Kiwi
import time
from quickspacer import Spacer

class BERT_Augmentation():
    def __init__(self):
        self.model_name = 'monologg/koelectra-base-v3-generator'
        self.model = transformers.AutoModelForMaskedLM.from_pretrained(self.model_name)
        self.tokenizer = transformers.AutoTokenizer.from_pretrained(self.model_name)
        self.unmasker = transformers.pipeline("fill-mask", model=self.model, tokenizer=self.tokenizer)
        random.seed(42)
    def random_masking_replacement(self, sentence, ratio=0.15):
        """Masking random eojeol of the sentence, and recover them using PLM.

        Args:
            sentence (str): Source sentence
            ratio (int): Ratio of masking

        Returns:
          str: Recovered sentence
        """
        
        span = int(round(len(sentence.split()) * ratio))
        
        ## 품질 유지를 위해, 문장의 어절 수가 4 이하라면 원문장을 그대로 리턴합니다.
        if len(sentence.split()) <= 4:
            return sentence

        mask = self.tokenizer.mask_token
        unmasker = self.unmasker

        unmask_sentence = sentence
        ## 처음과 끝 부분을 [MASK]로 치환 후 추론할 때의 품질이 좋지 않음.
        random_idx = random.randint(1, len(unmask_sentence.split()) - span)
        
        unmask_sentence = unmask_sentence.split()
        ## del unmask_sentence[random_idx:random_idx+span]
        cache = []
        for _ in range(span):
            ## 처음과 끝 부분을 [MASK]로 치환 후 추론할 때의 품질이 좋지 않음.
            while cache and random_idx in cache:
                random_idx = random.randint(1, len(unmask_sentence) - 2)
            cache.append(random_idx)
            unmask_sentence[random_idx] = mask
            unmask_sentence = unmasker(" ".join(unmask_sentence))[0]['sequence']
            unmask_sentence = unmask_sentence.split()
        unmask_sentence = " ".join(unmask_sentence)
        unmask_sentence = unmask_sentence.replace("  ", " ")

        return unmask_sentence.strip()

    def random_masking_insertion(self, sentence, ratio=0.15):
        
        span = int(round(len(sentence.split()) * ratio))
        mask = self.tokenizer.mask_token
        unmasker = self.unmasker
        
        ## Recover
        unmask_sentence = sentence
        
        for _ in range(span):
            unmask_sentence = unmask_sentence.split()
            random_idx = random.randint(0, len(unmask_sentence)-1)
            unmask_sentence.insert(random_idx, mask)
            unmask_sentence = unmasker(" ".join(unmask_sentence))[0]['sequence']

        unmask_sentence = unmask_sentence.replace("  ", " ")

        return unmask_sentence.strip()

class AdverbAugmentation():
    def __init__(self):
        self.kiwi = Kiwi()
        self.spacing = Spacer().space
    def _adverb_detector(self, sentence):

        ## POS info
        pos_list = [(x[0], x[1]) for x in self.kiwi.tokenize(sentence)] # (token, pos)
        
        adverb_list = []
        for pos in pos_list:
            if pos[1] == "MAG" and len(pos[0]) > 1: ## 1음절 부사는 제외함.
                adverb_list.append(pos[0])
        return adverb_list

    def _get_gloss(self, word):
        res = requests.get("https://dic.daum.net/search.do?q=" + word, timeout=5)
        time.sleep(random.uniform(0.5,2.5))
        soup = BeautifulSoup(res.content, "html.parser")
        try:
            ## 첫 번째 뜻풀이.
            meaning = soup.find('span', class_='txt_search')
        except AttributeError:
            return word
        if not meaning:
            return word
        
        ## parsing 결과에서 한글만 추출
        meaning = re.findall('[가-힣]+', str(meaning))
        meaning = ' '.join(meaning)
        
        ## 띄어쓰기 오류 교정 (위 에 -> 위에)
        ## meaning = spell_checker.check(meaning).as_dict()['checked'].strip()
        meaning = self.spacing([meaning.replace(" ", "")])
        return meaning[0].strip()
    
    def adverb_gloss_replacement(self, sentence):
        adverb_list = self._adverb_detector(sentence)
        if adverb_list:
            ## 부사들 중에서 1개만 랜덤으로 선택합니다.
            adverb = random.choice(adverb_list)
            try:
                gloss = self._get_gloss(adverb)
                sentence = sentence.replace(adverb, gloss)
            except:
                pass
        return sentence

## 함수 정의 부분
BERT_aug = BERT_Augmentation()
random_masking_insertion = BERT_aug.random_masking_insertion

adverb_aug = AdverbAugmentation()
adverb_gloss_replacement = adverb_aug.adverb_gloss_replacement

## 함수 적용 부분
augmented_data = []

## tqdm 적용
for index, row in tqdm(df.iterrows(), desc="Augmenting Data", total=len(df)):
    original_text = row['answer_text']

    ## Apply random masking and insertion (함수 이름이 제대로 정의되어 있어야 함)
    bert_aug_text = random_masking_insertion(original_text)

    ## Apply adverb gloss replacement (함수 이름이 제대로 정의되어 있어야 함)
    adverb_aug_text = adverb_gloss_replacement(original_text)

    ## Add the augmented texts along with the original row's metadata
    augmented_data.append({
        'answer_text': bert_aug_text,
        'gender': row['gender'],
        'ageRange': row['ageRange'],
        'experience': row['experience'],
        'combined_label': row['combined_label'],
        'combined_label_encoded': row['combined_label_encoded']
    })

    augmented_data.append({
        'answer_text': adverb_aug_text,
        'gender': row['gender'],
        'ageRange': row['ageRange'],
        'experience': row['experience'],
        'combined_label': row['combined_label'],
        'combined_label_encoded': row['combined_label_encoded']
    })

## Convert augmented data into a DataFrame
augmented_df = pd.DataFrame(augmented_data)

## Concatenate the original and augmented DataFrames
final_df = pd.concat([df, augmented_df]).reset_index(drop=True)

## Augmented 데이터를 CSV로 저장
final_df.to_csv('/kaggle/working/augmented_interview.csv', index=False)

---

3. 클래스 불균형에 따른 가중치 조정

 

우리가 예측할 라벨의 클래스가 불균형한 문제가 있기 때문에 적은 클래스에 가중치를 높게 설정하도록 하겠습니다.

아래 코드처럼 적용할 수 있습니다.

from sklearn.utils.class_weight import compute_class_weight

## 클래스별 개수 확인 및 가중치 계산
class_labels = df['combined_label_encoded'].values
class_weights = compute_class_weight('balanced', classes=np.unique(class_labels), y=class_labels)
class_weights = torch.tensor(class_weights, dtype=torch.float).to(device)

## 손실 함수 수정: 클래스별 가중치를 포함하는 손실 함수 정의
from torch.nn import CrossEntropyLoss

## 모델의 forward pass를 수정하여 가중치 적용
class WeightedModelForSequenceClassification(AutoModelForSequenceClassification):
    def forward(self, input_ids=None, attention_mask=None, labels=None, **kwargs):
        ## 모델의 forward pass 실행
        outputs = super().forward(input_ids=input_ids, attention_mask=attention_mask, labels=labels, **kwargs)
        
        ## 로짓(logits)과 레이블(labels) 추출
        logits = outputs.logits
        
        ## 손실 계산 시 클래스 가중치 적용
        if labels is not None:
            loss_fct = CrossEntropyLoss(weight=class_weights)
            loss = loss_fct(logits.view(-1, self.num_labels), labels.view(-1))
            outputs = (loss,) + outputs[1:]

        return outputs

## WeightedModelForSequenceClassification으로 모델 변경
model = WeightedModelForSequenceClassification.from_pretrained('Junhoee/Kobigbird_HR', num_labels=len(np.unique(class_labels))).to(device)

 

---

4. 증강 후 재학습 및 추론 결과

데이터 증강 마친 후 다시 학습을 마치고 추론 결과 보여드립니다.

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch

## 모델과 토크나이저 로드
model = AutoModelForSequenceClassification.from_pretrained('Junhoee/Kobigbird_HR').to(device)
tokenizer = AutoTokenizer.from_pretrained("Junhoee/Kobigbird_HR")

## test_df의 answer_text에 대해 추론 수행
def predict_combined_label(test_df):
    all_predictions = []
    
    ## 배치 단위로 데이터를 처리할 수 있도록 설정 (메모리 절약을 위해)
    batch_size = 16
    for i in range(0, len(test_df), batch_size):
        ## answer_text 추출 및 토크나이징
        batch_texts = test_df['answer_text'].iloc[i:i+batch_size].tolist()
        inputs = tokenizer(batch_texts, padding=True, truncation=True, return_tensors="pt", max_length=512)

        ## 입력 데이터를 GPU로 이동
        inputs = {key: val.to(device) for key, val in inputs.items()}

        ## 추론 수행
        with torch.no_grad():
            outputs = model(**inputs)

        ## 로짓(logits) 출력
        logits = outputs.logits

        ## 예측 클래스 추출
        predicted_class = torch.argmax(logits, dim=1).cpu().numpy()
        all_predictions.extend(predicted_class)

    return all_predictions

## 예측 수행
test_df['predicted_combined_label_encoded'] = predict_combined_label(test_df)

## 실제 레이블과 예측 레이블이 일치하는지 여부 확인 (True/False)
correct_predictions = test_df['combined_label_encoded'] == test_df['predicted_combined_label_encoded']

## 일치하는 행의 개수를 셈
num_correct = correct_predictions.sum()

## 전체 데이터에서의 정확도 계산
accuracy = round(num_correct / len(test_df), 2) * 100
print("정확도 : ", accuracy)

 

그래서 현재 정확도 점수는 78점이구요.

다음 포스팅에는 78점보다 더 높여보도록 고민을 한 번 해보겠습니다.

어떤 걸 더 해야 올라갈지 고민이긴 합니다..

아무튼 이번 포스팅은 여기서 마치겠습니다~