意味的類似性マッチングの基礎

Embedding表現の深化

自然言語処理や大規模言語モデル（LLM）のアプリケーションにおいて、関連する「コンテキスト」を取得することは極めて重要です。このコンテキスト取得を実現する中核技術がEmbeddingであり、関連する情報をベクトル空間にマッピングし、その類似度を基に最適な情報を引き出すプロセスを「意味的類似性マッチング」と呼びます。

トークン化とベクトル表現

自然言語をモデルに入力するためには、まずテキストをトークンと呼ばれる単位に分割する必要があります。日本語や中国語の場合、文字単位、単語単位、あるいはN-gram単位など、さまざまな分割アプローチが存在します。

• 元のテキスト：人工知能が世界をより良くする
• 文字ベース：人 工 知 能 が 世 界 を よ り 良 く す る
• 単語ベース：人工知能 が 世界 を より 良く する
• 文字Bi-gramベース：人工 工知 知能 能が が世 世界 界を をよ より り良 良く くす する

近年のLLMでは、主に「文字＋単語」のハイブリッド方式が採用されています。これにより、語彙の意味表現を精度よく捉えつつ、未知の単語（アウトオブボキャブラリー）に対しても文字単位で柔軟に対応できます。

トークン化の後、各トークンをコンピュータが処理可能な数値に変換します。語彙リスト（ボキャブラリ）を用意し、各トークンをインデックス番号に変換したのち、固定次元D（例：256, 768など）の密なベクトル（Embedding）にマッピングします。

import numpy as np
rand_generator = np.random.default_rng(42)
# 語彙サイズV=16、次元D=256の埋め込み行列を初期化
embed_matrix = rand_generator.uniform(size=(16, 256))
assert embed_matrix.shape == (16, 256)

初期化された行列は、学習プロセスを通じて更新されます。かつてはWord2Vecのように、学習後の単語ベクトルが固定される静的なアプローチが主流でしたが、「リンゴ（水果）」と「リンゴ（企業）」のような多義語の区別ができないという課題がありました。現在では、文脈に応じて動的に表現を生成するTransformerベースのモデル（BERTやGPTなど）が主流となっており、[CLS]トークンや平均化プーリング、注意力メカニズムを用いた重み付けにより、柔軟で高精度な文書表現を獲得しています。

類似度の尺度

テキストの意味的類似度を評価するには、一般的にコサイン類似度が用いられます。これは2つのベクトルがなす角のコサイン値を計算するもので、ベクトルの大きさに影響されずに方向の近さを測定できます。

\[ \text{cosine}(x, y) = \frac{x \cdot y}{\|x\| \|y\|} = \frac{\sum_{i=1}^{N} x_i y_i}{\sqrt{\sum_{i=1}^{N} x_i^2} \sqrt{\sum_{i=1}^{N} y_i^2}} \]

import numpy as np

vec_p = np.array([0.1, 0.2, 0.3])
vec_q = np.array([0.2, 0.3, 0.4])

dot_product = np.dot(vec_p, vec_q)
norm_p = np.linalg.norm(vec_p)
norm_q = np.linalg.norm(vec_q)

cos_sim = dot_product / (norm_p * norm_q)
# cos_simは約0.99258

OpenAI APIの活用

Embedding API

OpenAIが提供するAPIを使用して、テキストのベクトル表現を取得できます。APIキーは環境変数から安全に読み込むことが推奨されます。

import os
import openai

openai.api_key = os.environ.get("OPENAI_API_KEY")

def fetch_embedding(text_data, model_id="text-embedding-ada-002"):
    res = openai.Embedding.create(input=[text_data], model=model_id)
    return res.data[0].embedding

vec = fetch_embedding("私はあなたが好きです")
assert len(vec) == 1536

ユーティリティ関数を用いると、ベクトルの取得と類似度計算をより簡潔に記述できます。

from openai.embeddings_utils import get_embedding, cosine_similarity

 Phrase_a = "私はあなたが好きです"
 Phrase_b = "私はあなたを愛しています"
 Phrase_c = "私はあなたが嫌いです"

 emb_a = get_embedding(Phrase_a, engine="text-embedding-ada-002")
 emb_b = get_embedding(Phrase_b, engine="text-embedding-ada-002")
 emb_c = get_embedding(Phrase_c, engine="text-embedding-ada-002")

 print(cosine_similarity(emb_a, emb_b)) # 高い類似度
 print(cosine_similarity(emb_a, emb_c)) # 対義語でも構造が似ていればある程度の類似度が発生する

