텍스트 분류(2) - 다양한 모델링 기법

모델링

 

 

1. 머신러닝 모델 :  선형 회귀 모델, 랜덤 포레스트 모델 등등

2. 딥러닝 모델 : 합성곱 신경망(CNN), 순환 신경망(RNN) 등등

 

---

 

TF-IDF 를 활용한 모델 구현

TF-IDF 단어 임베딩 이후, Logistic Regression 모델을 활용한 감성 분류 모델을 만들 것.(머신러닝 감성 분석)

 

DATA_IN_PATH = './data_in/'
TRAIN_CLEAN_DATA = 'train_clean.csv'

train_data = pd.read_csv(DATA_IN_PATH + TRAIN_CLEAN_DATA)

reviews = list(train_data['review'])
sentiments = list(train_data['sentiment'])

 

전처리된 데이터를 불러오고 review 데이터, label 데이터 따로 reviews, sentiments 변수에 저장해둔다.

 

from sklearn.feature_extraction.text import TfidfVectorizer

vectorizer = TfidfVectorizer(min_df = 0.0, analyzer = "char", sublinear_tf = True,
                            ngram_range = (1,3), max_features = 5000)

X = vectorizer.fit_transform(reviews)

 

사이킷런 패키지를 이용해서 단어들을 tfidf 벡터화 시킨다.

 

from sklearn.model_selection import train_test_split
import numpy as np

RANDOM_SEED = 42
TEST_SPLIT = 0.2

y = np.array(sentiments)


X_train, X_eval, y_train, y_eval = train_test_split(X, y, test_size = TEST_SPLIT,
                                                   random_state = RANDOM_SEED)

 

사이킷런 패키지를 이용해서 훈련 데이터와 테스트 데이터로 나누고 그 비중은 테스트 데이터를 20% 로 설정한다.

 

from sklearn.linear_model import LogisticRegression

lgs = LogisticRegression(class_weight = 'balanced')
lgs.fit(X_train, y_train)

print(lgs.score(X_eval, y_eval))

 

사이킷런을 활용해서 로지스틱 회귀 분석 모델을 만든다.

앞서 각 단어들에 대해서 tf-idf 벡터화 시켜 X변수에 저장한 것을 이용한다.

 

이 모델은 TF-IDF 방식의 단어 벡터화를 통해서 만든 Logistic Regression 방식이다.

 

---

word2vec을 활용한 모델 구현

 

DATA_IN_PATH = './data_in/'
TRAIN_CLEAN_DATA = 'train_clean.csv'

train_data = pd.read_csv(DATA_IN_PATH + TRAIN_CLEAN_DATA)

reviews = list(train_data['review'])
sentiments = list(train_data['sentiment'])

 

위와 똑같이 데이터를 불러오기

 

#word2vec 을 하기 위해서는 문장을 단어 단위로 바꿔줘야한다.
sentences = []

for review in reviews:
    sentences.append(review.split())

 

reviews 에는 각각의 리뷰들이 담겨있는데, 리뷰별로 단어크기로 tokenizing 해서 그 정보를 sentence 에 저장.

 

#학습 시 필요한 파라미터

num_features = 300
min_word_count = 40
num_workers = 4
context = 10
downsampling = 1e-3

 

단어에 대해서 임베딩 작업을 진행할건데, 여기서 우리가 사용할 임베딩 기술은 word2vec 기술.

word2vec 에 사용할 파라미터 값을 설정한다.

 

import logging
logging.basicConfig(format = '%(asctime)s : %(levelname)s : %(message)s',
                   level = logging.INFO)

 

word2vec 모델 학습의 진행 과정을 나타내기 위해서 logging 모듈 사용하기.

 

from gensim.models import Word2Vec

print('Training model....')

model = Word2Vec()
model = Word2Vec(sentences,
                 workers=num_workers,
                 vector_size = num_features,
                 min_count = min_word_count,
                 window = context,
                 sample = downsampling)
                 
                 
model.wv['moment']

 

gensim 패키지에서 word2vec 임베딩 기술을 사용할 수 있다.

word2vec 객체를 만들고 그 안에 word2vec 시킬 텍스트 데이터를 넣고 파라미터 값을 설정한다.

(window 크기 및 단어 벡터의 크기 설정, 빈도수가 적은 단어는 임베딩 작업 제외)

