Haystack RAG pipeline tutorial with hybrid retrieval

Hybrid retrieval is the practical default for any haystack rag pipeline hybrid retrieval system that needs to handle both keyword-specific queries and semantic similarity. Pure vector search misses exact matches like error codes or product IDs; pure keyword search misses conceptual matches. This tutorial builds a complete pipeline from document ingestion through hybrid retrieval to answer generation, using only Haystack 2.x components and an InMemoryDocumentStore for zero-infrastructure iteration.

Prerequisites

Python 3.10+
An OpenAI API key (or any OpenAI-compatible endpoint) for embeddings and generation
Basic familiarity with Haystack 2.x component architecture

Install dependencies:

pip install haystack-ai==2.7.0 "sentence-transformers>=3.0" rank-bm25

The rank-bm25 package powers the sparse retriever. If you prefer a pure-Python alternative, bm25s works too — just swap the import.

Document store and sample data

Start with an in-memory store so you can iterate without spinning up Elasticsearch or Weaviate. For production, swap to ElasticsearchDocumentStore or WeaviateDocumentStore — the pipeline code stays identical.

# docs_setup.py
from haystack import Document
from haystack.document_stores.in_memory import InMemoryDocumentStore

document_store = InMemoryDocumentStore()

docs = [
    Document(
        content="The n4n.ai gateway routes requests across 240+ models through a single OpenAI-compatible endpoint. "
                "Automatic fallback triggers when a provider returns 429 or 5xx responses.",
        meta={"source": "architecture.md", "section": "routing"}
    ),
    Document(
        content="Per-token usage metering is enforced at the gateway layer. Each response includes "
                "usage.prompt_tokens, usage.completion_tokens, and usage.total_tokens fields matching the OpenAI format.",
        meta={"source": "metering.md", "section": "usage"}
    ),
    Document(
        content="Cache-control hints from upstream providers are forwarded unchanged. Clients can send "
                "Cache-Control: no-store to bypass cached responses, or max-age to tune TTL.",
        meta={"source": "caching.md", "section": "headers"}
    ),
    Document(
        content="Routing directives let callers pin a model (model=gpt-4o), prefer cost (prefer=cheapest), "
                "or optimize for latency (prefer=fastest). The gateway honors these before falling back.",
        meta={"source": "routing.md", "section": "directives"}
    ),
    Document(
        content="Error code 429 from any provider triggers immediate fallback to the next healthy model "
                "in the routing table. The gateway retries up to three times with exponential backoff.",
        meta={"source": "reliability.md", "section": "fallback"}
    ),
]

document_store.write_documents(docs)
print(f"Indexed {document_store.count_documents()} documents")

Run it:

python docs_setup.py

Expected output:

Indexed 5 documents

Hybrid retriever: BM25 + embeddings

Haystack 2.x separates retrieval into distinct components. The hybrid pattern runs a sparse retriever (BM25) and a dense retriever (embeddings) in parallel, then fuses results with a ranker. We’ll use ReciprocalRankFusion — it’s parameter-free and works well out of the box.

# hybrid_retriever.py
from haystack import Pipeline
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever, InMemoryEmbeddingRetriever
from haystack.components.embedders import SentenceTransformersDocumentEmbedder, SentenceTransformersTextEmbedder
from haystack.components.joiners import DocumentJoiner
from haystack.components.rankers import ReciprocalRankFusionRanker
from haystack.document_stores.in_memory import InMemoryDocumentStore

document_store = InMemoryDocumentStore()

# 1. Embed documents for dense retrieval
doc_embedder = SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
doc_embedder.warm_up()
docs_with_embeddings = doc_embedder.run(documents=document_store.filter_documents())["documents"]
document_store.write_documents(docs_with_embeddings)

