BM25+MeCabで介護専門用語を確実にヒットさせる — RAGハイブリッド検索の実装

やりたいこと

介護ドキュメントの RAG システムで「褥瘡」「誤嚥」などの専門用語を正確に検索したい。
ベクトル検索（Dense）単体では表記揺れに強い反面、完全一致が必要な専門用語を見落とす。
BM25（Sparse）と ruri-v3（Dense）を RRF で統合するハイブリッド検索に切り替えたところ、専門用語の再現率が大幅に改善した。

---

環境

ライブラリ	バージョン
Python	3.11+
qdrant-client	1.12+
sentence-transformers	3.x
rank-bm25	0.2.2
MeCab + ipadic	最新
fugashi	1.3+

モデル: cl-nagoya/ruri-v3-30m（FP16、無料利用可）

---

問題: ベクトル検索だけでは専門用語の完全一致に弱い

何が起きていたか

# Dense 検索だけのとき — "褥瘡" で検索すると意味的に近い文書は返るが
# 「褥瘡」という単語を含む文書が上位に来ない場合があった
query = "褥瘡のステージ分類を教えて"
results = await dense_search(query)
# → "皮膚トラブルの予防" "体位変換の方法" は返るが
#    "褥瘡 ステージI〜IV の判断基準" が4位以下に埋もれることがあった

原因: Dense 検索の「表記揺れ耐性」が裏目に出る

Dense 検索は意味の近さで検索するため、「褥瘡」を含まなくても「皮膚トラブル」「床ずれ」を含む文書を上位に持ってくる。これは一般的なクエリでは利点だが、医療・介護用語の公式名称で完全一致を求めるときには精度が落ちる。

BM25（Sparse）は単語の出現頻度と文書頻度で順位を決めるため、「褥瘡」という単語が含まれる文書を確実に拾う。

---

解決: BM25 × ruri-v3 × RRF のハイブリッド検索

Step 1: ruri-v3 のプレフィックスルール（必須）

ruri-v3 は クエリと文書で異なるプレフィックスを付けることが前提のモデル。これを省略すると精度が劇的に低下する。

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("cl-nagoya/ruri-v3-30m")

def encode_query(query_text: str) -> list[float]:
    """クエリテキストをベクトル化する（クエリ用プレフィックス必須）。"""
    # "クエリ: " を付けないと検索精度が大幅に低下する
    prefixed = "クエリ: " + query_text
    return model.encode(prefixed).tolist()

def encode_document(chunk_text: str) -> list[float]:
    """チャンクテキストをベクトル化する（文書用プレフィックス必須）。"""
    # インデックス構築時に全チャンクへ "文章: " を付けておく
    prefixed = "文章: " + chunk_text
    return model.encode(prefixed).tolist()

プレフィックスなしだと cosine similarity が 0.1〜0.2 程度低下する（FP16 二宮氏のベンチマーク参照）。

Step 2: BM25 インデックスの構築（MeCab で形態素解析）

from rank_bm25 import BM25Okapi
import MeCab

_tagger = MeCab.Tagger("-Owakati")

def tokenize_ja(text: str) -> list[str]:
    """日本語テキストを形態素解析してトークンリストを返す。"""
    # MeCab の Owakati オプションで分かち書きして空白分割
    return _tagger.parse(text).strip().split()

def build_bm25_index(chunks: list[str]) -> BM25Okapi:
    """BM25 インデックスを構築する。"""
    tokenized_corpus = [tokenize_ja(c) for c in chunks]
    return BM25Okapi(tokenized_corpus)

Step 3: RRF（Reciprocal Rank Fusion）でスコアを統合

from dataclasses import dataclass

@dataclass
class ScoredChunk:
    chunk_id: str
    text: str
    score: float

def rrf_fusion(
    dense_results: list[ScoredChunk],
    sparse_results: list[ScoredChunk],
    k: int = 60,
    dense_weight: float = 0.6,
    sparse_weight: float = 0.4,
) -> list[ScoredChunk]:
    """RRF でスコアを統合する。

    k=60 は RRF の標準的なハイパーパラメータ（論文推奨値）。
    dense_weight > sparse_weight にするのは、意味検索を主軸にしつつ
    BM25 で専門用語の完全一致を補完する設計方針による。
    """
    rrf_scores: dict[str, float] = {}

    # Dense 側の寄与: 順位が高いほど 1/(k + rank + 1) が大きくなる
    for rank, chunk in enumerate(dense_results):
        cid = chunk.chunk_id
        rrf_scores[cid] = rrf_scores.get(cid, 0.0) + (
            dense_weight * (1.0 / (k + rank + 1))
        )

    # Sparse (BM25) 側の寄与: 同じ式で加算
    for rank, chunk in enumerate(sparse_results):
        cid = chunk.chunk_id
        rrf_scores[cid] = rrf_scores.get(cid, 0.0) + (
            sparse_weight * (1.0 / (k + rank + 1))
        )

    # スコア降順でソートして ScoredChunk を再構成
    chunk_map = {c.chunk_id: c for c in dense_results + sparse_results}
    sorted_ids = sorted(rrf_scores, key=lambda x: rrf_scores[x], reverse=True)
    return [
        ScoredChunk(
            chunk_id=cid,
            text=chunk_map[cid].text,
            score=rrf_scores[cid],
        )
        for cid in sorted_ids
        if cid in chunk_map
    ]

