「東京で犬と泊まれる温泉旅館を教えて」

この質問に答えるには、少なくとも3つの条件で検索する必要がある。「東京 温泉旅館」「ペット可 宿泊」「犬 同伴 旅館」。1つのクエリでベクトル検索をかけても、3条件すべてを満たす結果が上位に来る保証はない。

クエリファンアウト（Query Fan-Out）は、この問題に対するシンプルな解法だ。ユーザーの質問を複数のサブクエリに分解し、それぞれを並列で検索して、結果をマージする。RAGパイプラインや検索エージェントで、検索精度を上げるための定番パターンになっている。

なぜ1つのクエリでは足りないのか

ベクトル検索は、クエリと文書の意味的な近さを測る。だが「意味的に近い」と「質問に正確に答えている」は別の話だ。

例えば「Pythonの非同期処理でデータベース接続プールを管理するベストプラクティス」という質問。このクエリをそのまま埋め込みモデルに渡すと、ベクトルは3つの概念（非同期処理、DB接続プール、ベストプラクティス）の平均的な方向を向く。結果として、どの概念にも中途半端にしかマッチしないドキュメントが返ってくる。

これが「Lost in the Middle」問題の一因だ。長い質問ほど、ベクトル空間上での表現が曖昧になる。

クエリファンアウトの仕組み

処理の流れは3ステップで構成される。

1. クエリ分解（Decomposition）

元の質問をLLMで複数のサブクエリに分解する。

import openai

def decompose_query(question: str, n: int = 3) -> list[str]:
    """質問を複数のサブクエリに分解する"""
    response = openai.chat.completions.create(
        model='gpt-4o-mini',
        messages=[
            {
                'role': 'system',
                'content': (
                    'ユーザーの質問を、ベクトル検索に適した'
                    f'{n}個のサブクエリに分解してください。'
                    '各サブクエリは1つの概念に焦点を当ててください。'
                    'JSON配列で返してください。'
                )
            },
            {'role': 'user', 'content': question}
        ],
        response_format={'type': 'json_object'}
    )
    import json
    return json.loads(response.choices[0].message.content)['queries']

入力と出力の例:

入力: 「Pythonの非同期処理でデータベース接続プールを管理するベストプラクティス」

出力:
- "Python asyncio 非同期処理 基本パターン"
- "データベース 接続プール 設定 管理"
- "非同期 DB接続 ベストプラクティス SQLAlchemy asyncpg"

2. 並列検索（Parallel Retrieval）

分解したサブクエリを並列でベクトルDBに投げる。

import asyncio
from typing import Any

async def parallel_search(
    queries: list[str],
    vector_store: Any,
    top_k: int = 5
) -> list[list[dict]]:
    """複数クエリを並列でベクトル検索する"""
    async def search_one(query: str) -> list[dict]:
        results = await vector_store.asimilarity_search(
            query, k=top_k
        )
        return [
            {
                'content': doc.page_content,
                'metadata': doc.metadata,
                'query': query
            }
            for doc in results
        ]

    tasks = [search_one(q) for q in queries]
    return await asyncio.gather(*tasks)

ポイントはasyncio.gatherで並列実行すること。3つのサブクエリを直列で検索すると3倍の時間がかかるが、並列なら最も遅いクエリの時間で済む。

3. 結果のマージとリランキング（Merge & Rerank）

複数の検索結果をマージする。ここが一番工夫が必要な部分だ。

def merge_results(
    all_results: list[list[dict]],
    strategy: str = 'rrf'
) -> list[dict]:
    """複数の検索結果をマージする"""
    if strategy == 'rrf':
        return reciprocal_rank_fusion(all_results)
    elif strategy == 'dedupe_top':
        return dedupe_and_sort(all_results)
    else:
        raise ValueError(f'未対応の戦略: {strategy}')

def reciprocal_rank_fusion(
    all_results: list[list[dict]],
    k: int = 60
) -> list[dict]:
    """Reciprocal Rank Fusionでスコアを統合する"""
    scores: dict[str, float] = {}
    doc_map: dict[str, dict] = {}

    for results in all_results:
        for rank, doc in enumerate(results):
            doc_id = doc['content'][:100]  # 簡易的なID
            if doc_id not in doc_map:
                doc_map[doc_id] = doc
            # RRFスコア: 1 / (k + rank)
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)

    sorted_ids = sorted(scores, key=scores.get, reverse=True)
    return [doc_map[doc_id] for doc_id in sorted_ids]