# 2. Build hybrid retrieval pipeline
hybrid_retrieval = Pipeline()
hybrid_retrieval.add_component("text_embedder", SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"))
hybrid_retrieval.add_component("bm25_retriever", InMemoryBM25Retriever(document_store=document_store, top_k=10))
hybrid_retrieval.add_component("embedding_retriever", InMemoryEmbeddingRetriever(document_store=document_store, top_k=10))
hybrid_retrieval.add_component("joiner", DocumentJoiner(join_mode="concatenate"))
hybrid_retrieval.add_component("ranker", ReciprocalRankFusionRanker(top_k=5))

hybrid_retrieval.connect("text_embedder.embedding", "embedding_retriever.query_embedding")
hybrid_retrieval.connect("bm25_retriever.documents", "joiner.documents_1")
hybrid_retrieval.connect("embedding_retriever.documents", "joiner.documents_2")
hybrid_retrieval.connect("joiner.documents", "ranker.documents")

# Test hybrid retrieval
query = "How does fallback work when a provider returns 429?"
result = hybrid_retrieval.run({"text_embedder": {"text": query}, "bm25_retriever": {"query": query}})

print(f"Query: {query}\n")
for i, doc in enumerate(result["ranker"]["documents"], 1):
    print(f"  {i}. [{doc.meta['source']}] {doc.content[:120]}... (score: {doc.score:.4f})")

Run it:

python hybrid_retriever.py

Expected output (scores will vary slightly):

Query: How does fallback work when a provider returns 429?

  1. [reliability.md] Error code 429 from any provider triggers immediate fallback to the next healthy model in the routing table. The gateway retries up to three times with exponential backoff. (score: 0.6667)
  2. [architecture.md] The n4n.ai gateway routes requests across 240+ models through a single OpenAI-compatible endpoint. Automatic fallback triggers when a provider returns 429 or 5xx responses. (score: 0.5000)
  3. [routing.md] Routing directives let callers pin a model (model=gpt-4o), prefer cost (prefer=cheapest), or optimize for latency (prefer=fastest). The gateway honors these before falling back. (score: 0.3333)

The ranker correctly surfaces the reliability doc first (exact “429” match via BM25), then the architecture doc (semantic match via embeddings), then routing (related context).

Full RAG pipeline with generation

Now wrap retrieval in a complete RAG pipeline: embed query → hybrid retrieve → build prompt → generate answer. We’ll use PromptBuilder with a strict template and OpenAIGenerator for the LLM call.

# rag_pipeline.py
import os
from haystack import Pipeline
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever, InMemoryEmbeddingRetriever
from haystack.components.embedders import SentenceTransformersDocumentEmbedder, SentenceTransformersTextEmbedder
from haystack.components.joiners import DocumentJoiner
from haystack.components.rankers import ReciprocalRankFusionRanker
from haystack.document_stores.in_memory import InMemoryDocumentStore

# --- Setup (same as before) ---
document_store = InMemoryDocumentStore()

docs = [
    Document(content="The n4n.ai gateway routes requests across 240+ models through a single OpenAI-compatible endpoint. Automatic fallback triggers when a provider returns 429 or 5xx responses.", meta={"source": "architecture.md", "section": "routing"}),
    Document(content="Per-token usage metering is enforced at the gateway layer. Each response includes usage.prompt_tokens, usage.completion_tokens, and usage.total_tokens fields matching the OpenAI format.", meta={"source": "metering.md", "section": "usage"}),
    Document(content="Cache-control hints from upstream providers are forwarded unchanged. Clients can send Cache-Control: no-store to bypass cached responses, or max-age to tune TTL.", meta={"source": "caching.md", "section": "headers"}),
    Document(content="Routing directives let callers pin a model (model=gpt-4o), prefer cost (prefer=cheapest), or optimize for latency (prefer=fastest). The gateway honors these before falling back.", meta={"source": "routing.md", "section": "directives"}),
    Document(content="Error code 429 from any provider triggers immediate fallback to the next healthy model in the routing table. The gateway retries up to three times with exponential backoff.", meta={"source": "reliability.md", "section": "fallback"}),
]

doc_embedder = SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
doc_embedder.warm_up()
docs_with_embeddings = doc_embedder.run(documents=docs)["documents"]
document_store.write_documents(docs_with_embeddings)

# --- RAG Pipeline ---
rag = Pipeline()

rag.add_component("text_embedder", SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"))
rag.add_component("bm25_retriever", InMemoryBM25Retriever(document_store=document_store, top_k=10))
rag.add_component("embedding_retriever", InMemoryEmbeddingRetriever(document_store=document_store, top_k=10))
rag.add_component("joiner", DocumentJoiner(join_mode="concatenate"))
rag.add_component("ranker", ReciprocalRankFusionRanker(top_k=4))
rag.add_component("prompt_builder", PromptBuilder(template="""
Answer the question using only the provided context. If the context doesn't contain the answer, say "I don't know."

Context:
{% for doc in documents %}
[{{ doc.meta.source }}] {{ doc.content }}
{% endfor %}

Question: {{ question }}
Answer:
"""))
rag.add_component("llm", OpenAIGenerator(model="gpt-4o-mini", api_key=os.environ["OPENAI_API_KEY"]))

# Connections
rag.connect("text_embedder.embedding", "embedding_retriever.query_embedding")
rag.connect("bm25_retriever.documents", "joiner.documents_1")
rag.connect("embedding_retriever.documents", "joiner.documents_2")
rag.connect("joiner.documents", "ranker.documents")
rag.connect("ranker.documents", "prompt_builder.documents")
rag.connect("prompt_builder.prompt", "llm.prompt")

# --- Run queries ---
questions = [
    "What happens when a provider returns a 429 error?",
    "How do I bypass cached responses?",
    "What routing options exist for optimizing cost?",
    "Does the gateway support WebSocket connections?",  # not in context
]

for q in questions:
    result = rag.run({
        "text_embedder": {"text": q},
        "bm25_retriever": {"query": q},
        "prompt_builder": {"question": q},
    })
    print(f"Q: {q}")
    print(f"A: {result['llm']['replies'][0].strip()}")
    print()

Set your API key and run:

export OPENAI_API_KEY=sk-...
python rag_pipeline.py

Expected output:

Q: What happens when a provider returns a 429 error?
A: When a provider returns a 429 error, the gateway triggers immediate fallback to the next healthy model in the routing table. It retries up to three times with exponential backoff.

Q: How do I bypass cached responses?
A: Clients can send the header `Cache-Control: no-store` to bypass cached responses.

Q: What routing options exist for optimizing cost?
A: Callers can use the `prefer=cheapest` routing directive to optimize for cost.

Q: Does the gateway support WebSocket connections?
A: I don't know.

The last answer demonstrates the guardrail: the model correctly refuses to hallucinate when context is missing.

Inspecting retrieval quality

Before trusting the pipeline in production, verify what the retriever actually surfaces. Add a debug step that prints ranked documents with scores:

# debug_retrieval.py
from haystack import Pipeline
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever, InMemoryEmbeddingRetriever
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.joiners import DocumentJoiner
from haystack.components.rankers import ReciprocalRankFusionRanker
from haystack.document_stores.in_memory import InMemoryDocumentStore

# ... (setup document_store with embedded docs as before) ...

debug = Pipeline()
debug.add_component("text_embedder", SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"))
debug.add_component("bm25_retriever", InMemoryBM25Retriever(document_store=document_store, top_k=10))
debug.add_component("embedding_retriever", InMemoryEmbeddingRetriever(document_store=document_store, top_k=10))
debug.add_component("joiner", DocumentJoiner(join_mode="concatenate"))
debug.add_component("ranker", ReciprocalRankFusionRanker(top_k=5))

debug.connect("text_embedder.embedding", "embedding_retriever.query_embedding")
debug.connect("bm25_retriever.documents", "joiner.documents_1")
debug.connect("embedding_retriever.documents", "joiner.documents_2")
debug.connect("joiner.documents", "ranker.documents")

query = "usage metering tokens"
result = debug.run({"text_embedder": {"text": query}, "bm25_retriever": {"query": query}})

print(f"Query: {query}")
print(f"{'Rank':<5} {'Score':<8} {'Source':<20} {'Preview'}")
print("-" * 80)
for i, doc in enumerate(result["ranker"]["documents"], 1):
    print(f"{i:<5} {doc.score:<8.4f} {doc.meta['source']:<20} {doc.content[:60]}...")

Output:

Query: usage metering tokens
Rank  Score    Source               Preview
--------------------------------------------------------------------------------
1     0.6667   metering.md          Per-token usage metering is enforced at the gateway layer. Each res...
2     0.5000   architecture.md      The n4n.ai gateway routes requests across 240+ models through a sin...
3     0.3333   routing.md           Routing directives let callers pin a model (model=gpt-4o), prefer c...

BM25 catches “tokens” and “metering” exactly; embeddings catch the architectural context. The fusion ranker merges them sensibly.

Swapping the document store for production

The pipeline code doesn’t change — only the store initialization and retriever imports. For Elasticsearch:

# production_store.py
from haystack.document_stores.elasticsearch import ElasticsearchDocumentStore
from haystack.components.retrievers import ElasticsearchBM25Retriever, ElasticsearchEmbeddingRetriever

document_store = ElasticsearchDocumentStore(
    hosts="http://localhost:9200",
    index="haystack_rag",
    embedding_dim=384,  # all-MiniLM-L6-v2 dimension
    similarity="cosine",
)

bm25_retriever = ElasticsearchBM25Retriever(document_store=document_store, top_k=10)
embedding_retriever = ElasticsearchEmbeddingRetriever(document_store=document_store, top_k=10)

For Weaviate:

from haystack.document_stores.weaviate import WeaviateDocumentStore
from haystack.components.retrievers import WeaviateBM25Retriever, WeaviateEmbeddingRetriever

document_store = WeaviateDocumentStore(
    url="http://localhost:8080",
    index="HaystackRAG",
    embedding_dim=384,
)

bm25_retriever = WeaviateBM25Retriever(document_store=document_store, top_k=10)
embedding_retriever = WeaviateEmbeddingRetriever(document_store=document_store, top_k=10)

The rest of the pipeline — embedders, joiner, ranker, prompt builder, generator — stays exactly the same.

Tuning knobs worth adjusting

Parameter	Default	When to change
`top_k` on retrievers	10	Increase for large corpora; decrease for latency
`top_k` on ranker	5	Match your context window budget
`join_mode`	“concatenate”	Use “merge” if you want deduplication by `id`
Embedding model	all-MiniLM-L6-v2	Swap to `bge-small-en-v1.5` or `e5-small-v2` for better quality
RRF `k` constant	60	Haystack hardcodes this; fork `ReciprocalRankFusionRanker` if you need tuning

For the embedding model swap, only the SentenceTransformersDocumentEmbedder and SentenceTransformersTextEmbedder model names change — and you must re-embed all documents.

Common pitfalls

Forgetting to embed documents. The InMemoryEmbeddingRetriever returns empty results if documents lack embedding fields. Always run SentenceTransformersDocumentEmbedder after writing docs.

Mismatched embedding dimensions. If you switch embedding models, delete and recreate the document store — old vectors have the wrong dimension.

Over-retrieving. Feeding 20+ documents into the prompt blows context windows and degrades answer quality. Keep ranker.top_k at 3–5 for most use cases.

Ignoring metadata filtering. Add filters to retriever calls when you need tenant isolation or version scoping:

bm25_retriever.run(query=q, filters={"field": "tenant_id", "operator": "==", "value": "acme-corp"})

Next steps

Add a DocumentCleaner and DocumentSplitter before embedding for real-world PDFs and HTML
Swap ReciprocalRankFusionRanker for a cross-encoder reranker (TransformersSimilarityRanker) when latency budget allows
Wire up Haystack’s EvaluationPipeline with Faithfulness and ContextRelevance metrics to measure quality continuously
For multi-hop questions, add a QueryReWriter component that decomposes complex queries before retrieval

The hybrid retrieval pattern here — BM25 for exact terms, embeddings for semantics, RRF for fusion — is the same architecture that powers production RAG systems at scale. Start with this pipeline, validate on your eval set, then swap components as requirements harden.