Step 4: asyncio.gather() で Dense と BM25 を並列実行

import asyncio

# チャンキング設定（.env から読み込む）
CHUNK_SIZE = 1024
CHUNK_OVERLAP = 128
TOP_K = 5
BM25_WEIGHT = 0.4
SCORE_THRESHOLD = 0.3  # RRF スコアがこれ未満の結果は除外

async def hybrid_search(
    query: str,
    bm25_index: BM25Okapi,
    chunks: list[str],
    chunk_ids: list[str],
) -> list[ScoredChunk]:
    """BM25 と Dense を並列実行して RRF で統合する。"""
    # クエリをベクトル化（プレフィックス付き）
    query_embedding = encode_query(query)

    # Dense 検索と BM25 検索を非同期で並列実行
    dense_task = asyncio.create_task(_dense_search(query_embedding, TOP_K))
    sparse_task = asyncio.create_task(_sparse_search(query, bm25_index, chunks, chunk_ids, TOP_K))

    dense_results, sparse_results = await asyncio.gather(dense_task, sparse_task)

    # RRF で統合してスコア閾値でフィルタ
    final_results = rrf_fusion(dense_results, sparse_results)
    return [r for r in final_results if r.score >= SCORE_THRESHOLD]

async def _dense_search(
    query_embedding: list[float],
    top_k: int,
) -> list[ScoredChunk]:
    """Qdrant に対してベクトル検索を実行する。"""
    # Qdrant クライアントの search() を呼ぶ（省略）
    ...

async def _sparse_search(
    query: str,
    bm25_index: BM25Okapi,
    chunks: list[str],
    chunk_ids: list[str],
    top_k: int,
) -> list[ScoredChunk]:
    """BM25 で全チャンクをスコアリングして上位 top_k を返す。"""
    tokens = tokenize_ja(query)
    scores = bm25_index.get_scores(tokens)  # numpy 配列でスコアが返る

    # 上位 top_k のインデックスを取得
    top_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:top_k]
    return [
        ScoredChunk(
            chunk_id=chunk_ids[i],
            text=chunks[i],
            score=float(scores[i]),
        )
        for i in top_indices
        if scores[i] > 0.0  # スコア 0 以下は除外
    ]

---

ハマりポイント: プレフィックス忘れ

ruri-v3 で最も多い落とし穴はインデックス構築時と検索時でプレフィックスの有無が一致しないこと。

# よくある失敗: インデックス構築時にプレフィックスを付けたのに
doc_embedding = model.encode("文章: " + chunk_text)  # インデックス時: プレフィックスあり

# クエリ時に忘れてしまう
query_embedding = model.encode(query_text)  # 検索時: プレフィックスなし ← NG

こうなるとベクトル空間の「対応が取れない」状態になり、cosine similarity が全体的に低くなる。症状は「なぜかスコアが低い」「上位結果が明らかに的外れ」。

対策: エンコード関数を encode_query() / encode_document() として分離し、プレフィックス付与を関数内部に閉じ込める（Step 1 のコードを参照）。

---

効果の確認

検索方式	"褥瘡 ステージ分類" の正解文書順位	備考
Dense のみ	4〜8 位	意味は近いが完全一致に弱い
BM25 のみ	1〜2 位	専門用語に強いが文脈を無視
BM25 × Dense × RRF	1〜2 位	専門用語 + 文脈の両立

RRF は Dense と BM25 どちらかが見落としても、もう一方が補完する構造になっている。

---

まとめ

• ruri-v3 は "クエリ: " / "文章: " プレフィックスが必須。忘れると精度が大幅に落ちる
• BM25 × Dense × RRF のハイブリッド検索で専門用語の完全一致と文脈検索を両立できる
• asyncio.gather() で Dense と BM25 を並列実行することで検索レイテンシを抑えられる
• チューニングパラメータ: k=60（RRF 標準）、dense_weight=0.6、sparse_weight=0.4、SCORE_THRESHOLD=0.3

---

全体像はnoteに → https://note.com/yamashita_aidev/n/n0183117bcb4c