はじめに

Hugging Faceにあった日本語用LUKEを使用してレコメンドモデルを作成してみました。
本記事では、model.pyの作成までを解説します！

Sentence-LUKEとは

Studio Ousia様によるLUKEモデルをベースに学習したSentence-LUKEモデルです。
日本語Sentence-BERTモデル（バージョン2）と比べて定量的な精度が同等〜0.5pt程度高く、定性的な精度は本モデルの方が高い結果でした。

目標

anacondaの仮想環境内で日本語用Sentence-LUKEを使い、必要なデータのスクレイピングから文章と単語の類似度を測るところ、レコメンドまでをやっていきます。
今回はこちらのサイトからスクレイピングをして、おすすめクラフトビールのレコメンドをターミナルへ出力できるところまでを行います。

環境

python 3.10.12
transformers 4.29.2
VSCode(model.pyファイルを作成してください！)

手順

仮想環境の作成

まずは、仮想環境の作成＆有効化をします！

仮想環境の作成＆有効化

仮想環境の作成

 conda create -n (仮想環境名なんでも良き) python=3.10

仮想環境の有効化

 conda activate (先ほどの仮想環境名)

ライブラリのインストール

次に、anacondaの仮想環境に下記4つをインストールします。

ライブラリ集

transformers

 conda install transformers

sentencepiece

 conda install sentencepiece

torch

 pip install torch

scipy

 pip install scipy

スクレイピングで集めたデータをCSV形式にする

今回は、上記のクラフトビールのレビューサイトからスクレイピングで集めたCSVファイルを用います！（GoogleスプレッドシートからCSVファイルで保存してください）

ダウンロードできたら、VSCodeにこのCSVファイルを移してください！

※スクレイピングの基礎について学びたい方は、こちらの記事も読んでみてください！

Sentence-LUKEを試す

次に、Sentence-LUKEを実際に試してみることにします！

お試しコード

model.py

from transformers import MLukeTokenizer, LukeModel
import torch


class SentenceLukeJapanese:
    def __init__(self, model_name_or_path, device=None):
        self.tokenizer = MLukeTokenizer.from_pretrained(model_name_or_path)
        self.model = LukeModel.from_pretrained(model_name_or_path)
        self.model.eval()

        if device is None:
            device = "cuda" if torch.cuda.is_available() else "cpu"
        self.device = torch.device(device)
        self.model.to(device)

    def _mean_pooling(self, model_output, attention_mask):
        token_embeddings = model_output[0] #First element of model_output contains all token embeddings
        input_mask_expanded = attention_mask.unsqueeze(-1).expand(token_embeddings.size()).float()
        return torch.sum(token_embeddings * input_mask_expanded, 1) / torch.clamp(input_mask_expanded.sum(1), min=1e-9)


    def encode(self, sentences, batch_size=8):
        all_embeddings = []
        iterator = range(0, len(sentences), batch_size)
        for batch_idx in iterator:
            batch = sentences[batch_idx:batch_idx + batch_size]

            encoded_input = self.tokenizer.batch_encode_plus(batch, padding="longest", 
                                           truncation=True, return_tensors="pt").to(self.device)
            model_output = self.model(**encoded_input)
            sentence_embeddings = self._mean_pooling(model_output, encoded_input["attention_mask"]).to('cpu')

            all_embeddings.extend(sentence_embeddings)

        return torch.stack(all_embeddings)


MODEL_NAME = "sonoisa/sentence-luke-japanese-base-lite"
model = SentenceLukeJapanese(MODEL_NAME)

sentences = ["暴走したAI", "暴走した人工知能"]
sentence_embeddings = model.encode(sentences, batch_size=8)

print("Sentence embeddings:", sentence_embeddings)

出力結果

Sentence embeddings: tensor([[ 0.0655, -0.4808,  0.1523,  ..., -0.2441, -0.4688,  0.0680],
        [ 0.1478, -0.9534,  0.2588,  ..., -0.1766, -0.2504, -0.2852]],
       grad_fn=<StackBackward0>)

