機械学習Advent Calendar 2023

BERTでembeddingした文書をクラスタリングしてWordCloudsで分析する

Posted at 2023-12-18

皆さんこんにちは。今日は「BERTでembeddingした文書をクラスタリングしてWordCloudsで分析する」というテーマで雑談したいと思います。いつものように、ChatGPT先生にも来ていただいています。

assistant:

こんにちは。今日のテーマはとても面白そうですね。BERTを使った文章の埋め込みからクラスタリングを行い、その結果をWordCloudsで分析するという内容は、自然言語処理と可視化のテクニックが合わさった内容となっていますね。どのようなステップでそれを実現するのでしょうか？

user:

まずは、日本語の文書データを手に入れます。今回は、『名大会話コーパス』https://mmsrv.ninjal.ac.jp/nucc/ を使いたいと思います。

assistant:

その選択は良いと思います。名大会話コーパスは自然な日本語の会話データが収録された資源ですから、自然言語処理の素材としては適したものでしょう。このデータセットをどのようにしてBERTでembeddingする予定ですか？

user:

実は既に、ChatGPT先生に助けていただきながら生成したPythonコードがあります。以下のコードについて、日本語で詳しく解説をお願いします。

import os
from typing import List


def read_text_files(folder_path: str) -> List[str]:
    """
    指定されたディレクトリ内のすべてのテキストファイルの内容をリストとして返す関数

    Parameters:
    folder_path : str : テキストファイルが格納されたディレクトリへのパス

    Returns:
    texts : List[str] : 各テキストファイルの内容のリスト
    """

    # 内容を保存するためのリストを初期化
    texts = []

    # 指定されたディレクトリ内のすべてのファイルについて処理を行う
    for file_name in os.listdir(folder_path):
        # ファイルパスを生成
        file_path = os.path.join(folder_path, file_name)

        # テキストファイルのみを対象とする
        if file_path.endswith(".txt"):
            try:
                # テキストファイルを開き、内容を読み込む
                with open(file_path, "r", encoding="utf-8") as file:
                    text = file.read()
                    texts.append(text)
            except Exception as e:
                print(f"Could not read file {file_path} due to error {e}")
    return texts


folder_path = "nucc/"  # 検索したいディレクトリのパスを指定
# 指定されたディレクトリ中のすべてのテキストファイルを読み込む
texts = read_text_files(folder_path)

assistant:

このコードは、指定されたディレクトリ内に存在するテキストファイルすべてを開き、内容をリストとして保存する処理を行っています。関数 read_text_files は、ファイルパスを引数として受け取り、そのファイルパス以下のテキストファイルすべての内容をリストで返します。

処理の詳細は以下の通りです：

os.listdir(folder_path) で指定されたディレクトリ内のすべてのファイル名を取得します。
それぞれのファイルに対して、ファイルパスを生成します。これは os.path.join(folder_path, file_name) で実現されます。
ファイルがテキストファイルであるか（ファイル名が ".txt" で終わるか）を確認します。これは file_path.endswith(".txt") で行われます。
条件を満たす各テキストファイルを開き（open(file_path, "r", encoding="utf-8") as file）、その内容を読み取ります（file.read()）。
テキストファイルの内容はリスト texts に追加されます。
ファイルの読み取り中に何らかの問題が発生した場合は、そのファイル名とエラーを表示します。

以上の結果、textsは読み込んだ全てのテキストファイルの内容を持つリストです。folder_path = "nucc/" の部分でディレクトリを指定し、texts = read_text_files(folder_path)で関数を呼び出し、ファイルの内容を読み込んでいます。

user:

ありがとうございます。続いて、以下のコードについて、日本語で詳しく解説をお願いします。

from typing import List

import torch
from nltk.corpus import stopwords
from transformers import BertModel, BertTokenizer


def preprocess_text(texts: List[str]) -> List[str]:
    stop_words = set(stopwords.words("english"))
    processed_texts = []
    for text in texts:
        words = text.split()
        words = [word for word in words if word not in stop_words]
        text = " ".join(words)
        processed_texts.append(text)
    return processed_texts


