Haystack 2.0: Document Indexing・Hybrid Retrieval・スマートQ&Aで構築するドキュメント処理パイプライン

通常のRAGパイプラインの問題点

以前、LangChainで社内Q&Aシステムを構築したことがある — 動きはしたが、パイプラインの途中に新しい処理ステップを追加しようとしたとたん、すべてが混乱し始めた。コールバックがコールバックにネストされ、データがどこを流れているのかを理解するだけでデバッグに半日かかってしまった。

Haystack（deepset製）はまさにこの痛点を解決する。パイプラインは有向グラフ（DAG）で、各コンポーネントが明示的に接続される。前処理ステップを追加したい？コンポーネントを差し込めばいい。Embeddingモデルを交換したい？一行変えるだけ。残りのロジックには触れなくていい。

この記事では3つの実践的なパートに直接入る：適切なドキュメントのインデックス化、精度向上のためのHybrid Retrieval、そして本番対応のQ&Aパイプライン。

Quick Start — 5分で動くQ&Aパイプライン

インストール

pip install haystack-ai openai

最もシンプルなQ&Aパイプライン

from haystack import Document, Pipeline
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import PromptBuilder
from haystack.document_stores.in_memory import InMemoryDocumentStore

# 1. Document store + インデックス化
store = InMemoryDocumentStore()
store.write_documents([
    Document(content="HaystackはAIドキュメント処理パイプラインを構築するPythonフレームワークです。"),
    Document(content="BM25はキーワードベースの検索アルゴリズムで、技術文書に適しています。"),
    Document(content="Embedding retrievalはベクトル類似度を使って意味的に関連するドキュメントを検索します。"),
])

# 2. パイプラインの構築
template = """
以下のドキュメントに基づいて、日本語で質問に答えてください：

{% for doc in documents %}
{{ doc.content }}
{% endfor %}

質問: {{ question }}
"""

pipe = Pipeline()
pipe.add_component("retriever", InMemoryBM25Retriever(document_store=store))
pipe.add_component("prompt", PromptBuilder(template=template))
pipe.add_component("llm", OpenAIGenerator(model="gpt-4o-mini"))

pipe.connect("retriever", "prompt.documents")
pipe.connect("prompt", "llm")

# 3. 実行
result = pipe.run({
    "retriever": {"query": "Haystackとは何ですか？"},
    "prompt": {"question": "Haystackとは何ですか？"}
})
print(result["llm"]["replies"][0])

この構造はネストされたチェーンよりはるかに明確だ。各コンポーネントはインプットを受け取り、処理し、アウトプットを返す — パイプラインがそれらを接続する。デバッグも各ステップのテストも容易だ。

Document Indexing — 最初から正しく行う

RAGシステムの大半が低品質な結果を出す原因は、LLMが悪いからではなく、インデックス化が間違っているからだ。チャンクが大きすぎる、メタデータがない、PDFから変換したテキストがクリーニングされていない、といった問題が原因だ。

PDFの完全なIndexingパイプライン

from haystack import Pipeline
from haystack.components.converters import PyPDFToDocument
from haystack.components.preprocessors import DocumentCleaner, DocumentSplitter
from haystack.components.embedders import OpenAIDocumentEmbedder
from haystack.components.writers import DocumentWriter
from haystack.document_stores.in_memory import InMemoryDocumentStore
from pathlib import Path

store = InMemoryDocumentStore()

indexing_pipe = Pipeline()
indexing_pipe.add_component("converter", PyPDFToDocument())
indexing_pipe.add_component("cleaner", DocumentCleaner(
    remove_empty_lines=True,
    remove_extra_whitespaces=True,
))
indexing_pipe.add_component("splitter", DocumentSplitter(
    split_by="sentence",
    split_length=5,   # 1チャンクあたり5文
    split_overlap=1,  # コンテキストの欠落を防ぐため1文オーバーラップ
))
indexing_pipe.add_component("embedder", OpenAIDocumentEmbedder(
    model="text-embedding-3-small"
))
indexing_pipe.add_component("writer", DocumentWriter(document_store=store))

indexing_pipe.connect("converter", "cleaner")
indexing_pipe.connect("cleaner", "splitter")
indexing_pipe.connect("splitter", "embedder")
indexing_pipe.connect("embedder", "writer")

# ディレクトリ内のすべてのPDFをインデックス化
pdf_files = list(Path("./docs").glob("*.pdf"))
indexing_pipe.run({"converter": {"sources": pdf_files}})
print(f"{store.count_documents()}チャンクをインデックス化しました")

技術文書には、split_by="sentence"とsplit_length=5のコンビをよく使う — 小さいチャンクはretrieval精度を上げ、1文のoverlapでチャンク境界でのコンテキスト欠落を防ぐ。

後でフィルタリングするためのメタデータ追加

from haystack import Document

docs = [
    Document(
        content="Dockerに関する技術ドキュメントの内容...",
        meta={
            "source": "docker-guide.pdf",
            "category": "devops",
            "language": "vi",
        }
    )
]
store.write_documents(docs)