これで["暴走したAI", "暴走した人工知能"]の２つの文章をベクトル化することに成功しました！

実際にレコメンドをしてみた

ここから、先ほどのCSVファイルからレコメンドを行い、おすすめのクラフトビールを出してみたいと思います！

まずは、model.pyを以下のサンプルコードに書き換えてみてください！
その後、ターミナルで　python model.pyを実行し、今の自分の気分を書き込んでみてください！
※試しに、私は低アルコールと入力してみました！

サンプルコード

model.py

from transformers import MLukeTokenizer, LukeModel
import sentencepiece as spm
import torch
import scipy.spatial
import pandas as pd

class SentenceLukeJapanese:
    def __init__(self, model_name_or_path, device=None):
        self.tokenizer = MLukeTokenizer.from_pretrained(model_name_or_path)
        self.model = LukeModel.from_pretrained(model_name_or_path)
        self.model.eval()

        if device is None:
            device = "cuda" if torch.cuda.is_available() else "cpu"
        self.device = torch.device(device)
        self.model.to(device)

    def _mean_pooling(self, model_output, attention_mask):
        token_embeddings = model_output[0]
        input_mask_expanded = (
            attention_mask.unsqueeze(-1).expand(token_embeddings.size()).float()
        )
        return torch.sum(token_embeddings * input_mask_expanded, 1) / torch.clamp(
            input_mask_expanded.sum(1), min=1e-9
        )

    @torch.no_grad()
    def encode(self, sentences, batch_size=8):
        all_embeddings = []
        iterator = range(0, len(sentences), batch_size)
        for batch_idx in iterator:
            batch = sentences[batch_idx: batch_idx + batch_size]

            encoded_input = self.tokenizer.batch_encode_plus(
                batch, padding="longest", truncation=True, return_tensors="pt"
            ).to(self.device)
            model_output = self.model(**encoded_input)
            sentence_embeddings = self._mean_pooling(
                model_output, encoded_input["attention_mask"]
            ).to("cpu")

            all_embeddings.extend(sentence_embeddings)

        return torch.stack(all_embeddings)

def recommend_beer(model, csv_file_path, query):
    """
    理想のビールを推薦する関数

    Args:
        model: SentenceLukeJapanese モデル
        csv_file_path (str): CSVファイルのパス
        query (str): どんなビールが飲みたいかの気分を表す文章

    Returns:
        dict: おすすめビールの名前、説明、類似度
    """
    # CSVデータの読み込み
    data = pd.read_csv(csv_file_path)
    target_column_name = 'introduction_text'
    sentences = data[target_column_name].tolist()

    # 気分をリストに追加
    sentences.append(query)

    # 文のエンコード（ベクトル化）
    sentence_embeddings = model.encode(sentences, batch_size=8)

    # 類似度計算
    distances = scipy.spatial.distance.cdist(
        [sentence_embeddings[-1]], sentence_embeddings[:-1], metric="cosine"
    )[0]

    # 類似度上位の結果を取得
    results = sorted(enumerate(distances), key=lambda x: x[1])

    if len(results) == 0:
        return {"error": "条件に一致するビールが見つかりませんでした"}

    closest_index, similarity = results[0]

    # 結果を返却
    if isinstance(closest_index, int):
        return {
            "name": data.iloc[closest_index]['name'],
            "description": data.iloc[closest_index][target_column_name],
            "similarity": round((1 - similarity) * 100, 2),  # パーセンテージ形式
        }
    else:
        return {"error": "インデックスが不正です"}


# 既存モデルの読み込み
MODEL_NAME = "sonoisa/sentence-luke-japanese-base-lite"
model = SentenceLukeJapanese(MODEL_NAME)

# CSVファイルパス
csv_file_path = 'beersking_kansei.csv'


# 入力クエリ
query = input('どんなビールが飲みたい？：')

# 関数呼び出し
result = recommend_beer(model, csv_file_path, query)

# 結果表示
if "error" in result:
    print(result["error"])
