文章ベクトルSCDVの実装例のご紹介

import re
import numpy as np
import pandas as pd
import matplotlib.pylab as plt
import seaborn as sns
from sklearn.feature_extraction.text import CountVectorizer, TfidfVectorizer
from sklearn import datasets, manifold, mixture, model_selection
from gensim.models import Word2Vec
from gensim.models.doc2vec import Doc2Vec, TaggedDocument
from xgboost import XGBClassifier

# 適当にトピックカテゴリを選択
categories = [
    'alt.atheism',
    'comp.graphics',
    'rec.sport.baseball',
    'sci.space',
    'talk.politics.guns'
]
train = datasets.fetch_20newsgroups(subset='train', categories=categories)
train.data = np.array(train.data, dtype=object)

# それぞれ含まれる文章数をカウント
for i, c in enumerate(categories):
    indices = np.where(train.target == i)
    print(c + ':\t{}'.format(len(train.data[indices])))

alt.atheism:    480
comp.graphics:  584
rec.sport.baseball: 597
sci.space:  593
talk.politics.guns: 546

# BoW, tf-idf, average Word2Vec, Doc2Vec, SCDV
features_num = 200
min_word_count = 10
context = 5
downsampling = 1e-3
epoch_num = 10

# Analyzer
def analyzer(text):
    stop_words = ['i', 'a', 'an', 'the', 'to', 'and', 'or', 'if', 'is', 'are', 'am', 'it', 'this', 'that', 'of', 'from', 'in', 'on']
    text = text.lower() # 小文字化
    text = text.replace('\n', '') # 改行削除
    text = text.replace('\t', '') # タブ削除
    text = re.sub(re.compile(r'[!-\/:-@[-`{-~]'), ' ', text) # 記号をスペースに置き換え
    text = text.split(' ') # スペースで区切る

    words = []
    for word in text:
        if (re.compile(r'^.*[0-9]+.*$').fullmatch(word) is not None): # 数字が含まれるものは除外
            continue
        if word in stop_words: # ストップワードに含まれるものは除外
            continue
        if len(word) < 2: # 1文字、0文字（空文字）は除外
            continue
        words.append(word)

    return words

# BoW
corpus = train.data
count_vectorizer = CountVectorizer(analyzer=analyzer, min_df=min_word_count, binary=True)
bows = count_vectorizer.fit_transform(corpus)
bows.shape # (2800, 5445)

# t-SNEで圧縮して可視化
tsne_bow = manifold.TSNE(n_components=2, random_state=42).fit_transform(bows.toarray())

df_tsne_bow = pd.DataFrame({
    'x': tsne_bow[:, 0],
    'y': tsne_bow[:, 1],
    'category': train.target,
})
df_tsne_bow.plot.scatter(x='x', y='y', c='category', colormap='viridis', figsize=(7, 5), s=20)
plt.show()

# tf-idf
corpus = train.data
tfidf_vectorizer = TfidfVectorizer(analyzer=analyzer, min_df=min_word_count)
tfidfs = tfidf_vectorizer.fit_transform(corpus)
tfidfs.shape # (2800, 5445)

# t-SNEで圧縮して可視化
tsne_tfidf = manifold.TSNE(n_components=2, random_state=42).fit_transform(tfidfs.toarray())

df_tsne_tfidf = pd.DataFrame({
    'x': tsne_tfidf[:, 0],
    'y': tsne_tfidf[:, 1],
    'category': train.target,
})
df_tsne_tfidf.plot.scatter(x='x', y='y', c='category', colormap='viridis', figsize=(7, 5), s=20)
plt.show()

# Averaged-Word2Vec
corpus = [analyzer(text) for text in train.data]
word2vecs = Word2Vec(
    sentences=corpus, epochs=epoch_num, vector_size=features_num,
    min_count=min_word_count, window=context, sample=downsampling, seed=42,
)

def avg_vector(text):
    vecs = [word2vecs.wv[w] for w in analyzer(text) if w in word2vecs.wv.key_to_index]
    if len(vecs) == 0:  # 語彙に含まれる単語が無い場合はゼロベクトル
        return np.zeros(features_num, dtype=np.float32)
    return np.mean(vecs, axis=0)

avg_word2vec = np.array([avg_vector(text) for text in train.data])
avg_word2vec.shape # (2800, 200)

# t-SNEで圧縮して可視化
tsne_avg_word2vec = manifold.TSNE(n_components=2, random_state=42).fit_transform(avg_word2vec)

