n4nAI
Open console →
Framework tutorialsTutorialHaystack RAG Pipelines

Haystack PromptBuilder tutorial for RAG pipelines

Build a production-ready RAG pipeline with Haystack's PromptBuilder — from basic templates to advanced conditional logic, with runnable code at every step.

n4n Team3 min read677 words

Audio narration

Coming soon — every post will get a voice note here.

Haystack’s PromptBuilder is the component that turns retrieved documents into structured prompts for your generator. This haystack promptbuilder rag pipeline tutorial walks through building a complete RAG system: from a minimal working example to a production pipeline with citation handling, fallback logic, and token-aware truncation. You’ll end up with code you can drop into a service.

Prerequisites

Python 3.10+ and a virtual environment. Install the core packages:

pip install haystack-ai==2.6.0 \
            sentence-transformers==3.0.1 \
            rank-bm25==0.2.2 \
            openai==1.35.0

You need an OpenAI API key (or any OpenAI-compatible endpoint) for the generator. Export it:

export OPENAI_API_KEY="sk-..."

If you prefer a local model, swap the generator for HuggingFaceLocalGenerator — the PromptBuilder usage stays identical.

Minimal PromptBuilder RAG pipeline

Start with the simplest working pipeline: an in-memory document store, BM25 retrieval, and a single PromptBuilder feeding GPT-4o-mini.

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

document_store = InMemoryDocumentStore()
document_store.write_documents([
    Document(content="Haystack 2.0 introduces a component-based architecture where each piece — retriever, builder, generator — is a replaceable class."),
    Document(content="PromptBuilder uses Jinja2 templates. You can loop over documents, conditionally include sections, and access metadata."),
    Document(content="The generator receives the rendered prompt string. It does not know about documents unless you pass them in the template."),
])

retriever = InMemoryBM25Retriever(document_store=document_store, top_k=3)

template = """
Answer the question using only the provided documents.

Documents:
{% for doc in documents %}
[{{ loop.index }}] {{ doc.content }}
{% endfor %}

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

prompt_builder = PromptBuilder(template=template)
generator = OpenAIGenerator(model="gpt-4o-mini")

rag = Pipeline()
rag.add_component("retriever", retriever)
rag.add_component("prompt_builder", prompt_builder)
rag.add_component("generator", generator)

rag.connect("retriever.documents", "prompt_builder.documents")
rag.connect("prompt_builder.prompt", "generator.prompt")

question = "What architecture does Haystack 2.0 use?"
result = rag.run({"retriever": {"query": question}, "prompt_builder": {"question": question}})
print(result["generator"]["replies"][0])

Run it:

python minimal_rag.py

Expected output (abridged):

Haystack 2.0 uses a component-based architecture where each piece — retriever, builder, generator — is a replaceable class.

The pipeline ran three components in sequence. The retriever fetched three documents, PromptBuilder rendered them into a single string, and the generator returned an answer grounded in those documents.

Adding citations and metadata

Production RAG needs citations. Extend the template to include document IDs and scores, then parse them post-generation.

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

document_store = InMemoryDocumentStore()
docs = [
    Document(content="Haystack 2.0 uses a component-based architecture.", meta={"source": "docs/architecture.md", "page": 1}),
    Document(content="PromptBuilder templates are Jinja2 with custom filters.", meta={"source": "docs/promptbuilder.md", "page": 3}),
    Document(content="Generators only see the rendered prompt string.", meta={"source": "docs/generators.md", "page": 2}),
]
document_store.write_documents(docs)

retriever = InMemoryBM25Retriever(document_store=document_store, top_k=3)

template = """
Answer the question using only the provided documents. Cite sources inline like [1], [2].

Documents:
{% for doc in documents %}
[{{ loop.index }}] (source: {{ doc.meta.source }}, page: {{ doc.meta.page }}) {{ doc.content }}
{% endfor %}

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

prompt_builder = PromptBuilder(template=template)
generator = OpenAIGenerator(model="gpt-4o-mini")

rag = Pipeline()
rag.add_component("retriever", retriever)
rag.add_component("prompt_builder", prompt_builder)
rag.add_component("generator", generator)
rag.connect("retriever.documents", "prompt_builder.documents")
rag.connect("prompt_builder.prompt", "generator.prompt")

question = "How do PromptBuilder templates work?"
result = rag.run({"retriever": {"query": question}, "prompt_builder": {"question": question}})
reply = result["generator"]["replies"][0]
print("Raw reply:\n", reply)