위의 코드를 진행하면 텍스트 데이터의 각 단어에 대해서 일정한 크기로 벡터화 시킨다.

 

wv 함수를 사용하면 원하는 단어에 대한 단어 벡터 값을 확인할 수 있다.

 

model_name = "300features_40minwords_10context"
model.save(model_name)

model 이름을 지정학 저정해둔다.

 

 

#입력값을 같은 형태로 만들어야 한다.

def get_features(words, model, num_features):
    
    #출력 벡터 초기화
    
    feature_vector = np.zeros((num_features), dtype = np.float32)
    
    num_words = 0
    
    #word2vec 단어사전에 리뷰의 단어가 포함 됐는지 확인
    index2word_set = set(model.wv.index_to_key)
    
    #리뷰 하나당 하나의 벡터를 만들기 위해서 단어 벡터들을 모두 더하고 평균도출.
    for w in words:
        if w in index2word_set:
            num_words += 1
            feature_vector = np.add(feature_vector, model.wv[w])
            
    
    feature_vector = np.divide(feature_vector, num_words)
    return feature_vector

 

문서를 벡터화 시키는 과정(단어 벡터 평균 이용)

 

앞서 리뷰 한 개에 대해서 word2vec 을 이용해 단어 벡터화 시켜봤다.

이제 이를 이용해서 리뷰 전체에 대해서 벡터화 시켜보자.

그 방식으로 리뷰의 모든 구성 단어들의 벡터에 대해서 평균을 구하는 방식을 사용한다.

예를 들어서 "I like banana" 라는 문서가 있다고 해보자. word2vec을 이용해서 단어 벡터의 길이를 3으로 만들었다 가정하자.

[1,2,3].[4,5,6],[7,8,9] 로 각각의 단어들이 벡터화가 됐을 때, 이 값들을 모두 더한다.

[12,15,18] 의 값이 도출되고 단어의 개수 3으로 나누어 평균 벡터값을 구한다.

최종 [4,5,6] 이 "I like banana" 문서의 벡터값이다.

 

즉, word2vec 으로 문서의 각 단어들을 벡터화 시키고 이 값들을 평균내어 문서의 벡터 값으로 사용한다.

 

def get_dataset(reviews, model, num_features):
    dataset = list()
    
    for s in reviews:
        dataset.append(get_features(s, model, num_features))
        
    reviewFeatureVecs = np.stack(dataset)
    
    return reviewFeatureVecs
    
    
test_data_vect = get_dataset(sentences, model, num_features)

 

전체 리뷰 문서 각각에 대해서 벡터 값들을 구해보자.

그리고 test_data_vect 변수에 각각의 리뷰 데이터를 벡터화 시킨 값을 집어 넣자.

 

from sklearn.model_selection import train_test_split

X = test_data_vect
y = np.array(sentiments)

RANDOM_SEED = 42
TEST_SPLIT = 0.2

X_train, X_eval, y_train, y_eval = train_test_split(X,y test_size = TEST_SPLIT,
                                                   random_state = RANDOM_STATE)

 

마찬가지로 사이킷런을 이용해서 학습 데이터와 트레인 데이터를 나눈다.

 

from sklearn.linear_model import LogisticRegression

lgs  = LogisticRegression(class_weight = 'balanced')
lgs.fit(X_train, y_train)
lgs.score(X_eval, y_eval)

 

마찬가지로 Logistic Regression 을 이용해서 감성 값에 대한 분류를 진행한다.

그리고 성능을 평가한다.

 

이 모델은 word2vec 임베딩 방식을 사용해서 만든 Logistic Regression 모델이다.

 

---

랜덤 포레스트 분류 모델

 

이번에는 임베딩 방식도 바꾸고 모델 종류도 바꿔보자. 다른 머신러닝 분류 모델을 사용해보자.

임베딩 방식으로는 빈도기반 CountVectorizer 방식을 사용.

 

import pandas as pd

DATA_IN_PATH = './data_in/'
TRAIN_CLEAN_DATA = 'train_clean.csv'

train_data = pd.read_csv(DATA_IN_PATH + TRAIN_CLEAN_DATA)
reviews = train_data['review']
y = train_data['sentiment']

 

마찬가지로 데이터를 불러온다.

 

from sklearn.feature_extraction.text import CountVectorizer

vectorizer = CountVectorizer(analyzer = 'word', max_features = 5000) #분석 단위를 하나의 단어 단위로, 차원 수를 5000으로 한정지었다.