ChatGPTとプロンプトによる判定

Embeddingを使わず、ChatGPT（GPT-3.5-turboなど）を利用して直接類似度を判定・説明させるアプローチもあります。

prompt_text = "以下の3つの文の類似度を判定し、JSON形式で出力してください。\n1: 私はあなたが好きです\n2: 私はあなたを愛しています\n3: 私はあなたが嫌いです"

response = openai.ChatCompletion.create(
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": prompt_text}]
)

print(response.choices[0].message.content)

実践的なタスクと応用

質問応答（QA）システム

QAタスクの基本は、ユーザーの質問に対して、既存のQAデータベースから最も類似する質問を検索し、その回答を返すことです。ここではKaggleのQuoraデータセットを用いて実装します。

import pandas as pd
from openai.embeddings_utils import get_embedding, cosine_similarity
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity as calc_cosine_similarity

qa_df = pd.read_csv("Kaggle_related_questions.csv")
# データフレームの前三列：Questions, Followers, Answered, Link

少量のデータであれば、すべてのEmbeddingをメモリに展開し、NumPyのベクトル化演算で高速に類似度を計算できます。

knowledge_base = []
for idx, row in qa_df.iterrows():
    emb_vec = get_embedding(row["Questions"], engine="text-embedding-ada-002")
    knowledge_base.append({
        "query": row["Questions"],
        "vector": emb_vec,
        "resource": row["Link"]
    })

def search_answer(user_input, top_k=1):
    input_vec = get_embedding(user_input, engine="text-embedding-ada-002")
    base_matrix = np.array([item["vector"] for item in knowledge_base])
    query_matrix = np.expand_dims(input_vec, axis=0)
    
    sim_scores = calc_cosine_similarity(base_matrix, query_matrix).flatten()
    best_indices = sim_scores.argsort()[-top_k:][::-1]
    
    return [knowledge_base[i] for i in best_indices]

データ規模が大きい場合、Redisなどのベクトルデータベースを利用するのが効率的です。DockerでRedisを起動し、インデックスを構築してベクトルを格納します。

import redis
from redis.commands.search.field import TextField, VectorField
from redis.commands.search.indexDefinition import IndexDefinition
from redis.commands.search.query import Query

r_client = redis.Redis(host="localhost", port=6379)

VEC_DIM = 1536
IDX_NAME = "qa_idx"

q_field = TextField(name="question")
a_field = TextField(name="answer")
v_field = VectorField(name="vec", algorithm="HNSW", attributes={"TYPE": "FLOAT32", "DIM": VEC_DIM, "DISTANCE_METRIC": "COSINE"})

schema = (q_field, v_field, a_field)
search_idx = r_client.ft(IDX_NAME)

try:
    search_idx.info()
except:
    search_idx.create_index(schema, definition=IndexDefinition(prefix=[IDX_NAME + ":"]))

# データ登録
for idx, row in qa_df.iterrows():
    vec_bytes = np.array(get_embedding(row["Questions"], engine="text-embedding-ada-002"), dtype=np.float32).tobytes()
    r_client.hset(f"{IDX_NAME}:{idx}", mapping={
        "question": row["Questions"],
        "vec": vec_bytes,
        "answer": row["Link"]
    })

検索時にはKNNクエリを用いて、類似度が最も高いドキュメントを取得します。

def query_redis(user_query, k=3):
    q_vec = np.array(get_embedding(user_query, engine="text-embedding-ada-002"), dtype=np.float32).tobytes()
    query_obj = Query(f"*=>[KNN {k} @vec $q_vec AS similarity]").sort_by("similarity").paging(0, k).dialect(2)
    res = search_idx.search(query_obj, {"q_vec": q_vec})
    
    for doc in res.docs:
        dist = float(doc.similarity)
        cos_val = 1 - dist # 距離を類似度に変換
        print(f"ID: {doc.id}, Q: {doc.question}, Score: {cos_val:.3f}")

query_redis("Is Kaggle alive?")

クラスタリング分析

テキストのEmbeddingを用いることで、意味的に近いドキュメントをグループ化できます。DBPediaデータセットを利用し、カテゴリごとの分布を可視化・クラスタリングしてみます。

