AI アプリケーション開発において、検索精度は成功の鍵を握ります。特に RAG（Retrieval-Augmented Generation）システムでは、関連性の高い情報を正確に取得できなければ、LLM が適切な回答を生成できません。

しかし、従来の単一の検索手法だけでは、複雑な質問や多様なデータ形式に対応するのが難しいのが現状です。そこで注目されているのが、Haystack フレームワークを活用したハイブリッド検索と再ランキングの組み合わせでしょう。

本記事では、Haystack を使った RAG アーキテクチャ設計の実践的な手法を、コード例とともに詳しくご紹介します。初心者の方でも実装できるよう、段階的に解説していきますね。

背景

RAG システムの重要性

RAG は、大規模言語モデル（LLM）の回答生成能力と、外部データベースからの情報検索を組み合わせた技術です。LLM 単体では学習データに含まれない最新情報や専門知識に対応できませんが、RAG を導入することで、常に最新かつ正確な情報に基づいた回答が可能になります。

企業の社内文書検索、カスタマーサポート、法律文書の分析など、幅広い分野で活用が進んでいるのです。

Haystack フレームワークとは

Haystack は deepset 社が開発した、本番環境対応の AI オーケストレーションフレームワークです。RAG システムの構築に特化した設計となっており、以下の特徴を持ちます。

#	特徴	説明
1	モジュール性	コンポーネントを自由に組み合わせてパイプラインを構築可能
2	ベンダーニュートラル	OpenAI、Anthropic、Hugging Face など複数のサービスに対応
3	透明性	各処理ステップを検査・デバッグ・最適化できる
4	スケーラビリティ	Kubernetes 対応で本番環境にデプロイしやすい

これにより、開発者は特定のベンダーにロックインされることなく、最適な技術スタックを選択できます。

以下の図は、Haystack の基本的なアーキテクチャを示しています。

mermaidCopy codeflowchart TB
    user["ユーザー"] -->|質問| pipeline["Haystack<br/>パイプライン"]
    pipeline --> retriever["Retriever<br/>コンポーネント"]
    pipeline --> ranker["Ranker<br/>コンポーネント"]
    pipeline --> generator["Generator<br/>コンポーネント"]

    retriever -->|検索| vectordb[("Vector<br/>Database")]
    ranker -->|再順序付け| docs["検索結果<br/>ドキュメント"]
    generator -->|プロンプト| llm["LLM<br/>(GPT-4など)"]

    llm -->|回答| user

図で理解できる要点：

• パイプラインは複数のコンポーネントで構成される
• 各コンポーネントは独立して交換可能
• データフローが明確に定義されている

検索技術の進化

従来の検索技術は、主に 2 つのアプローチに分かれていました。

**キーワードベース検索（Sparse 検索）**は、BM25 などのアルゴリズムを用いて、文書内の単語の出現頻度に基づいてスコアリングします。固有名詞や専門用語の検索に強いですが、同義語や文脈を理解できません。

一方、**ベクトル検索（Dense 検索）**は、文章を高次元ベクトル空間に埋め込み、意味的な類似度で検索を行います。文脈理解に優れていますが、固有名詞の完全一致には弱い傾向があるのです。

これら両方のアプローチを組み合わせることで、より高精度な検索が実現できるようになりました。

課題

単一検索手法の限界

従来の RAG システムでは、ベクトル検索のみ、またはキーワード検索のみを使用するケースが多く見られました。しかし、この単一手法では以下のような課題が生じます。

ベクトル検索のみの場合：

• 固有名詞（人名、地名、製品名）の完全一致検索が苦手
• 専門用語や略語の検索精度が低下
• 計算コストが高く、大規模データベースでのレスポンス時間が長い

キーワード検索のみの場合：

• 同義語や言い換え表現に対応できない
• 文脈や意味の理解が不可能
• 質問文とドキュメントの表現が異なると検索できない

実際の運用では、これらの問題により検索精度が 60〜70% 程度に留まることも珍しくありません。

検索結果の順序最適化の必要性

複数の検索手法を組み合わせた場合、さらに別の課題が浮上します。それぞれの検索手法が異なるスコアリング基準を持つため、結果をマージする際に順序が最適化されないのです。

以下の図は、ハイブリッド検索における課題を示しています。

mermaidCopy codeflowchart LR
    query["ユーザー<br/>クエリ"] --> sparse["Sparse検索<br/>(BM25)"]
    query --> dense["Dense検索<br/>(Vector)"]

    sparse -->|スコア: 0.8, 0.6, 0.5| merge["結果マージ"]
    dense -->|スコア: 0.9, 0.7, 0.4| merge

    merge -->|順序が<br/>不整合| problem["課題:<br/>関連性が低い<br/>文書が上位に"]

