Hybrid Search — 密ベクトル検索とBM25スパース検索の融合

import numpy as np
import re
from collections import Counter
from typing import List, Dict, Tuple


class BM25:
    """BM25スコアリング関数のスクラッチ実装"""

    def __init__(self, k1: float = 1.5, b: float = 0.75):
        self.k1 = k1
        self.b = b
        self.corpus_size = 0
        self.avgdl = 0.0
        self.doc_freqs: Dict[str, int] = {}  # 単語 -> 出現文書数
        self.doc_lens: List[int] = []
        self.term_freqs: List[Dict[str, int]] = []  # 各文書の単語出現回数

    def _tokenize(self, text: str) -> List[str]:
        """簡易トークナイザ（英語・日本語混在対応）"""
        # 英単語は小文字化してスペース分割、日本語は1文字ずつ分割
        text = text.lower()
        # 英単語を抽出
        tokens = re.findall(r'[a-z0-9]+|[^\s\x00-\x7F]', text)
        return tokens

    def fit(self, corpus: List[str]):
        """コーパスからIDF等の統計量を計算"""
        self.corpus_size = len(corpus)
        self.doc_lens = []
        self.term_freqs = []

        for doc in corpus:
            tokens = self._tokenize(doc)
            self.doc_lens.append(len(tokens))
            tf = Counter(tokens)
            self.term_freqs.append(dict(tf))
            for token in set(tokens):
                self.doc_freqs[token] = self.doc_freqs.get(token, 0) + 1

        self.avgdl = sum(self.doc_lens) / self.corpus_size

    def _idf(self, term: str) -> float:
        """BM25のIDF計算"""
        n = self.doc_freqs.get(term, 0)
        return np.log((self.corpus_size - n + 0.5) / (n + 0.5) + 1.0)

    def get_scores(self, query: str) -> np.ndarray:
        """クエリに対する全文書のBM25スコアを計算"""
        query_tokens = self._tokenize(query)
        scores = np.zeros(self.corpus_size)

        for q in query_tokens:
            idf = self._idf(q)
            for i in range(self.corpus_size):
                tf = self.term_freqs[i].get(q, 0)
                dl = self.doc_lens[i]
                numerator = tf * (self.k1 + 1)
                denominator = tf + self.k1 * (1 - self.b + self.b * dl / self.avgdl)
                scores[i] += idf * numerator / denominator

        return scores

    def get_top_k(self, query: str, k: int = 10) -> List[Tuple[int, float]]:
        """上位k件の(文書インデックス, スコア)を返す"""
        scores = self.get_scores(query)
        top_indices = np.argsort(scores)[::-1][:k]
        return [(int(idx), float(scores[idx])) for idx in top_indices]

class DenseRetriever:
    """密ベクトル検索（コサイン類似度ベース）"""

    def __init__(self, use_random: bool = False, dim: int = 384):
        """
        use_random=True: ランダム埋め込み（依存ライブラリ不要）
        use_random=False: sentence-transformersを使用
        """
        self.use_random = use_random
        self.dim = dim
        self.doc_embeddings = None

        if not use_random:
            from sentence_transformers import SentenceTransformer
            self.model = SentenceTransformer('all-MiniLM-L6-v2')

    def _embed(self, texts: List[str]) -> np.ndarray:
        """テキストをベクトルに変換"""
        if self.use_random:
            np.random.seed(42)
            embeddings = np.random.randn(len(texts), self.dim)
        else:
            embeddings = self.model.encode(texts, convert_to_numpy=True)
        # L2正規化
        norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
        return embeddings / norms

    def fit(self, corpus: List[str]):
        """コーパスの埋め込みベクトルを事前計算"""
        self.doc_embeddings = self._embed(corpus)

    def get_scores(self, query: str) -> np.ndarray:
        """クエリに対する全文書のコサイン類似度を計算"""
        query_emb = self._embed([query])  # (1, dim)
        # 正規化済みなので内積 = コサイン類似度
        scores = (self.doc_embeddings @ query_emb.T).flatten()
        return scores

    def get_top_k(self, query: str, k: int = 10) -> List[Tuple[int, float]]:
        """上位k件の(文書インデックス, スコア)を返す"""
        scores = self.get_scores(query)
        top_indices = np.argsort(scores)[::-1][:k]
        return [(int(idx), float(scores[idx])) for idx in top_indices]

