[파이토치로 배우는 자연어처리] 레스토랑 리뷰 감성 분류하기 - (3) 감성 분류 모델 만들기

이 글은 한빛미디어의 '파이토치로 배우는 자연어처리' 글을 읽고 정리한 것입니다.

저번 포스트까지 텍스트 데이터를 정제하고, 단어-인덱스를 매핑하는 사전을 만들고, 사전을 바탕으로 리뷰 문장을 벡터화하는 과정을 마쳤다. 이번 포스트에서는 이렇게 가공해 둔 문장을 가지고 모델을 만들어 학습시키고 성능을 측정해 보도록 하겠다.

GitHub - mori8/NLP-Pytorch-practice: PyTorch Zero To All(by Sung Kim) 강의와 '파이토치로 배우는 자연어 처리' 책

 

ReviewClassifier 모델

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

 

 

훈련 과정

본격적으로 훈련 과정에 들어가기 전, 몇 개의 헬퍼 함수를 만들어 놓자.

• set_seed_everywhere(): 다른 환경에서도 동일한 결과를 출력하기 위해 Random Seed를 모두 동일하게 설정한다.
• handle_dirs(): dirpath가 존재하는 디렉터리인지 확인하고, 존재하지 않는다면 직접 만든다.

def set_seed_everywhere(seed, cuda):
  np.random.seed(seed)  # Python의 Random으로 PyTorch transforms의 random seed 고정
  torch.manual_seed(seed)  # PyTorch의 random seed 고정
  if cuda:
    torch.cuda.manual_seed_all(seed)  # CUDA와 CuDNN random seed 고정
# 참고: https://tootouch.github.io/experiment/reproduction_pytorch/


def handle_dirs(dirpath):
  if not os.path.exists(dirpath):
    os.makedirs(dirpath)

다음으로, 모델 생성과 훈련 과정에서 사용할 변수들을 Namespace 안에 설정한다.

args = Namespace(
    frequency_cutoff=25,
    # 학습 데이터셋에서 25번 이상 출현한 단어만 학습, 나머지는 UNK으로 처리
    model_state_file='model_pth',
    # 모델 상태를 저장할 파일 이름
    review_csv='/content/drive/MyDrive/NLP-Pytorch/data/yelp/reviews_with_splits_lite.csv',
    # 전처리를 마친 리뷰 csv 파일의 위치
    save_dir='model_storage/ch3/yelp/',
    # vectorizer_file과 model_state_file을 저장할 디렉터리
    vectorizer_file='vectorizer.json',
    # vectorizer를 저장할 파일 이름
    batch_size=128,
    # 배치 사이즈
    early_stopping_criteria=5,
    # 오버피팅 방지를 위한 조기 종료 기준
    learning_rate=0.001,
    # 학습율
    num_epochs=100,
    # 학습을 반복할 횟수
    seed=1337,
    # 시드
    catch_keyboard_interrupt=True,
    # 키보드 인터럽트 처리?
    cuda=True,
    # GPU 사용?
    expand_filepaths_to_save_dir=True,
    # filepaths 앞에 save_dir을 붙여?
    reload_from_files=False,
    # 체크포인트에서 훈련을 다시 시작하는 경우?
)

파일 경로와 gpu 설정을 마친다.

if args.expand_filepaths_to_save_dir:
  args.vectorizer_file = os.path.join(args.save_dir, args.vectorizer_file)
  args.model_state_file = os.path.join(args.save_dir, args.model_state_file)

  print("파일 경로: ")
  print("\t{}".format(args.vectorizer_file))
  print("\t{}".format(args.model_state_file))

if not torch.cuda.is_available():
  args.cuda = False

print("CUDA 사용 여부: {}".format(args.cuda))

args.device = torch.device("cuda" if args.cuda else "cpu")
set_seed_everywhere(args.seed, args.cuda)
handle_dirs(args.save_dir)

훈련을 도와줄 헬퍼 함수들을 정의한다.

• make_train_state(): 훈련 과정 중 훈련 상태를 저장할 변수들의 묶음을 만든다.
• update_train_state(): 훈련 상태를 업데이트한다. 성능이 향상되면 현재 모델을 저장하여 최상의 모델을 사용할 수 있도록 한다.
• compute_accuracy(): 정확도를 계산한다.

def make_train_state(args):
  return {'stop_early': False,
          'early_stopping_step': 0,
          'early_stopping_best_val': 1e8,
          'learning_rate': args.learning_rate,
          'epoch_index': 0,
          'train_loss': [],
          'train_acc': [],
          'val_loss': [],
          'val_acc': [],
          'test_loss': -1,
          'test_acc': -1,
          'model_filename': args.model_state_file}