図で理解できる要点：

• 異なるスコアリング基準が混在
• マージ時に真の関連性が失われる
• LLM に不適切な文脈が渡される可能性

具体的には、以下の問題が発生します。

#	課題	影響
1	スコアの基準が異なる	単純な統合では優先順位を決められない
2	関連性の低い文書が上位に	LLM が誤った情報を基に回答を生成
3	コンテキストウィンドウの無駄遣い	限られたトークン数で不要な文書を送信

これらの課題を解決するには、統一された基準で文書を再評価する再ランキングの仕組みが不可欠です。

パフォーマンスと精度のトレードオフ

高精度な検索を実現しようとすると、計算コストが増大し、レスポンス時間が長くなってしまいます。

例えば、100 万件のドキュメントに対してベクトル検索を行う場合、すべての候補に対して高精度な再ランキングモデルを適用すると、1 クエリあたり数秒から数十秒かかることもあるでしょう。

ユーザー体験を損なわずに高精度を実現するには、検索パイプラインの各段階で適切な候補数（top_k パラメータ）を設定し、計算リソースを効率的に配分する必要があります。

解決策

ハイブリッド検索の実装

ハイブリッド検索は、Sparse 検索と Dense 検索を組み合わせることで、両者の長所を活かす手法です。Haystack では、複数の Retriever を並列実行し、結果をマージできます。

以下のアーキテクチャ図は、ハイブリッド検索の全体像を示しています。

mermaidCopy codeflowchart TB
    query["ユーザークエリ"] --> embedder["Query<br/>Embedder"]
    query --> bm25["BM25<br/>Retriever"]

    embedder --> vector["Vector<br/>Retriever"]

    bm25 -->|top_k=30| joiner["Document<br/>Joiner"]
    vector -->|top_k=30| joiner

    joiner -->|統合: 30-50件| ranker["Ranker<br/>(再ランキング)"]
    ranker -->|top_k=5| prompt["Prompt<br/>Builder"]
    prompt --> llm["LLM<br/>Generator"]
    llm --> answer["最終回答"]

図で理解できる要点：

• 2 つの検索手法が並列実行される
• Joiner で結果を統合
• Ranker で最終的な順序を最適化

まず、必要なパッケージをインストールしましょう。

bashCopy code# Haystack のコアパッケージをインストール
yarn add haystack-ai

# ベクトルストアとして Elasticsearch を使用する場合
yarn add elasticsearch-haystack

次に、基本的なパイプライン構成を定義します。

pythonCopy code# 必要なコンポーネントをインポート
from haystack import Pipeline
from haystack.components.retrievers import InMemoryBM25Retriever
from haystack.components.retrievers import InMemoryEmbeddingRetriever

Document Store の初期化を行います。これはドキュメントを保存・検索するためのストレージです。

pythonCopy code# インメモリのドキュメントストアを作成
from haystack.document_stores.in_memory import InMemoryDocumentStore

# ドキュメントストアの初期化
document_store = InMemoryDocumentStore()

# サンプルドキュメントを追加
documents = [
    {"content": "Haystack は RAG システム構築フレームワークです。", "meta": {"source": "doc1"}},
    {"content": "ハイブリッド検索は精度を 30% 向上させます。", "meta": {"source": "doc2"}},
    {"content": "再ランキングにより関連性の高い文書が上位に表示されます。", "meta": {"source": "doc3"}}
]

# ドキュメントストアに書き込み
document_store.write_documents(documents)

BM25 Retriever を設定します。これはキーワードベースの検索を担当します。

pythonCopy code# BM25 検索コンポーネントの初期化
bm25_retriever = InMemoryBM25Retriever(
    document_store=document_store,
    top_k=30  # 上位30件を取得
)

ベクトル検索のための Embedder を設定します。

pythonCopy code# テキスト埋め込みコンポーネントのインポート
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.embedders import SentenceTransformersDocumentEmbedder

# クエリ用の Embedder を初期化
text_embedder = SentenceTransformersTextEmbedder(
    model="sentence-transformers/all-MiniLM-L6-v2"
)

# ドキュメント用の Embedder を初期化
doc_embedder = SentenceTransformersDocumentEmbedder(
    model="sentence-transformers/all-MiniLM-L6-v2"
)

ドキュメントに埋め込みを追加し、ベクトル検索を可能にします。

pythonCopy code# ドキュメントに埋め込みを追加
documents_with_embeddings = doc_embedder.run(documents)
document_store.write_documents(documents_with_embeddings["documents"])

