[NLP] 레스토랑 리뷰 감성 분류하기 (3) (feat.파이토치로 배우는 자연어 처리) - 훈련 및 평가, 추론, 분석

-- 본 포스팅은 파이토치로 배우는 자연어 처리 (한빛미디어) 책을 참고해서 작성된 글입니다.
-- 소스코드 ) https://github.com/rickiepark/nlp-with-pytorch

GitHub - rickiepark/nlp-with-pytorch: <파이토치로 배우는 자연어 처리>(한빛미디어, 2021)의 소스 코드를

 

<PREVIOUS> 

https://didu-story.tistory.com/83?category=952805 

[NLP] 레스토랑 리뷰 감성 분류하기 (1) (feat.파이토치로 배우는 자연어 처리)

https://didu-story.tistory.com/86?category=952805 

 

 

▶ 레스토랑 리뷰 감성 분류하기

앞의 (1) , (2) 포스팅에서 데이터를 전처리하고, 데이터를 파이토치에서 활용 가능하게 만들어주는 여러가지 클래스에 대해서 살펴보았다. (여기 이해하는데 개오래걸림 ;;...)

이제 간단한 퍼셉트론 모델을 활용해서 본격적인 감성분류를 진행해보자.

 

1. 퍼셉트론 분류기 정의하기

• ReviewClassifier 클래스는 파이토치의 Module 클래스를 상속하고 단일 출력을 만드는 Linear층 하나를 생성하도록 할 것이다.
• 마지막에는 비선형 활성화 함수로 시그모이드 함수를 사용할 것이다.
• forward() 메서드
  • 선택적으로 시그모이드 함수를 적용하는 매개변수를 만듦
  • 이진분류 문제에서는 이진 크로스피 엔트로피 손실(BCELoss)가 가장 적절하지만, 시그모이드와 손실함수를 사용할 경우에 수치 안정성의 이슈가 발생한다고 한다.
  • 파이토치는 시그모이드 없이 간편하게 사용할 수있고, 수치적으로 안정된 계산을 위한 BSEWithLogitsLoss()를 사용할 수있다.

class ReviewClassifier(nn.Module):
    """ 간단한 퍼셉트론 기반 분류기 """
    def __init__(self, num_features):
        """
        매개변수:
            num_features (int): 입력 특성 벡트의 크기
        """
        super(ReviewClassifier, self).__init__()
        self.fc1 = nn.Linear(in_features=num_features, 
                             out_features=1)

    def forward(self, x_in, apply_sigmoid=False):
        """ 분류기의 정방향 계산
        
        매개변수:
            x_in (torch.Tensor): 입력 데이터 텐서 
                x_in.shape는 (batch, num_features)입니다.
            apply_sigmoid (bool): 시그모이드 활성화 함수를 위한 플래그
                크로스-엔트로피 손실을 사용하려면 False로 지정합니다
        반환값:
            결과 텐서. tensor.shape은 (batch,)입니다.
        """
        y_out = self.fc1(x_in).squeeze()
        ## 선택적으로 시그모이드 함수를 적용하기 위한 과정
        if apply_sigmoid:
            y_out = torch.sigmoid(y_out)
        return y_out

 

2. 모델 훈련

2.1 퍼셉트론 분류기를 위한 하이퍼 파라미터와 프로그램 옵션을 설정

from argparse import Namespace

args = Namespace(
    # 날짜와 경로 정보
    frequency_cutoff=25,
    model_state_file='model.pth',
    review_csv='data/yelp/reviews_with_splits_lite.csv',
    # review_csv='data/yelp/reviews_with_splits_full.csv',
    save_dir='model_storage/ch3/yelp/',
    vectorizer_file='vectorizer.json',
    # 모델 하이퍼파라미터 없음
    # 훈련 하이퍼파라미터
    batch_size=128,
    early_stopping_criteria=5,
    learning_rate=0.001,
    num_epochs=100,
    seed=1337,
    # 실행 옵션
    catch_keyboard_interrupt=True,
    cuda=True,
    expand_filepaths_to_save_dir=True,
    reload_from_files=False,
)

 

