RAG クエリルーティング実践：マルチベクトルストア連携とインテリジェント検索分散

深夜 2 時、本番環境のアラートがまた鳴り響きました。ログを確認すると、ユーザーが「サプライヤーのストライキが株価に与える影響は何ですか」と質問した際、システムが断片的なニュース記事を返し、そのうち 2 件は競合他社の情報でした。チャットで顧客が詰め寄ります。「なぜあなたたちの AI はこんなに愚かなんですか？」

同じ RAG システムでも、「2023 Q3 の華東地区の売上高はいくらですか」という質問には瞬時に正確な回答を返し、上司から「最も頼りになるチーム」と褒められました。しかし、「サプライヤーのストライキが株価に与える影響」という質問に対しては、完全に失敗してしまいました。

問題の根本原因はシンプルです。前者は単純な事実照会で、ベクトル検索で対応可能。後者はマルチホップ推論が必要です。サプライヤー、ストライキイベント、株価変動、これら 3 つの要素間の関係はナレッジグラフに隠されています。同じ検索戦略ですべてのクエリに対応するのは、1 つの鍵で全てのドアを開けようとするようなもの。開かないか、無理やり破壊するかのどちらかです。

「インテリジェントルーター」が必要です。システムが質問の特徴に基づいて、最適な検索パスを自動的に選択できるように。

この記事では、3 つの主要なアプローチについて解説します：論理ルーティング（LLM による意図分析）、意味ルーティング（埋め込み空間でのファジーマッチング）、EnsembleRetriever（RRF アルゴリズムによる融合）。いずれも実際に失敗を経験し、本番環境で効果を検証してきました。まず明確にしておきますが、「最高」のソリューションは存在しません。「最適」なシナリオに適したソリューションがあるだけです。

第 1 章：なぜクエリルーティングが必要なのか？ ——「単一ベクトルストア」から「マルチソース連携」へ

以前、ある企業のナレッジベースシステム構築を支援しました。財務データベース（MySQL）、技術文書ライブラリ（ベクトルストア）、人員関係グラフ（Neo4j）の 3 つのデータソースがありました。最初のソリューションはシンプルでした。すべてのデータを 1 つのベクトルストアに格納するのです。

結果はどうなったでしょうか？「華東地区の売上高」という単純な質問には、財務報告書の表から正確に回答を取得できました。しかし、「サプライヤーのストライキがどの製品ラインに影響するか」という質問には、バラバラのニュース記事を返し、ユーザーは首をかしげるばかりでした。

その後、すべてのクエリがベクトル検索に適しているわけではないと気づきました。ある質問は SQL クエリの方が高速かつ正確。ある質問はナレッジグラフで関係を繋げる必要があります。またある質問は Web 検索で最新情報を取得する必要があります。同じ検索戦略を使用すると、「能力不足」か「過剰設計」のどちらかになります。

1.1 単一ベクトルストア検索の限界：2 つの実シナリオ比較

シナリオ A：単純な事実照会（ベクトル検索で十分）

ユーザーの質問：「2023 Q3 の華東地区の売上高はいくらですか？」

システムの動作：ベクトル検索が財務報告書の表を特定し、「華東地区 Q3 売上高は 1.2 億元」と直接回答。全プロセスは約 300ms で、ユーザーは満足しています。

もし無理にナレッジグラフ推論モジュールを呼び出すとどうなるでしょうか？GPU リソースを浪費するだけでなく、500ms のレイテンシが追加されます。ロケットで宅配便を届けるようなもの。届きますが、不要です。

シナリオ B：複雑な推論クエリ（マルチホップ検索が必要）

ユーザーの質問：「サプライヤーのストライキが株価に与える影響は何ですか？」

システムの動作：ベクトル検索が断片的なニュースを返します。「XX 社の株価が 5% 下落」、「サプライヤーのストライキイベント報道」。しかし LLM は中間の論理チェーンを欠いています。どのサプライヤー？誰に供給している？ストライキ期間は？株価はどれくらい下落した？これらの情報は異なる文書に散らばっており、LLM は答えを捏造しやすくなります。

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