# ベクトル検索コンポーネントの初期化
embedding_retriever = InMemoryEmbeddingRetriever(
    document_store=document_store,
    top_k=30  # 上位30件を取得
)

再ランキングの実装

再ランキングは、検索結果を統一された基準で再評価し、真に関連性の高い文書を上位に配置する手法です。Haystack では、各種 Ranker コンポーネントを提供しています。

Document Joiner で複数の検索結果を統合します。

pythonCopy code# ドキュメント統合コンポーネントのインポート
from haystack.components.joiners import DocumentJoiner

# Joiner の初期化（重複排除機能付き）
document_joiner = DocumentJoiner(
    join_mode="merge",  # マージモード
    weights=[0.5, 0.5]  # BM25 と Vector を同等の重みで統合
)

再ランキングモデルを設定します。ここでは NVIDIA NeMo Retriever を使用した例を示します。

pythonCopy code# NVIDIA Ranker のインポート
from haystack_integrations.components.rankers.nvidia import NvidiaRanker

# 再ランキングコンポーネントの初期化
ranker = NvidiaRanker(
    model="nvidia/nv-rerankqa-mistral-4b-v3",  # 再ランキングモデル
    top_k=5,  # 最終的に上位5件を選択
    api_key="your-nvidia-api-key"  # API キー
)

より軽量な代替手段として、Sentence Transformers ベースの Ranker も使用できます。

pythonCopy code# Sentence Transformers Ranker のインポート
from haystack.components.rankers import TransformersSimilarityRanker

# 軽量な再ランキングコンポーネント
similarity_ranker = TransformersSimilarityRanker(
    model="cross-encoder/ms-marco-MiniLM-L-6-v2",
    top_k=5
)

パイプラインの構築

これまで定義したコンポーネントを組み合わせて、完全なハイブリッド検索パイプラインを構築します。

パイプラインオブジェクトを作成し、コンポーネントを追加していきます。

pythonCopy code# パイプラインの初期化
pipeline = Pipeline()

# 各コンポーネントをパイプラインに追加
pipeline.add_component("text_embedder", text_embedder)
pipeline.add_component("bm25_retriever", bm25_retriever)
pipeline.add_component("embedding_retriever", embedding_retriever)
pipeline.add_component("document_joiner", document_joiner)
pipeline.add_component("ranker", similarity_ranker)

コンポーネント間の接続を定義し、データフローを確立します。

pythonCopy code# クエリ埋め込みと検索の接続
pipeline.connect("text_embedder.embedding", "embedding_retriever.query_embedding")

# 両方の Retriever の結果を Joiner に接続
pipeline.connect("bm25_retriever.documents", "document_joiner.documents")
pipeline.connect("embedding_retriever.documents", "document_joiner.documents")

# Joiner の結果を Ranker に接続
pipeline.connect("document_joiner.documents", "ranker.documents")

以下の図は、構築したパイプラインのデータフローを示しています。

mermaidCopy codesequenceDiagram
    participant U as ユーザー
    participant P as Pipeline
    participant E as TextEmbedder
    participant B as BM25Retriever
    participant V as VectorRetriever
    participant J as Joiner
    participant R as Ranker

    U->>P: クエリ送信
    P->>E: テキスト埋め込み生成
    P->>B: BM25検索実行
    P->>V: ベクトル検索実行

    B->>J: 検索結果(30件)
    V->>J: 検索結果(30件)

    J->>R: 統合結果(40-50件)
    R->>P: 再ランキング結果(5件)
    P->>U: 最終結果を返却

図で理解できる要点：

• 並列処理により検索速度を最適化
• Joiner で重複を排除しつつ統合
• Ranker が最終的な順序を決定

最適化のポイント

パイプラインの性能を最大化するには、各段階での top_k パラメータの調整が重要です。

#	段階	推奨 top_k	理由
1	初期検索（BM25/Vector）	20-50	幅広い候補を確保
2	統合後	30-60	重複排除後の候補数
3	再ランキング後	3-10	LLM のコンテキストに最適な数

検索精度を評価するための指標設定も忘れずに行いましょう。

pythonCopy code# 評価用のインポート
from haystack.evaluation import EvaluationResult

# MRR（Mean Reciprocal Rank）と NDCG で評価
# これにより再ランキングの効果を定量的に測定できる

具体例

実践的なユースケース

企業の技術文書検索システムを構築する例を見ていきましょう。製品マニュアル、API ドキュメント、トラブルシューティングガイドなど、多様な文書から適切な情報を検索するシステムです。

まず、エンドツーエンドのパイプラインを構築します。

pythonCopy code# 完全な RAG パイプラインの構築
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator

# プロンプトテンプレートの定義
prompt_template = """
以下のコンテキスト情報を使用して、質問に回答してください。

コンテキスト:
{% for doc in documents %}
- {{ doc.content }}
{% endfor %}

質問: {{ query }}

回答:
"""

プロンプトビルダーと LLM ジェネレーターをパイプラインに追加します。

pythonCopy code# プロンプトビルダーの初期化
prompt_builder = PromptBuilder(template=prompt_template)

# OpenAI Generator の初期化
generator = OpenAIGenerator(
    model="gpt-4",
    api_key="your-openai-api-key"
)

# パイプラインに追加
pipeline.add_component("prompt_builder", prompt_builder)
pipeline.add_component("generator", generator)

Ranker と Prompt Builder を接続し、最終的な回答生成までのフローを完成させます。

pythonCopy code# Ranker の出力を Prompt Builder に接続
pipeline.connect("ranker.documents", "prompt_builder.documents")

# Prompt Builder の出力を Generator に接続
pipeline.connect("prompt_builder.prompt", "generator.prompt")

実際にパイプラインを実行してみましょう。

pythonCopy code# クエリの実行
query = "Haystack でハイブリッド検索を実装する方法は？"

# パイプラインに入力を渡す
result = pipeline.run({
    "text_embedder": {"text": query},
    "bm25_retriever": {"query": query},
    "prompt_builder": {"query": query}
})

# 結果の取得
answer = result["generator"]["replies"][0]
retrieved_docs = result["ranker"]["documents"]

print(f"回答: {answer}")
print(f"\n関連ドキュメント数: {len(retrieved_docs)}")

検索結果を分析し、各文書のスコアを確認できます。

pythonCopy code# 各ドキュメントのスコアを表示
for idx, doc in enumerate(retrieved_docs, 1):
    print(f"\n[{idx}] スコア: {doc.score:.4f}")
    print(f"内容: {doc.content[:100]}...")
    print(f"ソース: {doc.meta.get('source', 'unknown')}")

パフォーマンス測定

再ランキングの効果を定量的に測定するためのコードを実装します。

pythonCopy code# 評価用のデータセット準備
test_queries = [
    "RAG システムとは何ですか？",
    "ハイブリッド検索の利点は？",
    "再ランキングの仕組みを教えてください"
]

# 正解ドキュメント ID のマッピング
ground_truth = {
    test_queries[0]: ["doc1", "doc2"],
    test_queries[1]: ["doc2", "doc5"],
    test_queries[2]: ["doc3", "doc7"]
}

Recall@K と NDCG の計算ロジックを実装します。

pythonCopy code# メトリクス計算関数
def calculate_recall_at_k(retrieved_docs, relevant_docs, k=5):
    """
    Recall@K を計算
    retrieved_docs: 検索結果のドキュメント ID リスト
    relevant_docs: 正解のドキュメント ID リスト
    """
    top_k = retrieved_docs[:k]
    relevant_found = len(set(top_k) & set(relevant_docs))
    recall = relevant_found / len(relevant_docs)
    return recall

NDCG（Normalized Discounted Cumulative Gain）の実装です。

pythonCopy codeimport math

def calculate_ndcg_at_k(retrieved_docs, relevant_docs, k=5):
    """
    NDCG@K を計算
    上位の関連文書ほど高いスコアを与える
    """
    dcg = 0.0
    for i, doc_id in enumerate(retrieved_docs[:k], 1):
        if doc_id in relevant_docs:
            dcg += 1.0 / math.log2(i + 1)

    # 理想的な DCG を計算
    idcg = sum(1.0 / math.log2(i + 1) for i in range(1, min(k, len(relevant_docs)) + 1))

    return dcg / idcg if idcg > 0 else 0.0

実際に評価を実行し、再ランキングの効果を測定します。

pythonCopy code# 再ランキングあり/なしでの比較
results_with_reranking = []
results_without_reranking = []

for query in test_queries:
    # 再ランキングありで実行
    result_with = pipeline.run({
        "text_embedder": {"text": query},
        "bm25_retriever": {"query": query}
    })

    # 再ランキングなし（Joiner の直後まで）の結果を取得
    result_without = result_with["document_joiner"]["documents"]

    # メトリクス計算
    docs_with_ids = [doc.meta["source"] for doc in result_with["ranker"]["documents"]]
    docs_without_ids = [doc.meta["source"] for doc in result_without[:5]]

    relevant = ground_truth[query]

    recall_with = calculate_recall_at_k(docs_with_ids, relevant)
    recall_without = calculate_recall_at_k(docs_without_ids, relevant)

    ndcg_with = calculate_ndcg_at_k(docs_with_ids, relevant)
    ndcg_without = calculate_ndcg_at_k(docs_without_ids, relevant)

    print(f"\nクエリ: {query}")
    print(f"Recall@5 - 再ランキングあり: {recall_with:.2%}")
    print(f"Recall@5 - 再ランキングなし: {recall_without:.2%}")
    print(f"NDCG@5 - 再ランキングあり: {ndcg_with:.4f}")
    print(f"NDCG@5 - 再ランキングなし: {ndcg_without:.4f}")