Reciprocal Rank Fusion（RRF）は、各検索結果内での順位をスコアに変換して合算する手法だ。複数のクエリで上位に出たドキュメントほどスコアが高くなる。Elasticsearchのrank_featureクエリや、Pineconeのハイブリッド検索でも内部的に同じ考え方が使われている。

全体をつなげる

上の3つを組み合わせた完成形がこうなる。

async def query_fan_out(
    question: str,
    vector_store: Any,
    n_queries: int = 3,
    top_k: int = 5
) -> list[dict]:
    """クエリファンアウトの全体フロー"""
    # 1. 分解
    sub_queries = decompose_query(question, n=n_queries)
    print(f"サブクエリ: {sub_queries}")

    # 2. 並列検索
    all_results = await parallel_search(
        sub_queries, vector_store, top_k=top_k
    )

    # 3. マージ
    merged = merge_results(all_results, strategy='rrf')

    return merged[:top_k]

どこで使うと効果が高いか

RAGパイプライン

最も典型的なユースケース。LangChainやLlamaIndexには、クエリ分解のモジュールが組み込まれている。

LangChainの場合:

from langchain.retrievers import MultiQueryRetriever
from langchain_openai import ChatOpenAI

retriever = MultiQueryRetriever.from_llm(
    retriever=vector_store.as_retriever(),
    llm=ChatOpenAI(model='gpt-4o-mini'),
)

# 内部でクエリファンアウトが実行される
docs = retriever.invoke('Pythonの非同期DBプール管理')

MultiQueryRetrieverは内部でクエリ分解→並列検索→重複排除をやってくれる。自前で実装する前に、まずこれを試すのがいい。

検索エージェント

エージェントが複数のツール（Web検索、社内Wiki、DB）を持っている場合、クエリファンアウトの考え方を拡張して「どのツールにどのサブクエリを投げるか」まで判断させることができる。これはルーティング付きファンアウトと呼ばれることもある。

ECサイトの検索

「赤い ワンピース 夏用 1万円以下」のような複合条件の検索。属性ごとにサブクエリを生成し、フィルタ条件として構造化することで、ベクトル検索とメタデータフィルタのハイブリッド検索が可能になる。

注意すべきトレードオフ

レイテンシ

クエリ分解にLLM呼び出しが入る分、通常の検索より遅くなる。GPT-4o-miniなら分解に200〜500ms、検索自体は並列で50〜200ms程度。合計で0.5〜1秒の追加レイテンシが発生する。リアルタイム性が求められる場面では、分解をキャッシュするか、ルールベースで事前定義しておく手もある。

コスト

サブクエリの数だけベクトル検索のコストが増える。Pineconeなら読み取りユニット消費、OpenAI Embeddingsなら埋め込みAPI呼び出しが増える。3分解なら3倍。これはトラフィック量と相談になる。

分解の質

LLMに分解を任せると、たまに的外れなサブクエリが混じる。「東京で犬と泊まれる温泉旅館」を「東京の観光スポット」に分解されても困る。プロンプトのチューニングと、分解結果のバリデーションが実運用では必要だ。

クエリファンアウトと関連手法の比較

これらは排他的ではない。クエリファンアウトとハイブリッド検索を組み合わせて、各サブクエリでベクトル＋キーワードの両方を実行するのが、現時点で最も精度が出やすい構成だ。

試すなら

1. まずLangChainのMultiQueryRetrieverで効果を確認する
2. 効果があるなら、分解プロンプトをドメインに特化させる
3. レイテンシが問題になったら、頻出クエリパターンの分解結果をキャッシュする
4. 本番投入前に、ファンアウトあり/なしでリコール率を比較する

検索精度の改善は、RAGの回答品質に直結する。クエリファンアウトは実装コストが低い割に効果が出やすいので、検索結果に不満があるなら最初に試す価値がある。