RAG クエリルーティング実践：複数ベクトルデータベース連携とインテリジェント検索分散

深夜3時、本番環境のアラートがまた鳴り響いた。

モニタリングダッシュボードを眺めながら、ユーザーの「Q3の華東地区売上」というクエリのレスポンスタイムが12秒に急上昇していることに気づいた。デモ環境では問題なく動いていたのに、本番に上げたらなぜ失敗するのか？さらに深刻なのは、ログを見ると、単純な事実確認のクエリに答えるために、システムが完全なマルチホップ推論プロセスを呼び出していたことだ。

「これじゃ蚊を叩くのに核爆弾を使うようなものだ」。同僚が覗き込んで言った言葉は、実に的を得ていた。

その時、問題は検索精度ではなく、検索戦略にあることに気づいた。従来のRAGは「思考停止の検索器」のようなものだ。ユーザーが何を聞いても、同じベクトル検索＋大規模言語モデル生成のフローを通る。単純なクエリは過剰に処理され、複雑なクエリは十分なサポートを受けられない。

この記事で共有したいのは、RAGシステムに「交通整理役」——クエリルーター——を設ける方法だ。クエリの特徴に基づいて、リクエストを最適な検索パスに振り分ける。高速パスは単純な事実を処理し、ディープパスは複雑な推論を処理し、マルチソース検索は回答の包括性を保証する。

正直に言えば、このアプローチが私のプロジェクトを救ってくれた。

1. なぜクエリルーティングが必要なのか？ — 「思考停止検索」からインテリジェント分散へ

まず、私が経験した失敗から話そう。

昨年、あるEコマース企業のカスタマーサービスRAGシステムを構築した。ナレッジベースには、商品情報、アフター販売ポリシー、物流ルール、マーケティングキャンペーンの4つのカテゴリのデータを詰め込んだ。テスト時は問題なかったが、本番稼働後、苦情が倍増した。

ログを調査すると、典型的な問題が見つかった。ユーザーが「返金はどのくらいかかりますか」と聞いたのに、システムは配送期間とセールルールを返してきた。答えが間違っているわけではないが、答えが焦点を欠いている——無関係な情報が多すぎて、ユーザーの判断を妨げているのだ。

これが従来のRAGの最初の痛点だ：ナレッジ干渉。すべてのビジネスシナリオのデータを1つのベクトルデータベースに混ぜると、検索結果は当然のことながらごった煮状態になる。ユーザーがシナリオAについて質問しても、システムはシナリオBの「類似」コンテンツを返す可能性がある。

2つ目の痛点はさらに隠れている：レスポンス効率。

このクエリを見てみよう。「Q3の華東地区売上はいくら？」これは本質的に単純な事実確認クエリで、データベースを直接検索するか、キーワード検索で十分だ。しかし従来のRAGはどうするか？ベクトルエンコーディング、コサイン類似度計算、Top-K検索、大規模言語モデル生成……一連のプロセスで1〜2秒かかり、リソース消費も大きい。

3つ目の痛点：クエリ意図の誤判定。

ユーザーが「再生時間が最も短い動画を探して」と言ったとき、SelfQueryRetrieverは「再生時間」がメタデータフィルタ条件であることを理解できないかもしれない。ユーザーが「ストライキ事件は株価に影響しましたか」と聞いた場合、これはマルチホップ推論（まずストライキ事件を見つけ、関連企業を見つけ、最後に株価推移を調べる）が必要で、単一のベクトル検索では対応できない。

だから、「状況を見極める」ルーターが必要だ。クエリの特徴を識別し、最適な検索パスを選択する。

Alibaba Cloud - 開発者向けクラウドの第一候補、ライトサーバー15%OFF、プロジェクトの迅速な立ち上げを支援

15%割引を入手 →広告

これはレストランの注文システムに似ている。ファストフードカウンターは簡単な注文を処理し（高速）、通常のキッチンは複雑な料理を処理し（ディープ）、デリバリーウィンドウは配送ニーズを処理する（マルチチャンネル）。それぞれが役割を担い、効率が最大化される。

2. クエリルーティングの核心アーキテクチャ — 3層分散モデル

クエリルーティングの核心思想は実にシンプルだ：階層的に処理し、各層が役割を担う。

