RAGの検索精度を改善する：ハイブリッド検索とリランキングの導入方法

RAGを実装していると、「キーワードは一致しているのに意味の違う文書が返る」「言い換えた質問では検索できない」という問題に遭遇します。

私が小さなFAQデータで検証したところ、ベクトル検索だけに頼るよりも、キーワード検索とベクトル検索を組み合わせ、最後にリランキングする構成のほうが検索結果を確認しやすくなりました。

本記事では、次の3段階で検索精度を改善する方法を紹介します。

1. BM25によるキーワード検索
2. embeddingによるベクトル検索
3. 上位候補に対するリランキング

ベクトル検索だけでは不足するケース

ベクトル検索は、文章の意味が近い文書を探すのが得意です。一方で、次のような情報は苦手になることがあります。

• エラーコードや製品名など、完全一致が重要な語句
• 数値やバージョン番号
• 否定表現
• 同じ単語を使っていても意味が異なる文書

例えば、ユーザーが「API-401の原因」と検索した場合、意味的には近いものの「API-403」や「API-500」を説明する文書が混ざる可能性があります。

逆にBM25のようなキーワード検索だけでは、「ログインできない」と「認証に失敗する」のような言い換えに弱くなります。

そこで、両方の検索結果を候補として集めます。

質問
  ├─ BM25検索       ─┐
  └─ ベクトル検索    ─┼─ 候補文書を統合
                       └─ リランキング ─ 最終結果

環境を準備する

今回は、検索エンジンを別途構築せず、Python上で動作を確認します。実運用ではElasticsearchやOpenSearchなどに置き換えられます。

python -m venv .venv
.venv\Scripts\activate

pip install rank-bm25 sentence-transformers fugashi unidic-lite numpy

日本語のBM25検索では、単純な空白区切りでは十分に分割できません。ここではfugashiを使って形態素単位に分割します。

扱う文書は、次のようなFAQを想定します。

documents = [
    {
        "id": "faq-001",
        "text": "API-401は認証情報が無効な場合に発生します。APIキーの有効期限と設定値を確認してください。"
    },
    {
        "id": "faq-002",
        "text": "API-403は認証後の権限が不足している場合に発生します。ロール設定を確認してください。"
    },
    {
        "id": "faq-003",
        "text": "API-500はサーバー内部エラーです。アプリケーションログとデータベース接続を確認してください。"
    },
    {
        "id": "faq-004",
        "text": "ログインできない場合は、ユーザー名、パスワード、二要素認証の設定を確認してください。"
    },
]

BM25とベクトル検索を組み合わせる

検索からリランキングまでの処理全体を図にすると次の通りです。

次のコードでは、BM25のスコアとembeddingの類似度を正規化し、重み付きで統合します。

import numpy as np
from fugashi import Tagger
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer

tagger = Tagger()

def tokenize(text: str) -> list[str]:
    return [
        word.surface
        for word in tagger(text)
        if word.surface.strip()
    ]

texts = [doc["text"] for doc in documents]
tokenized_docs = [tokenize(text) for text in texts]

bm25 = BM25Okapi(tokenized_docs)

embedding_model = SentenceTransformer(
    "intfloat/multilingual-e5-small"
)

doc_vectors = embedding_model.encode(
    [f"passage: {text}" for text in texts],
    normalize_embeddings=True
)

def min_max_normalize(values: np.ndarray) -> np.ndarray:
    minimum = values.min()
    maximum = values.max()

    if maximum == minimum:
        return np.ones_like(values)

    return (values - minimum) / (maximum - minimum)

def hybrid_search(
    query: str,
    top_k: int = 5,
    keyword_weight: float = 0.4
):
    query_tokens = tokenize(query)
    keyword_scores = np.array(
        bm25.get_scores(query_tokens)
    )

    query_vector = embedding_model.encode(
        [f"query: {query}"],
        normalize_embeddings=True
    )[0]

    semantic_scores = doc_vectors @ query_vector

    keyword_scores = min_max_normalize(keyword_scores)
    semantic_scores = min_max_normalize(semantic_scores)

    hybrid_scores = (
        keyword_weight * keyword_scores
        + (1 - keyword_weight) * semantic_scores
    )

    candidate_indices = np.argsort(
        hybrid_scores
    )[::-1][:top_k]

    return [
        {
            **documents[index],
            "keyword_score": float(keyword_scores[index]),
            "semantic_score": float(semantic_scores[index]),
            "hybrid_score": float(hybrid_scores[index]),
        }
        for index in candidate_indices
    ]

