Haystack: Production-Ready RAG & NLP Pipeline Framework

Building a RAG Pipeline with Haystack 2.x

Define a pipeline by connecting components explicitly — indexing and querying are separate pipelines in Haystack.

pip install haystack-ai

from haystack import Document, Pipeline
from haystack.components.embedders import OpenAIDocumentEmbedder, OpenAITextEmbedder
from haystack.components.writers import DocumentWriter
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator
from haystack.document_stores.in_memory import InMemoryDocumentStore

# --- INDEXING PIPELINE ---
document_store = InMemoryDocumentStore()

docs = [
    Document(content="Haystack uses a component-based pipeline architecture.", meta={"source": "docs"}),
    Document(content="RAG combines retrieval with LLM generation for grounded answers.", meta={"source": "blog"}),
    Document(content="Hybrid search merges BM25 sparse and dense vector retrieval.", meta={"source": "docs"}),
]

indexing_pipeline = Pipeline()
indexing_pipeline.add_component("embedder",    OpenAIDocumentEmbedder(model="text-embedding-3-small"))
indexing_pipeline.add_component("writer",      DocumentWriter(document_store=document_store))

# Connect component outputs to inputs
indexing_pipeline.connect("embedder.documents", "writer.documents")

# Run indexing
indexing_pipeline.run({"embedder": {"documents": docs}})
print(f"Indexed {document_store.count_documents()} documents")

RAG Query Pipeline: Retrieve + Generate

The query pipeline embeds the user question, retrieves relevant documents, and passes them to an LLM for generation.

rag_template = """
Answer the question based on the provided documents.
If the answer is not in the documents, say "I don't know."

Documents:
{% for doc in documents %}
  - {{ doc.content }}
{% endfor %}

Question: {{ question }}
Answer:
"""

# --- QUERY PIPELINE ---
query_pipeline = Pipeline()
query_pipeline.add_component("text_embedder", OpenAITextEmbedder(model="text-embedding-3-small"))
query_pipeline.add_component("retriever",     InMemoryEmbeddingRetriever(document_store=document_store, top_k=3))
query_pipeline.add_component("prompt_builder",PromptBuilder(template=rag_template))
query_pipeline.add_component("llm",           OpenAIGenerator(model="gpt-4o-mini"))

# Wire components together
query_pipeline.connect("text_embedder.embedding",  "retriever.query_embedding")
query_pipeline.connect("retriever.documents",       "prompt_builder.documents")
query_pipeline.connect("prompt_builder.prompt",     "llm.prompt")

# Run query
result = query_pipeline.run({
    "text_embedder": {"text": "How does Haystack handle retrieval?"},
    "prompt_builder": {"question": "How does Haystack handle retrieval?"}
})
print(result["llm"]["replies"][0])

Hybrid Retrieval: BM25 + Embedding + DocumentJoiner

Run sparse and dense retrieval in parallel, then merge results with Reciprocal Rank Fusion.

from haystack.components.retrievers.in_memory import (
    InMemoryBM25Retriever,
    InMemoryEmbeddingRetriever
)
from haystack.components.joiners import DocumentJoiner
from haystack.components.rankers import MetaFieldRanker

hybrid_pipeline = Pipeline()

# Add both retriever types
hybrid_pipeline.add_component("text_embedder",    OpenAITextEmbedder(model="text-embedding-3-small"))
hybrid_pipeline.add_component("bm25_retriever",   InMemoryBM25Retriever(document_store=document_store, top_k=10))
hybrid_pipeline.add_component("embed_retriever",  InMemoryEmbeddingRetriever(document_store=document_store, top_k=10))
hybrid_pipeline.add_component("joiner",           DocumentJoiner(
    join_mode="reciprocal_rank_fusion",   # merge with RRF
    top_k=5
))
hybrid_pipeline.add_component("prompt_builder",   PromptBuilder(template=rag_template))
hybrid_pipeline.add_component("llm",              OpenAIGenerator(model="gpt-4o-mini"))

# Connect: two parallel retrieval branches → joiner → LLM
hybrid_pipeline.connect("text_embedder.embedding",  "embed_retriever.query_embedding")
hybrid_pipeline.connect("bm25_retriever.documents", "joiner.documents")
hybrid_pipeline.connect("embed_retriever.documents","joiner.documents")
hybrid_pipeline.connect("joiner.documents",          "prompt_builder.documents")
hybrid_pipeline.connect("prompt_builder.prompt",     "llm.prompt")

result = hybrid_pipeline.run({
    "text_embedder":  {"text": "hybrid retrieval BM25 vector"},
    "bm25_retriever": {"query": "hybrid retrieval BM25 vector"},
    "prompt_builder": {"question": "What is hybrid retrieval?"}
})

Haystack vs LangChain vs LlamaIndex