IDF+Word2Vec & Word2VecVectorizer 클래스 & embed 모듈화 정리

멀티캠퍼스 데이터분석 6월 8일 수업 내용 — IDF+Word2Vec 혼합 벡터화, Word2VecVectorizer 커스텀 클래스, GridSearchCV 연동, embed 모듈화, 연습문제

---

📌 목차

1. IDF + Word2Vec 혼합 벡터화
2. 전역 변수 활용 & global 키워드
3. Word2VecVectorizer — 커스텀 클래스
4. Pipeline + GridSearchCV 연동
5. embed 모듈화 — 라이브러리로 분리
6. 연습문제 — embed 모듈 활용 감성 분류

---

1. IDF + Word2Vec 혼합 벡터화

왜 IDF를 혼합하는가?

• Word2Vec은 단문에서 효과적인 벡터화 방법
• 그러나 단문에서 단어들의 출현 횟수는 대부분 1회 → TF 값이 큰 의미를 가지지 못함
• IDF (전체 문서 중 희귀한 단어에 높은 가중치) 와 Word2Vec (의미 기반 벡터) 를 결합하여 성능 향상

import numpy as np
import pandas as pd
from gensim.models import Word2Vec
from sklearn.svm import SVC
from sklearn.metrics import classification_report
from sklearn.feature_extraction.text import TfidfVectorizer
from konlpy.tag import Komoran

docs = [
    '오늘 날씨가 좋다 여행 가고 싶다',
    '기온이 너무 올라서 아무것도 하기 싫다',
    '수업이 너무 지루하고 졸리다',
    '음식이 너무 맛이 없고 서비스도 별로다',
    '영화가 너무 재미있어서 시간이 가는 줄 몰랐다'
]
target = [1, 0, 0, 0, 1]

토큰화 함수 — try/except로 안정성 확보

def tokenize(text):
    try:
        komoran   = Komoran()
        allow_pos = ['NNP', 'NNG', 'VV', 'VA', 'SL', 'MAG']
        tokens    = []
        for word, pos in komoran.pos(text):
            if pos in allow_pos:
                tokens.append(word)
    except Exception as e:
        print('Komoran 사용이 불가:', e)
        tokens = text.split()   # fallback: 공백 기준 분리
    return tokens

💡 try/except 사용 이유
KoNLPy는 JDK 의존성이 있어 환경에 따라 실행 불가능한 경우가 있습니다.
예외 발생 시 text.split() 으로 fallback 처리하여 코드가 중단되지 않도록 합니다.

Word2Vec 학습 & 문장 벡터 평균

tokens = [tokenize(doc) for doc in docs]

w2v = Word2Vec(
    sentences   = tokens,
    vector_size = 100,
    window      = 5,
    min_count   = 1,
    epochs      = 100,
    sg          = 1,
    workers     = 2,
    seed        = 42
)
wv = w2v.wv

def sent_embed_mean(token):
    """단어 벡터의 평균으로 문장 벡터 생성"""
    vector = []
    for word in token:
        if word in wv.index_to_key:
            vector.append(wv[word])
    if vector:
        return np.mean(vector, axis=0)
    else:
        return np.zeros(wv.vector_size)   # 미등록 단어만 있을 경우 0벡터

IDF 가중치 추출

# TfidfVectorizer로 IDF 값만 추출
tfidf_vec = TfidfVectorizer(tokenizer=tokenize, lowercase=False).fit(docs)

# {단어: idf값} 딕셔너리 생성
idf_weight = dict(zip(
    tfidf_vec.get_feature_names_out(),
    tfidf_vec.idf_
))

IDF 가중치 적용 문장 벡터

def sent_embed_wv_idf(token):
    """단어 벡터 × IDF 가중치 평균으로 문장 벡터 생성"""
    vector = []
    idf    = []

    for word in token:
        if word in wv.index_to_key and word in idf_weight:
            vector.append(wv[word] * idf_weight[word])
            idf.append(idf_weight[word])

    if vector:
        # IDF 가중 평균 = 합산 / IDF 합 (1e-9로 분모 0 방지)
        return np.sum(vector, axis=0) / (np.sum(idf) + 1e-9)
    else:
        return np.zeros(wv.vector_size)

💡 1e-9 를 더하는 이유
모든 단어의 IDF가 0인 경우 분모가 0이 되는 것을 방지합니다.
극히 작은 값을 더해 수치 안정성을 확보합니다.

모델 학습 & 예측 함수

from sklearn.model_selection import train_test_split

def run_model(X, y, model, test_size=0.2, stratify=None):
    """데이터 분할 → 학습 → 예측 → 평가 지표 반환"""
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=test_size, stratify=stratify, random_state=42
    )
    model.fit(X_train, y_train)
    pred   = model.predict(X_test)
    return classification_report(pred, y_test)