results = hybrid_search(
    "API-401が返る原因を知りたい",
    top_k=3,
    keyword_weight=0.4
)

for result in results:
    print(result["id"], result["hybrid_score"], result["text"])

intfloat/multilingual-e5-smallでは、文書側にpassage:、検索クエリ側にquery:を付けています。この形式を使うモデルでは、プレフィックスの有無が結果に影響するため注意が必要です。

keyword_weightを大きくすると、エラーコードや製品名の一致を重視できます。社内FAQでは、最初に0.3〜0.5程度から試すと調整しやすいと感じました。

リランキングで候補の順番を調整する

ハイブリッド検索は、広い文書集合から候補を集める処理です。しかし、BM25とembeddingのスコアを足し合わせただけでは、質問と回答の関係性を十分に判定できない場合があります。

そこで、上位候補だけをCross Encoder型のリランカーに渡します。

pip install torch

from sentence_transformers import CrossEncoder

reranker = CrossEncoder(
    "BAAI/bge-reranker-v2-m3"
)

def rerank(query: str, candidates: list[dict]) -> list[dict]:
    pairs = [
        [query, candidate["text"]]
        for candidate in candidates
    ]

    scores = reranker.predict(pairs)

    reranked = []
    for candidate, score in zip(candidates, scores):
        item = dict(candidate)
        item["rerank_score"] = float(score)
        reranked.append(item)

    return sorted(
        reranked,
        key=lambda item: item["rerank_score"],
        reverse=True
    )

candidates = hybrid_search(
    "API-401が返る原因を知りたい",
    top_k=5,
    keyword_weight=0.4
)

final_results = rerank(
    "API-401が返る原因を知りたい",
    candidates
)

for result in final_results[:3]:
    print(result["id"], result["rerank_score"])
    print(result["text"])

リランカーは、検索対象の全文書に適用すると処理時間が増えます。そのため、まずBM25とベクトル検索で10〜50件程度に絞り、その候補だけをリランキングする設計にします。

私の検証では、候補数を増やしすぎると、最終結果に関係の薄い文書まで入りやすくなりました。まずはtop_k=10程度で確認し、回答に必要な文書が候補から漏れる場合だけ増やすのが扱いやすいです。

精度を確認するときのポイント

検索精度を改善するときは、印象だけで判断せず、質問と正解文書の対応表を作ると調整しやすくなります。

evaluation_cases = [
    {
        "query": "API-401の原因",
        "expected_id": "faq-001",
    },
    {
        "query": "ログインできない",
        "expected_id": "faq-004",
    },
]

for case in evaluation_cases:
    candidates = hybrid_search(
        case["query"],
        top_k=5,
        keyword_weight=0.4
    )
    results = rerank(case["query"], candidates)

    returned_ids = [item["id"] for item in results[:3]]
    hit = case["expected_id"] in returned_ids

    print(case["query"], "hit@3 =", hit, returned_ids)

確認する指標は、まず次の2つで十分です。

• Hit@K: 正解文書が上位K件に含まれるか
• MRR: 正解文書が上位の何番目に出たか

調整する順番としては、以下がおすすめです。

1. 文書の分割単位を確認する
2. BM25とベクトル検索の重みを変える
3. 候補数を変える
4. リランカーを追加する
5. それでも改善しない場合はembeddingモデルを見直す

特に文書分割は重要です。1チャンクが長すぎると、質問に関係する情報が埋もれます。逆に短すぎると、前提条件や手順が分断されます。見出し、質問、回答をひとまとまりにするなど、元データの構造を活かして分割するのが効果的でした。

まとめ

RAGの検索精度を改善するには、ベクトル検索だけで解決しようとせず、検索処理を役割ごとに分けるのが有効です。

• BM25で固有名詞やエラーコードを拾う
• ベクトル検索で言い換えや意味の近さを拾う
• リランカーで質問と文書の関係性を再評価する
• 評価用の質問と正解文書を用意して重みを調整する

最初から複雑な検索基盤を作る必要はありません。小さなデータセットでkeyword_weight、候補数、チャンク分割を変えながら結果を比較すると、自分のデータに合う構成を見つけやすくなります。