# Extract citations for downstream use
citations = re.findall(r'\[(\d+)\]', reply)
print("\nCited doc indices:", citations)

Output:

Raw reply:
 PromptBuilder templates use Jinja2 syntax with custom filters [1]. They can loop over documents and access metadata like source and page [2].

Cited doc indices: ['1', '2']

The generator now emits bracketed numbers that map back to the retrieved list. Your API layer can replace those with links or expand them into full citations.

Token-aware truncation

LLMs have context limits. A haystack promptbuilder rag pipeline tutorial must handle documents that exceed the model’s window. Haystack provides TokenCountTruncator — but you can also implement truncation inside the template using a custom filter.

First, add a token counter utility:

# token_utils.py
import tiktoken

_enc = tiktoken.encoding_for_model("gpt-4o-mini")
MAX_PROMPT_TOKENS = 12000  # leave room for completion

def count_tokens(text: str) -> int:
    return len(_enc.encode(text))

def truncate_documents(documents: list, question: str, template: str, max_tokens: int = MAX_PROMPT_TOKENS) -> list:
    """
    Greedily drop lowest-ranked documents until the rendered prompt fits.
    """
    from haystack.components.builders import PromptBuilder
    builder = PromptBuilder(template=template)
    for i in range(len(documents), 0, -1):
        prompt = builder.run(documents=documents[:i], question=question)["prompt"]
        if count_tokens(prompt) <= max_tokens:
            return documents[:i]
    return [documents[0]]  # always keep at least one

Wire it into the pipeline as a custom component:

# truncation_component.py
from haystack import component
from typing import List
from haystack import Document
from token_utils import truncate_documents

@component
class DocumentTruncator:
    def __init__(self, template: str, max_tokens: int = 12000):
        self.template = template
        self.max_tokens = max_tokens

    @component.output_types(documents=List[Document])
    def run(self, documents: List[Document], question: str):
        truncated = truncate_documents(documents, question, self.template, self.max_tokens)
        return {"documents": truncated}

Now insert it between retriever and PromptBuilder:

# rag_with_truncation.py
from haystack import Pipeline
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator
from haystack.document_stores.in_memory import InMemoryDocumentStore
from truncation_component import DocumentTruncator

document_store = InMemoryDocumentStore()
# ... write many long documents ...
retriever = InMemoryBM25Retriever(document_store=document_store, top_k=10)

template = """
{% for doc in documents %}
[{{ loop.index }}] {{ doc.content }}
{% endfor %}
Question: {{ question }}
Answer:
"""

truncator = DocumentTruncator(template=template, max_tokens=12000)
prompt_builder = PromptBuilder(template=template)
generator = OpenAIGenerator(model="gpt-4o-mini")

rag = Pipeline()
rag.add_component("retriever", retriever)
rag.add_component("truncator", truncator)
rag.add_component("prompt_builder", prompt_builder)
rag.add_component("generator", generator)

rag.connect("retriever.documents", "truncator.documents")
rag.connect("truncator.documents", "prompt_builder.documents")
rag.connect("prompt_builder.prompt", "generator.prompt")

result = rag.run({
    "retriever": {"query": "complex question"},
    "truncator": {"question": "complex question"},
    "prompt_builder": {"question": "complex question"},
})

The truncator receives the raw document list and the question, renders a trial prompt, and drops tail documents until it fits. This keeps the generator from silently truncating or erroring.

Conditional sections and fallback logic

Real queries vary: some need citations, some need a “no answer found” path, some need a summary prefix. PromptBuilder handles this with Jinja2 conditionals and Haystack’s ConditionalRouter.

Add a router that checks retrieval score:

# conditional_rag.py
from haystack import Pipeline
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator
from haystack.components.routers import ConditionalRouter
from haystack.document_stores.in_memory import InMemoryDocumentStore

document_store = InMemoryDocumentStore()
document_store.write_documents([
    Document(content="Haystack is an LLM framework.", meta={"source": "intro"}),
    Document(content="PromptBuilder builds prompts.", meta={"source": "pb"}),
])

retriever = InMemoryBM25Retriever(document_store=document_store, top_k=3)

# Two templates: one for high-confidence, one for low-confidence
high_conf_template = """
High-confidence answer (retrieval score > 0.5).
Documents:
{% for doc in documents %}
[{{ loop.index }}] {{ doc.content }}
{% endfor %}
Question: {{ question }}
Answer with citations:
"""