def reciprocal_rank_fusion(
    rankings: List[List[Tuple[int, float]]],
    k: int = 60
) -> List[Tuple[int, float]]:
    """
    Reciprocal Rank Fusion

    Parameters
    ----------
    rankings : list of list of (doc_id, score)
        各検索システムのランキング結果
    k : int
        RRFの定数パラメータ（デフォルト: 60）

    Returns
    -------
    list of (doc_id, rrf_score)
        RRFスコアの降順でソートされた結果
    """
    rrf_scores: Dict[int, float] = {}

    for ranking in rankings:
        for rank, (doc_id, _score) in enumerate(ranking, start=1):
            if doc_id not in rrf_scores:
                rrf_scores[doc_id] = 0.0
            rrf_scores[doc_id] += 1.0 / (k + rank)

    # スコア降順でソート
    sorted_results = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
    return sorted_results


def convex_combination(
    results_list: List[List[Tuple[int, float]]],
    alpha: float = 0.5
) -> List[Tuple[int, float]]:
    """
    Convex Combination（Min-Max正規化 + 重み付き合計）

    Parameters
    ----------
    results_list : list of list of (doc_id, score)
        [密ベクトル検索結果, BM25結果] の順
    alpha : float
        密ベクトル検索の重み（0〜1）

    Returns
    -------
    list of (doc_id, combined_score)
    """
    weights = [alpha, 1 - alpha]
    normalized_scores: Dict[int, float] = {}

    for w, results in zip(weights, results_list):
        if not results:
            continue
        scores = [s for _, s in results]
        s_min, s_max = min(scores), max(scores)
        denom = s_max - s_min if s_max != s_min else 1.0

        for doc_id, score in results:
            norm_score = (score - s_min) / denom
            if doc_id not in normalized_scores:
                normalized_scores[doc_id] = 0.0
            normalized_scores[doc_id] += w * norm_score

    sorted_results = sorted(normalized_scores.items(), key=lambda x: x[1], reverse=True)
    return sorted_results

class HybridSearcher:
    """BM25 + 密ベクトル検索のハイブリッド検索"""

    def __init__(self, use_random_embeddings: bool = False):
        self.bm25 = BM25(k1=1.5, b=0.75)
        self.dense = DenseRetriever(use_random=use_random_embeddings)
        self.corpus: List[str] = []

    def fit(self, corpus: List[str]):
        """コーパスのインデックスを構築"""
        self.corpus = corpus
        self.bm25.fit(corpus)
        self.dense.fit(corpus)

    def search(
        self,
        query: str,
        top_k: int = 10,
        method: str = "rrf",
        alpha: float = 0.5,
        rrf_k: int = 60
    ) -> List[Tuple[int, float, str]]:
        """
        ハイブリッド検索を実行

        Returns
        -------
        list of (doc_id, score, document_text)
        """
        bm25_results = self.bm25.get_top_k(query, k=top_k)
        dense_results = self.dense.get_top_k(query, k=top_k)

        if method == "rrf":
            fused = reciprocal_rank_fusion(
                [bm25_results, dense_results], k=rrf_k
            )
        elif method == "convex":
            fused = convex_combination(
                [dense_results, bm25_results], alpha=alpha
            )
        else:
            raise ValueError(f"Unknown method: {method}")

        results = [
            (doc_id, score, self.corpus[doc_id])
            for doc_id, score in fused[:top_k]
        ]
        return results

import numpy as np
from typing import List, Tuple, Dict
import re
from collections import Counter


