Building a RAG pipeline with Haystack and n4n.ai

A haystack rag pipeline n4n.ai setup lets you prototype retrieval-augmented generation against 240+ models through a single OpenAI-compatible endpoint. This guide walks through a working Haystack 2.x pipeline that indexes local Markdown, retrieves with local embeddings, and generates answers via that gateway. You’ll end with a runnable script and a clear way to verify retrieval grounding.

Step 1: Set up your environment and install Haystack

Create a clean virtual environment and install the packages you need. Haystack 2.x ships core orchestration in haystack-ai; embedders and the OpenAI generator live in the same package but pull in sentence-transformers for local vectorization.

python -m venv venv
source venv/bin/activate
pip install haystack-ai sentence-transformers markdown

Set your gateway key as an environment variable. The endpoint uses a standard API key passed through the Authorization header, exactly like OpenAI’s API.

export N4N_API_KEY="sk-your-key-here"

Verify the install by importing Haystack and checking the document store class exists:

from haystack.document_stores.in_memory import InMemoryDocumentStore
print(InMemoryDocumentStore)  # should print class object, no error

Step 2: Configure the generator for the gateway

Building a haystack rag pipeline n4n.ai requires pointing Haystack’s OpenAI-compatible chat generator at the gateway’s base URL. Haystack’s OpenAIChatGenerator accepts an api_base_url parameter, so no custom client wrapper is necessary.

import os
from haystack.components.generators.openai import OpenAIChatGenerator

generator = OpenAIChatGenerator(
    api_key=os.environ["N4N_API_KEY"],
    api_base_url="https://api.n4n.ai/v1",
    model="openai/gpt-4o-mini",
    generation_kwargs={"temperature": 0.1}
)

The gateway forwards provider cache-control hints and applies automatic fallback if the underlying provider is rate-limited or degraded, so a single model string resolves without extra branching in your code. Per-token usage is metered on the gateway side; your Haystack pipeline does not need to instrument it.

One practical note: keep temperature low for RAG. You want deterministic grounding, not creative drift. If you later swap to a different provider model, only the model argument changes.

Step 3: Prepare and index documents

For prototyping, InMemoryDocumentStore avoids external services. The indexing pipeline converts Markdown to Haystack Document objects, cleans whitespace, splits into overlapping chunks, embeds them locally, and writes to the store.

from haystack import Pipeline
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.converters import MarkdownToDocument
from haystack.components.preprocessors import DocumentCleaner, RecursiveCharacterTextSplitter
from haystack.components.embedders import SentenceTransformersDocumentEmbedder
from haystack.components.writers import DocumentWriter

doc_store = InMemoryDocumentStore()

indexing = Pipeline()
indexing.add_component("converter", MarkdownToDocument())
indexing.add_component("cleaner", DocumentCleaner())
indexing.add_component("splitter", RecursiveCharacterTextSplitter(split_length=200, split_overlap=20))
indexing.add_component("embedder", SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"))
indexing.add_component("writer", DocumentWriter(doc_store))

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

indexing.run({"converter": {"sources": ["./docs/architecture.md", "./docs/api.md"]}})
print(f"Indexed {doc_store.count_documents()} chunks")

The all-MiniLM-L6-v2 model produces 384-dimensional vectors and runs comfortably on CPU. RecursiveCharacterTextSplitter respects paragraph and sentence boundaries better than fixed-size cuts. Run this script once to populate the store; for a repeatable build, wrap it in a function that drops and recreates the store.

If your ./docs folder is empty, the pipeline will index zero documents and the later query will fail loudly—that’s the correct signal something is misconfigured.

Step 4: Build the query pipeline

The query side mirrors indexing: embed the question, retrieve top-k chunks, pack them into a prompt, and call the generator. Reuse the generator instance from Step 2.

from haystack import Pipeline
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.components.builders import PromptBuilder

template = """
You are a technical assistant. Use only the provided documents to answer.

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

Question: {{ question }}
Answer concisely. If the documents do not contain the answer, say 'unknown'.
"""

query_pipe = Pipeline()
query_pipe.add_component("text_embedder", SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"))
query_pipe.add_component("retriever", InMemoryEmbeddingRetriever(doc_store, top_k=3))
query_pipe.add_component("prompt_builder", PromptBuilder(template=template))
query_pipe.add_component("generator", generator)

query_pipe.connect("text_embedder", "retriever")
query_pipe.connect("retriever", "prompt_builder")
query_pipe.connect("prompt_builder", "generator")

The PromptBuilder renders Jinja2 syntax; documents and question are the output/input names defined by the connected components. Because the gateway is OpenAI-compatible, the same generator works for any model it routes to—change model to "anthropic/claude-3-haiku" and the pipeline code stays identical.

Step 5: Run a query and verify success

Execute the pipeline and assert on both retrieval and generation. Verification means the retriever returned non-empty contexts and the generator produced a non-empty string grounded in those contexts.

question = "What are the rate limit headers in the API?"
result = query_pipe.run({
    "text_embedder": {"text": question},
    "prompt_builder": {"question": question}
})

retrieved = result["retriever"]["documents"]
answer = result["generator"]["replies"][0]

assert len(retrieved) > 0, "Retriever returned no documents"
assert answer.strip(), "Generator returned empty answer"
print(f"Retrieved {len(retrieved)} docs")
for i, doc in enumerate(retrieved):
    print(f"--- chunk {i} ---\n{doc.content[:200]}")
print("Answer:", answer)

If you see Retrieved 3 docs and a coherent answer that cites details from your Markdown, the pipeline works. For stricter grounding, compute token overlap between answer and retrieved[].content, or run Haystack’s FaithfulnessEvaluator on a labeled question set. That evaluator scores whether the answer is supported by the retrieved context—a useful CI gate before shipping.

Step 6: Tune chunking and prompt structure

Default split_length=200 is arbitrary. Smaller chunks (100–150 words) improve retrieval precision but increase embedder calls and may split tables mid-row. Larger chunks (300–400) reduce calls but risk exceeding the model’s context window once you concatenate top_k of them.

The prompt template controls hallucination. The “say unknown” directive in Step 4 forces the model to abstain when retrieval misses. Because the gateway meters per token, verbose system prompts and long document dumps cost real money at scale—keep the template tight and strip low-value document metadata before packing.

Test retrieval quality directly by printing retriever scores (cosine similarity). If top scores are below 0.3 on a normalized index, your embedding model or chunk size is mismatched to the query style.

Step 7: Production hardening

Move the document store to a persistent backend (Chroma, Qdrant, or Elasticsearch) before serving real traffic. The gateway already handles provider fallback and cache-control forwarding, so your main task is to make the Haystack pipeline stateless: load the generator as a singleton with a shared HTTP connection pool, and run indexing as a batch job triggered by docs changes.

Expose the query pipeline behind a thin FastAPI endpoint:

from fastapi import FastAPI

app = FastAPI()

@app.post("/ask")
def ask(q: str):
    res = query_pipe.run({"text_embedder": {"text": q}, "prompt_builder": {"question": q}})
    return {"answer": res["generator"]["replies"][0], "docs": len(res["retriever"]["documents"])}

Add a startup health check that sends a minimal completion to the gateway. If it fails, crash the process instead of serving degraded responses. Log the usage field from the generator response if you need per-request cost attribution downstream.

That is a complete path from zero to a verified Haystack RAG pipeline. The same structure ports to other Haystack integrations—only the generator endpoint and model string change.