# カテゴリフィルタ付きでRetrieval
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
retriever = InMemoryBM25Retriever(document_store=store)
results = retriever.run(
    query="Dockerの設定",
    filters={"field": "meta.category", "operator": "==", "value": "devops"}
)

Hybrid Retrieval — BM25とEmbeddingの組み合わせ

BM25は正確なキーワードに強い：固有名詞、エラーコード、具体的なコマンド。Embedding retrievalは異なる表現でも同じ意味を持つ質問に強い。Hybridは両方を組み合わせる — これが実際にretrieval品質が明らかに向上する理由だ。

from haystack import Pipeline
from haystack.components.retrievers.in_memory import (
    InMemoryBM25Retriever,
    InMemoryEmbeddingRetriever
)
from haystack.components.embedders import OpenAITextEmbedder
from haystack.components.joiners import DocumentJoiner
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator

template = """
以下のドキュメントに基づいて、質問に答えてください：
{% for doc in documents %}
- {{ doc.content }}
{% endfor %}
質問: {{ question }}
"""

query_pipe = Pipeline()
query_pipe.add_component("text_embedder", OpenAITextEmbedder(
    model="text-embedding-3-small"
))
query_pipe.add_component("embedding_retriever", InMemoryEmbeddingRetriever(
    document_store=store, top_k=10
))
query_pipe.add_component("bm25_retriever", InMemoryBM25Retriever(
    document_store=store, top_k=10
))
query_pipe.add_component("joiner", DocumentJoiner(
    join_mode="reciprocal_rank_fusion"  # RRF: 最良のランキング結合
))
query_pipe.add_component("prompt", PromptBuilder(template=template))
query_pipe.add_component("llm", OpenAIGenerator(model="gpt-4o-mini"))

# 接続
query_pipe.connect("text_embedder.embedding", "embedding_retriever.query_embedding")
query_pipe.connect("embedding_retriever", "joiner")
query_pipe.connect("bm25_retriever", "joiner")
query_pipe.connect("joiner", "prompt.documents")
query_pipe.connect("prompt", "llm")

# 実行
result = query_pipe.run({
    "text_embedder": {"text": "コンテナが起動しないエラーをデバッグするには？"},
    "bm25_retriever": {"query": "コンテナが起動しないエラーをデバッグするには？"},
    "prompt": {"question": "コンテナが起動しないエラーをデバッグするには？"}
})
print(result["llm"]["replies"][0])

社内ナレッジベースプロジェクトでの測定結果：Hybrid retrievalはEmbeddingのみと比べてrecallを約15〜20%改善する。製品名、エラーコード、具体的なコマンドを含むクエリで差が最も顕著に出る — これがEmbeddingよりBM25が優れている部分だ。

応用 — Persistent StoreとProduction Tips

InMemoryの代わりにChromaDBを使用

pip install chroma-haystack

from chroma_haystack import ChromaDocumentStore

store = ChromaDocumentStore(
    collection_name="my_docs",
    persist_path="./chroma_db"  # 再起動後もデータが保持される
)

バージョン管理と再利用のためのパイプラインのシリアライズ

import yaml

# パイプライン設定の保存
with open("pipeline.yaml", "w") as f:
    yaml.dump(query_pipe.to_dict(), f)

# 後で読み込む — 最初から再構築不要
from haystack import Pipeline
with open("pipeline.yaml") as f:
    pipe_dict = yaml.safe_load(f)
query_pipe = Pipeline.from_dict(pipe_dict)

各コンポーネントを独立してデバッグ

# パイプライン全体を実行せずにretrieverをテスト
retriever = InMemoryBM25Retriever(document_store=store)
docs = retriever.run(query="Docker", top_k=3)

print(f"{len(docs['documents'])}件のドキュメントが見つかりました")
for doc in docs["documents"]:
    print(f"Score: {doc.score:.3f} | {doc.content[:100]}...")

これはHaystackがLangChainより優れていると感じる点だ：各コンポーネントを個別にテストできること。システムが誤った回答を返したとき、問題がindexing、retrieval、それともpromptにあるのかをすぐに特定できる — チェーン全体のログダンプは不要だ。

Haystackを選ぶ際の実践的なヒント

• Haystackを使うのは、明確なパイプラインが必要な場合、複数のドキュメントソースがある場合、パイプラインのシリアライズ/バージョン管理が必要な場合、またはチームで同時にメンテナンスする場合。
• LangChainを使うのは、多くの既存インテグレーションで素早くプロトタイプを作りたい場合や、LangChainエコシステムに慣れている場合。
• カスタムコードを使うのは、パイプラインが1〜2ステップだけで、フレームワークへの依存を避けたい場合。

混同しやすい点がある：Haystack 2.0はv1とアーキテクチャが全く異なる。読んでいるチュートリアルのAPIが違って見えたら、それは古いv1だ。現在のパッケージはhaystack-aiであり、farm-haystackではない。V2はより型安全で、パイプラインがより明示的で、デバッグがはるかに簡単だ。