# ---- BM25 クラス（上と同一） ----
class BM25:
    def __init__(self, k1=1.5, b=0.75):
        self.k1, self.b = k1, b
        self.corpus_size = 0
        self.avgdl = 0.0
        self.doc_freqs, self.doc_lens, self.term_freqs = {}, [], []

    def _tokenize(self, text):
        return re.findall(r'[a-z0-9]+|[^\s\x00-\x7F]', text.lower())

    def fit(self, corpus):
        self.corpus_size = len(corpus)
        self.doc_lens, self.term_freqs = [], []
        for doc in corpus:
            tokens = self._tokenize(doc)
            self.doc_lens.append(len(tokens))
            tf = Counter(tokens)
            self.term_freqs.append(dict(tf))
            for t in set(tokens):
                self.doc_freqs[t] = self.doc_freqs.get(t, 0) + 1
        self.avgdl = sum(self.doc_lens) / self.corpus_size

    def _idf(self, term):
        n = self.doc_freqs.get(term, 0)
        return np.log((self.corpus_size - n + 0.5) / (n + 0.5) + 1.0)

    def get_scores(self, query):
        scores = np.zeros(self.corpus_size)
        for q in self._tokenize(query):
            idf = self._idf(q)
            for i in range(self.corpus_size):
                tf = self.term_freqs[i].get(q, 0)
                dl = self.doc_lens[i]
                num = tf * (self.k1 + 1)
                den = tf + self.k1 * (1 - self.b + self.b * dl / self.avgdl)
                scores[i] += idf * num / den
        return scores

    def get_top_k(self, query, k=10):
        scores = self.get_scores(query)
        idx = np.argsort(scores)[::-1][:k]
        return [(int(i), float(scores[i])) for i in idx]


# ---- 簡易密ベクトル検索（TF-IDFベースの疑似意味検索） ----
class SimpleDenseRetriever:
    """
    sentence-transformers不要の疑似密ベクトル検索
    TF-IDFベクトルにランダム射影を加えて「意味検索的な」振る舞いを模擬
    """
    def __init__(self, dim=64, seed=42):
        self.dim = dim
        self.seed = seed
        self.doc_embeddings = None

    def _build_vocab(self, corpus):
        vocab = set()
        for doc in corpus:
            vocab.update(re.findall(r'[a-z0-9]+|[^\s\x00-\x7F]', doc.lower()))
        self.vocab = sorted(vocab)
        self.word2idx = {w: i for i, w in enumerate(self.vocab)}

    def _tfidf_vector(self, text, doc_freqs, n_docs):
        tokens = re.findall(r'[a-z0-9]+|[^\s\x00-\x7F]', text.lower())
        tf = Counter(tokens)
        vec = np.zeros(len(self.vocab))
        for t, c in tf.items():
            if t in self.word2idx:
                idf = np.log((n_docs + 1) / (doc_freqs.get(t, 0) + 1))
                vec[self.word2idx[t]] = c * idf
        return vec

    def fit(self, corpus):
        self._build_vocab(corpus)
        doc_freqs = {}
        for doc in corpus:
            for t in set(re.findall(r'[a-z0-9]+|[^\s\x00-\x7F]', doc.lower())):
                doc_freqs[t] = doc_freqs.get(t, 0) + 1
        self.doc_freqs = doc_freqs
        self.n_docs = len(corpus)

        # TF-IDFベクトルを計算
        tfidf_matrix = np.array([
            self._tfidf_vector(doc, doc_freqs, len(corpus)) for doc in corpus
        ])
        # ランダム射影で次元削減（意味空間の模擬）
        np.random.seed(self.seed)
        self.proj = np.random.randn(len(self.vocab), self.dim) / np.sqrt(self.dim)
        self.doc_embeddings = tfidf_matrix @ self.proj
        # L2正規化
        norms = np.linalg.norm(self.doc_embeddings, axis=1, keepdims=True)
        norms[norms == 0] = 1.0
        self.doc_embeddings /= norms

    def get_scores(self, query):
        vec = self._tfidf_vector(query, self.doc_freqs, self.n_docs)
        q_emb = vec @ self.proj
        q_norm = np.linalg.norm(q_emb)
        if q_norm > 0:
            q_emb /= q_norm
        return (self.doc_embeddings @ q_emb).flatten()

    def get_top_k(self, query, k=10):
        scores = self.get_scores(query)
        idx = np.argsort(scores)[::-1][:k]
        return [(int(i), float(scores[i])) for i in idx]