2.2 데이터셋, 모델, 손실, 옵티마이저, 훈련상태 딕셔너리 생성

import torch.optim as optim

def make_train_state(args):
  return {'epoch_index': 0,
          'train_loss': [],
          'train_acc': [],
          'val_loss': [],
          'val_acc': [],
          'test_loss': -1,
          'test_acc': -1}
train_state = make_train_state(args)

if not torch.cuda_is_available():
  args.cuda = False
args.device = torch.device("cuda" if args.cuda else "cpu")

# 데이터셋과 Vectorizer
dataset = ReviewDataset.load_dataset_and_make_vectorizer(args.review_csv)
vectorizer = dataset.get_vecgtorizer()

# 모델
classifier = ReviewClassifier(num_features=len(vectorizer.review_vocab))
classifier = classifier.to(args.device)

# 손실함수와 옵티마이저
loss_func = nn.BCEWithLogitsLoss()
optimizer = optim.Adam(classifier.parameters(), lr = args.learning_rage)

• args객체를 매개변수로 받아서 훈련 상태를 초기화하는 함수 작성 (make_train_state)
• 데이터셋과 모델생성
  • ReviewDataset 클래스에서 vectorizer를 생성해준다.
• 손실함수는 BCEWithLogistitsLoss() 사용
• 옵티마이저는 Adam

 

2.3 훈련 반복

    # 에포크 횟수만큼 for문을 반복할 것이다. (args에서 정의)
    for epoch_index in range(args.num_epochs):
        train_state['epoch_index'] = epoch_index

        # 훈련 세트에 대한 순회
        # 훈련 세트와 배치 제너레이터 준비, 손실과 정확도를 0으로 설정
        dataset.set_split('train')
        batch_generator = generate_batches(dataset, 
                                           batch_size=args.batch_size, 
                                           device=args.device)
        running_loss = 0.0
        running_acc = 0.0
        classifier.train()

        for batch_index, batch_dict in enumerate(batch_generator):
            # 훈련 과정은 5단계로 이루어집니다

            # --------------------------------------
            # 단계 1. 그레이디언트를 0으로 초기화합니다
            optimizer.zero_grad()

            # 단계 2. 출력을 계산합니다
            y_pred = classifier(x_in=batch_dict['x_data'].float())

            # 단계 3. 손실을 계산합니다
            loss = loss_func(y_pred, batch_dict['y_target'].float())
            loss_t = loss.item()
            running_loss += (loss_t - running_loss) / (batch_index + 1)

            # 단계 4. 손실을 사용해 그레이디언트를 계산합니다
            loss.backward()

            # 단계 5. 옵티마이저로 가중치를 업데이트합니다
            optimizer.step()
            # -----------------------------------------
            
            # 정확도를 계산합니다
            acc_t = compute_accuracy(y_pred, batch_dict['y_target'])
            running_acc += (acc_t - running_acc) / (batch_index + 1)

            # 진행 바 업데이트
            train_bar.set_postfix(loss=running_loss, 
                                  acc=running_acc, 
                                  epoch=epoch_index)
            train_bar.update()

        train_state['train_loss'].append(running_loss)
        train_state['train_acc'].append(running_acc)

        # 검증 세트에 대한 순회

        # 검증 세트와 배치 제너레이터 준비, 손실과 정확도를 0으로 설정
        dataset.set_split('val')
        batch_generator = generate_batches(dataset, 
                                           batch_size=args.batch_size, 
                                           device=args.device)
        running_loss = 0.
        running_acc = 0.
        classifier.eval()

        for batch_index, batch_dict in enumerate(batch_generator):

            # 단계 1. 출력을 계산합니다
            y_pred = classifier(x_in=batch_dict['x_data'].float())

            # 단계 2. 손실을 계산합니다
            loss = loss_func(y_pred, batch_dict['y_target'].float())
            loss_t = loss.item()
            running_loss += (loss_t - running_loss) / (batch_index + 1)

            # 단계 3. 정확도를 계산합니다
            acc_t = compute_accuracy(y_pred, batch_dict['y_target'])
            running_acc += (acc_t - running_acc) / (batch_index + 1)
            
            val_bar.set_postfix(loss=running_loss, 
                                acc=running_acc, 
                                epoch=epoch_index)
            val_bar.update()

        train_state['val_loss'].append(running_loss)
        train_state['val_acc'].append(running_acc)