train_data_features = vectorizer.fit_transform(reviews)

 

사이킷런 패키지에 있는 CounVectorizer 를 사용해서 단어들에 대해서 벡터화 시킨다.

 

from sklearn.model_selection import train_test_split
TEST_SIZE = 0.2
RANDOM_SEED = 42

train_input, eval_input, train_label, eval_label = train_test_split(train_data_features, y, test_size = TEST_SIZE,
                                                                   random_state = RANDOM_SEED)

 

마찬가지로 사이킷런 패키지를 활용해서 train 데이터와 test 데이터를 분리한다.

 

from sklearn.ensemble import RandomForestClassifier


forest = RandomForestClassifier(n_estimators = 100)

forest.fit(train_input, train_label)

 

사이킷런 패키지를 통해서 랜덤 포레스트 모델을 생성하고 학습시킨다.

 

from sklearn.metrics import accuracy_score

predict_result = forest.predict(eval_input)
accuracy = accuracy_score(predict_result, eval_label)
print('랜덤 포레스트 정확도 : {:.2f}'.format(accuracy))

 

분류 모델의 성능 측정 기준인 accuracy_score 를 활용해서 랜덤포레스트 분류 모델의 성능을 평가한다.

 

---

순환 신경망을 사용한 텍스트 분류 모델

 

지금까지는 다양한 임베딩 방식과 머신러닝 기반 분류 모델을 활용한 분류 모델을 구현해봤다.

지금부터는 딥러닝 기반의 분류 모델을 만들어보자.

(RNN, CNN)

 

import numpy as np
from tensorflow.keras.preprocessing.sequence import pad_sequences
import json

DATA_IN_PATH = './data_in/'
DATA_OUT_PATH = './data_out/'
TRAIN_INPUT_DATA = 'train_input.npy'
TRAIN_LABEL_DATA = 'train_label.npy'
DATA_CONFIGS = 'data_configs.json'

train_input = np.load(open(DATA_IN_PATH + TRAIN_INPUT_DATA, 'rb'))
train_input = pad_sequences(train_input, maxlen = train_input.shape[1])
train_label = np.load(open(DATA_IN_PATH + TRAIN_LABEL_DATA, 'rb'))
prepro_configs = json.load(open(DATA_IN_PATH + DATA_CONFIGS, 'r',encoding = 'UTF-8'))

 

앞서와 마찬가지로 데이터를 불러온다.

이때 train 데이터에 대해서 pad_sequence 를 해주어 길이를 통일한다.

그리고 tokenizer 로 각 단어에 id를 부여한 작업을 진행했었는데 그 결과물인 data_configs 를 불러온다.

data_configs 안에는 각 단어에 대한 id정보와 전체 단어의 개수에 대한 정보가 포함되어 있다.

 

model_name = 'rnn_classifier'
BATCH_SIZE = 128
NUM_EPOCHS = 5
VALID_SPLIT = 0.1
MAX_LEN = train_input.shape[1]


kargs = {'model_name' : model_name,
        'vocab_size' : prepro_configs['vocab_size'],
        'embedding_dimension' : 100,
        'dropout_rate' : 0.2,
        'lstm_dimension' : 150,
        'dense_dimension' : 150,
        'output_dimension' : 1}

 

class 로 RNN 분류모델을 구현해볼건데, 그 전에 하이퍼 파라미터 값들을 kargs 변수에 딕셔너리 형태로 저장해둔다.

128개씩 꺼내서 학습을 시키는데 epoch 을 5를 주어 5번 진행한다. validation data는 10%만 사용한다.

 

 