def get_embeddings(texts: List[str]) -> torch.Tensor:
    tokenizer = BertTokenizer.from_pretrained("bert-base-multilingual-cased")
    model = BertModel.from_pretrained("bert-base-multilingual-cased")

    # GPUが使用可能であればGPUを利用する
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    model.to(device)

    embeddings = []

    # テキストの前処理
    processed_texts = preprocess_text(texts)

    for text in processed_texts:
        encoded_input = tokenizer(
            text, padding=True, truncation=True, max_length=512, return_tensors="pt"
        ).to(device)

        with torch.no_grad():
            outputs = model(**encoded_input)
            last_hidden_state = outputs.last_hidden_state
            embedding = torch.mean(last_hidden_state, dim=1)
            embeddings.append(embedding)

    embeddings = torch.cat(embeddings, dim=0).cpu()  # 最後に結果をcpuに戻す
    embeddings = embeddings.numpy()

    return embeddings


embeddings = get_embeddings(texts)  # BERTを使用してテキストをベクトル化

assistant:

このコードは、まずテキストデータを前処理し、その上でBERTを利用してテキストデータをベクトル化するという処理を行っています。

まずpreprocess_text関数では、引数として受け取ったテキストリストに対して以下の前処理を行います：

nltkのストップワードリストを用いて、テキストから英語のストップワードを除去します。ストップワードとは、"the", "is", "in"など、文章中で頻繁に使用されるがその単語が出現することによる情報量が少ないとされる単語のことを指します。

次にget_embeddings関数では、前処理済みの各テキストに対してBERTを使用してベクトル化を行います。詳細な流れは以下のとおりです：

まず、from_pretrained("bert-base-multilingual-cased")を用いて、マルチリンガル（多言語対応）で事前学習されたBERTのモデルとトークナイザーを取得します。
使用するデバイスを選択します。可能であればGPUを利用しますが、使用不可能であればCPUを用います。
前回の関数で定義した前処理の関数preprocess_textを用いてテキストデータを前処理します。
各テキストに対して以下の処理を行い、結果の埋め込みをリストに追加します：
1. tokenizerを用いてテキストをトークン化し、オプションでパディング、トランケーション、最大長指定を行い、テンソル型に変換します。
2. 作成した入力を用いてBERTモデルを通過させ、出力を得ます。ここではwith torch.no_grad():を使用して、推論時の勾配計算を行わないようにしています。
3. 得られた出力の最後の隠れ層の平均値をとり、それを各テキストの埋め込みとして利用します。
最後に、結果として得られた埋め込みのリストをテンソルとして結合し、それをnumpy配列に変換して、結果を返します。

この結果として得られるembeddingsは、各テキストをBERTでベクトル化した結果となります。

user:

ありがとうございます。続いて、以下のコードについて、日本語で詳しく解説をお願いします。

from typing import List

from numpy import ndarray
from sklearn.cluster import SpectralClustering


def perform_clustering(
    embeddings: ndarray,
    n_clusters: int = 9,
    affinity: str = "nearest_neighbors",
    random_state: int = 0,
) -> ndarray:
    """
    SpectralClusteringを用いてクラスタリングおよびラベルの予測を行います。

    Parameters:
    embeddings : ndarray : ベクトル表現されたテキストデータ
    n_clusters : int : クラスタ数 (default is 9)
    affinity : str : SpectralClusteringで使用するアフィニティ (default is "nearest_neighbors")
    random_state : int : ランダムシード (default is 0)

    Returns:
    labels : ndarray : クラスタリング結果のラベル
    """

    # クラスタリングオブジェクトの生成
    clt = SpectralClustering(
        n_clusters=n_clusters, affinity=affinity, random_state=random_state
    )

    # ベクトルを使用してクラスタリング
    clt.fit(embeddings)

    # ラベルの予測
    labels = clt.fit_predict(embeddings)

    return labels


# SpectralClusteringを使用してテキストのクラスタリングを実行し、クラスタリング結果のラベルを取得
labels = perform_clustering(embeddings)
print(labels)

assistant:

このコードは、SpectralClusteringを用いて、BERTによってベクトル化された文書をいくつかのクラスタに分ける処理を行っています。

spectral_clustering関数は以下のような流れで処理を行います。

scikit-learnのSpectralClusteringオブジェクトを生成します。ここでは、クラスタ数、アフィニティ（類似度の指標）、ランダムシードを指定しています。
生成したオブジェクトに対し、fitメソッドでベクトル化されたデータに対してクラスタリングを実行します。
同じく生成したオブジェクトに対し、fit_predictメソッドでデータに対してクラスタリングを行った後、各データ点に対するクラスタラベルを予測します。

