AI 인공지능
yelp 리뷰 감성 분류 (1)
블레이크-정
2023. 3. 30. 17:12
이 글은 파이토치로 배우는 자연어처리(O'REILLY, 한빛미디어)를 공부한 내용을 바탕으로 작성하였습니다.
yelp 레스토랑 리뷰 데이터셋
리뷰 텍스트(review)와 그에 맞는 평가 레이블(rating)를 준비하고, train, valid, test (0.7, 0.15, 0.15)로 분류
from argparse import Namespace
from collections import Counter
import json
import os
import re
import string
import numpy as np
import pandas as pd
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
import tqdm
Dataset
review dataset 로드하고, ReviewVectorizer 객체 생성
class ReviewDataset(Dataset):
def __init__(self, review_df, vectorizer):
"""
매개변수:
review_df (pandas.DataFrame): 데이터셋
vectorizer (ReviewVectorizer): ReviewVectorizer 객체
"""
self.review_df = review_df
self._vectorizer = vectorizer
self.train_df = self.review_df[self.review_df.split=='train']
self.train_size = len(self.train_df)
self.val_df = self.review_df[self.review_df.split=='val']
self.validation_size = len(self.val_df)
self.test_df = self.review_df[self.review_df.split=='test']
self.test_size = len(self.test_df)
self._lookup_dict = {'train': (self.train_df, self.train_size),
'val': (self.val_df, self.validation_size),
'test': (self.test_df, self.test_size)}
self.set_split('train')
@classmethod
def load_dataset_and_make_vectorizer(cls, review_csv):
""" 데이터셋을 로드하고 새로운 ReviewVectorizer 객체를 만듭니다
매개변수:
review_csv (str): 데이터셋의 위치
반환값:
ReviewDataset의 인스턴스
"""
review_df = pd.read_csv(review_csv)
train_review_df = review_df[review_df.split=='train']
return cls(review_df, ReviewVectorizer.from_dataframe(train_review_df))
@classmethod
def load_dataset_and_load_vectorizer(cls, review_csv, vectorizer_filepath):
""" 데이터셋을 로드하고 새로운 ReviewVectorizer 객체를 만듭니다.
캐시된 ReviewVectorizer 객체를 재사용할 때 사용합니다.
매개변수:
review_csv (str): 데이터셋의 위치
vectorizer_filepath (str): ReviewVectorizer 객체의 저장 위치
반환값:
ReviewDataset의 인스턴스
"""
review_df = pd.read_csv(review_csv)
vectorizer = cls.load_vectorizer_only(vectorizer_filepath)
return cls(review_df, vectorizer)
@staticmethod
def load_vectorizer_only(vectorizer_filepath):
""" 파일에서 ReviewVectorizer 객체를 로드하는 정적 메서드
매개변수:
vectorizer_filepath (str): 직렬화된 ReviewVectorizer 객체의 위치
반환값:
ReviewVectorizer의 인스턴스
"""
with open(vectorizer_filepath) as fp:
return ReviewVectorizer.from_serializable(json.load(fp))
def save_vectorizer(self, vectorizer_filepath):
""" ReviewVectorizer 객체를 json 형태로 디스크에 저장합니다
매개변수:
vectorizer_filepath (str): ReviewVectorizer 객체의 저장 위치
"""
with open(vectorizer_filepath, "w") as fp:
json.dump(self._vectorizer.to_serializable(), fp)
def get_vectorizer(self):
""" 벡터 변환 객체를 반환합니다 """
return self._vectorizer
def set_split(self, split="train"):
""" 데이터프레임에 있는 열을 사용해 분할 세트를 선택합니다
매개변수:
split (str): "train", "val", "test" 중 하나
"""
self._target_split = split
self._target_df, self._target_size = self._lookup_dict[split]
def __len__(self):
return self._target_size
def __getitem__(self, index):
""" 파이토치 데이터셋의 주요 진입 메서드
매개변수:
index (int): 데이터 포인트의 인덱스
반환값:
데이터 포인트의 특성(x_data)과 레이블(y_target)로 이루어진 딕셔너리
"""
row = self._target_df.iloc[index]
review_vector = \
self._vectorizer.vectorize(row.review)
rating_index = \
self._vectorizer.rating_vocab.lookup_token(row.rating)
return {'x_data': review_vector,
'y_target': rating_index}
def get_num_batches(self, batch_size):
""" 배치 크기가 주어지면 데이터셋으로 만들 수 있는 배치 개수를 반환합니다
매개변수:
batch_size (int)
반환값:
배치 개수
"""
return len(self) // batch_sizeVocabulary
토큰을 정수로 매핑하는 단계
token_to_idx 과 idx_to_token 두 가지 딕셔너리 반환
add_token() : 새로운 토큰 추가하기
lookup_token() : 특정 토큰에 해당하는 인덱스 추출
lookup_index() : 특정 인덱스에 해당하는 토큰 추출
class Vocabulary(object):
""" 매핑을 위해 텍스트를 처리하고 어휘 사전을 만드는 클래스 """
def __init__(self, token_to_idx=None, add_unk=True, unk_token="<UNK>"):
"""
매개변수:
token_to_idx (dict): 기존 토큰-인덱스 매핑 딕셔너리
add_unk (bool): UNK 토큰을 추가할지 지정하는 플래그
unk_token (str): Vocabulary에 추가할 UNK 토큰
"""
if token_to_idx is None:
token_to_idx = {}
self._token_to_idx = token_to_idx
self._idx_to_token = {idx: token
for token, idx in self._token_to_idx.items()}
self._add_unk = add_unk
self._unk_token = unk_token
self.unk_index = -1
if add_unk:
self.unk_index = self.add_token(unk_token)
def to_serializable(self):
""" 직렬화할 수 있는 딕셔너리를 반환합니다 """
return {'token_to_idx': self._token_to_idx,
'add_unk': self._add_unk,
'unk_token': self._unk_token}
@classmethod
def from_serializable(cls, contents):
""" 직렬화된 딕셔너리에서 Vocabulary 객체를 만듭니다 """
return cls(**contents)
def add_token(self, token):
""" 토큰을 기반으로 매핑 딕셔너리를 업데이트합니다
매개변수:
token (str): Vocabulary에 추가할 토큰
반환값:
index (int): 토큰에 상응하는 정수
"""
if token in self._token_to_idx:
index = self._token_to_idx[token]
else:
index = len(self._token_to_idx)
self._token_to_idx[token] = index
self._idx_to_token[index] = token
return index
def add_many(self, tokens):
""" 토큰 리스트를 Vocabulary에 추가합니다.
매개변수:
tokens (list): 문자열 토큰 리스트
반환값:
indices (list): 토큰 리스트에 상응되는 인덱스 리스트
"""
return [self.add_token(token) for token in tokens]
def lookup_token(self, token):
""" 토큰에 대응하는 인덱스를 추출합니다.
토큰이 없으면 UNK 인덱스를 반환합니다.
매개변수:
token (str): 찾을 토큰
반환값:
index (int): 토큰에 해당하는 인덱스
노트:
UNK 토큰을 사용하려면 (Vocabulary에 추가하기 위해)
`unk_index`가 0보다 커야 합니다.
"""
if self.unk_index >= 0:
return self._token_to_idx.get(token, self.unk_index)
else:
return self._token_to_idx[token]
def lookup_index(self, index):
""" 인덱스에 해당하는 토큰을 반환합니다.
매개변수:
index (int): 찾을 인덱스
반환값:
token (str): 인텍스에 해당하는 토큰
에러:
KeyError: 인덱스가 Vocabulary에 없을 때 발생합니다.
"""
if index not in self._idx_to_token:
raise KeyError("Vocabulary에 인덱스(%d)가 없습니다." % index)
return self._idx_to_token[index]
def __str__(self):
return "<Vocabulary(size=%d)>" % len(self)
def __len__(self):
return len(self._token_to_idx)Vectorizer
vocabulary에서 토큰-정수 매핑한 정보를 바탕으로 정수 인덱스를 1로 하는 원-핫 벡터 생성
메모리가 낭비되므로, 25번 이상 등장한 단어로 제한
class ReviewVectorizer(object):
""" 어휘 사전을 생성하고 관리합니다 """
def __init__(self, review_vocab, rating_vocab):
"""
매개변수:
review_vocab (Vocabulary): 단어를 정수에 매핑하는 Vocabulary
rating_vocab (Vocabulary): 클래스 레이블을 정수에 매핑하는 Vocabulary
"""
self.review_vocab = review_vocab
self.rating_vocab = rating_vocab
def vectorize(self, review):
""" 리뷰에 대한 원-핫 벡터를 만듭니다
매개변수:
review (str): 리뷰
반환값:
one_hot (np.ndarray): 원-핫 벡터
"""
one_hot = np.zeros(len(self.review_vocab), dtype=np.float32)
for token in review.split(" "):
if token not in string.punctuation:
one_hot[self.review_vocab.lookup_token(token)] = 1
return one_hot
@classmethod
def from_dataframe(cls, review_df, cutoff=25):
""" 데이터셋 데이터프레임에서 Vectorizer 객체를 만듭니다
매개변수:
review_df (pandas.DataFrame): 리뷰 데이터셋
cutoff (int): 빈도 기반 필터링 설정값
반환값:
ReviewVectorizer 객체
"""
review_vocab = Vocabulary(add_unk=True)
rating_vocab = Vocabulary(add_unk=False)
# 점수를 추가합니다
for rating in sorted(set(review_df.rating)):
rating_vocab.add_token(rating)
# count > cutoff인 단어를 추가합니다
word_counts = Counter()
for review in review_df.review:
for word in review.split(" "):
if word not in string.punctuation:
word_counts[word] += 1
for word, count in word_counts.items():
if count > cutoff:
review_vocab.add_token(word)
return cls(review_vocab, rating_vocab)
@classmethod
def from_serializable(cls, contents):
""" 직렬화된 딕셔너리에서 ReviewVectorizer 객체를 만듭니다
매개변수:
contents (dict): 직렬화된 딕셔너리
반환값:
ReviewVectorizer 클래스 객체
"""
review_vocab = Vocabulary.from_serializable(contents['review_vocab'])
rating_vocab = Vocabulary.from_serializable(contents['rating_vocab'])
return cls(review_vocab=review_vocab, rating_vocab=rating_vocab)
def to_serializable(self):
""" 캐싱을 위해 직렬화된 딕셔너리를 만듭니다
반환값:
contents (dict): 직렬화된 딕셔너리
"""
return {'review_vocab': self.review_vocab.to_serializable(),
'rating_vocab': self.rating_vocab.to_serializable()}DataLoader
Dataset을 배치 단위로 순회하며 반환
generate_batches() : 현재 이용할 데이터를 지정된 device로 보내는 함수
def generate_batches(dataset, batch_size, shuffle=True,
drop_last=True, device="cpu"):
"""
파이토치 DataLoader를 감싸고 있는 제너레이터 함수.
걱 텐서를 지정된 장치로 이동합니다.
"""
dataloader = DataLoader(dataset=dataset, batch_size=batch_size,
shuffle=shuffle, drop_last=drop_last)
for data_dict in dataloader:
out_data_dict = {}
for name, tensor in data_dict.items():
out_data_dict[name] = data_dict[name].to(device)
yield out_data_dict
이렇게 데이터를 모델에 입력하도록 클래스를 구성했다. 다음 게시물에서 분류기 모델을 만들고, 훈련을 준비해보자.