システム全体を3層アーキテクチャで設計した：

ユーザークエリ
    ↓
┌─────────────────────────────────┐
│  ルーティング層：シナリオ分類      │
│  (LLM / Semantic Router)         │
│  → python_docs / js_docs / go_docs│
└─────────────────────────────────┘
    ↓
┌─────────────────────────────────┐
│  検索層：各シナリオ専用ベクトルDB   │
│  Chroma(python) / Chroma(js)...  │
│  → top-k chunks                  │
└─────────────────────────────────┘
    ↓
┌─────────────────────────────────┐
│  マージ層：RRF アルゴリズム統合   │
│  RRF(d) = Σ 1/(k + rank(d))      │
│  → 最終回答                      │
└─────────────────────────────────┘
    ↓
回答生成

ルーティング層はアーキテクチャ全体の「脳」だ。その役割は明確だ：ユーザークエリを分析し、どの検索パスを通るべきか判断する。2つの主要なアプローチがある——LLM論理分析とSemantic Router意味マッチング。後で詳しく説明する。

検索層は「手」だ。各ビジネスシナリオに独立したベクトルインデックスがある。例えば、PythonドキュメントDB、JavaScriptドキュメントDB、GoドキュメントDB。ルーティング層がどのDBで検索するかを決定し、検索層が具体的な検索を実行する。

マージ層は「審判」だ。クエリが複数のシナリオに関わる場合、各検索器から返された結果をマージしてソートする必要がある。ここで使用するのはRRF（Reciprocal Rank Fusion）アルゴリズム——シンプルだが効果的なマルチソースランキング手法だ。

LangChainのEnsembleRetrieverでこのアーキテクチャをどう実装するか見てみよう：

from langchain.retrievers import EnsembleRetriever
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# Pythonドキュメントベクトルデータベースを初期化
python_store = Chroma(
    persist_directory="./chroma_python",
    embedding_function=OpenAIEmbeddings()
)
python_retriever = python_store.as_retriever(
    search_kwargs={"k": 5}
)

# JavaScriptドキュメントベクトルデータベースを初期化
js_store = Chroma(
    persist_directory="./chroma_js",
    embedding_function=OpenAIEmbeddings()
)
js_retriever = js_store.as_retriever(
    search_kwargs={"k": 5}
)

# RRFで複数の検索器をマージ
ensemble_retriever = EnsembleRetriever(
    retrievers=[python_retriever, js_retriever],
    c=60  # RRFパラメータ、古典的な値
)

# 検索を実行
docs = ensemble_retriever.invoke("非同期コールバックの処理方法は？")
print(f"{len(docs)}個のドキュメントチャンクを検索")

このコードの核心はEnsembleRetrieverだ。2つの検索器を同時に呼び出し、RRFアルゴリズムで結果をマージする。c=60は経験値——大きすぎるとランキングが平均に近づき、小さすぎると上位の結果の重みが大きくなりすぎる。

実測の結果、このアーキテクチャで検索精度が72%から92%に向上した。ただし、レスポンスタイムの増加という代償がある——マルチ検索器の並列クエリにはより多くの計算リソースが必要だ。

3. 3つのルーティング戦略の実践 — ロジカル、セマンティック、メタデータ

ルーティング層はどうやってクエリがどのパスを通るべきか判断するのか？3つの主要なアプローチがあり、それぞれ適用シーンが異なる。

3.1 ロジカルルーティング：LLMをディスパッチャーに

最も直接的なアイデア：LLMにクエリの意図を分析させ、データソースを選択させる。

LangChainの実装はシンプルだ：

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_deepseek import ChatDeepSeek

# ルーティングプロンプト
system_prompt = """あなたはプログラミングクエリルーティングの専門家です。
ユーザーの質問に関連するプログラミング言語に基づいて、適切なデータソースにルーティングしてください：

- Python関連の質問 → python_docs
- JavaScript関連の質問 → js_docs
- Go関連の質問 → golang_docs
- 判断できない → general_docs

データソース名のみを返してください。他の内容は含めないでください。"""

# ルーティングチェーンを構築
prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "{query}")
])

llm = ChatDeepSeek(model="deepseek-chat", temperature=0)
router_chain = prompt | llm | StrOutputParser()

