はじめに
Haystack は、Python で検索エンジンや質問応答システム、RAG（Retrieval‑Augmented Generation）アプリ、AIエージェントを構築するためのオープンソースフレームワークです。ユーザーからの質問に対して、まず自分のドキュメントストアから関連文書を検索し、その内容をもとに LLM や BERT などが自然な文章で回答を返す、という流れを少ないコードで実現できます。ここでは、Haystack の基本用語を日本語で丁寧に説明しながら、15章構成で徐々に本格的な RAG システムへ発展させる流れを紹介します。
第1章 Haystackとは何か ― 目的と特徴
Haystack は、「質問応答や検索を、自分のデータに対して行えるようにする」ことを目的としたフレームワークです。一般的なチャットボットはウェブ全体などで学習されたモデルに依存しますが、Haystack を使うと、自分の PDF・社内Wiki・マニュアルなどを検索対象として取り込み、その内容に基づく回答を返すシステムを作れます。構成要素を組み合わせるスタイルなので、小さな PoC から大規模な本番システムまで、スケールに応じて柔軟に設計できる点が大きな特徴です。
# 最小限のインポート例
import haystack
print("Haystack version:", haystack.__version__)
第2章 基本用語 ― Node・Component・Pipeline
Haystack では、処理の最小単位を「コンポーネント」あるいは「ノード」と呼び、それらをつなぎ合わせたものが「パイプライン」です。コンポーネントは、ドキュメントを保存する DocumentStore、類似文書を検索する Retriever、埋め込みを作る Embedder、LLM を呼び出す PromptNode など役割ごとに分かれています。パイプラインは、これらのノード同士を「どのノードの出力を次のノードに渡すか」という形で接続したもので、クエリを入力すると内部の流れに従って結果が返ってきます。
from haystack import Pipeline
from haystack.document_stores import InMemoryDocumentStore
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers import InMemoryEmbeddingRetriever
document_store = InMemoryDocumentStore()
embedder = SentenceTransformersTextEmbedder(model_name="sentence-transformers/all-MiniLM-L6-v2")
retriever = InMemoryEmbeddingRetriever(document_store=document_store)
pipeline = Pipeline()
pipeline.add_component("embedder", embedder)
pipeline.add_component("retriever", retriever)
pipeline.connect("embedder", "retriever")
第3章 Document と DocumentStore ― 知識を格納する仕組み
Haystack で扱うテキストは Document というオブジェクトで表現され、本文を表す content と、タイトルやタグなど任意の情報を入れられる meta、一意な id を持ちます。これら Document を多数保存する場所が DocumentStore で、メモリ上に保存する簡易版から、Elasticsearch・OpenSearch・Milvus などの外部ベクタDBを使う大規模版まで、用途に応じた実装が提供されています。まずは InMemoryDocumentStore に少量の文書を入れて振る舞いを理解し、慣れてきたら外部ストアに切り替えるのが現実的なステップです。
from haystack import Document
from haystack.document_stores import InMemoryDocumentStore
document_store = InMemoryDocumentStore()
docs = [
Document(content="HaystackはRAGアプリを構築するPythonフレームワークです。", meta={"source": "intro"}),
Document(content="DocumentStoreはテキストを保存し、検索のためのインデックスを管理します。", meta={"source": "docs"}),
]
document_store.write_documents(docs)
print("保存された件数:", document_store.count_documents())
第4章 Embedding と Retriever ― 類似検索の基礎
RAG の中心となるのが「埋め込み（Embedding）」と「リトリーバ（Retriever）」です。Embeddings は文章をベクトル化したもので、意味的に近い文章同士が近い位置に配置されるよう学習されています。Retriever は、クエリベクトルとストア内の文書ベクトルを比較し、最も似ている文書を上位 K 件返す役割を持ちます。Haystack では SentenceTransformers 系のモデルと InMemoryEmbeddingRetriever を組み合わせて、シンプルな類似検索をすぐに試せます。
from haystack.document_stores import InMemoryDocumentStore
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers import InMemoryEmbeddingRetriever
from haystack import Document
store = InMemoryDocumentStore()
store.write_documents([
Document(content="Haystackは質問応答や検索システムに使われます。"),
Document(content="PythonでRAGシステムを作るときにHaystackは有用です。"),
])
embedder = SentenceTransformersTextEmbedder(model_name="sentence-transformers/all-MiniLM-L6-v2")
retriever = InMemoryEmbeddingRetriever(document_store=store)
docs = store.get_all_documents()
embs = embedder.run({"texts": [d.content for d in docs]})["embeddings"]
store.update_embeddings(docs, embs)
query = "Haystackで質問応答を作りたい"
q_emb = embedder.run({"texts": [query]})["embeddings"][0]
result = retriever.run({"query_embedding": q_emb, "top_k": 2})
for d in result["documents"]:
print("ヒット:", d.content)
第5章 PromptNode と PromptTemplate ― LLMへの橋渡し
LLM を使って最終的な文章回答を生成するのが PromptNode で、そのひな型となるプロンプトを表現するのが PromptTemplate です。PromptNode はモデルの種類（OpenAI や HuggingFace など）と API キーを指定して初期化し、テンプレート内に {context} や {query} といったプレースホルダを書いておくことで、実行時にドキュメントや質問を埋め込んだプロンプトを自動生成してくれます。この組み合わせにより、「常に日本語で答える」「必ず根拠を明示する」といったポリシーをプロンプト設計でコントロールできます。
import os
from haystack.nodes import PromptNode, PromptTemplate
os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"
template = PromptTemplate(
prompt=(
"次のコンテキストを読み、ユーザーの質問に日本語で丁寧に答えてください。\n\n"
"コンテキスト:\n{context}\n\n"
"質問:\n{query}\n\n"
"答え:"
)
)
prompt_node = PromptNode(
model_name_or_path="gpt-4o-mini",
api_key=os.environ["OPENAI_API_KEY"],
default_prompts={"qa_ja": template},
)
ctx = "Haystackは自前のドキュメントをLLMと組み合わせるためのフレームワークです。"
q = "Haystackを使うと何ができますか？"
result = prompt_node.run(prompt_template="qa_ja", context=ctx, query=q)
print(result["replies"][0])
第6章 最初のRAG QAパイプラインを組み立てる
ここまでの要素を組み合わせると、シンプルな RAG 質問応答パイプラインを作ることができます。流れとしては、ユーザーの質問を埋め込みに変換して Retriever で類似文書を取り出し、その文書をまとめてコンテキストとして PromptNode に渡し、LLM が最終回答を生成します。パイプラインオブジェクトにコンポーネントと接続関係を登録すれば、あとは質問文字列だけを入力として渡し、結果として回答テキストと参照文書の一覧を受け取ることができます。
from haystack import Pipeline, Document
from haystack.document_stores import InMemoryDocumentStore
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers import InMemoryEmbeddingRetriever
from haystack.nodes import PromptNode, PromptTemplate
import os
store = InMemoryDocumentStore()
store.write_documents([
Document(content="HaystackはPythonでRAGシステムを構築するためのフレームワークです。"),
Document(content="DocumentStore, Retriever, PromptNodeを組み合わせてQAパイプラインを作ります。"),
])
embedder = SentenceTransformersTextEmbedder(model_name="sentence-transformers/all-MiniLM-L6-v2")
retriever = InMemoryEmbeddingRetriever(document_store=store)
docs = store.get_all_documents()
embs = embedder.run({"texts": [d.content for d in docs]})["embeddings"]
store.update_embeddings(docs, embs)
os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"
prompt_node = PromptNode(
model_name_or_path="gpt-4o-mini",
api_key=os.environ["OPENAI_API_KEY"],
default_prompts={
"rag_ja": PromptTemplate(
prompt=(
"次のドキュメントを参考に、質問に日本語で答えてください。\n\n"
"ドキュメント:\n{documents}\n\n"
"質問:\n{query}\n\n"
"答え:"
)
)
},
)
pipe = Pipeline()
pipe.add_component("embedder", embedder)
pipe.add_component("retriever", retriever)
pipe.add_component("llm", prompt_node)
pipe.connect("embedder", "retriever")
pipe.connect("retriever", "llm.documents")
question = "Haystackはどんな用途に向いていますか？"
q_emb = embedder.run({"texts": [question]})["embeddings"][0]
ret = retriever.run({"query_embedding": q_emb, "top_k": 2})
answer = prompt_node.run(
prompt_template="rag_ja",
documents="\n".join([d.content for d in ret["documents"]]),
query=question,
)
print("回答:", answer["replies"][0])
第7章 抽出型QA（Extractive QA）とReaderの役割
生成系 LLM を使った回答だけでなく、「文書の中から答えになる部分をそのまま抜き出す」抽出型 QA も Haystack の重要なユースケースです。ここでは BERT 系モデルなどを Reader として使い、Retriever が返した候補文書に対して「もっとも答えらしいスパン」を探し出します。抽出型 QA の利点は、回答の出典が文書中の具体的な位置として明確に示されることであり、監査性や根拠提示が重要な業務では特に有用です。
from haystack.document_stores import InMemoryDocumentStore
from haystack.nodes import FARMReader, BM25Retriever
from haystack.pipelines import ExtractiveQAPipeline
from haystack import Document
store = InMemoryDocumentStore()
store.write_documents([
Document(content="Pythonは高水準プログラミング言語で、Guido van Rossumによって開発されました。"),
Document(content="HaystackはPythonを使った質問応答フレームワークです。"),
])
retriever = BM25Retriever(document_store=store)
reader = FARMReader(model_name_or_path="deepset/roberta-base-squad2", use_gpu=False)
pipeline = ExtractiveQAPipeline(reader=reader, retriever=retriever)
prediction = pipeline.run(
query="Pythonを開発した人物は誰ですか？",
params={"Retriever": {"top_k": 2}, "Reader": {"top_k": 1}}
)
for ans in prediction["answers"]:
print("答え:", ans.answer)
print("スコア:", ans.score)
print("文脈:", ans.context)
第8章 RAGユーティリティとチュートリアルを活用する
Haystack には、最初の RAG パイプラインを素早く構築するためのユーティリティ関数や公式チュートリアルが用意されています。サンプルデータを自動的に DocumentStore に読み込み、推奨構成の Retriever と PromptNode を組み合わせたパイプラインをワンステップで生成する仕組みもあります。これらを使うことで、細かい設定に悩む前に「一度動くものを体験する」ことができ、そこから自分のユースケースに合わせてコンポーネントを差し替えたり、プロンプトを調整したりする流れへスムーズに移行できます。
# 擬似コード的なイメージ
from haystack.document_stores import InMemoryDocumentStore
from haystack.utils import add_example_data, build_pipeline
store = InMemoryDocumentStore()
add_example_data(store, "data/GoT_getting_started")
pipeline = build_pipeline(
provider="openai",
api_key="YOUR_API_KEY",
document_store=store,
)
result = pipeline.run(query="Who is Arya Stark's father?")
print(result)
第9章 Haystack 2系のPipelineシステムとYAML構成
Haystack 2 系では、パイプラインの入出力やノードの接続関係を introspection できるようになり、構成を YAML として保存・読み込みする仕組みも整備されています。これにより、開発環境では Python コードでパイプラインを組み、運用環境では YAML ファイルとして設定管理しつつ、必要に応じて再ロードする、といったワークフローが取りやすくなりました。特に複数のパイプラインを同じクラスタで動かす場合、構成管理をコードから切り離しておくと保守性が高まります。
from haystack import Pipeline
from haystack.components.preprocessors import DocumentCleaner
from haystack.document_stores import InMemoryDocumentStore
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers import InMemoryEmbeddingRetriever
store = InMemoryDocumentStore()
cleaner = DocumentCleaner()
embedder = SentenceTransformersTextEmbedder(model_name="sentence-transformers/all-MiniLM-L6-v2")
retriever = InMemoryEmbeddingRetriever(document_store=store)
pipe = Pipeline()
pipe.add_component("cleaner", cleaner)
pipe.add_component("embedder", embedder)
pipe.add_component("retriever", retriever)
pipe.connect("cleaner", "embedder")
pipe.connect("embedder", "retriever")
print("入力:", pipe.inputs())
print("出力:", pipe.outputs())
yaml_config = pipe.dumps()
print(yaml_config[:200], "...")
第10章 外部ベクタDBとの統合 ― MilvusやOpenSearchを使う
大規模データを扱う場合、InMemoryDocumentStore ではメモリ不足になりがちなので、Milvus や Weaviate、OpenSearch といったベクタDBをバックエンドに使う DocumentStore を選びます。これらのストアは高次元ベクトルを高速に検索するためのインデックスを提供し、数百万〜数億行規模の文書でも、類似検索をミリ秒〜数百ミリ秒で行えるよう設計されています。Haystack 側から見ると、インスタンス化時のクラスを変えるだけで同じ API を利用できるため、プロトタイプから本番スケールへ移行しやすいのが利点です。
from haystack.document_stores import MilvusDocumentStore
from haystack import Document
document_store = MilvusDocumentStore(
host="localhost",
port=19530,
index="haystack_docs",
embedding_dim=384,
)
docs = [
Document(content="Milvusを使うと大規模なベクタ検索を高速に行えます。"),
Document(content="HaystackはMilvusや他のベクタDBと連携可能です。"),
]
document_store.write_documents(docs)
print("Milvus内の件数:", document_store.count_documents())
第11章 Webサービス化 ― REST API とチャットUIへの接続
実際の利用では、Haystack のパイプラインを Web サービスとして公開し、フロントエンドのチャットUIや業務アプリから HTTP 経由で呼び出すケースが多くなります。Python では FastAPI や Flask を使ってエンドポイントを定義し、その中で pipeline.run() を呼び出して回答を生成する形が典型的です。フロント側から見ると単なる「質問を投げてテキスト回答を得る API」として扱えるため、バックエンドのベクタDB や LLM の種類を後から差し替えても、クライアントの実装を変えずに済むような設計にしやすくなります。
from fastapi import FastAPI
from pydantic import BaseModel
from haystack import Pipeline
app = FastAPI()
qa_pipeline = Pipeline.load_from_yaml("rag_pipeline.yaml")
class Query(BaseModel):
question: str
@app.post("/ask")
def ask(query: Query):
result = qa_pipeline.run({"query": query.question})
answers = result.get("answers") or result.get("replies") or []
first = answers[0] if answers else None
return {"answer": getattr(first, "answer", str(first))}
第12章 チューニングと評価 ― top_k・閾値・プロンプト設計
Haystack を実務に投入するには、Retrieval の top_k、スコア閾値、プロンプトテンプレート、LLM の温度や最大トークン数など、多くのパラメータをチューニングする必要があります。たとえば top_k を上げると回答の根拠になりうる文書が増える一方で、プロンプトが長くなりAPIコストや応答時間が増えるため、適切な折り合いをつけなければなりません。また、評価用にクエリと正解回答のペアを用意し、どの設定が最もユーザー体験を向上させるかを継続的に測る「フィードバックループ」を回すことも重要です。
from typing import List, Dict, Any
class FeedbackLogger:
def __init__(self):
self.logs: List[Dict[str, Any]] = []
def log(self, query: str, answer: str, useful: bool):
self.logs.append({"query": query, "answer": answer, "useful": useful})
def accuracy(self) -> float:
if not self.logs:
return 0.0
return sum(1 for l in self.logs if l["useful"]) / len(self.logs)
logger = FeedbackLogger()
logger.log("Haystackとは？", "RAGフレームワークの説明", True)
logger.log("別の質問", "あまり役立たない回答", False)
print("簡易的な満足度:", logger.accuracy())
第13章 公式チュートリアルとコミュニティで学ぶ
Haystack には多数の公式チュートリアルやサンプルパイプラインが用意されており、検索・抽出型QA・RAG・エージェント・外部DB連携など、ユースケースごとに段階的な学習ができるようになっています。GitHub 上のチュートリアルリポジトリには Notebook が公開されており、実際にコードを実行しながら理解を深めることが可能です。また、フォーラムや GitHub Discussions には実務でのノウハウやトラブルシューティングの議論が蓄積されているため、困ったときには過去の Q&A を検索したり自分で質問したりすることで、解決のヒントを得られます。
import subprocess
repo_url = "https://github.com/deepset-ai/haystack-tutorials.git"
subprocess.run(["git", "clone", repo_url])
print("haystack-tutorials リポジトリを取得しました。")
print("cd haystack-tutorials して Jupyter でノートブックを開きましょう。")
第14章 エージェントとツール連携 ― 高度なワークフロー
Haystack の新しいバージョンでは、検索や RAG にとどまらず、外部APIやコード実行、計算エンジンなど複数のツールを組み合わせる「エージェント」的な構成も取り扱えるようになっています。エージェントは LLM を中心に据え、ユーザーからの指示を解析して「このタスクは検索が必要」「この部分は計算が必要」といった判断を行い、適切なツールを順番に呼び出していきます。その結果、単純なQ&Aだけでなく、社内システムと対話しながら情報を集約・変換・要約するような高度なワークフローを自然な会話インターフェースで提供できます。
from haystack import Pipeline
agent_pipeline = Pipeline.load_from_yaml("agent_pipeline.yaml")
def chat():
print("Haystackエージェントとの対話を開始します。終了するには 'exit' と入力してください。")
while True:
q = input("あなた: ")
if q.strip().lower() == "exit":
break
result = agent_pipeline.run({"query": q})
answer = result.get("answer") or result.get("replies", [""])[0]
print("エージェント:", answer)
# chat()
第15章 まとめ ― 自分のデータで小さく始める
Haystack は、LLM・埋め込みモデル・ベクタDB・Webフレームワークなど現代の AI 技術をつなぐ「ハブ」のような存在であり、自分のドキュメントに対して高品質な検索・質問応答を行いたいときの強力な選択肢です。まずは InMemoryDocumentStore と少量のテキストファイルを使ってローカル環境に簡単な QA パイプラインを作り、どのように Retriever と LLM が連携するのかを体験してみると理解が深まります。その後、データ量や要件に応じて外部ベクタDB・クラウド環境・エージェント構成へと発展させていけば、業務で通用する本格的な RAG システムを段階的に育てていくことができます。
from haystack import Document, Pipeline
from haystack.document_stores import InMemoryDocumentStore
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers import InMemoryEmbeddingRetriever
from haystack.nodes import PromptNode, PromptTemplate
import os
store = InMemoryDocumentStore()
text = open("docs/manual_ja.txt", encoding="utf-8").read()
store.write_documents([Document(content=text, meta={"source": "manual"})])
embedder = SentenceTransformersTextEmbedder(model_name="sentence-transformers/all-MiniLM-L6-v2")
retriever = InMemoryEmbeddingRetriever(document_store=store)
docs = store.get_all_documents()
embs = embedder.run({"texts": [d.content for d in docs]})["embeddings"]
store.update_embeddings(docs, embs)
os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"
prompt_node = PromptNode(
model_name_or_path="gpt-4o-mini",
api_key=os.environ["OPENAI_API_KEY"],
default_prompts={
"qa_manual": PromptTemplate(
prompt=(
"以下のマニュアル内容を参考に、質問に日本語でわかりやすく答えてください。\n\n"
"マニュアル:\n{documents}\n\n"
"質問:\n{query}\n\n"
"答え:"
)
)
},
)
pipe = Pipeline()
pipe.add_component("embedder", embedder)
pipe.add_component("retriever", retriever)
pipe.add_component("llm", prompt_node)
pipe.connect("embedder", "retriever")
pipe.connect("retriever", "llm.documents")
question = "このシステムのログの確認方法を教えてください。"
q_emb = embedder.run({"texts": [question]})["embeddings"][0]
ret = retriever.run({"query_embedding": q_emb, "top_k": 3})
answer = prompt_node.run(
prompt_template="qa_manual",
documents="\n".join([d.content for d in ret["documents"]]),
query=question,
)
print("回答:", answer["replies"][0])