def predict_sentence_list(sentences, model, vec_type='wv'):
    """새로운 문장 리스트에 대한 감성 예측"""
    X_test = []
    for sentence in sentences:
        token = tokenize(sentence)
        if vec_type == 'wv':
            vec = sent_embed_mean(token)
        elif vec_type == 'idf':
            vec = sent_embed_wv_idf(token)
        else:
            print('vec_type이 맞지 않습니다')
            return ''
        X_test.append(vec)

    preds  = model.predict(X_test)
    result = []
    for sentence, pred in zip(sentences, preds):
        label = '긍정' if pred == 1 else '부정'
        result.append([sentence, label])
    return result

실행 & 비교

# 벡터화
X_embed_wv  = [sent_embed_mean(token)    for token in tokens]
X_embed_idf = [sent_embed_wv_idf(token) for token in tokens]

svc = SVC(random_state=42)

print('=== Word2Vec 평균 ===')
print(run_model(X_embed_wv, target, svc))

print('=== Word2Vec + IDF 가중치 ===')
print(run_model(X_embed_idf, target, svc))

# 새 문장 예측
new_sentences = [
    '영화가 너무 지루해서 돈이 아깝다',
    '날씨가 너무 별로다',
    '기온이 좋아서 어디론가 떠나고 싶다'
]

print(predict_sentence_list(new_sentences, svc, 'wv'))
print(predict_sentence_list(new_sentences, svc, 'idf'))

---

2. 전역 변수 활용 & global 키워드

함수 안에서 외부(전역) 변수를 수정하려면 global 키워드가 필요합니다.

a = "Hello"

def func_1(text):
    # 방법 1 — globals() 함수 사용
    # globals()['a'] = text

    # 방법 2 — global 키워드 사용 ✅ 권장
    global a
    a = text
    return a

print(a)          # Hello
print(func_1('Hi'))  # Hi
print(a)          # Hi (전역 변수가 변경됨)

⚠️ 전역 변수 수정 주의사항
전역 변수를 무분별하게 수정하면 코드 추적이 어려워집니다.
모듈화 시 global 을 사용하여 wv, idf_weight 같은 공유 자원을 설정하는 패턴이 자주 사용됩니다.

---

3. Word2VecVectorizer — 커스텀 클래스

gensim의 Word2Vec 은 sklearn의 fit() / transform() 인터페이스가 없어 Pipeline/GridSearchCV와 직접 연동이 불가능합니다. sklearn의 BaseEstimator, TransformerMixin 을 상속받아 커스텀 클래스를 만듭니다.

from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.pipeline import Pipeline
from sklearn.model_selection import GridSearchCV, StratifiedKFold
from sklearn.linear_model import LogisticRegression
from gensim.models import Word2Vec
import numpy as np

상속 클래스 역할

상속 클래스	제공 기능
BaseEstimator	get_params() / set_params() 자동 제공
TransformerMixin	fit() + transform() 만 구현하면 fit_transform() 자동 제공

Word2VecVectorizer 전체 코드

class Word2VecVectorizer(BaseEstimator, TransformerMixin):

    def __init__(self, tokenizer=None, vector_size=100, min_count=5,
                 window=5, sg=1, epochs=100, workers=1, seed=42):
        self.tokenizer   = tokenizer
        self.vector_size = vector_size
        self.min_count   = min_count
        self.window      = window
        self.sg          = sg
        self.epochs      = epochs
        self.workers     = workers
        self.seed        = seed
        self.model       = None   # 학습 전 빈 공간 초기화
        self.voca        = None

    def to_token(self, sentences):
        """토큰화 — tokenizer가 None이면 split() 사용"""
        if self.tokenizer is None:
            return [sentence.split() for sentence in sentences]
        else:
            return [self.tokenizer(sentence) for sentence in sentences]

    def fit(self, X, y=None):
        """Word2Vec 학습"""
        sentences  = self.to_token(X)
        self.model = Word2Vec(
            sentences   = sentences,
            vector_size = self.vector_size,
            window      = self.window,
            min_count   = self.min_count,
            sg          = self.sg,
            epochs      = self.epochs,
            workers     = self.workers,
            seed        = self.seed
        )
        self.voca = self.model.wv.key_to_index.keys()
        return self

    def doc_vec(self, token):
        """문장 내 단어 벡터 평균 계산"""
        vectors = []
        for word in token:
            if word in self.model.wv.index_to_key:
                vectors.append(self.model.wv[word])
        if vectors:
            return np.mean(vectors, axis=0)
        else:
            return np.zeros(self.model.vector_size)

    def transform(self, X):
        """토큰화 + 벡터화 (문장 목록 → 벡터 행렬)"""
        sentences = self.to_token(X)
        return np.array([self.doc_vec(token) for token in sentences])