from sklearn.decomposition import PCA
from sklearn.cluster import KMeans
import matplotlib.pyplot as plt

dbpedia_df = pd.read_csv("DBPEDIA_val.csv")
sampled_df = dbpedia_df.sample(200)
filtered_df = sampled_df[sampled_df.l1.isin(["Place", "Work", "Species"])]

filtered_df["embedding"] = filtered_df.text.apply(lambda x: get_embedding(x, engine="text-embedding-ada-002"))

# PCAで1536次元から3次元に圧縮
pca_model = PCA(n_components=3)
X = np.array(filtered_df.embedding.tolist())
X_3d = pca_model.fit_transform(X)

fig = plt.figure(figsize=(10, 8))
ax = fig.add_subplot(111, projection="3d")

for cat in filtered_df.l1.unique():
    subset = X_3d[filtered_df.l1 == cat]
    ax.scatter(subset[:, 0], subset[:, 1], subset[:, 2], label=cat)

ax.legend()
plt.show()

# KMeansによるクラスタリング
kmeans_model = KMeans(n_clusters=3, n_init="auto")
cluster_labels = kmeans_model.fit_predict(X)

レコメンドシステム

ユーザーの行動履歴や嗜好に基づいてコンテンツを推奨するシステムも、意味的類似性の応用です。AG Newsデータセットを用いて、シンプルなレコメンドパイプラインを構築します。

from dataclasses import dataclass
from typing import List, Dict

@dataclass
class Account:
    account_id: str

@dataclass
class PreferenceProfile:
    account_id: str
    fav_categories: List[int]

@dataclass
class ContentItem:
    item_uid: str
    meta: Dict

@dataclass
class InteractionLog:
    log_type: str
    details: Dict

@dataclass
class UserActivity:
    user: Account
    item: ContentItem
    log: InteractionLog
    timestamp: str

# インスタンス化
act1 = Account("user_001")
pref1 = PreferenceProfile("user_001", [1, 2])
itm1 = ContentItem("doc_001", {"id": 1, "cat": "Sports", "title": "Swimming Championship Results"})
log1 = InteractionLog("view", {"duration": 120})
ua1 = UserActivity(act1, itm1, log1, "2023-10-01 10:00:00")

レコメンドの「リコール（召回）」フェーズでは、ユーザーの行動履歴に基づく推奨、嗜好カテゴリに基づく推奨、人気コンテンツの推奨などを組み合わせます。

class Recommender:
    def __init__(self, catalog_df):
        self.catalog = catalog_df
        
    def recall_by_action(self, account, limit):
        action_data = self._fetch_actions(account)
        target_item = self._extract_interest_item(action_data)
        return self._find_similar_items(target_item, limit)
        
    def recall_by_preference(self, account, limit):
        prefs = self._fetch_preferences(account)
        chosen_cat = random.choice(prefs.fav_categories)
        return self.catalog[self.catalog["Class Index"] == chosen_cat].sample(limit)
        
    def recall_trending(self, limit):
        return self.catalog.sample(limit)
        
    def _find_similar_items(self, target_row, limit):
        cat_id = target_row["Class Index"]
        target_vec = np.expand_dims(target_row["embedding"], 0)
        subset = self.catalog[self.catalog["Class Index"] == cat_id]
        matrix = np.array(subset.embedding.tolist())
        
        scores = calc_cosine_similarity(matrix, target_vec).flatten()
        top_ids = scores.argsort()[-(limit+1):-1][::-1]
        return subset.iloc[top_ids]

    def execute_pipeline(self, account):
        action_recs = self.recall_by_action(account, 5)
        trending_recs = self.recall_trending(3)
        return pd.concat([action_recs, trending_recs])

# データの準備
news_df = pd.read_csv("AG_News.csv").sample(100)
news_df["embedding"] = news_df.apply(lambda r: get_embedding(r.Title + " " + r.Description, engine="text-embedding-ada-002"), axis=1)

engine = Recommender(news_df)
recommendations = engine.execute_pipeline(act1)

このように、リコールモジュールを組み合わせることで、多様な推薦結果を生成できます。実際のプロダクション環境では、さらに多様性や鮮度を考慮した再ランキング（ソート）処理や、ユーザーのリアルタイムフィードバックの反映、A/Bテストによる評価などが加わります。