class RNNClassifier(tf.keras.Model):
    def __init__(self, **kargs):
        #부모 클래스에 있는 __init__ 정보를 받아온다. (Super)
        #Super 함수를 통해서 부모 클래스에 __init__ 함수 인자에 모델 이름을 넣으면 tf.eras.Model 을 상속받는 자식들은 모두 모델 이름하에
        #공통적으로 사용된다.
        super(RNNClassifier, self).__init__(name = kargs['model_name']) 
        self.embedding = tf.keras.layers.Embedding(input_dim = kargs['vocab_size'],
                                         output_dim = kargs['embedding_dimension'])
        self.lstm_1_layer = tf.keras.layers.LSTM(kargs['lstm_dimension'],
                                                return_sequences = True)
        self.lstm_2_layer = tf.keras.layers.LSTM(kargs['lstm_dimension'])
        self.dropout = tf.keras.layers.Dropout(kargs['dropout_rate'])
        self.fc1 = tf.keras.layers.Dense(units = kargs['dense_dimension'],
                               activation = tf.keras.activations.tanh)
        self.fc2 = tf.keras.layers.Dense(units = kargs['output_dimension'],
                               activation = tf.keras.activations.sigmoid)
        
        
    def call(self, x):
        x = self.embedding(x)
        x = self.dropout(x)
        x = self.lstm_1_layer(x)
        x = self.lstm_2_layer(x)
        x = self.dropout(x)
        x = self.fc1(x)
        x = self.dropout(x)
        x = self.fc2(x)
        
        return x

 

RNN 모델에 단어 개수만큼 입력 데이터가 들어올텐데 이를 Embedding layer 를 사용해서 100개의 출력으로 바꾼다.

RNN 신경망 층을 두 개의 층으로 설정한다.

Dropout 을 통해서 신경망의 오버피팅을 막는다. 이때 학습시 훈련 시키지 않을 노드의 개수를 정해야 하는데 이를 0.2%로 정했다.

완전연결노드를 정한다.

hidden layer 와 출력 layer 를 정하는데 hidden layer 에는 150개의 노드를 설정하고, output 은 1로 정한다.

hidden layer 활성화 함수로는 tanh 함수를 사용, 분류 모델이기 때문에 출력노드의 활성화 함수는 sigmoid 함수를 사용한다.

 

model = RNNClassifier(**kargs)
model.compile(optimizer = tf.keras.optimizers.Adam(1e-4),
             loss = tf.keras.losses.BinaryCrossentropy(),
             metrics = [tf.keras.metrics.BinaryAccuracy(name = 'accuracy')])

 

학습 방식에 대해서 .compile() 로 정의내린다.

최적화 방식은 Adam 방식을 사용, 손실 함수로는 이진분류 문제이기 때문에 이진 교차 손실 함수를 사용한다.

그리고 평가 방식으로는 accuracy 를 사용한다.

즉 정확도를 기준으로 손실을 계산하고 이에 대해서 모델을 개선해나간다.

 

from tensorflow.keras.callbacks import EarlyStopping, ModelCheckpoint
import os

earlystop_callback = EarlyStopping(monitor = 'val_accuracy', min_delta = 0.0001, patience = 2)
checkpoint_path = DATA_OUT_PATH + model_name + '/weights.h5'
checkpoint_dir = os.path.dirname(checkpoint_path)

if os.path.exists(checkpoint_dir):
    print("{} -- Folder already exits \n".format(checkpoint_dir))
else:
    os.makedirs(checkpoint_dir, exist_ok = True)
    print('{} -- Folder create complete \n'.format(checkpoint_dir))
    
    
cp_callback = ModelCheckpoint(checkpoint_path, monitor = 'val_accuracy', verbose = 1, save_best_only = True,
                             save_weights_only = True)


history = model.fit(train_input, train_label, batch_size = BATCH_SIZE, epochs = NUM_EPOCHS,
                    validation_split = VALID_SPLIT, callbacks = [earlystop_callback, cp_callback])

 

학습에 대해서 callback 을 설정  가능하다.

tensorflow keras callbacks 에서 EarlyStopping, ModelCheckpoint 모듈을 사용한다.

EarlyStopping 은 과적합을 막기 위해서 사용하는데, 만일 학습을 하면서 test데이터의 성능이 그 전보다 더 나빠지면 학습을 중단한다. (과적합의 특징은 train 데이터의 성능은 계속 좋아지지만 test데이터의 성능은 좋아지다가 다시 나빠지게 된다.)

ModelCheckpoint 는 epoch 마다의 모델 성능을 저장하는데, 이때 파라미터를 설정하면 최고의 성능을 보이는 모델을 저장할 수 있다.

---

합성곱 신경망을 사용한 텍스트 분류 모델

 

model_name = 'cnn_classifier_en'
BATCH_SIZE = 512
NUM_EPOCHS = 2
VALID_SPLIT = 0.1
MAX_LEN = train_input.shape[1]

kargs = {'model_name' : model_name,
        'vocab_size' : prepro_configs['vocab_size'],
        'embedding_size' : 128,
        'num_filters' : 100,
        'dropout_rate' : 0.5,
        'hidden_dimension' : 250,
        'output_dimension' : 1}