def update_train_state(args, model, train_state):
  """ 훈련 상태를 업데이트합니다.

  Components:
  - 조기 종료: 과대 적합 방지
  - 모델 체크포인트: 더 나은 모델을 저장합니다

  :param args: 메인 매개변수
  :param model: 훈련할 모델
  :param train_state: 훈련 상태를 담은 딕셔너리
  :returns: 새로운 훈련 상태
  """

  # 적어도 한 번 모델을 저장합니다
  if train_state['epoch_index'] == 0:
    torch.save(model.state_dict(), train_state['model_filename'])
    train_state['stop_early'] = False

  # 성능이 향상되면 모델을 저장합니다
  elif train_state['epoch_index'] >= 1:
    loss_tm1, loss_t = train_state['val_loss'][-2:]

    # 손실이 나빠지면
    if loss_t >= train_state['early_stopping_best_val']:
      # 조기 종료 단계 업데이트
      train_state['early_stopping_step'] += 1
    # 손실이 감소하면
    else:
      # 최상의 모델 저장
      if loss_t < train_state['early_stopping_best_val']:
        torch.save(model.state_dict(), train_state['model_filename'])

      # 조기 종료 단계 재설정
      train_state['early_stopping_step'] = 0

    # 조기 종료 여부 확인
    train_state['stop_early'] = \
      train_state['early_stopping_step'] >= args.early_stopping_criteria

  return train_state

def compute_accuracy(y_pred, y_target):
    y_target = y_target.cpu()
    y_pred_indices = (torch.sigmoid(y_pred)>0.5).cpu().long()#.max(dim=1)[1]
    n_correct = torch.eq(y_pred_indices, y_target).sum().item()
    return n_correct / len(y_pred_indices) * 100

다음으로, 데이터셋과 Vectorizer를 준비한다. 전에 만들어 둔 Vectorizer가 있다면 체크포인트에서 훈련을 다시 시작할 수도 있다. 우리가 만든 간단한 분류 모델인 ReviewClassifier 객체를 생성하고, loss function과 optimizer를 설정한다. scheduler로 학습 과정에서 learning rate를 조정할 수 있도록 한다.

if args.reload_from_files:
    # 체크포인트에서 훈련을 다시 시작
    print("데이터셋과 Vectorizer를 로드합니다")
    dataset = ReviewDataset.load_dataset_and_load_vectorizer(args.review_csv,
                                                            args.vectorizer_file)
else:
    print("데이터셋을 로드하고 Vectorizer를 만듭니다")
    # 데이터셋과 Vectorizer 만들기
    dataset = ReviewDataset.load_dataset_and_make_vectorizer(args.review_csv)
    dataset.save_vectorizer(args.vectorizer_file)    
vectorizer = dataset.get_vectorizer()

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_rate)
scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer=optimizer,
                                                 mode='min', factor=0.5,
                                                 patience=1)
train_state = make_train_state(args)

훈련에 들어가기 전, 먼저 tqdm 모듈로 전체적인 학습 과정과 훈련 데이터 학습 과정, validataion 데이터셋을 통한 검증 과정을 프로그래스바로 시각화할 수 있도록 한다.

# tqdm: 진행률 프로세스바 제공
epoch_bar = tqdm.notebook.tqdm(desc='training routine',
                               total=args.num_epochs,
                               position=0)
dataset.set_split('train')
train_bar = tqdm.notebook.tqdm(desc='split=train',
                               total=dataset.get_num_batches(args.batch_size),
                               position=1,
                               leave=True)
dataset.set_split('val')
val_bar = tqdm.notebook.tqdm(desc='split=val',
                               total=dataset.get_num_batches(args.batch_size),
                               position=1,
                               leave=True)

tqdm 모듈을 사용하면 아래처럼 진행율을 직관적으로 확인 가능하다.

 

이제 본격적으로 훈련을 시작해보자. args.num_epochs에 설정해 둔 값만큼 학습을 반복한다. 저번 포스트에서 만든 generate_batches로 훈련(train) 데이터셋을 args.batch_size 개수만큼의 배치로 나눈다. 배치(batch)는 모델의 가중치를 한 번 업데이트할 때 사용되는 데이터의 묶음을 의미한다. 학습은 5단계로 진행되며, 각 단계에 대한 설명은 코드블록 내에 주석으로 달아놓았다.
훈련이 끝나면 검증(val) 데이터셋으로 만들어진 모델을 평가한다. 검증 과정은 3단계로 진행된다. 훈련과 검증을 정해진 epochs 개수만큼 반복하면서 진행율을 업데이트한다. 정해진 epochs를 다 돌았거나, KeyboardInterrupt를 받았거나, 조기 종료 조건을 만족한 경우 학습을 마친다.