# ---- 融合関数 ----
def reciprocal_rank_fusion(rankings, k=60):
    rrf_scores = {}
    for ranking in rankings:
        for rank, (doc_id, _) in enumerate(ranking, start=1):
            rrf_scores[doc_id] = rrf_scores.get(doc_id, 0.0) + 1.0 / (k + rank)
    return sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)


def convex_combination(results_list, alpha=0.5):
    weights = [alpha, 1 - alpha]
    combined = {}
    for w, results in zip(weights, results_list):
        if not results:
            continue
        scores = [s for _, s in results]
        s_min, s_max = min(scores), max(scores)
        denom = s_max - s_min if s_max != s_min else 1.0
        for doc_id, score in results:
            norm = (score - s_min) / denom
            combined[doc_id] = combined.get(doc_id, 0.0) + w * norm
    return sorted(combined.items(), key=lambda x: x[1], reverse=True)


# ---- 合成データセット ----
corpus = [
    # 0-4: XGBoost関連
    "XGBoost v2.1 hyperparameter tuning guide for gradient boosting",
    "Gradient boosting theory and optimization techniques overview",
    "Random forest vs XGBoost performance comparison benchmark",
    "XGBoost feature importance and SHAP value interpretation",
    "Ensemble learning methods including bagging and boosting",
    # 5-9: 深層学習関連
    "PyTorch deep learning framework tutorial for beginners",
    "Convolutional neural network CNN image classification",
    "Transfer learning with pretrained models in computer vision",
    "Recurrent neural network RNN for sequence modeling",
    "Transformer architecture and self-attention mechanism",
    # 10-14: NLP関連
    "BERT fine-tuning for text classification NLP task",
    "Word embedding Word2Vec and GloVe comparison",
    "Sentiment analysis using natural language processing",
    "Named entity recognition NER with spaCy tutorial",
    "Text preprocessing tokenization and lemmatization guide",
    # 15-19: データ分析関連
    "Pandas DataFrame operations for data analysis",
    "Exploratory data analysis EDA visualization techniques",
    "Feature engineering for machine learning pipelines",
    "Data cleaning and handling missing values in datasets",
    "Statistical hypothesis testing t-test and chi-square",
]

# クエリと正解文書のペア
# (query, relevant_doc_ids)
queries_and_relevance = [
    # BM25が得意: キーワード完全一致
    ("XGBoost hyperparameter", [0, 3]),
    ("CNN image classification", [6, 7]),
    ("Pandas DataFrame", [15, 16]),
    ("spaCy NER tutorial", [13, 14]),
    # 密ベクトル検索が得意: 意味的一致
    ("gradient tree model tuning", [0, 1]),
    ("deep neural network for pictures", [6, 7]),
    ("analyzing text sentiment", [12, 10]),
    ("preparing data for ML", [17, 18]),
    # 両方が必要: キーワード + 意味
    ("XGBoost optimization best practices", [0, 1, 3]),
    ("NLP text feature extraction", [10, 14, 11]),
    ("ensemble model comparison benchmark", [4, 2, 1]),
    ("missing data preprocessing pipeline", [18, 17, 14]),
]


def compute_mrr(rankings, relevant_ids, top_k=10):
    """MRR（Mean Reciprocal Rank）を計算"""
    rrs = []
    for ranking, rel_ids in zip(rankings, relevant_ids):
        ranked_ids = [doc_id for doc_id, _ in ranking[:top_k]]
        rr = 0.0
        for rank, doc_id in enumerate(ranked_ids, start=1):
            if doc_id in rel_ids:
                rr = 1.0 / rank
                break
        rrs.append(rr)
    return np.mean(rrs)


# ---- 実験実行 ----
bm25 = BM25(k1=1.5, b=0.75)
bm25.fit(corpus)

dense = SimpleDenseRetriever(dim=64, seed=42)
dense.fit(corpus)