# ルーティングを実行
query = "Pythonで非同期クローラを実装する方法は？"
datasource = router_chain.invoke({"query": query})
print(f"ルーティング結果: {datasource}")  # 出力: python_docs

ロジカルルーティングの利点は柔軟性だ。LLMは複雑なクエリの意図を理解できる。例えば、「PythonとGoの並行モデルを比較して」のようなマルチ言語の質問。LLMは複数のデータソースを返し、Ensemble検索を行うことができる。

欠点も明確だ：遅い。各ルーティングでLLM APIを呼び出す必要があり、0.5〜1秒のレイテンシーが増加する。さらに、APIコストが蓄積する。

3.2 セマンティックルーティング：ベクトルマッチングでLLM呼び出しを代替

スピードを追求するなら、Semantic Routerがより良い選択だ。

その原理は「ファジーif/else」のようなものだ。いくつかのルーティングルール（各ルールにはサンプル質問のセットが含まれる）を事前に定義し、ユーザークエリとこれらのサンプルのベクトル類似度をマッチングする。最も一致するルールが、クエリが通るべきパスだ。

semantic-routerライブラリを使って実装する：

from semantic_router import Route, RouteLayer
from semantic_router.encoders import OpenAIEncoder

# ルーティングルールを定義
python_route = Route(
    name="python_docs",
    utterances=[
        "Pythonでファイルを読み込む方法",
        "Pythonのデコレータの使い方",
        "Pythonで非同期プログラミングを実装する方法",
        "Pythonのリスト内包表記の構文",
    ]
)

js_route = Route(
    name="js_docs",
    utterances=[
        "JavaScriptの非同期コールバックの処理方法",
        "JSでDOMを操作する方法",
        "Node.jsのイベントループメカニズム",
        "JSのPromiseとasync/awaitの違い",
    ]
)

# ルーティング層を作成
route_layer = RouteLayer(
    encoder=OpenAIEncoder(),
    routes=[python_route, js_route]
)

# ルーティングを実行（LLM呼び出し不要）
query = "Pythonのジェネレータの使い方は？"
result = route_layer(query)
print(f"ルーティング結果: {result.name}")  # 出力: python_docs

セマンティックルーティングのレスポンスタイムはLLMルーティングより3〜5倍速い。実測の結果、OpenAI Embedding APIのレスポンスタイムは約100msだが、LLM呼び出しは500ms以上かかる。

ただし、制限もある：ルーティングルールを事前に定義する必要がある。ユーザーのクエリタイプが事前定義の範囲を超えると、ルーティングは失敗する（Noneを返す）。だから、適用シーンはクエリタイプが比較的固定されているビジネスだ。

3.3 メタデータルーティング：構造化フィールドベースのフィルタリング

ナレッジベースに豊富なメタデータ（ドキュメント分類、言語タグ、タイムスタンプなど）がある場合、SelfQueryRetrieverで精密なフィルタリングを実現できる。

from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain.chains.query_constructor.base import AttributeInfo
from langchain_openai import ChatOpenAI

# メタデータフィールドを定義
metadata_field_info = [
    AttributeInfo(
        name="category",
        description="ドキュメント分類：tutorial, api, guide, troubleshooting",
        type="string"
    ),
    AttributeInfo(
        name="language",
        description="プログラミング言語：python, javascript, golang",
        type="string"
    ),
    AttributeInfo(
        name="date",
        description="ドキュメント公開日",
        type="date"
    )
]

# 検索器を作成
llm = ChatOpenAI(model="gpt-4", temperature=0)
retriever = SelfQueryRetriever.from_llm(
    llm=llm,
    vectorstore=vectorstore,
    document_contents="プログラミング技術ドキュメント",
    metadata_field_info=metadata_field_info,
    verbose=True
)

# クエリが自動的にメタデータフィルタに変換される
query = "Pythonのチュートリアルドキュメント、最新のもの"
docs = retriever.invoke(query)
# 底層で自動的にフィルタ条件が生成される：
# category == "tutorial" AND language == "python"
# 日付で降順ソート

メタデータルーティングの利点は精密さだ。LLMが自然言語クエリを構造化フィルタ条件に変換し、その後ベクトル検索を行う。欠点はメタデータの品質への要求が高いこと——ドキュメントに分類や言語タグがない場合、このアプローチは使えない。