15%割引を入手 →広告

正しいアプローチ：ナレッジグラフで「サプライヤー → ストライキイベント → 契約関係 → 株価変動」を繋げ、論理チェーンを明確にします。しかし問題は、「この質問にはナレッジグラフを使うべき」とシステムが自動的に判断するにはどうすればよいかです。

これがクエリルーティングが解決すべき核心課題です。

1.2 クエリ特徴の 4 次元分析

プロジェクトで活用しているシンプルな判断フレームワークを紹介します。クエリの 4 つの次元に基づいて検索戦略を選択します。

例えば、「華東地区の売上高」はシングルホップ、構造化、静的データなので、SQL クエリが最速。「サプライヤーのストライキが株価に与える影響」は高コンテキスト依存、マルチホップ、非構造化データなので、ナレッジグラフが適しています。

ここまで読んで、「各クエリで同時に 3 つのデータベースを検索し、結果をマージすればいいのでは？」と考えるかもしれません。確かに可能ですが、コストが爆発します。各クエリで 3 つの検索器を呼び出すと、レイテンシが 200-500ms 増加し、LLM 呼び出しコストも倍増します。上司がコストを気にしない場合を除き。

より賢いアプローチは、システムが「時流を見極め」、クエリ特徴に基づいて検索パスを動的に選択することです。これがクエリルーターの価値です。精度、効率、コストのバランスを見つけることです。

第 2 章：論理ルーティング —— LLM による意図分析とデータソース選択

論理ルーティングは最も直感的なアプローチです。LLM に「選択リスト」を与え、質問を分析させ、リストから最もマッチするデータソースを選択させます。

病院の受付に例えると、看護師が「どこが痛いですか？」と聞き、「胃が痛い」と答えれば消化器内科、「頭が痛い」と答えれば神経内科に案内します。論理ルーティングの LLM はこの看護師のようなもので、症状（クエリ）に基づいて最適な診療科（データソース）を選択します。

実装方法：LangChain + Structured Output

まず完全なコードを見てから、経験した失敗について話しましょう。

from langchain_core.prompts import ChatPromptTemplate
from langchain_deepseek import ChatDeepSeek
from pydantic import BaseModel, Field
from typing import Literal

# データソース列挙の定義（LLM の出力の曖昧さを回避）
class DataSource(BaseModel):
    """データソース選択結果"""
    source: Literal["finance_db", "tech_docs", "knowledge_graph", "web_search", "general_search"] = Field(
        description="選択されたデータソース"
    )