• 내부 for loop : 미니 배치에 대해서 반복 수행
  • 미니배치 : 예측 - 손실계산 - 정확도 계산 
• 외부 for loop : 내부반복문을 여러번 반복한다. 내부 반복문에서 미니배치마다 손실을 계산하고 옵티마이저가 모델 파라미터를 업데이트 해준다. 

 

 

2.4 평가, 추론, 분석

2.4.1 테스트 데이터로 평가하기

• 사용 데이터를 val 대신 test로 지정

dataset.set_split('test')
batch_generator = generate_batches(dataset, 
                                    batch_size=args.batch_size, 
                                    device=args.device)
running_loss = 0.
running_acc = 0.
classifier.eval()

for batch_index, batch_dict in enumerate(batch_generator):

    # 단계 1. 출력을 계산합니다
    y_pred = classifier(x_in=batch_dict['x_data'].float())

    # 단계 2. 손실을 계산합니다
    loss = loss_func(y_pred, batch_dict['y_target'].float())
    loss_batch = loss.item()
    running_loss += (loss_batch - running_loss) / (batch_index + 1)

    # 단계 3. 정확도를 계산합니다
    acc_batch = compute_accuracy(y_pred, batch_dict['y_target'])
    running_acc += (acc_batch - running_acc) / (batch_index + 1)
    
train_state['test_loss'].append(running_loss)
train_state['test_acc'].append(running_acc)

print("Test loss: {:.3f}".format(train_state['test_loss']))
print("Test Accuracy: {:.2f}".format(train_state['test_acc']))

 

2.4.2 새로운 데이터 포인트 추론하여 분류하기

# 정규식을 사용하여 text를 토큰화
def preprocess_text(text):
    text = text.lower()
    text = re.sub(r"([.,!?])", r" \1 ", text)
    text = re.sub(r"[^a-zA-Z.,!?]+", r" ", text)
    return text

def predict_rating(review, classifier, vectorizer, decision_threshold=0.5):
    """ 리뷰 점수 예측하기
    
    매개변수:
        review (str): 리뷰 텍스트
        classifier (ReviewClassifier): 훈련된 모델
        vectorizer (ReviewVectorizer): Vectorizer 객체
        decision_threshold (float): 클래스를 나눌 결정 경계
    """
    review = preprocess_text(review)
    
    vectorized_review = torch.tensor(vectorizer.vectorize(review))
    result = classifier(vectorized_review.view(1, -1))
    
    probability_value = torch.sigmoid(result).item()
    index = 1
    if probability_value < decision_threshold:
        index = 0

    return vectorizer.rating_vocab.lookup_index(index)

test_review = "this is a pretty awesome book"

#위의 문장을 예측해서 분류해보기
classifier = classifier.cpu()
prediction = predict_rating(test_review, classifier, vectorizer, decision_threshold=0.5)
print("{} -> {}".format(test_review, prediction))

 

2.4.3 모델 가중치 분석

훈련이 끝난 뒤 모델이 잘 작동하는지 알아보기 위해 가중치를 분석해보자.

# 가중치 정렬
fc1_weights = classifier.fc1.weight.detach()[0]
_, indices = torch.sort(fc1_weights, dim=0, descending=True)
indices = indices.numpy().tolist()

# 긍정적인 상위 20개 단어
print("긍정 리뷰에 영향을 미치는 단어:")
print("--------------------------------------")
for i in range(20):
    print(vectorizer.review_vocab.lookup_index(indices[i]))
    
print("====\n\n\n")

# 부정적인 상위 20개 단어
print("부정 리뷰에 영향을 미치는 단어:")
print("--------------------------------------")
indices.reverse()
for i in range(20):
    print(vectorizer.review_vocab.lookup_index(indices[i]))