low_conf_template = """
Low-confidence answer (retrieval score <= 0.5).
I found limited relevant information. Here is what I have:
{% for doc in documents %}
- {{ doc.content }}
{% endfor %}
Question: {{ question }}
Answer honestly, stating uncertainty:
"""

high_builder = PromptBuilder(template=high_conf_template, required_variables=["documents", "question"])
low_builder = PromptBuilder(template=low_conf_template, required_variables=["documents", "question"])
generator = OpenAIGenerator(model="gpt-4o-mini")

router = ConditionalRouter(
    routes=[
        {"condition": "{{ documents[0].score > 0.5 }}", "output": "{{ documents }}", "output_name": "high_conf_docs", "output_type": "list[Document]"},
        {"condition": "True", "output": "{{ documents }}", "output_name": "low_conf_docs", "output_type": "list[Document]"},
    ]
)

rag = Pipeline()
rag.add_component("retriever", retriever)
rag.add_component("router", router)
rag.add_component("high_builder", high_builder)
rag.add_component("low_builder", low_builder)
rag.add_component("generator", generator)

rag.connect("retriever.documents", "router.documents")
rag.connect("router.high_conf_docs", "high_builder.documents")
rag.connect("router.low_conf_docs", "low_builder.documents")
rag.connect("high_builder.prompt", "generator.prompt")
rag.connect("low_builder.prompt", "generator.prompt")

# Pass question to both builders via run kwargs
result = rag.run({
    "retriever": {"query": "What is Haystack?"},
    "high_builder": {"question": "What is Haystack?"},
    "low_builder": {"question": "What is Haystack?"},
})
print(result["generator"]["replies"][0])

The router inspects the top document’s BM25 score (available as doc.score after retrieval) and routes to the appropriate builder. Both builders feed the same generator — the pipeline merges their prompt outputs automatically because only one branch executes per run.

Streaming responses

For latency-sensitive UIs, stream tokens instead of waiting for the full completion. OpenAIGenerator supports streaming via a callback:

# streaming_rag.py
from haystack import Pipeline
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator
from haystack.document_stores.in_memory import InMemoryDocumentStore

document_store = InMemoryDocumentStore()
document_store.write_documents([
    Document(content="Streaming sends tokens as they arrive."),
    Document(content="Use a callback to process each chunk."),
])
retriever = InMemoryBM25Retriever(document_store=document_store, top_k=2)

template = """
{% for doc in documents %}
{{ doc.content }}
{% endfor %}
Question: {{ question }}
Answer:
"""
prompt_builder = PromptBuilder(template=template)
generator = OpenAIGenerator(model="gpt-4o-mini", streaming_callback=lambda chunk: print(chunk, end="", flush=True))

rag = Pipeline()
rag.add_component("retriever", retriever)
rag.add_component("prompt_builder", prompt_builder)
rag.add_component("generator", generator)
rag.connect("retriever.documents", "prompt_builder.documents")
rag.connect("prompt_builder.prompt", "generator.prompt")

rag.run({"retriever": {"query": "How does streaming work?"}, "prompt_builder": {"question": "How does streaming work?"}})
print()  # newline after stream

Output appears token-by-token:

Streaming sends tokens as they arrive. Use a callback to process each chunk.

The PromptBuilder usage is unchanged — streaming is a generator concern.

Evaluating prompt quality

Before shipping, measure whether your template actually helps. Use Haystack’s evaluation harness with a small labeled set:

# eval_prompt.py
from haystack import Pipeline
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.eval import EvaluationRunResult, evaluate
from haystack.eval.metrics import AnswerExactMatch, AnswerF1

document_store = InMemoryDocumentStore()
document_store.write_documents([
    Document(content="Paris is the capital of France.", meta={"id": "1"}),
    Document(content="Berlin is the capital of Germany.", meta={"id": "2"}),
])

retriever = InMemoryBM25Retriever(document_store=document_store, top_k=2)
template = "Docs: {% for d in documents %}{{ d.content }} {% endfor %} Q: {{ question }} A:"
prompt_builder = PromptBuilder(template=template)
generator = OpenAIGenerator(model="gpt-4o-mini")

rag = Pipeline()
rag.add_component("retriever", retriever)
rag.add_component("prompt_builder", prompt_builder)
rag.add_component("generator", generator)
rag.connect("retriever.documents", "prompt_builder.documents")
rag.connect("prompt_builder.prompt", "generator.prompt")