queries = [q for q, _ in queries_and_relevance]
relevance = [r for _, r in queries_and_relevance]
top_k = 10

# 各手法のランキングを取得
bm25_rankings = [bm25.get_top_k(q, k=top_k) for q in queries]
dense_rankings = [dense.get_top_k(q, k=top_k) for q in queries]
rrf_rankings = [
    reciprocal_rank_fusion([b, d], k=60)
    for b, d in zip(bm25_rankings, dense_rankings)
]
convex_rankings = [
    convex_combination([d, b], alpha=0.5)
    for b, d in zip(bm25_rankings, dense_rankings)
]

# MRR計算
mrr_bm25 = compute_mrr(bm25_rankings, relevance, top_k)
mrr_dense = compute_mrr(dense_rankings, relevance, top_k)
mrr_rrf = compute_mrr(rrf_rankings, relevance, top_k)
mrr_convex = compute_mrr(convex_rankings, relevance, top_k)

print("=== MRR（Mean Reciprocal Rank） ===")
print(f"BM25 単独:           {mrr_bm25:.4f}")
print(f"Dense 単独:          {mrr_dense:.4f}")
print(f"Hybrid (RRF):        {mrr_rrf:.4f}")
print(f"Hybrid (Convex 0.5): {mrr_convex:.4f}")

=== MRR（Mean Reciprocal Rank） ===
BM25 単独:           0.8750
Dense 単独:          0.5417
Hybrid (RRF):        0.8750
Hybrid (Convex 0.5): 0.7917

import matplotlib.pyplot as plt
import matplotlib
matplotlib.rcParams['font.size'] = 10

fig, axes = plt.subplots(3, 4, figsize=(18, 12))
fig.suptitle('Search Rankings by Query (Top-5)', fontsize=14, fontweight='bold')

query_labels = [q for q, _ in queries_and_relevance]
methods = {
    'BM25': bm25_rankings,
    'Dense': dense_rankings,
    'RRF': rrf_rankings,
}

for idx, (query, rel_ids) in enumerate(queries_and_relevance):
    ax = axes[idx // 4][idx % 4]
    method_names = list(methods.keys())
    top5_per_method = []

    for method_name in method_names:
        ranking = methods[method_name][idx]
        top5_ids = [doc_id for doc_id, _ in ranking[:5]]
        top5_per_method.append(top5_ids)

    # 各手法のTop-5をテーブル形式で表示
    cell_colors = []
    cell_text = []
    for m_idx, (m_name, top5) in enumerate(zip(method_names, top5_per_method)):
        row = []
        colors_row = []
        for rank_pos, doc_id in enumerate(top5):
            row.append(f"D{doc_id}")
            if doc_id in rel_ids:
                colors_row.append('#2ecc71')  # 正解: 緑
            else:
                colors_row.append('#ecf0f1')  # 不正解: 灰色
        cell_text.append(row)
        cell_colors.append(colors_row)

    ax.axis('off')
    table = ax.table(
        cellText=cell_text,
        rowLabels=method_names,
        colLabels=['1st', '2nd', '3rd', '4th', '5th'],
        cellColours=cell_colors,
        loc='center',
        cellLoc='center'
    )
    table.auto_set_font_size(False)
    table.set_fontsize(8)
    table.scale(1.0, 1.5)

    # クエリ名を短縮して表示
    short_query = query[:30] + "..." if len(query) > 30 else query
    ax.set_title(short_query, fontsize=9, pad=20)

plt.tight_layout(rect=[0, 0, 1, 0.95])
plt.savefig('hybrid_search_comparison.png', dpi=150, bbox_inches='tight')
plt.show()

import matplotlib.pyplot as plt
import numpy as np

# RRFのkパラメータの感度分析
k_values = [1, 5, 10, 20, 40, 60, 80, 100, 200]
mrr_by_k = []

for k_val in k_values:
    rrf_k_rankings = [
        reciprocal_rank_fusion([b, d], k=k_val)
        for b, d in zip(bm25_rankings, dense_rankings)
    ]
    mrr_val = compute_mrr(rrf_k_rankings, relevance, top_k)
    mrr_by_k.append(mrr_val)

# Convex Combinationのalphaパラメータの感度分析
alpha_values = np.linspace(0, 1, 21)
mrr_by_alpha = []

for alpha_val in alpha_values:
    convex_a_rankings = [
        convex_combination([d, b], alpha=alpha_val)
        for b, d in zip(bm25_rankings, dense_rankings)
    ]
    mrr_val = compute_mrr(convex_a_rankings, relevance, top_k)
    mrr_by_alpha.append(mrr_val)

fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 5))