df_tsne_avg_word2vec = pd.DataFrame({
    'x': tsne_avg_word2vec[:, 0],
    'y': tsne_avg_word2vec[:, 1],
    'category': train.target,
})
df_tsne_avg_word2vec.plot.scatter(x='x', y='y', c='category', colormap='viridis', figsize=(7, 5), s=20)
plt.show()

# Doc2Vec
corpus = [TaggedDocument(words=analyzer(text), tags=[i]) for i, text in enumerate(train.data)]
doc2vec_model = Doc2Vec(
    documents=corpus, dm=1, epochs=epoch_num, vector_size=features_num,
    min_count=min_word_count, window=context, sample=downsampling, seed=42
) # dm == 1 -> dmpv, dm != 1 -> DBoW
doc2vecs = np.array([doc2vec_model.infer_vector(analyzer(text)) for text in train.data])
doc2vecs.shape # (2800, 200)

# t-SNEで圧縮して可視化
tsne_doc2vec = manifold.TSNE(n_components=2, random_state=42).fit_transform(doc2vecs)

df_tsne_doc2vec = pd.DataFrame({
    'x': tsne_doc2vec[:, 0],
    'y': tsne_doc2vec[:, 1],
    'category': train.target,
})
df_tsne_doc2vec.plot.scatter(x='x', y='y', c='category', colormap='viridis', figsize=(7, 5), s=20)
plt.show()

word_vectors = word2vecs.wv.vectors
clusters_num = 60
gmm = mixture.GaussianMixture(n_components=clusters_num, covariance_type='tied', max_iter=50, random_state=42)
gmm.fit(word_vectors)

idf_dic = dict(zip(tfidf_vectorizer.get_feature_names_out(), tfidf_vectorizer.idf_))
assign_dic = dict(zip(word2vecs.wv.index_to_key, gmm.predict(word_vectors)))
soft_assign_dic = dict(zip(word2vecs.wv.index_to_key, gmm.predict_proba(word_vectors)))

word_topic_vecs = {}
for word in assign_dic:
    word_topic_vecs[word] = np.zeros(features_num*clusters_num, dtype=np.float32)
    for i in range(0, clusters_num):
        try:
            word_topic_vecs[word][i*features_num:(i+1)*features_num] = word2vecs.wv[word]*soft_assign_dic[word][i]*idf_dic[word]
        except:
            continue

scdvs = np.zeros((len(train.data), clusters_num*features_num), dtype=np.float32)

a_min = 0
a_max = 0

for i, text in enumerate(train.data):
    tmp = np.zeros(clusters_num*features_num, dtype=np.float32)
    words = analyzer(text)
    for word in words:
        if word in word_topic_vecs:
            tmp += word_topic_vecs[word]
    norm = np.sqrt(np.sum(tmp**2))
    if norm > 0:
        tmp /= norm
    a_min += min(tmp)
    a_max += max(tmp)
    scdvs[i] = tmp

p = 0.04
a_min = a_min*1.0 / len(train.data)
a_max = a_max*1.0 / len(train.data)
thres = (abs(a_min)+abs(a_max)) / 2
thres *= p

scdvs[abs(scdvs) < thres] = 0
scdvs.shape # (2800, 12000)

tsne_scdv = manifold.TSNE(n_components=2, random_state=42).fit_transform(scdvs)

df_tsne_scdv = pd.DataFrame({
    'x': tsne_scdv[:, 0],
    'y': tsne_scdv[:, 1],
    'category': train.target,
})
df_tsne_scdv.plot.scatter(x='x', y='y', c='category', colormap='viridis', figsize=(7, 5), s=20)
plt.show()

model = XGBClassifier()

scoring = ['accuracy']
cv_trial_num = 8

records = []
datasets_to_compare = [
    ('BoW', bows.toarray()),
    ('tfidf', tfidfs.toarray()),
    ('avg_Word2Vec', avg_word2vec),
    ('Doc2Vec', doc2vecs),
    ('SCDV', scdvs),
]

# XGBoostは目的変数が0始まりの連番である必要があるためそのまま利用
for name, X in datasets_to_compare:
    cv_rlts = model_selection.cross_validate(
        model, X, train.target, scoring=scoring, cv=cv_trial_num, return_train_score=True)
    for i in range(cv_trial_num):
        records.append({
            'name': name,
            'train_accuracy': cv_rlts['train_accuracy'][i],
            'valid_accuracy': cv_rlts['test_accuracy'][i],
            'time': cv_rlts['fit_time'][i],
        })

df_compare = pd.DataFrame(records)

plt.figure(figsize=(12, 5))
sns.boxplot(data=df_compare, y='name', x='valid_accuracy', orient='h', palette='viridis', linewidth=0.5, width=0.5)
plt.grid()
plt.title('validation accuracy')
plt.show()