사용 예시

komoran   = Komoran()
tokenizer = lambda x: [word for word in komoran.morphs(x)]

docs  = ['상품의 품질이 아주 좋다', '배송이 너무 느리다']
label = [1, 0]

vec_class = Word2VecVectorizer(tokenizer, min_count=1)
vec_class.fit(docs, label)
print(vec_class.transform(docs).shape)   # (2, 100)

# BaseEstimator 덕분에 자동 제공
print(vec_class.get_params())
vec_class.set_params(workers=2, sg=0)

---

4. Pipeline + GridSearchCV 연동

Word2VecVectorizer 가 sklearn 인터페이스를 갖추었으므로 Pipeline/GridSearchCV와 자유롭게 연동할 수 있습니다.

import pandas as pd

df = pd.read_csv('../data/ratings_train.txt', sep='\t')
df.dropna(inplace=True)
df.drop_duplicates('document', inplace=True)
df2 = df[:200]

pipe = Pipeline([
    ('emb',  Word2VecVectorizer(tokenizer=tokenizer, min_count=1)),
    ('logi', LogisticRegression())
])

pipe.fit(df2['document'].values, df2['label'].values)

# 예측 & 평가
df3    = df.tail(100)
X_test = df3['document'].values
y_test = df3['label'].values

pred = pipe.predict(X_test)
print(classification_report(pred, y_test))

# GridSearchCV — Pipeline 내 매개변수는 단계명__매개변수 형태로 접근
cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)

grid_params = {
    'emb__vector_size' : [100, 200],
    'emb__sg'          : [0, 1],
    'logi__C'          : [0.8, 1.0]
}

grid = GridSearchCV(estimator=pipe, param_grid=grid_params, cv=cv, verbose=1)
grid.fit(df2['document'].values, df2['label'].values)

print('최적 파라미터:', grid.best_params_)
print('최적 Score  :', grid.best_score_)

pred = grid.predict(X_test)
print(classification_report(pred, y_test))

---

5. embed 모듈화 — 라이브러리로 분리

앞서 만든 함수들을 embed.py 파일로 분리하여 재사용 가능한 모듈로 만듭니다.

embed.py 전체 구조

# embed.py
from konlpy.tag import Komoran
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
from gensim.models import Word2Vec
from sklearn.feature_extraction.text import TfidfVectorizer

# 전역 변수 선언 (다른 함수에서 공유)
wv         = None
idf_weight = None

def tokenize(text):
    """토큰화 — Komoran 사용 불가 시 split() fallback"""
    try:
        komoran   = Komoran()
        allow_pos = ['NNP', 'NNG', 'VV', 'VA', 'SL', 'MAG']
        tokens    = [word for word, pos in komoran.pos(text) if pos in allow_pos]
    except Exception as e:
        print('Komoran 사용이 불가:', e)
        tokens = text.split()
    return tokens

def global_vari_set(docs):
    """wv, idf_weight 전역 변수 설정"""
    tokens = [tokenize(doc) for doc in docs]

    w2v = Word2Vec(
        sentences=tokens, vector_size=100, window=5,
        min_count=1, epochs=100, sg=1, workers=2, seed=42
    )

    global wv
    wv = w2v.wv

    tfidf_vec = TfidfVectorizer(tokenizer=tokenize, lowercase=False).fit(docs)

    global idf_weight
    idf_weight = dict(zip(
        tfidf_vec.get_feature_names_out(),
        tfidf_vec.idf_
    ))

    return tokens

def sent_embed_mean(token):
    """단어 벡터 평균 → 문장 벡터"""
    vector = [wv[word] for word in token if word in wv.index_to_key]
    return np.mean(vector, axis=0) if vector else np.zeros(wv.vector_size)

def sent_embed_wv_idf(token):
    """단어 벡터 × IDF 가중치 평균 → 문장 벡터"""
    vector, idf = [], []
    for word in token:
        if word in wv.index_to_key and word in idf_weight:
            vector.append(wv[word] * idf_weight[word])
            idf.append(idf_weight[word])
    if vector:
        return np.sum(vector, axis=0) / (np.sum(idf) + 1e-9)
    return np.zeros(wv.vector_size)

def run_model(X, y, model, test_size=0.2, stratify=None):
    """학습 → 예측 → classification_report 반환"""
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=test_size, stratify=stratify, random_state=42
    )
    model.fit(X_train, y_train)
    return classification_report(model.predict(X_test), y_test)