Railway - モダンなデプロイプラットフォーム、煩わしい運用から解放、数分で公開

今すぐ始める →広告

4. 複数ベクトルデータベース連携の実践 — EnsembleRetriever詳細解説

これまで説明したルーティング戦略は「どのDBで検索するか」という問題を解決した。しかし、一部のクエリは複数のビジネスシナリオに関わる。例えば、「PythonとJavaScriptの非同期プログラミングソリューションを比較して」。この場合、複数のDBを同時に検索し、結果をマージする必要がある。

EnsembleRetrieverの核心はRRF（Reciprocal Rank Fusion）アルゴリズムだ。

RRF アルゴリズムの原理

RRFの式はシンプルだ：

RRF(d) = Σ 1/(k + rank(d))

ここで：

• dはあるドキュメント
• rank(d)はそのドキュメントのある検索器でのランキング（1から開始）
• kは平滑化パラメータ、典型的な値は60

例を見てみよう。2つの検索器が同じドキュメントに対して異なるランキングを持つと仮定する：

• 検索器A：ドキュメントXは2位 → 貢献スコア 1/(60+2) = 0.0156
• 検索器B：ドキュメントXは5位 → 貢献スコア 1/(60+5) = 0.0154
• 合計スコア = 0.0156 + 0.0154 = 0.031

すべてのドキュメントをこの方法で計算し、合計スコアでソートする。

なぜRRFは単純な加重平均より良いのか？それはランキング位置を考慮し、元の類似度スコアではないからだ。異なる検索器のスコア範囲は大きく異なる可能性がある（例えば、ベクトル検索は0〜1のコサイン類似度を返すが、BM25は別のスコア体系を返す）。直接加重平均するとバイアスが生じる。RRFはこの問題を回避する。

高密 + 疎密ハイブリッド検索

実際のプロジェクトでは、高密検索（ベクトル検索）と疎密検索（BM25）を組み合わせることが多い。

ベクトル検索は意味マッチングに優れ、BM25はキーワードマッチングに優れている。両者は相互に補完し、より良い効果をもたらす。

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import Chroma

# 疎密検索：BM25
bm25_retriever = BM25Retriever.from_texts(
    documents_text_list,
    k=5
)

# 高密検索：ベクトル
vector_retriever = Chroma.from_texts(
    documents_text_list,
    embedding=OpenAIEmbeddings()
).as_retriever(search_kwargs={"k": 5})

# ハイブリッド検索
ensemble = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6],  # BM25の重み0.4、ベクトルの重み0.6
    c=60
)

# 検索を実行
query = "LangChain Agent ツール呼び出し"
docs = ensemble.invoke(query)

ここでパラメータ調整のコツがある：重みの配分だ。

私たちの経験では：

• 意味理解クエリ（「インテリジェントQ&Aを実装する方法」など）：ベクトルの重みを高く（0.6〜0.7）
• キーワード精密マッチングクエリ（「Python 3.11の新機能」など）：BM25の重みを高く（0.5〜0.6）
• 一般的なシナリオ：バランスの取れた重み（0.5/0.5）

検索品質の評価

EnsembleRetrieverが本当に効果があるかどうかを知るには？TruLensで評価できる。

from trulens_eval import Feedback, TruChain
from trulens_eval.feedback.provider.openai import OpenAI

provider = OpenAI()

# 評価指標を定義
relevance_feedback = Feedback(
    provider.relevance,
    name="Answer Relevance"
).on_input_output()

context_relevance_feedback = Feedback(
    provider.context_relevance,
    name="Context Relevance"
).on_input().on(context)

# 評価チェーンを登録
tru_recorder = TruChain(
    chain=ensemble_retriever_chain,
    feedbacks=[relevance_feedback, context_relevance_feedback],
    feedback_mode="with_chain"
)

# 評価を実行
with tru_recorder as recording:
    response = ensemble_retriever_chain.invoke({"query": test_query})

TruLensは「回答関連性」と「コンテキスト関連性」の2つの指標を提供する。実測の結果、単一ベクトル検索のコンテキスト関連性は平均0.72だが、Ensemble後は0.91に向上した。

