작년에 이 분야를 배우기 시작하면서 진행했던 감성 분석은 기본 중의 기본 같은 느낌이었다.
당시에는 뭘 어떻게 해야하는 지? 데이터는 어떻게 다뤄야하는 지? 뭐가 어떻게 돌아가는 건지? 등등
개념이 제대로 잡히지 않은 상태에서 진행했던 프로젝트였다.
(물론 지금도 제대로 잡혔다는 건 아니지만...적어도 1년동안 발전했으니까)
나는 당시 프로젝트를 되게 자랑스럽게 생각했지만 부족한 부분이 많았다.
이후로 더 공부하다보니 요즘 자연어처리는 Transformer계열이 대세라고 했다. 사실 대세가 된 지 좀 시간이 흘렀다.
어찌되었든 대학원에서도 자연어처리 수업을 들으며 그리고 개인적으로 공부하고 논문을 읽어보며 100% 이해는 못했지만 어느정도 이해는 했고 직접 써봐야겠다는 생각이 들었다. 그렇게 작년에 하지 못했던 7가지 감성 분석을 진행해보았다.
(RTX3070 8GB 로컬에서 작업하였으며 대략 1epoch당 12분이상 걸렸습니다.)
먼저, 데이터를 준비해야한다!
Low-Resource 데이터에 대한 NLP방법에 대한 논문(A Survey on Recent Approaches for Natural Language Processing in Low-Resource Scenarios)을 발표준비하고 있는 입장에서 이것까지 활용했으면 더 좋았겠지만 목적은 저번에 못해낸 7가지 분류니까 저번에 사용했던 한국어 단발성 대화세트와 연속성 대화세트 두 개를 사용했다.
필요한 모듈 임포트 및 cuda와 kobert모델 세팅
import torch
from torch import nn
import torch.nn.functional as F
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
import gluonnlp as nlp
import numpy as np
from tqdm.notebook import tqdm
import pandas as pd
import numpy as np
from kobert.utils import get_tokenizer
from kobert.pytorch_kobert import get_pytorch_kobert_model
from transformers import AdamW
from transformers.optimization import get_cosine_schedule_with_warmup
device = torch.device("cuda:0")
bertmodel, vocab = get_pytorch_kobert_model()
데이터 2개가 쪼개져있으니까 이를 합쳐주자
심플한 내용이니 한번에 다루겠다.
short_sen = pd.read_csv('단발성 대화 정리.csv')
long_sen = pd.read_csv('연속적 대화 정리.csv')
short_sen = short_sen.rename(columns={"발화":'sentence','감정':'str','감정.1':'Emotion'})
long_sen = long_sen.rename(columns={"발화":'sentence','감정_str':'str','감정_int':'Emotion'})
total_data = pd.concat([short_sen, long_sen], axis=0)
total_data = total_data.replace({'Emotion' : 7}, 0)
emotion_dict = {}
for i in sorted(emotion_list):
temp = long_sen[long_sen['Emotion']==i].sample(1)['str'].values[0]
# print(temp)
emotion_dict[i] = temp
total_data = pd.concat([short_sen, long_sen], axis=0)두 데이터셋의 컬럼명이 달랐기 때문에 통일시켜준 다음에 concat을 통해 합쳐주었고 각 숫자별로 어떠한 상태를 나타내는 지에 대해서 emotion_dict을 만들었다. 이는 추후에 predict로직에서 활용한다.
label값이 1~7로 들어가있는데 7 -> 0으로 변환시켜주었다. 왜 이런 것을 하는 지에 대해서는 아래에 분류클래스를 만드는 부분에서 설명하겠다.
모아놓은 데이터를 ['문장','라벨'] 의 꼴로 처리하는 작업을 진행해준다.
data_list = []
for q, label in zip(total_data['sentence'], total_data['Emotion']) :
data = []
data.append(q)
data.append(str(label))
data_list.append(data)
그리고 train과 test로 분리해서 진행을 한다! 전부 다 넣어도 상관없을 수 있지만 학습이 잘 되어가는 지 확인하기 위함이다. test_size와 random_state는 본인 취향에 맞게 설정해주면 된다. random_state는 굳이 안넣어도 되지만 개인적으로는 계속 반복해도 동일한 값을 보는 것을 선호해서 처음에는 넣어주는 편이다.
from sklearn.model_selection import train_test_split
dataset_train, dataset_test = train_test_split(data_list, test_size=0.25, random_state=1993)
BERT에 집어넣기 위한 dataset을 만들어주기 위해 클래스를 선언해준다.
class BERTDataset(Dataset):
def __init__(self, dataset, sent_idx, label_idx, bert_tokenizer, max_len, pad, pair):
transform = nlp.data.BERTSentenceTransform(
bert_tokenizer, max_seq_length=max_len, pad=pad, pair=pair)
self.sentences = [transform([i[sent_idx]]) for i in dataset]
self.labels = [np.int32(i[label_idx]) for i in dataset]
def __getitem__(self, i):
return (self.sentences[i] + (self.labels[i], ))
def __len__(self):
return (len(self.labels))그리고 파라미터를 한번 정립을 해주자
max_len = 128
batch_size = 10
warmup_ratio = 0.1
num_epochs = 5
max_grad_norm = 1
log_interval = 200
learning_rate = 5e-5로컬에서 작업을 하다보니 gpu메모리가 적어서 batch_size를 10으로 잡아주었고 epochs도 너무 오래걸리다보니 5번만 진행하였다. 그 이외의 값은 베이스에서 추가로 건들지 않았다.
tokenizer = get_tokenizer()
tok = nlp.data.BERTSPTokenizer(tokenizer, vocab, lower=False)
data_train = BERTDataset(dataset_train, 0, 1, tok, max_len, True, False)
data_test = BERTDataset(dataset_test, 0, 1, tok, max_len, True, False)토크나이저와 패딩를 통하게 되면 데이터는 다음과 같이 변한다.
train_dataloader = torch.utils.data.DataLoader(data_train, batch_size=batch_size, num_workers=5)
test_dataloader = torch.utils.data.DataLoader(data_test, batch_size=batch_size, num_workers=5)torch형식의 데이터로 변환해주고 학습모델을 만들어준다.
class BERTClassifier(nn.Module):
def __init__(self,
bert,
hidden_size = 768,
num_classes=7, # 분류갯수
dr_rate=None,
params=None):
super(BERTClassifier, self).__init__()
self.bert = bert
self.dr_rate = dr_rate
self.classifier = nn.Linear(hidden_size , num_classes)
if dr_rate:
self.dropout = nn.Dropout(p=dr_rate)
def gen_attention_mask(self, token_ids, valid_length):
attention_mask = torch.zeros_like(token_ids)
for i, v in enumerate(valid_length):
attention_mask[i][:v] = 1
return attention_mask.float()
def forward(self, token_ids, valid_length, segment_ids):
attention_mask = self.gen_attention_mask(token_ids, valid_length)
_, pooler = self.bert(input_ids = token_ids, token_type_ids = segment_ids.long(), attention_mask = attention_mask.float().to(token_ids.device))
if self.dr_rate:
out = self.dropout(pooler)
return self.classifier(out)
위에서 1~7로 잡은 것이 아니라 0~6으로 잡은 이유가 여기서 나온다. 만약 label값이 1~7이 되면 동일하게 7개긴 하지만 num_class는 8이상으로 넣어주어야한다. 그래서 데이터의 label을 0부터 시작하도록 처리해주어야한다.
이후로 모델세팅과 학습에 필요한 작업들을 세팅해준다.
model = BERTClassifier(bertmodel, dr_rate=0.5).to(device)
# Prepare optimizer and schedule (linear warmup and decay)
no_decay = ['bias', 'LayerNorm.weight']
optimizer_grouped_parameters = [
{'params': [p for n, p in model.named_parameters() if not any(nd in n for nd in no_decay)], 'weight_decay': 0.01},
{'params': [p for n, p in model.named_parameters() if any(nd in n for nd in no_decay)], 'weight_decay': 0.0}
]
optimizer = AdamW(optimizer_grouped_parameters, lr=learning_rate)
loss_fn = nn.CrossEntropyLoss()
t_total = len(train_dataloader) * num_epochs
warmup_step = int(t_total * warmup_ratio)
scheduler = get_cosine_schedule_with_warmup(optimizer, num_warmup_steps=warmup_step, num_training_steps=t_total)
def calc_accuracy(X,Y):
max_vals, max_indices = torch.max(X, 1)
train_acc = (max_indices == Y).sum().data.cpu().numpy()/max_indices.size()[0]
return train_accif torch.cuda.is_available():
device = torch.device("cuda")
print('There are %d GPU(s) available.' % torch.cuda.device_count())
print('We will use the GPU:', torch.cuda.get_device_name(0))
else:
device = torch.device("cpu")
print('No GPU available, using the CPU instead.')
이제 학습을 시작한다.
for e in range(num_epochs):
train_acc = 0.0
test_acc = 0.0
model.train()
for batch_id, (token_ids, valid_length, segment_ids, label) in enumerate(tqdm(train_dataloader)):
optimizer.zero_grad()
token_ids = token_ids.long().to(device)
segment_ids = segment_ids.long().to(device)
valid_length= valid_length
label = label.long().to(device)
out = model(token_ids, valid_length, segment_ids)
loss = loss_fn(out, label)
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), max_grad_norm)
optimizer.step()
scheduler.step() # Update learning rate schedule
train_acc += calc_accuracy(out, label)
if batch_id % log_interval == 0:
print("epoch {} batch id {} loss {} train acc {}".format(e+1, batch_id+1, loss.data.cpu().numpy(), train_acc / (batch_id+1)))
print("epoch {} train acc {}".format(e+1, train_acc / (batch_id+1)))
model.eval()
for batch_id, (token_ids, valid_length, segment_ids, label) in enumerate(tqdm(test_dataloader)):
token_ids = token_ids.long().to(device)
segment_ids = segment_ids.long().to(device)
valid_length= valid_length
label = label.long().to(device)
out = model(token_ids, valid_length, segment_ids)
test_acc += calc_accuracy(out, label)
print("epoch {} test acc {}".format(e+1, test_acc / (batch_id+1)))
Acc 0.5까지는 쭉쭉 올라가는데 그 이후부터는 미세하게 등반하기 시작한다.
모델학습이 완료되면 predict를 해야되는데 입력한 문장을 토큰화작업을 한 이후에 입력해야한다!
#토큰화
tokenizer = get_tokenizer()
tok = nlp.data.BERTSPTokenizer(tokenizer, vocab, lower=False)
def predict(predict_sentence):
data = [predict_sentence, '0']
dataset_another = [data]
another_test = BERTDataset(dataset_another, 0, 1, tok, max_len, True, False)
test_dataloader = torch.utils.data.DataLoader(another_test, batch_size=batch_size, num_workers=5)
model.eval()
for batch_id, (token_ids, valid_length, segment_ids, label) in enumerate(test_dataloader):
token_ids = token_ids.long().to(device)
segment_ids = segment_ids.long().to(device)
valid_length= valid_length
label = label.long().to(device)
out = model(token_ids, valid_length, segment_ids)
# test_eval=[]
for i in out:
logits=i
logits = logits.detach().cpu().numpy()
emotion = emotion_dict[np.argmax(logits)]
print(f">> 입력하신 내용의 감정은 {emotion}입니다.")
quit을 대소문자 구분없이 입력하면 종료되도록 코드를 만들어주고 실행하면 된다!
while True:
sentence = input("하고싶은 말을 입력해주세요 : ")
if sentence.upper() == "QUIT":
print("감정 분석을 종료합니다.")
break
predict(sentence)
print("\n")
아주 간단하게 감성분석을 진행해보았다. 추가할 내용이 있다면 추가적으로 수정 보완해서 글을 완성시키겠다!
끝!