# RRF k パラメータ
ax1.plot(k_values, mrr_by_k, 'o-', color='#3498db', linewidth=2, markersize=8)
ax1.axhline(y=mrr_bm25, color='#e74c3c', linestyle='--', label=f'BM25 only ({mrr_bm25:.3f})')
ax1.axhline(y=mrr_dense, color='#2ecc71', linestyle='--', label=f'Dense only ({mrr_dense:.3f})')
ax1.set_xlabel('RRF parameter k')
ax1.set_ylabel('MRR')
ax1.set_title('RRF: Sensitivity to k')
ax1.set_xscale('log')
ax1.legend()
ax1.grid(True, alpha=0.3)

# Convex Combination alpha パラメータ
ax2.plot(alpha_values, mrr_by_alpha, 'o-', color='#9b59b6', linewidth=2, markersize=6)
ax2.axhline(y=mrr_bm25, color='#e74c3c', linestyle='--', label=f'BM25 only ({mrr_bm25:.3f})')
ax2.axhline(y=mrr_dense, color='#2ecc71', linestyle='--', label=f'Dense only ({mrr_dense:.3f})')
ax2.set_xlabel('alpha (weight for dense retriever)')
ax2.set_ylabel('MRR')
ax2.set_title('Convex Combination: Sensitivity to alpha')
ax2.legend()
ax2.grid(True, alpha=0.3)

plt.tight_layout()
plt.savefig('hybrid_search_sensitivity.png', dpi=150, bbox_inches='tight')
plt.show()

import numpy as np
import matplotlib.pyplot as plt

# TF飽和の可視化
tf_values = np.linspace(0, 20, 200)
k1_values = [0.5, 1.2, 1.5, 2.0, 5.0]

fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 5))

# 左: k1による飽和の違い
for k1 in k1_values:
    bm25_tf = tf_values * (k1 + 1) / (tf_values + k1)
    ax1.plot(tf_values, bm25_tf, label=f'k1={k1}', linewidth=2)

# TF-IDF（線形）を比較用にプロット
ax1.plot(tf_values, tf_values, '--', color='gray', alpha=0.5, label='TF-IDF (linear)')
ax1.set_xlabel('Term Frequency (tf)')
ax1.set_ylabel('TF Component Score')
ax1.set_title('BM25 TF Saturation for Different k1')
ax1.legend()
ax1.set_xlim(0, 20)
ax1.set_ylim(0, 7)
ax1.grid(True, alpha=0.3)

# 右: 文書長正規化の効果
dl_ratios = np.linspace(0.2, 3.0, 200)  # |d| / avgdl
b_values = [0, 0.25, 0.5, 0.75, 1.0]
k1_fixed = 1.5
tf_fixed = 3

for b in b_values:
    denominator = tf_fixed + k1_fixed * (1 - b + b * dl_ratios)
    score = tf_fixed * (k1_fixed + 1) / denominator
    ax2.plot(dl_ratios, score, label=f'b={b}', linewidth=2)

ax2.axvline(x=1.0, color='gray', linestyle=':', alpha=0.5, label='avg doc length')
ax2.set_xlabel('Document Length / Average Document Length')
ax2.set_ylabel('TF Component Score (tf=3)')
ax2.set_title('BM25 Document Length Normalization')
ax2.legend()
ax2.grid(True, alpha=0.3)

plt.tight_layout()
plt.savefig('bm25_parameters.png', dpi=150, bbox_inches='tight')
plt.show()