# tqdm: 진행률 프로세스바 제공
epoch_bar = tqdm.notebook.tqdm(desc='training routine',
                               total=args.num_epochs,
                               position=0)
dataset.set_split('train')
train_bar = tqdm.notebook.tqdm(desc='split=train',
                               total=dataset.get_num_batches(args.batch_size),
                               position=1,
                               leave=True)
dataset.set_split('val')
val_bar = tqdm.notebook.tqdm(desc='split=val',
                               total=dataset.get_num_batches(args.batch_size),
                               position=1,
                               leave=True)

try:
  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):
      # 단계 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)

    # validation 세트에 대한 순회
    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)

    train_state = update_train_state(args=args, model=classifier,
                                     train_state=train_state)
    scheduler.step(train_state["val_loss"][-1])

    train_bar.n = 0
    val_bar.n = 0
    epoch_bar.update()

    if train_state['stop_early']:
      break

    train_bar.n = 0
    val_bar.n = 0
    epoch_bar.update()
except KeyboardInterrupt:
  print("Exiting loop")

이제 우리가 학습시킨 모델을 시험해 볼 차례다.
우리는 위에서 만든 update_train_state() 함수를 통해 학습 과정에서 성능이 가장 좋은 모델을 저장해뒀다. 이 모델은 train_state['model_filename']의 위치에 저장되어 있으므로, 이 파일을 불러와 가장 성능이 좋은 모델을 가져온다.

# 가장 좋은 모델을 사용해 테스트 세트의 손실과 정확도를 계산합니다
classifier.load_state_dict(torch.load(train_state['model_filename']))
classifier = classifier.to(args.device)

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):
    # 출력을 계산합니다
    y_pred = classifier(x_in=batch_dict['x_data'].float())

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

    # 정확도를 계산합니다
    acc_t = compute_accuracy(y_pred, batch_dict['y_target'])
    running_acc += (acc_t - running_acc) / (batch_index + 1)

train_state['test_loss'] = running_loss
train_state['test_acc'] = running_acc

이제 테스트된 모델의 loss와 정확도는 각각 train_state['test_loss']와 train_state['test_acc']에 저장되어 있다. 테스트 loss와 정확도를 출력해 보자.

print("테스트 손실: {:.3f}".format(train_state['test_loss']))
print("테스트 정확도: {:.2f}".format(train_state['test_acc']))

테스트 손실: 0.214
테스트 정확도: 91.84

0.214의 Loss와 91.84%의 정확도를 기록했다. 귯

 

 

추론

이제 모델 학습은 끝났다. 학습된 모델에 리뷰 텍스트를 넣어 해당 리뷰가 긍정적인 리뷰인지, 부정적인 리뷰인지 판단해 보자.
리뷰 텍스트를 넣어보기 전에, 우리가 만든 모델은 전처리된 텍스트 데이터를 입력받으므로 텍스트 전처리를 위한 preprocess_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

모든 준비는 끝났다. 이제 리뷰 텍스트와 훈련된 모델, Vectorizer 객체, positive와 negative를 나눌 경계값(threshold)을 매개변수로 받아 positive 혹은 negative를 반환하는 predict_rating() 함수를 만들어 보자.

def predict_rating(review, classifier, vectorizer, decision_threshold=0.5):
  """리뷰 점수 예측하기

  매개변수:
    review(str): 리뷰 텍스트
    classifier(ReviewClassifier): 훈련된 모델
    vectorizer(ReviewVectorizer): Vectorizer 객체
    decision_threshold(float): 클래스를 나눌 결정 경계. 얘보다 높으면 positive, 낮으면 negative
  """
  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)

이제 시험삼아 넣어볼 리뷰 문장을 준비해 보자. 이 책에서는 "this is a pretty awesome book"이라는 문장을 준비했다. classifier 모델을 준비하고, 이 모델을 예제 문장과 vectorizer와 함께 predict_rating() 함수에 넣으면 prediction 변수에 결과가 저장된다.

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))

결과는 this is a pretty awesome book -> positive이다. 참 똑똑하다

다음 글에서는 자연어 처리를 위한 피드포워드 신경망에 대해서 알아보도록 하자.