RNN 과 마찬가지로 하이퍼 파라미터 값을 미리 정해두고 kargs 딕셔너리에 저장해둔다.

 

class CNNClassifier(tf.keras.Model):
    
    def __init__(self,**kargs):
        super(CNNClassifier, self).__init__(name = kargs['model_name'])
        self.embedding = tf.keras.layers.Embedding(input_dim = kargs['vocab_size'],
                                                  output_dim = kargs['embedding_size'])
        
        self.conv_list = [tf.keras.layers.Conv1D(filters = kargs['num_filters'],
                                                kernel_size = kernel_size,
                                                padding = 'valid',
                                                activation = tf.keras.activations.relu,
                                                kernel_constraint = tf.keras.constraints.MaxNorm(max_value = 3.)) for kernel_size in [3,4,5]]
        
        self.pooling = tf.keras.layers.GlobalMaxPooling1D()
        self.dropout = tf.keras.layers.Dropout(kargs['dropout_rate'])
        self.fc1 = tf.keras.layers.Dense(units = kargs['hidden_dimension'],
                                        activation = tf.keras.activations.relu,
                                        kernel_constraint = tf.keras.constraints.MaxNorm(max_value = 3.))
        self.fc2 = tf.keras.layers.Dense(units = kargs['output_dimension'],
                                        activation = tf.keras.activations.sigmoid,
                                        kernel_constraint = tf.keras.constraints.MaxNorm(max_value = 3.))
        
    def call(self, x):
        x = self.embedding(x)
        x = self.dropout(x)
        x = tf.concat([self.pooling(conv(x)) for conv in self.conv_list], axis = 1)
        x = self.fc1(x)
        x = self.fc2(x)
        
        
        return x

 

RNN 과 마찬가지로 tf.keras.Model 을 상속받는 class 를 만들고 그 안에 계층을 설정한다.

여기서 유의해야할 코드로는 self.conv_list 코드다.

필터의 크기를 for 문을 이용해서 3,4,5 개 줬는데, 이 때문에 합성곱 결과가 각각의 필터의 크기에 따라서 총 3개가 도출됐다.

이 3개의 결과값들에 대해서 각각 가장 크기가 큰 값을 꺼내려고 하는데 이때 사용하는 함수가 pooling 이며, 그리고 이 pooling 결과 값을 하나의 결과값으로 바꾸려고 하는데 이 떄 사용한 함수가 tf.concat 함수다.

따라서 3개의 pooing 결과 값들이 합쳐진 벡터가 완전연결 신경망 층으로 전달이 된다.

 

 

 

#학습 방식 정의
model = CNNClassifier(**kargs)
model.compile(optimizer = tf.keras.optimizers.Adam(1e-4),
             loss = tf.keras.losses.BinaryCrossentropy(),
             metrics = [tf.keras.metrics.BinaryAccuracy(name = 'accuracy')])

#모델 학습

earlystop_callback = EarlyStopping(monitor = 'val_accuracy', min_delta = 0.0001, patience = 2)

checkpoint_path = DATA_OUT_PATH + model_name + '/weights.h5'
checkpoint_dir = os.path.dirname(checkpoint_path)

if os.path.exists(checkpoint_dir):
    print('{} -- Folder already exists \n'.format(checkpoint_dir))
else:
    os.makedirs(checkpoint_dir, exist_ok = True)
    print('{} -- Folder create compile \n'.format(checkpoint_dir))
    
cp_callback = ModelCheckpoint(checkpoint_path,
                             monitor = 'val_accuracy',
                             verbose = 1,
                             save_best_only = True,
                             save_weights_only = True)

history = model.fit(train_input, train_label, batch_size = BATCH_SIZE, epochs = NUM_EPOCHS,
                   validation_split = VALID_SPLIT, callbacks = [earlystop_callback, cp_callback])

 

모델을 생성하고 앞의 RNN과 마찬가지로 callback 들을 지정해주어 오버피팅을 막는다.

그리고 학습을 시킨다.

 

---

정리

 

여기까지 우리는 다양한 임베딩 방식을 사용한 머신러닝 분류 모델들을 구현해봤으며 (로지스틱 회귀, 랜덤 포레스트)

신경망을 활용한 분류 모델도 구현해봤다.(RNN, CNN).