else:
    print("\nオススメのクラフトビールは:")
    print("名前:", result["name"])
    print("説明:", result["description"])
    print("類似度スコア:", f"{result['similarity']}%")

出力結果

Query: 低アルコール

オススメのクラフトビールは:
CREW Republic「HOP JUNKIE」
キレのあるドライな飲み心地、華やかでバラエティ豊かなホップの香り、低アルコールでとても飲みやすい

ValueError: only one element tensors can be converted to Python scalarsが出る人向け

model.py

from transformers import MLukeTokenizer, LukeModel
import sentencepiece as spm
import torch
import scipy.spatial
import pandas as pd
import numpy as np


class SentenceLukeJapanese:
    def __init__(self, model_name_or_path, device=None):
        self.tokenizer = MLukeTokenizer.from_pretrained(model_name_or_path)
        self.model = LukeModel.from_pretrained(model_name_or_path)
        self.model.eval()

        if device is None:
            device = "cuda" if torch.cuda.is_available() else "cpu"
        self.device = torch.device(device)
        self.model.to(device)

    def _mean_pooling(self, model_output, attention_mask):
        token_embeddings = model_output[
            0
        ]  # First element of model_output contains all token embeddings
        input_mask_expanded = (
            attention_mask.unsqueeze(-1).expand(token_embeddings.size()).float()
        )
        return torch.sum(token_embeddings * input_mask_expanded, 1) / torch.clamp(
            input_mask_expanded.sum(1), min=1e-9
        )

    @torch.no_grad()
    def encode(self, sentences, batch_size=8):
        all_embeddings = []
        iterator = range(0, len(sentences), batch_size)
        for batch_idx in iterator:
            batch = sentences[batch_idx : batch_idx + batch_size]

            encoded_input = self.tokenizer.batch_encode_plus(
                batch, padding="longest", truncation=True, return_tensors="pt"
            ).to(self.device)
            model_output = self.model(**encoded_input)
            sentence_embeddings = self._mean_pooling(
                model_output, encoded_input["attention_mask"]
            ).to("cpu")

            all_embeddings.extend(sentence_embeddings)

        return torch.stack(all_embeddings)


# 既存モデルの読み込み
MODEL_NAME = "sonoisa/sentence-luke-japanese-base-lite"
model = SentenceLukeJapanese(MODEL_NAME)

# 説明文を入れるリストを作成
sentences = []

# CSVファイルのパスを指定
csv_file_path = 'beersking_kansei.csv'

# 読み込む列の名前を指定
target_column_name = 'introduction_text'

# CSVファイルをDataFrameとして読み込む
data = pd.read_csv(csv_file_path)

# 指定した列のデータをリストに追加
sentences = data[target_column_name].tolist()


# 標準入力で、理想のビールのイメージを文章で受け取る
query = input()
sentences.append(query)

# ビールの説明文、受け取った文章をエンコード（ベクトル表現に変換）
sentence_embeddings = model.encode(sentences, batch_size=8)
sentence_embeddings_np = sentence_embeddings.numpy()

# 類似度上位1つを出力
closest_n = 1

distances = scipy.spatial.distance.cdist(
    [sentence_embeddings_np[-1]], sentence_embeddings_np, metric="cosine"
)[0]

results = zip(range(len(distances)), distances)
results = sorted(results, key=lambda x: x[1])

print("\n\n======================\n\n")
print("Query:", query)
print("\nオススメのビールは:")
index= data[data['introduction_text']==sentences[results[1][0]].strip()].index[0]

for idx, distance in results[1 : closest_n + 1]:
    print(data.iloc[index,1])
   
    print(sentences[idx].strip())

出力結果

Query: 低アルコール

オススメのクラフトビールは:
CREW Republic「HOP JUNKIE」
キレのあるドライな飲み心地、華やかでバラエティ豊かなホップの香り、低アルコールでとても飲みやすい

次回予告

これで、ターミナル上でオススメのビールをレコメンドすることに成功しました！

次回は、実際にこのレコメンドモデルをFlaskアプリにしてみようと思います！