def predict_sentence_list(sentences, model, vec_type='wv'):
    """새 문장 리스트 예측 → [(문장, 레이블)] 반환"""
    X_test = []
    for sentence in sentences:
        token = tokenize(sentence)
        vec   = sent_embed_mean(token) if vec_type == 'wv' else sent_embed_wv_idf(token)
        X_test.append(vec)
    preds  = model.predict(X_test)
    return [[s, '긍정' if p == 1 else '부정'] for s, p in zip(sentences, preds)]

# 직접 실행 시에만 테스트 코드 실행
if __name__ == '__main__':
    docs   = ['오늘 날씨가 좋다 여행 가고 싶다', '영화가 너무 재미있어서 시간이 가는 줄 몰랐다']
    target = [1, 1]
    tokens = global_vari_set(docs)
    print(tokens)
    print(wv)
    print(idf_weight)

💡 if __name__ == '__main__': 의 역할
이 파일을 직접 실행할 때만 테스트 코드가 실행됩니다.
import embed 로 불러올 때는 테스트 코드가 실행되지 않습니다.

---

6. 연습문제 — embed 모듈 활용 감성 분류

문제 조건

• ratings_train.txt 파일 로드
• 결측치 제거, 중복 제거, 상위 500개 데이터 사용
• embed 모듈의 wv / idf 두 가지 벡터화 방법 비교
• SVC, LogisticRegression 두 모델 비교
• 최적 모델로 하위 100개 데이터 예측 & 분류 레포트 생성

풀이

import pandas as pd
from sklearn.linear_model import LogisticRegression
from sklearn.svm import SVC
from sklearn.metrics import classification_report
import embed

df = pd.read_csv('../data/ratings_train.txt', sep='\t')
df.dropna(inplace=True)
df.drop_duplicates('document', inplace=True)

df2 = df[:500]
X   = df2['document'].values
y   = df2['label'].values

# embed 모듈로 전역 변수(wv, idf_weight) 설정 + 토큰화
tokens = embed.global_vari_set(X)

# 벡터화 (2가지)
X_wv  = [embed.sent_embed_mean(token)    for token in tokens]
X_idf = [embed.sent_embed_wv_idf(token) for token in tokens]

# 4가지 모델 성능 비교
logi_wv  = LogisticRegression(random_state=42)
logi_idf = LogisticRegression(random_state=42)
svc_wv   = SVC(random_state=42)
svc_idf  = SVC(random_state=42)

print('=== Logistic + wv  ===') ; print(embed.run_model(X_wv,  y, logi_wv))
print('=== SVC + wv       ===') ; print(embed.run_model(X_wv,  y, svc_wv))
print('=== Logistic + idf ===') ; print(embed.run_model(X_idf, y, logi_idf))
print('=== SVC + idf      ===') ; print(embed.run_model(X_idf, y, svc_idf))

최적 모델로 예측 & 결과 DataFrame 생성

# 최적 모델(svc_idf)로 하위 100개 예측
X_test = df2.tail(100)['document'].values
y_test = df2.tail(100)['label'].values

pred    = embed.predict_sentence_list(X_test, svc_idf, 'idf')
pred_df = pd.DataFrame(pred, columns=['sentence', 'label'])

# 레이블 역변환 (문자 → 숫자)
pred_y = pred_df['label'].map({'부정': 0, '긍정': 1})
print(classification_report(pred_y, y_test))

display(pred_df)

💡 predict_sentence_list() 반환값은 문자형 레이블입니다.
classification_report() 와 비교하려면 map({'부정': 0, '긍정': 1}) 으로 숫자형으로 변환해야 합니다.

---

📎 핵심 개념 요약

개념	설명
IDF + Word2Vec	문맥 기반 Word2Vec 벡터 × IDF 가중치 결합
sent_embed_mean()	문장 내 단어 벡터 평균 → 문장 벡터
sent_embed_wv_idf()	단어 벡터 × IDF 가중치 평균 → 문장 벡터
tfidf_vec.idf_	TfidfVectorizer의 IDF 값 배열
dict(zip(피처, idf_))	{단어: IDF값} 딕셔너리 생성
1e-9	분모 0 방지를 위한 수치 안정성 값
BaseEstimator	get_params() / set_params() 자동 제공
TransformerMixin	fit() + transform() 구현 시 fit_transform() 자동 제공
Word2VecVectorizer	gensim을 sklearn Pipeline에 연동하기 위한 커스텀 클래스
emb__vector_size	Pipeline 내 단계 매개변수 접근 방식
global 변수명	함수 내에서 전역 변수 수정
globals()['변수명']	딕셔너리로 전역 변수 접근/수정
global_vari_set()	wv, idf_weight 전역 변수 설정 함수
if __name__ == '__main__':	직접 실행 시에만 테스트 코드 실행
embed.run_model()	분할 → 학습 → 예측 → 레포트 한번에 처리
pred_df['label'].map({'부정': 0, '긍정': 1})	문자 레이블 → 숫자 변환