ハイブリッド検索完全ガイド：BM25とベクター検索を組み合わせてRAGを改善する

score(D, Q) = Σ IDF(qi) × [f(qi, D) × (k1 + 1)] / [f(qi, D) + k1 × (1 - b + b × |D| / avgdl)]

ここで：
  qi        = クエリの各単語
  f(qi, D)  = ドキュメントDでの単語qiの出現頻度（Term Frequency）
  |D|       = ドキュメントDの長さ
  avgdl     = コーパス全体の平均ドキュメント長
  k1, b     = チューニングパラメータ（通常k1=1.5, b=0.75）
  IDF(qi)   = log((N - df + 0.5) / (df + 0.5))

ベクター検索の弱点：
- 正確なキーワードマッチングに弱い（モデル番号、製品コード）
- 「意味的ドリフト」：似ているが異なる概念を取得してしまう
- 希少語、新造語、略語に脆弱

BM25の弱点：
- 同義語を理解できない（"automobile"と"car"）
- 文脈を理解できない（語順を無視）
- タイポに脆弱
- 多言語処理に弱い

def reciprocal_rank_fusion(results_list: list, k: int = 60) -> list:
    """
    複数の順位付き結果リストをRRFで結合。

    RRFスコア = Σ 1/(k + rank_i)
    k=60は低順位アイテムの影響を和らげる定数

    Args:
        results_list: 各検索結果リストのリスト（doc_idの順序リスト）
        k: 順位安定化定数（デフォルト60）

    Returns:
        RRFスコアで降順ソートされたdoc_idのリスト
    """
    scores: dict = {}

    for results in results_list:
        for rank, doc_id in enumerate(results, start=1):
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank)

    return sorted(scores.keys(), key=lambda x: scores[x], reverse=True)


# 具体的な例
vector_results = ["doc_A", "doc_C", "doc_B"]  # 埋め込み検索結果
bm25_results   = ["doc_B", "doc_A", "doc_D"]  # BM25キーワード検索結果

# スコア計算：
# doc_A: 1/(60+1) + 1/(60+2) = 0.01639 + 0.01613 = 0.03252
# doc_B: 1/(60+3) + 1/(60+1) = 0.01587 + 0.01639 = 0.03226
# doc_C: 1/(60+2) = 0.01613
# doc_D: 1/(60+2) = 0.01613

fused = reciprocal_rank_fusion([vector_results, bm25_results])
print(fused)
# ['doc_A', 'doc_B', 'doc_C', 'doc_D']
# doc_Aは両方の検索で上位なので1位
# doc_BはBM25で1位だがベクターで3位 → 2位

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# 前提: docsはLangChain Documentオブジェクトのリスト
embedding_model = OpenAIEmbeddings()

# BM25リトリーバー作成
bm25_retriever = BM25Retriever.from_documents(
    docs,
    k=5  # 上位5件を返す
)

# ベクターストアとリトリーバー設定
vectorstore = FAISS.from_documents(docs, embedding_model)
vector_retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5}
)

# EnsembleRetriever：重み付きで両検索を組み合わせ
hybrid_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # BM25 40%, ベクター 60%
)

# 使用例
query = "iPhone 15 Pro Max 256GB 在庫"
results = hybrid_retriever.get_relevant_documents(query)

from elasticsearch import Elasticsearch

es_client = Elasticsearch(["http://localhost:9200"])

def hybrid_search_es(query: str, query_embedding: list, k: int = 5):
    response = es_client.search(
        index="documents",
        body={
            "query": {
                "bool": {
                    "should": [
                        # BM25キーワード検索
                        {
                            "match": {
                                "content": {
                                    "query": query,
                                    "boost": 0.4
                                }
                            }
                        }
                    ]
                }
            },
            # KNNベクター検索（ES 8.x以降）
            "knn": {
                "field": "content_vector",
                "query_vector": query_embedding,
                "k": k,
                "num_candidates": 100,
                "boost": 0.6
            },
            "size": k
        }
    )
    return response["hits"]["hits"]

import nltk
from nltk.tokenize import word_tokenize
from nltk.stem import PorterStemmer
from nltk.corpus import stopwords

nltk.download('punkt')
nltk.download('stopwords')

stemmer = PorterStemmer()
stop_words = set(stopwords.words('english'))

def technical_tokenizer(text: str) -> list:
    tokens = word_tokenize(text.lower())
    tokens = [
        stemmer.stem(token)
        for token in tokens
        if token.isalnum() and token not in stop_words
    ]
    return tokens

# カスタムトークナイザーを適用
bm25_retriever = BM25Retriever.from_documents(
    docs,
    preprocess_func=technical_tokenizer,
    k=5
)