この結果として得られるlabelsは、各文書が所属するクラスタを表す配列となります。

なお、SpectralClusteringは、データをグラフとみなし、そのグラフ上でクラスタリングを行う手法の一つで、似たようなデータが高い確率で同じクラスタに分けられ、異なるデータが低い確率で同じクラスタに分けられるようなクラスタリングを目指します。

また、アフィニティパラメータはクラスタリングの際の類似度の計算方法を指定します。ここでは"nearest_neighbors"が指定されており、これは各点が他の点とどの程度近くにあるか（すなわち、近傍の点との距離）を基に類似度を計算します。

以上の内容より、このコードは与えられたベクトル化された文書を、SpectralClusteringを用いて指定された数のクラスタに分割し、その結果を配列として出力します。

user:

ありがとうございます。最後に、以下のコードについて、日本語で詳しく解説をお願いします。

from collections import Counter
from typing import List

import matplotlib.pyplot as plt
import nltk
from nltk.corpus import stopwords
from sklearn.feature_extraction.text import CountVectorizer
from wordcloud import STOPWORDS, WordCloud

nltk.download("stopwords")

FONT_PATH = "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc"


def visualize_wordclouds(
    texts: List[str], labels: List[int], max_words: int = 100
) -> None:
    # ラベルの出現回数をカウントする
    label_counter = Counter(labels)

    # テキストデータをベクトル化
    vectorizer = CountVectorizer()
    vectorized_texts = vectorizer.fit_transform(texts)

    # ボキャブラリーと単語の出現回数を取得
    vocabulary = vectorizer.get_feature_names_out()
    word_frequency = vectorized_texts.sum(axis=0).A1.tolist()

    # ラベルごとの単語とその出現回数をまとめる
    word_freq_per_label = {label: {} for label in set(labels)}
    common_frequency = Counter()

    for label in set(labels):
        local_counter = Counter()
        for word, freq in zip(vocabulary, word_frequency):
            for text, lbl in zip(texts, labels):
                if lbl == label and word in text:
                    local_counter[word] += freq
        word_freq_per_label[label] = local_counter
        common_frequency += local_counter

    # 共通単語を特定し、全ラベルから削除
    common_words = {
        word for word, freq in common_frequency.items() if freq > len(set(labels))
    }
    for label in word_freq_per_label:
        for word in common_words:
            word_freq_per_label[label].pop(word, None)

    # WordCloudの作成と表示
    for label, word_freq in word_freq_per_label.items():
        # ストップワードを考慮し、最大単語数を設定
        wordcloud = WordCloud(
            width=800,
            height=400,
            background_color="white",
            max_words=max_words,
            font_path=FONT_PATH,
            stopwords=set(stopwords.words("english")) | STOPWORDS,
        ).generate_from_frequencies(word_freq)

        # 文書数をタイトルに含める
        doc_count = label_counter[label]

        # Matplotlibを用いてWordCloudを表示
        plt.figure(figsize=(8, 6))
        plt.imshow(wordcloud, interpolation="bilinear")
        plt.title(f"Cluster {label} (document count: {doc_count})")
        plt.axis("off")
        plt.show()


visualize_wordclouds(texts, labels, max_words=100)

assistant:

このコードでは、クラスタリングの結果を利用して、各クラスタにおける単語の使われ方をWordCloudとして可視化しています。

具体的な処理手順は以下のとおりです。

手始めに、クラスタラベルのリストから各クラスタの出現回数をカウントしています。これは後ほどWordCloudのタイトル表示に使用されます。
続いて、CountVectorizerを用いてテキストデータをベクトル化します。この操作によって、テキスト内に出現する各単語とその出現回数が抽出されます。
ベクトル化されたテキストから各クラスタにおける単語の出現頻度を記録します。具体的には、各クラスタに対して、そのクラスタに所属する文書に出現する単語とその出現回数を紐付けて記録していきます。また、全クラスタを通じた単語の出現頻度も同時に記録しています。
テキスト全体で頻出する単語（すなわち、文書全体で共通して使われている単語）を特定し、各クラスタにおける単語の出現頻度の記録から削除します。これによってWordCloudでは各クラスタ固有の単語が強調されることになります。
最後に、各クラスタについてWordCloudを生成し、表示します。WordCloudでは、単語の出現頻度に基づいて単語の視覚的なサイズが決定されます。そのため、一つのクラスタ内で頻繁に使われる単語ほど大きく表示されます。