本番環境への展開

Haystack パイプラインは YAML 形式でシリアライズでき、本番環境での管理が容易です。

pythonCopy code# パイプラインを YAML にエクスポート
pipeline_yaml = pipeline.dumps()

# ファイルに保存
with open("hybrid_search_pipeline.yaml", "w") as f:
    f.write(pipeline_yaml)

print("パイプライン設定を保存しました")

保存したパイプラインを読み込んで再利用できます。

pythonCopy code# 保存されたパイプラインを読み込み
from haystack import Pipeline

with open("hybrid_search_pipeline.yaml", "r") as f:
    pipeline_config = f.read()

# パイプラインを復元
loaded_pipeline = Pipeline.loads(pipeline_config)

# すぐに使用可能
result = loaded_pipeline.run({"query": "テストクエリ"})

Docker コンテナ化のための Dockerfile 例も見てみましょう。

dockerfileCopy code# Haystack アプリケーション用 Dockerfile
FROM python:3.10-slim

# 作業ディレクトリ設定
WORKDIR /app

# 依存パッケージのインストール
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

アプリケーションファイルとパイプライン設定をコピーし、実行します。

dockerfileCopy code# アプリケーションファイルをコピー
COPY app.py .
COPY hybrid_search_pipeline.yaml .

# ポート公開
EXPOSE 8000

# アプリケーション起動
CMD ["python", "app.py"]

FastAPI を使った REST API サーバーの実装例です。

pythonCopy code# FastAPI でパイプラインを公開
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# パイプラインをロード（起動時に1回だけ実行）
with open("hybrid_search_pipeline.yaml", "r") as f:
    pipeline = Pipeline.loads(f.read())

# リクエストモデル定義
class QueryRequest(BaseModel):
    query: str
    top_k: int = 5

検索エンドポイントを実装し、外部からアクセス可能にします。

pythonCopy code# 検索エンドポイント
@app.post("/search")
async def search(request: QueryRequest):
    """
    ハイブリッド検索 API
    """
    try:
        result = pipeline.run({
            "text_embedder": {"text": request.query},
            "bm25_retriever": {"query": request.query},
            "prompt_builder": {"query": request.query}
        })

        return {
            "answer": result["generator"]["replies"][0],
            "documents": [
                {
                    "content": doc.content,
                    "score": doc.score,
                    "source": doc.meta.get("source")
                }
                for doc in result["ranker"]["documents"][:request.top_k]
            ]
        }
    except Exception as e:
        return {"error": str(e)}

まとめ

Haystack を活用したハイブリッド検索と再ランキングの実装について、詳しく解説してきました。

RAG システムの検索精度を向上させるには、単一の検索手法に頼るのではなく、Sparse 検索と Dense 検索を組み合わせたハイブリッドアプローチが効果的です。実際のデータでは、ハイブリッド検索により検索精度が 30% 向上した事例も報告されています。

さらに再ランキングを導入することで、異なるスコアリング基準で取得された文書を統一的に評価し、真に関連性の高い情報を LLM に提供できます。NVIDIA NeMo Retriever などの高度な再ランキングモデルを使用すれば、Recall@5 が 6〜7% 改善されるでしょう。

Haystack のモジュール設計により、各コンポーネントを独立して選択・交換できるため、プロジェクトの要件に応じた最適な構成を柔軟に実現できます。YAML によるシリアライズ機能や Kubernetes 対応により、本番環境への展開も容易ですね。

今後 RAG システムを構築される際は、ぜひハイブリッド検索と再ランキングの組み合わせを検討してみてください。初期の実装コストは若干増えますが、得られる検索精度の向上は、ユーザー体験の大幅な改善につながります。

まずは小規模なプロトタイプから始め、評価指標（Recall、NDCG など）を測定しながら、段階的にシステムを最適化していくアプローチがおすすめです。

関連リンク

• Haystack 公式サイト
• Haystack GitHub リポジトリ
• Haystack ドキュメント
• NVIDIA NeMo Retriever との統合ガイド
• O'Reilly Guide - RAG in Production with Haystack
• Haystack チュートリアル - RAG パイプライン構築