5. パフォーマンス比較とベストプラクティス

理論をこれだけ話したが、実際の効果はどうなのか？

同じテストセット（500クエリ、4つのビジネスシナリオをカバー）で4つのアプローチを比較した：

主要なデータの解釈

セマンティックルーティングはロジカルルーティングより3〜4倍速い。セマンティックルーティングはEmbedding APIのみを呼び出す（約100ms）が、ロジカルルーティングはLLMを呼び出す（約800ms）必要があるからだ。クエリタイプが比較的固定されている場合、セマンティックルーティングが第一選択だ。

Ensemble RRFの精度が最も高い。マルチ検索器連携で異なる意味空間をカバーでき、RRFランキングが各検索器の強みをバランスよく活かす。代償はレスポンスタイムが単一検索器より少し長いこと——マルチ検索器の並列クエリにはより多くのリソースが必要だ。

ルーティングなしアプローチが最も遅い。これは直感に反するが、考えればわかる：単一DB検索は大量の無関係なドキュメントを返し、LLMはより多くのノイズから回答を抽出する必要があるため、生成速度が遅くなる。

ベストプラクティスのまとめ

私たちの失敗経験から、いくつかのアドバイスを提供する：

Vultr - 高性能NVMeクラウドサーバー、世界32拠点、Dockerワンクリック展開対応

価格を見る →広告

1. 単純なシナリオではセマンティックルーティングを使用

ビジネスシナリオが明確（例えばPythonドキュメントQ&Aのみ）、クエリタイプが固定（問題解決、コード例、トラブルシューティングの3種類）の場合、Semantic Routerを直接採用する。レスポンスが速く、コストも低い。

2. 複雑な推論にはロジカルルーティングを使用

クエリがマルチホップ推論やクロスドメイン比較に関わる場合、LLMの意味理解能力がより強い。例えば、「PythonとGoの並行モデルの違いは何か」の場合、LLMは2つのDBを同時に検索する必要があると判断できる。

3. マルチソース検索にはEnsembleを使用

ユーザーの意図が不明確な場合、複数のDBを同時に検索し、RRFでマージできる。単一ルーティングより安全だが、検索器の数を制御すること——3〜4個が上限で、それ以上増やすとレスポンスタイムが爆発的に増加する。

4. メタデータが豊富なDBにはSelfQueryを使用

ドキュメントに規範的なメタデータ（分類、言語、時間、作成者）がある場合、SelfQueryRetrieverは強力なツールだ。クエリの意図を自動的に解析し、精密なフィルタ条件を生成して、検索ノイズを減らすことができる。

5. 戦略を動的に調整

私たちが本番環境で運用しているのはハイブリッドアプローチだ：まずセマンティックルーティングで高速振り分けを行い（100ms）、マッチ度が閾値（例えば0.6）より低い場合、ロジカルルーティングにフォールバックして精密な判断を行う。これにより、大部分の単純なクエリは高速にレスポンスでき、少数の複雑なクエリも正しく処理される。

まとめ

RAGシステムを構築する際、多くの人はEmbeddingモデルの調整やプロンプトエンジニアリングに精力を費やすが、クエリルーティングという重要な要素を見落としている。

このクエリは高速パスを通るべきか、それともディープパスか？このシンプルな判断を正しく行うことで、レスポンスタイムを1.8秒から0.5秒に短縮し、検索精度を72%から92%に向上できる。

ルーティング戦略を選択する際、これらの原則を覚えておこう：

• 単純なクエリはセマンティックルーティング：高速、低コスト
• 複雑な推論はロジカルルーティング：LLMの意味理解がより正確
• マルチソース検索はEnsemble：RRFマージで回答の包括性を保証
• メタデータ豊富ならSelfQuery：精密フィルタリングでノイズを低減

私たちのチームは現在、本番環境でハイブリッドアプローチを運用している：セマンティックルーティングを第一層の振り分けとし、低信頼度クエリはロジカルルーティングにフォールバックし、マルチソースシナリオでは自動的にEnsemble検索をトリガーする。この組み合わせで、カスタマーサービスボットの問題解決率を68%から89%に向上させた。

完全なサンプルコードはGitHubリポジトリにあるので、クローンして試してみてください。質問があれば、コメント欄で交流しましょう。