지난번에 이런저런 전처리를 하고 textblob으로 감정분석하고 wordcloud를 만들어봤다.
이번에는 lstm으로 모델학습을 하고, 리뷰를 줘서 긍정/부정을 평가하도록 하겠다.
사실 지난주에 만들었는데 지금이라도 기록해놓지 않으면 코드를 잊어버릴거같아..
전체코드
import pandas as pd
df=pd.read_csv('파일주소') #지난번의 netflix review를 가져왔다.
df['content']=df['content'].fillna('')
df['label']=df['score'].apply(lambda x: 1 if x>=4 else (0 if x<=2 else ' '))
df=df[df['label']!=' ']
# 텍스트 전처리와 자연어 처리를 위한 라이브러리
import nltk
from textblob import TextBlob
# 토픽 모델링을 위한 라이브러리
import gensim
from gensim import corpora
from gensim.utils import simple_preprocess
processed=df['content'].apply(simple_preprocess)
from nltk.stem import WordNetLemmatizer
from nltk.corpus import wordnet
nltk.download('wordnet')
nltk.download('averaged_perceptron_tagger_eng')
lemmatizer=WordNetLemmatizer()
def get_wordnet_pos(tag):
if tag.startswith('J'):
return wordnet.ADJ
elif tag.startswith('V'):
return wordnet.VERB
elif tag.startswith('N'):
return wordnet.NOUN
elif tag.startswith('R'):
return wordnet.ADV
else:
return wordnet.NOUN
lemmatized = processed.apply(lambda x: [lemmatizer.lemmatize(word, get_wordnet_pos(tag))
for word, tag in nltk.pos_tag(x)])
import nltk
nltk.download('stopwords')
from nltk.corpus import stopwords
stop_words = set(stopwords.words('english'))
filtered_text = lemmatized.apply(lambda words: [word for word in words if word not in stop_words])
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, Dataset
label = df['label'].tolist()
from sklearn.model_selection import train_test_split
train_texts, test_texts, train_labels, test_labels = train_test_split(filtered_text, label, test_size=0.2, random_state=42)
train_texts = train_texts.tolist()
test_texts = test_texts.tolist()
dictionary = corpora.Dictionary(train_texts)
#데이터셋 정의
class ReviewDataset(Dataset):
def __init__(self, texts, labels, dictionary):
self.texts = texts
self.labels = labels
self.dictionary = dictionary
def __len__(self):
return len(self.texts)
def __getitem__(self, idx):
indexed_text = self.dictionary.doc2idx(self.texts[idx])
return torch.tensor(indexed_text, dtype=torch.long), self.labels[idx]
#패딩처리
from torch.nn.utils.rnn import pad_sequence
def collate_batch(batch):
texts, labels = zip(*batch)
texts_padded = pad_sequence(texts, batch_first=True, padding_value=0)
labels = torch.tensor(labels, dtype=torch.float)
return texts_padded, labels
BATCH_SIZE = 64
train_dataset = ReviewDataset(train_texts, train_labels, dictionary)
test_dataset = ReviewDataset(test_texts, test_labels, dictionary)
train_dataloader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True, collate_fn=collate_batch)
test_dataloader = DataLoader(test_dataset, batch_size=BATCH_SIZE, shuffle=False, collate_fn=collate_batch)
#모델정의
class LSTMModel(nn.Module):
def __init__(self, vocab_size, embed_dim, hidden_dim, output_dim):
super(LSTMModel, self).__init__()
self.embedding = nn.Embedding(vocab_size, embed_dim)
self.lstm = nn.LSTM(embed_dim, hidden_dim, batch_first=True)
self.fc = nn.Linear(hidden_dim, output_dim)
def forward(self, text):
embedded = self.embedding(text)
output, (hidden, cell) = self.lstm(embedded)
return self.fc(hidden[-1])
vocab_size = len(dictionary)
embed_dim = 64
hidden_dim = 128
output_dim = 1
#모델학
model = LSTMModel(vocab_size, embed_dim, hidden_dim, output_dim)
criterion = nn.BCEWithLogitsLoss()
optimizer = optim.Adam(model.parameters(), lr=0.01)
num_epochs = 20
for epoch in range(num_epochs):
model.train()
running_loss = 0.0
for text_batch, label_batch in train_dataloader:
optimizer.zero_grad()
predicted_label = model(text_batch).squeeze()
loss = criterion(predicted_label, label_batch)
loss.backward()
optimizer.step()
running_loss += loss.item()
if (epoch + 1) % 2 == 0:
print(f'Epoch [{epoch + 1}/{num_epochs}], Loss: {running_loss / len(train_dataloader):.4f}')
print('Training Completed')
# 단일 리뷰 텍스트를 인덱스 텐서로 변환하는 함수
def review_to_tensor(review, dictionary, max_len=50, padding_value=0):
# 텍스트를 사전을 통해 인덱스 리스트로 변환
indexed_text = dictionary.doc2idx(review.split(), unknown_word_index=padding_value)
# 패딩 적용
if len(indexed_text) < max_len:
indexed_text += [padding_value] * (max_len - len(indexed_text))
else:
indexed_text = indexed_text[:max_len]
# 텐서화
return torch.tensor(indexed_text, dtype=torch.long)
def predicted_review(review, model, dictionary, max_len=50):
# 모델을 평가 모드로 설정
model.eval()
# 리뷰를 텐서로 변환
review_tensor = review_to_tensor(review, dictionary, max_len)
# 배치 차원을 추가해 모델에 전달할 준비
review_tensor = review_tensor.unsqueeze(0) # [1, max_len] 형태로 만듦
# 모델로 예측 수행
with torch.no_grad():
logits = model(review_tensor)
probs = torch.sigmoid(logits)
predicted_label = (probs >= 0.5).int() # 확률을 0 또는 1로 변환
return predicted_label.item(), probs.item() # 예측 라벨과 확률 반환
#테스트
sample_review = "사용자가 입력하는 리뷰"
predicted_label, probability = predicted_review(sample_review, model, dictionary)
if predicted_label==1:
print(f"positive, Probability: {probability:.4f}" )
else:
print(f"negative, Probability: {probability:.4f}" )
전처리 단계에서 이전에도 했던 부분은 넘어감
df['label']=df['score'].apply(lambda x: 1 if x>=4 else (0 if x<=2 else ' '))
df=df[df['label']!=' ']
전처리에서 추가된 부분은 여기인데
score가 4점,5점이면 1(긍정)을, 1점,2점이면 0(부정)을 라벨링하도록 했다. 3점은 공백으로 남긴다.
그 결과를 label 컬럼에 저장하고, 공백인 부분을 제외하고 나머지만 선택했다.
3점을 제외한 이유: neutral한 리뷰는 긍정/부정을 구분하고 학습하는데 방해가 될 것 같아서
참고로 lambda를 적용할 때 조건문은 간결해야 한다. 쉼표나 elif를 쓸 수 없고 else 안에 괄호를 또 써서 이중조건문으로 만들어야 한다.
dictionary = corpora.Dictionary(train_texts)
주어진 텍스트의 단어에다 고유한 단어 인덱스를 매핑해줌
ex)
train_texts=[['i','love','you'],['i','do','not']]
dictionary = corpora.Dictionary(train_texts)
print(dictionary.token2id)
->결과
{'i': 0, 'love': 1, 'you': 2, 'do': 3, 'not': 4}
class ReviewDataset(Dataset):
def __init__(self, texts, labels, dictionary):
self.texts = texts
self.labels = labels
self.dictionary = dictionary
def __len__(self):
return len(self.texts)
def __getitem__(self, idx):
indexed_text = self.dictionary.doc2idx(self.texts[idx])
return torch.tensor(indexed_text, dtype=torch.long), self.labels[idx]
여기서 볼 것은 다음의 코드인데,
def __getitem__(self, idx):
indexed_text = self.dictionary.doc2idx(self.texts[idx])
return torch.tensor(indexed_text, dtype=torch.long), self.labels[idx]
doc2idx는 텍스트에 포함된 단어들을 사전(dictionary)에 저장된 고유 인덱스로 바꿔준다.
ex)
texts[0]=['i','love','you','i','do','not']
indexed_text =self.dictionary.doc2idx(self.texts[0])
indexed_text
[0, 1, 2, 0, 3, 4]
이 결과를 torch.tensor로 반환하고, 라벨도 같이 반환해주고 있다.
이전에 패딩처리할때도 비슷한 오류를 언급했는데, 데이터로더가 데이터를 가져올 때 text와 label을 같이 가져오기 때문에, 반환도 같이 해줘야 한다.
from torch.nn.utils.rnn import pad_sequence
def collate_batch(batch):
texts, labels = zip(*batch)
texts_padded = pad_sequence(texts, batch_first=True, padding_value=0)
labels = torch.tensor(labels, dtype=torch.float)
return texts_padded, labels
패딩처리 해주는 코드
texts, labels = zip(*batch)
zip은 리스트나 튜플을 병렬로 묶어주는 함수이고, *는 언패킹을 수행한다.
ex)
batch=[( ['i','love','you'] ,1),( ['i','do','not'] ,0)]
*batch= ( ['i','love','you'] ,1),( ['i','do','not'] ,0)
-> 두 개의 튜플이 됨
list(zip(*batch))
[ (['i','love','you'], ['i','do','not'] ), (1,0) ]
->text끼리, label끼리 묶임
texts_padded = pad_sequence(texts, batch_first=True, padding_value=0)
이전에도 봤던 패딩처리 해주는 함수
labels = torch.tensor(labels, dtype=torch.float)
라벨도 torch.tensor 실수형으로 반환해주고 있다.
Q 왜 실수일까?
A 0 또는 1의 이진분류문제라고 생각하기 쉽지만, 사실 이 모델은 0과 1 사이의 값을 내야 하는 문제이다.
0에 가까우면 부정리뷰, 1에 가까우면 긍정리뷰로 볼 것이며, 이진분류 손실함수인 BCELoss, BCEWithLogitsLoss는 레이블을 실수로 받기를 기대한다.
return texts_padded, labels
마찬가지로 텍스트와 라벨을 모두 반환해줘야 데이터로더가 오류없이 동작한다.
Q 왜 label은 torch.tensor로 변환했는데 texts_padded는 그냥 반환하나?
A pad_sequence로 패딩적용을 하면 이미 텐서형태로 나와서 추가로 변형할 필요가 없음
참고) 이전에 패딩줬던 코드
def collate_batch(batch):
reviews, ratings = [], []
for (review, rating) in batch:
reviews.append(torch.tensor(review, dtype=torch.long))
ratings.append(torch.tensor(rating, dtype=torch.long))
reviews_padded = pad_sequence(reviews, batch_first=True, padding_value=0)
return reviews_padded, torch.tensor(ratings, dtype=torch.long)
이 코드는 리뷰와 레이팅을 각각 따로 텐서로 변환해서 리스트에 추가하고
리뷰에는 패딩을 준 다음
패딩처리된 리뷰와, 레이팅을 반환하고 있다.
전자는 텍스트를 한번에 분리해서 동시에 패딩처리하면서 텐서형태로까지 변형
후자는 텍스트를 하나씩 텐서형태로 변형한 다음 패딩처리
전자가 더 간결한 코드이다. 반면 후자는 리뷰의 길이가 많이 다르거나, 데이터에 특별한 처리를 더해줘야 할 때 더 유리하다.
ratings.append(torch.tensor(rating, dtype=torch.long))
return reviews_padded, torch.tensor(ratings, dtype=torch.long)
ratings에 tensor 변환을 두 번이나?
ratings.append(torch.tensor(rating, dtype=torch.long))
이것의 결과는 [tensor1, tensor2, tensor3..] 즉 텐서들로 된 리스트
return reviews_padded, torch.tensor(ratings, dtype=torch.long)
이것의 결과는 tensor 즉 저 리스트를 텐서로 바꾼것
class LSTMModel(nn.Module):
def __init__(self, vocab_size, embed_dim, hidden_dim, output_dim):
...중략...
vocab_size = len(dictionary)
...후략...
vocab_size에는 임베딩할 단어의 전체 개수를 써줘야 한다.
그래서 생성한 사전에 있는 모든 단어의 수가 vocab_size 가 된다.
model = LSTMModel(vocab_size, embed_dim, hidden_dim, output_dim)
criterion = nn.BCEWithLogitsLoss()
optimizer = optim.Adam(model.parameters(), lr=0.01)
num_epochs = 20
for epoch in range(num_epochs):
model.train()
running_loss = 0.0
for text_batch, label_batch in train_dataloader:
optimizer.zero_grad()
predicted_label = model(text_batch).squeeze()
loss = criterion(predicted_label, label_batch)
loss.backward()
optimizer.step()
running_loss += loss.item()
if (epoch + 1) % 2 == 0:
print(f'Epoch [{epoch + 1}/{num_epochs}], Loss: {running_loss / len(train_dataloader):.4f}')
print('Training Completed')
모델학습코드
criterion = nn.BCEWithLogitsLoss()
처음 써보는 손실함수가 나왔다.
BCEWithLogitsLoss는 이진분류에 사용하는 손실함수인데, BCE( Binary Cross-Entropy)와 sigmoid함수를 결합한 것
모델에서 최종으로 출력한 로짓을 자동으로 sigmoid함수에 넣어서 확률로 계산해준다.
만약 BCE를 썼다면 최종출력으로 return self.fc(hidden[-1]) 대신 return sigmoid(self.fc(hidden[-1])) 를 써야 한다.
결과는..! probability가 1에 가까울수록 긍정, 0에 가까울수록 부정적인 리뷰이다.
느낀점
- 전에 netflix review 분석을 했을때는 1점과 5점짜리 리뷰가 많아서 예측결과도 1점 아니면 5점으로 나오곤 했는데, 1~2점을 묶고 4~5점을 묶어서 학습시키니 좀 더 잘 알아듣는 것 같다.
- 위에서 전처리했던 긴 과정을 모듈화해서 함수형태로 정의했으면 나중에 샘플리뷰 받아서 처리하기가 훨씬 편했을 것 같다. 다음부터는 모듈화를 적극적으로 활용해보는 것으로..
'머신러닝' 카테고리의 다른 글
| style transfer model (0) | 2024.11.05 |
|---|---|
| openAI API를 활용한 챗봇 구현 (0) | 2024.11.04 |
| sentence generator with GPT-2 (0) | 2024.10.31 |
| skin cancer HAM10000(2) (0) | 2024.10.30 |
| skin cancer HAM10000 (0) | 2024.10.29 |