# ルーティングプロンプトテンプレートの設定
system_prompt = """
あなたはプロフェッショナルなクエリルーティング専門家です。ユーザーの質問内容に基づいて、適切なデータソースにルーティングしてください：

- 財務データ、売上データに関する質問は "finance_db"（リレーショナルデータベース）を返す
- 技術文書、製品マニュアルに関する質問は "tech_docs"（ベクトルデータベース）を返す
- 人物関係、組織図に関する質問は "knowledge_graph"（グラフデータベース）を返す
- 最新のリアルタイム情報が必要な場合は "web_search" を返す
- 明確に判断できない場合は "general_search" を返す

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

prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "{question}"),
])

# DeepSeek モデルを使用（安価で高性能）
llm = ChatDeepSeek(model="deepseek-chat", temperature=0.1)
structured_llm = llm.with_structured_output(DataSource)

# ルーティングチェーンの構築
route_chain = prompt | structured_llm

# ルーティングのテスト
query1 = "2023 Q3 華東地区の総売上高はいくらですか？"
result1 = route_chain.invoke({"question": query1})
print(result1.source)  # 出力：finance_db

query2 = "サプライヤーのストライキが株価に与える影響は何ですか？"
result2 = route_chain.invoke({"question": query2})
print(result2.source)  # 出力：knowledge_graph

このコードには重要なディテールがあります。temperature=0.1 です。以前、0.7 に設定して失敗しました。同じクエリでも、ある時はナレッジグラフに、ある時は Web 検索にルーティングされました。その後、ルーターには安定性が必要で、ランダム性を持たせてはいけないと気づきました。

もう一つのディテールは Pydantic の DataSource 列挙です。最初は LLM に文字列を直接返させていましたが、「finance_db を調べるべきです」とか「finance_db か general_search のどちらかを調べるべきだと思います」といった返答が来てしまいました。これらの曖昧さは後続処理を複雑にします。Pydantic で列挙値を強制することで、すっきりとした結果になります。

長所と短所の比較

プロジェクトでテストした結果、論理ルーティングはデータソース 5 以下 の場合に最も効果的です。5 つを超えると、プロンプトが長くなり、LLM も混乱しやすくなります。例えば 10 個のデータソースがある場合、意味ルーティングまたは階層的な論理ルーティング（まず大分類、その後細分化）を検討すべきです。

第 3 章：意味ルーティング —— 埋め込み空間に基づく「ファジー If/Else」

意味ルーティングはより高速です。論理ルーティングは LLM が少し「考える」必要があります（500-800ms）。意味ルーティングは埋め込み類似度を直接計算し、50ms で完了します。10 倍以上高速です。

原理は「ファジーマッチング」に似ています。事前にサンプルクエリ（utterances）を定義します。例えば、「売上高を照会」「財務報告書データ」「収益状況はどうですか」などです。これらのクエリはすべて「財務照会」という意図を指します。ユーザーが質問すると、システムはユーザーの質問とこれらのサンプル間の意味的類似度を計算し、閾値を超えると対応するルートをトリガーします。

母親が「夜何を食べたい？」と聞き、あなたが「なんでもいい、辛すぎなければ」と答えるようなものです。母親の頭には「ファジーマッチングテーブル」があります。「辛すぎない」≈「トマトスクランブルエッグ」、「蒸し魚」、「冬瓜スープ」。意味ルーティングはこのファジーマッチングプロセスです。

実装方法：semantic-router ライブラリ + 事前定義 Utterances

コードは論理ルーティングよりもシンプルです。

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

# ルーティングルールの定義（意味的類似度閾値）
routes = [
    Route(
        name="finance_query",
        utterances=[
            "売上高を照会",
            "財務報告書データ",
            "収益状況はどうですか",
            "利益分析",
        ],
    ),
    Route(
        name="tech_support",
        utterances=[
            "製品の使い方",
            "技術文書はどこですか",
            "トラブルシューティング方法",
            "機能説明",
        ],
    ),
    Route(
        name="graph_query",
        utterances=[
            "誰と誰が協力関係にあるか",
            "組織図の関係",
            "上流下流のサプライチェーン",
            "人物関係グラフ",
        ],
    ),
]

# RouteLayer を作成（無料の HuggingFace 埋め込みモデルを使用）
encoder = HuggingFaceEncoder()
route_layer = RouteLayer(encoder=encoder, routes=routes)

# ルーティングのテスト（LLM 呼び出しなし、レスポンス ~50ms）
query1 = "サプライヤーのストライキが株価に与える影響は何ですか？"
route1 = route_layer(query1)
print(route1.name)  # 出力：graph_query

query2 = "2023 Q3 華東地区の売上高はいくらですか？"
route2 = route_layer(query2)
print(route2.name)  # 出力：finance_query

このコードの鍵は utterances です。各意図に対して 4-10 個のサンプルクエリを事前定義し、システムはユーザーの質問とこれらのサンプル間の意味的類似度を計算します。デフォルトの閾値は 0.85 です。つまり、類似度が 85% を超えるとルートがトリガーされます。

以前テストしたところ、utterances が少なすぎると（例えば 2 つだけ）、再現率が低くなります。逆に多すぎると（20 個以上）、計算コストが増加します。各意図に対して 4-10 個のサンプルを推奨します。一般的な表現方法をカバーしてください。

もう一つの利点は HuggingFaceEncoder です。これは無料のローカル埋め込みモデルで、OpenAI API を呼び出す必要がなく、コストがかかりません。クエリ量が多い場合、論理ルーティングは 1 回あたり $0.0001 に見えますが、1 日 10 万回のクエリで $10、1 ヶ月で $300 になります。意味ルーティングは完全に無料です。

長所と短所の比較

意味ルーティングには限界があります。「論理推論」タイプの意図判断を処理できません。例えば「クエリが財務データに関連し、かつ即時性が高い場合、リアルタイムデータベースを優先する」といった条件判断です。こうした論理判断は LLM に頼る必要があります。そのため実際のプロジェクトでは、意味ルーティングを意図分類（財務/技術/関係照会）に使用し、論理ルーティングで複雑な条件判断を処理するという組み合わせが効果的です。

第 4 章：EnsembleRetriever —— RRF アルゴリズムによるマルチ検索器の統合

前 2 つのアプローチは「1 つの検索器を選択」するものでしたが、EnsembleRetriever は「複数の検索器の結果を統合」します。

KiloClaw - 企業向けマネージド OpenClaw、AI エージェントのオーケストレーションとスマートルーティングを加速

今すぐ始める →広告

最も古典的なシナリオ：BM25（キーワードマッチング）+ ベクトル検索（意味マッチング）。ユーザーが「2023 Q3 売上高」と尋ねた場合、BM25 は「売上高」というキーワードを正確にマッチングできますが、「収益」という同義語を見逃す可能性があります。ベクトル検索は「収益」と「売上高」が同じ意味であることを理解できますが、無関係な財務文書を多数召回してしまう可能性があります。

この 2 つを組み合わせることで、再現率と精度の両方を向上させることができます。これが EnsembleRetriever の価値です。

実装方法：LangChain EnsembleRetriever + RRF アルゴリズム

コードの実装は意外なほどシンプルです。

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

# BM25 検索器を作成（キーワードマッチング）
bm25_retriever = BM25Retriever.from_texts(
    ["財務報告書2023 Q3", "華東地区売上データ", "サプライヤーリスト"],
    k=2,
)

# ベクトル検索器を作成（意味マッチング）
vectorstore = Chroma.from_texts(
    ["財務報告書2023 Q3", "華東地区売上データ", "サプライヤーリスト"],
    embedding=OpenAIEmbeddings(),
)
vector_retriever = vectorstore.as_retriever(k=2)

# EnsembleRetriever を構成（RRF アルゴリズム）
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6],  # BM25 ウェイト 0.4、ベクトルウェイト 0.6
)

# 検索のテスト
query = "2023 Q3 華東地区売上高"
docs = ensemble_retriever.invoke(query)
print(docs)  # 出力：BM25 とベクトル検索の融合結果（RRF スコア順）

核となるのは RRF（Reciprocal Rank Fusion）アルゴリズムです。高度に聞こえますが、原理は非常にシンプルです。

ある文書が BM25 で 1 位、ベクトル検索で 3 位だったとします。RRF の計算方法は：

BM25 1 位 → 1/(60+1) = 0.0164
ベクトル検索 3 位 → 1/(60+3) = 0.0159
総合スコア = 0.0164 + 0.0159 = 0.0323

k=60 は経験値で、プロジェクトに応じて調整できます。k が大きいほど順位差の影響が小さくなり、k が小さいほど上位の文書が優位になります。

なぜ RRF は有効なのか？

RRF の巧妙な点は、文書の元のスコア（異なる検索器のスコアは比較不可能）に依存せず、順位のみに依存することです。これにより、あらゆるタイプの検索器を統合できます。BM25、ベクトル検索、ナレッジグラフ検索、さらには Web 検索結果まで。

プロジェクトでテストした結果、純粋な BM25 の再現率は 70%、純粋なベクトル検索は 85%、EnsembleRetriever は 92% に達しました。コストは 50ms の増加のみです（2 つの検索器を並列呼び出し）。

長所と短所の比較

EnsembleRetriever には限界があります。「同じタイプ」の検索結果のみ統合できます。例えば、ベクトルストアとナレッジグラフを同時に検索したい場合、EnsembleRetriever は役立ちません。こうしたクロスデータソースのシナリオでは、論理ルーティングまたは意味ルーティングが必要です。

第 5 章：本番デプロイのコスト最適化戦略

これまでは「システムをより賢くする方法」について話してきました。この章では「システムをより経済的にする方法」について話します。私が経験した最大の失敗はコストの爆発です。初週、LLM 呼び出しコストが $500 になり、上司に解雇されかけました。

その後、3 つの戦略を学びました。Semantic Caching（意味キャッシング）、Tiered Retrieval（階層型検索）、Parallel Processing（並列処理）です。コストは週 $50 まで削減され、精度も向上しました。

5.1 Semantic Caching（意味キャッシング）

最もシンプルで効果的な戦略です。原理：一般的なクエリの埋め込みをキャッシュし、類似度 > 0.95 であればキャッシュされた回答を直接返し、LLM を呼び出さないようにします。

本番環境でテストした結果、キャッシュヒット率は 30-50% に達しました。レスポンス時間は ~500ms から ~50ms に短縮（キャッシュヒット時）、ユーザー体験が明確に向上しました。

from langchain.cache import InMemoryCache
from langchain.embeddings import CacheBackedEmbeddings
from langchain_openai import OpenAIEmbeddings

# 一般的なクエリの埋め込みをキャッシュ
underlying_embeddings = OpenAIEmbeddings()
cached_embeddings = CacheBackedEmbeddings.from_bytes_store(
    underlying_embeddings,
    InMemoryCache(),  # 本番環境では Redis を推奨
)

# キャッシュされた埋め込みを使用してルーティング
# 類似度 > 0.95 でキャッシュされた回答を直接返す

実際のデプロイでは Redis または Memcached を使用することを推奨します。InMemoryCache は使用しないでください（再起動で失われます）。定期的なキャッシュクリアも行います。例えば 30 日間アクセスがないクエリは自動的に期限切れにします。

5.2 Tiered Retrieval（階層型検索）

単純なクエリには安価なモデル、複雑なクエリには高価なモデルを使用します。最も直感的なコスト最適化戦略です。

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# 単純なクエリには安価なモデル
simple_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.1)

# 複雑なクエリには高価なモデル
complex_llm = ChatAnthropic(model="claude-opus-4-20250514", temperature=0.2)

# ルーティングロジック
if route_layer(query).name in ["finance_query", "tech_support"]:
    # 単純な意図には GPT-4o-mini
    response = simple_llm.invoke(query)
elif route_layer(query).name == "graph_query":
    # 複雑な推論には Claude Opus
    response = complex_llm.invoke(query)

コスト比較は明確です。

100 倍の差があります。システムの 80% が単純なクエリの場合、コストは元の 20% まで削減できます。

5.3 Parallel Processing（並列処理）

ハイブリッドルーティング（論理ルーティング + EnsembleRetriever）は 200-500ms のレイテンシを追加します。しかし、良いニュースは、複数の検索器を並列呼び出しできるため、レイテンシは少ししか増加しないことです。

import asyncio
from langchain_community.retrievers import BM25Retriever

# BM25 + Vector 検索器を並列呼び出し
async def parallel_retrieval(query):
    bm25_task = asyncio.create_task(bm25_retriever.invoke(query))
    vector_task = asyncio.create_task(vector_retriever.invoke(query))

    bm25_docs, vector_docs = await asyncio.gather(bm25_task, vector_task)

    # RRF 融合
    return ensemble_docs

実測結果：逐次呼び出しは 600ms、並列呼び出しは 320ms。ハイブリッドルーティングのレイテンシオーバーヘッドをほぼ相殺できます。

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

価格を見る →広告

この 3 つの戦略を組み合わせることで、システムコストは週 $500 から $50 まで削減され、レスポンス速度も向上しました。コスト最適化は「手抜き」ではなく、「リソースの賢い配分」です。

第 6 章：アプローチ比較と選定ガイド

ここまで説明してきましたが、「私のプロジェクトではどのアプローチを使うべきか？」と疑問に思うかもしれません。シンプルな比較表と判断ツリーを提供します。

3 つのアプローチの核心比較

選定判断ツリー

シンプルな判断フローを描きました。迅速に判断するのに役立ちます。

第 1 ステップ：異なるタイプのデータソースにルーティングする必要がありますか？
├─ はい → 第 2 ステップ：データソース数 5 以下？
│   ├─ はい → 【論理ルーティング】を選択（LLM による意図分析）
│   └─ いいえ → 第 2 ステップ：データソース数 20 以下？
│       ├─ はい → 【意味ルーティング】を選択（事前定義 utterances）
│       └─ いいえ → Multi-Agent コーディネーターが必要（本記事の範囲外）
└─ いいえ → 第 3 ステップ：同じタイプの検索器を統合する必要がありますか？
    ├─ はい → 【EnsembleRetriever】を選択（RRF 融合）
    └─ いいえ → 単一ベクトルストア検索で十分

実践的なアドバイス

あなたのプロジェクトが「企業ナレッジベースシステム」で、財務データベース、技術文書ライブラリ、ナレッジグラフの 3 つのデータソースがある場合、以下を推奨します。

1. まず意味ルーティングで意図分類（財務/技術/関係照会）を行う。高速で無料。
2. 次に論理ルーティングで特殊シナリオを処理（例えば、即時性が高いクエリを Web 検索にルーティング）。
3. 各データソース内で EnsembleRetrieverを使用（BM25 + ベクトル検索）、再現率を向上。
4. 最後にコスト最適化を追加（Semantic Caching、Tiered Retrieval）、コスト削減と速度向上。

この「3 層ルーティング」アーキテクチャは、3 つのプロジェクトで検証し、効果は安定しています。週約 $50 のコスト、レスポンス時間 < 800ms、ユーザー満足度 85% 以上。

あなたのプロジェクトが単一データソース（例えばベクトルストアのみ）の場合、急いでルーティングを導入しないでください。まず EnsembleRetriever で BM25 + ベクトル検索のハイブリッドを行い、再現率がニーズを満たすか確認してください。多くの場合、単一ベクトルストアの限界は検索戦略の最適化不足であり、そもそもルーティングは不要です。

まとめとアクションプラン

ここまで書いてきましたが、核心的なポイントをまとめます。

クエリルーティングの本質：クエリ特徴（コンテキスト依存、推論ホップ数、データタイプ、即時性）に基づいて検索パスを動的に選択すること。ナビゲーションアプリが交通状況に基づいて最適ルートを選択するように、固定ルートを盲目的に進むのではありません。

3 つのアプローチの適用シナリオ：

• 論理ルーティングはデータソースタイプが明確（5以下）、深い意図理解が必要なシナリオに適しています。
• 意味ルーティングは意図分類（20以下）、高速レスポンス、コスト重視のシナリオに適しています。
• EnsembleRetriever は同じタイプの検索器統合（BM25 + ベクトル）、再現率向上に適しています。

本番デプロイのコスト最適化：Semantic Caching、Tiered Retrieval、Parallel Processing の 3 つの戦略を組み合わせることで、コストを元の 10% まで削減し、レスポンス速度も向上します。

アクションプラン

RAG システムを構築中の場合、この順序で反復することを推奨します。

第 1 ステップ：ボトルネックの診断
既存システムの失敗ケースを分析し、「低コンテキスト依存」（ベクトル検索で十分）と「高コンテキスト依存」（マルチホップ推論が必要）に分類します。このステップを飛ばさないでください。そうしないと過剰設計になりがちです。

第 2 ステップ：アプローチの選択
データソース数、意図数、コスト予算に基づいて、論理/意味/EnsembleRetriever を選択します。最初から 3 つのアプローチを組み合わせず、まず単一アプローチの効果を検証してください。

第 3 ステップ：コスト最適化の追加
まず Semantic Caching（最もシンプルで効果が高い）、その後 Tiered Retrieval と Parallel Processing を検討します。コスト最適化は一足飛びではなく、継続的な反復プロセスです。

具体的なプロジェクトの疑問がある場合は、ぜひコメントで議論してください。私が経験した失敗が、あなたの回避に役立つかもしれません。

9 min read · 公開日: 2026年5月5日 · 更新日: 2026年5月5日

default

AI・インテリジェンス