# Labeled examples
questions = ["Capital of France?", "Capital of Germany?"]
ground_truth = ["Paris", "Berlin"]

results = []
for q, gt in zip(questions, ground_truth):
    out = rag.run({"retriever": {"query": q}, "prompt_builder": {"question": q}})
    pred = out["generator"]["replies"][0].strip()
    results.append({"question": q, "ground_truth": gt, "prediction": pred})

# Compute metrics
em = AnswerExactMatch()
f1 = AnswerF1()
for r in results:
    print(f"Q: {r['question']} | GT: {r['ground_truth']} | Pred: {r['prediction']} | EM: {em.compute(r['ground_truth'], r['prediction'])} | F1: {f1.compute(r['ground_truth'], r['prediction'])}")

Sample output:

Q: Capital of France? | GT: Paris | Pred: Paris | EM: 1.0 | F1: 1.0
Q: Capital of Germany? | GT: Berlin | Pred: Berlin | EM: 1.0 | F1: 1.0

Swap templates, re-run, compare. This is how you justify prompt engineering time.

Wiring it to an OpenAI-compatible gateway

If your organization routes model calls through a gateway (for fallbacks, usage metering, or cache-control forwarding), point OpenAIGenerator at that endpoint:

generator = OpenAIGenerator(
    model="gpt-4o-mini",
    api_base_url="https://api.n4n.ai/v1",  # example gateway endpoint
    api_key=os.getenv("GATEWAY_KEY"),
)

The rest of the pipeline — retriever, PromptBuilder, truncator, router — remains untouched. The gateway handles provider degradation and token accounting transparently.

Checklist before deploying

  • Template version control: Store PromptBuilder templates as .j2 files, not inline strings. Load them at startup.
  • Input validation: Ensure question is non-empty and under a character limit before it hits the template.
  • Observability: Log the rendered prompt (truncated to 500 chars) and the document IDs used. Correlate with generator latency.
  • Fallback chain: If the generator errors, retry with a smaller top_k or a simpler template.
  • Token budget: Set max_tokens on the generator and enforce it in the truncator. Monitor actual usage vs. budget.

What to build next

  • Swap BM25 for a dense retriever (InMemoryEmbeddingRetriever + SentenceTransformersDocumentEmbedder) and compare recall.
  • Add a DocumentJoiner to merge results from sparse and dense retrievers before PromptBuilder.
  • Implement a PromptBuilder that emits JSON Schema for structured extraction — useful for downstream agents.
  • Hook the pipeline into a FastAPI endpoint with request validation, streaming SSE responses, and per-tenant usage quotas.

The PromptBuilder is a small component, but it sits at the seam between retrieval and generation. Getting its template right — citations, truncation, conditionals, streaming — determines whether your RAG pipeline feels like a demo or a product.

Tagshaystackpromptbuilderragtutorial

Written by

n4n Team

The team building n4n — a single OpenAI-compatible API in front of 240+ models, with automatic fallback, load balancing and pay-per-token metering.

More from n4n Team →
Framework tutorialsHaystack RAG pipeline tutorial: enterprise document searchBuild a production-ready Haystack RAG pipeline for enterprise document search with hybrid retrieval, reranking, and observability.Framework tutorialsHaystack RAG pipeline tutorial: query routingBuild a Haystack RAG pipeline that classifies queries and routes them to specialized retrievers for better accuracy and lower latency.Framework tutorialsHaystack RAG pipeline tutorial: from PDF to answerBuild a production-ready Haystack RAG pipeline that extracts text from PDFs, indexes it, and answers questions with citations — complete with runnable code and expected outputs.Framework tutorialsHaystack RAG pipeline tutorial with hybrid retrievalBuild a production-ready Haystack RAG pipeline with hybrid retrieval combining BM25 and dense embeddings for better search relevance.Framework tutorialsBuild a RAG pipeline with Haystack 2.0 and n4n.aiHands-on tutorial to build rag pipeline haystack 2.0 n4n.ai using Haystack 2.0 and an OpenAI-compatible endpoint, with runnable code and expected output.Framework tutorialsChunking strategies for Haystack RAG pipelinesPractical guide to chunking strategies for Haystack RAG pipelines with code examples, tradeoffs, and evaluation methods.

All haystack rag pipelines posts →