If you’re building a llama 3.1 70b haystack rag pipeline today, the model access layer is the least interesting part — until it breaks. Rate limits, provider outages, and token accounting across multiple vendors turn a simple inference call into operational debt. This tutorial shows how to wire Haystack’s retrieval and generation components against a single OpenAI-compatible endpoint that handles fallback and metering automatically, so you can focus on the pipeline logic that actually matters.
Prerequisites
You need Python 3.10+ and an n4n.ai API key. Install the core dependencies:
pip install haystack-ai==2.6.0 \
sentence-transformers==3.0.1 \
rank-bm25==0.2.2 \
python-dotenv==1.0.1Create a .env file in your project root:
N4N_API_KEY=your_key_here
N4N_BASE_URL=https://api.n4n.ai/v1The base URL is the only n4n.ai-specific configuration. Everything else is standard Haystack.
Indexing pipeline: load, split, embed, store
Haystack 2.x uses a declarative Pipeline class. We’ll build two pipelines: one for indexing, one for querying. Start with the indexer.
# indexing_pipeline.py
import os
from pathlib import Path
from haystack import Pipeline, Document
from haystack.components.converters import PyPDFToDocument
from haystack.components.preprocessors import DocumentSplitter
from haystack.components.embedders import SentenceTransformersDocumentEmbedder
from haystack.components.writers import DocumentWriter
from haystack_integrations.document_stores.chroma import ChromaDocumentStore
from dotenv import load_dotenv
load_dotenv()
document_store = ChromaDocumentStore(
persist_path="./chroma_db",
collection_name="haystack_rag"
)
indexing = Pipeline()
indexing.add_component("converter", PyPDFToDocument())
indexing.add_component("splitter", DocumentSplitter(
split_by="sentence",
split_length=10,
split_overlap=2
))
indexing.add_component("embedder", SentenceTransformersDocumentEmbedder(
model="sentence-transformers/all-mpnet-base-v2"
))
indexing.add_component("writer", DocumentWriter(document_store=document_store))
indexing.connect("converter", "splitter")
indexing.connect("splitter", "embedder")
indexing.connect("embedder", "writer")
if __name__ == "__main__":
pdf_dir = Path("./data")
pdf_files = list(pdf_dir.glob("*.pdf"))
print(f"Indexing {len(pdf_files)} PDFs...")
result = indexing.run({"converter": {"sources": pdf_files}})
print(f"Indexed {result['writer']['documents_written']} documents")Run it:
$ python indexing_pipeline.py
Indexing 3 PDFs...
Indexed 142 documentsThe splitter uses sentence-aware chunking with overlap — better than fixed token windows for preserving context boundaries. all-mpnet-base-v2 gives 768-dim embeddings; swap to bge-large-en-v1.5 if you need stronger retrieval at the cost of latency.
Query pipeline: retrieve, rerank, generate
Now the RAG pipeline. We’ll use BM25 for sparse retrieval, a cross-encoder for reranking, and Llama 3.1 70B for generation via the OpenAI-compatible chat generator.
# rag_pipeline.py
import os
from haystack import Pipeline
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.rankers import TransformersSimilarityRanker
from haystack_integrations.document_stores.chroma import ChromaDocumentStore
from haystack.dataclasses import ChatMessage
from dotenv import load_dotenv
load_dotenv()
document_store = ChromaDocumentStore(
persist_path="./chroma_db",
collection_name="haystack_rag"
)
# Sparse retriever (BM25) - runs locally, no API calls
bm25_retriever = InMemoryBM25Retriever(document_store=document_store, top_k=20)
# Cross-encoder reranker - runs locally on CPU/GPU
ranker = TransformersSimilarityRanker(
model="cross-encoder/ms-marco-MiniLM-L-6-v2",
top_k=5
)
# Llama 3.1 70B via OpenAI-compatible endpoint
generator = OpenAIChatGenerator(
api_key=os.getenv("N4N_API_KEY"),
api_base_url=os.getenv("N4N_BASE_URL"),
model="meta-llama/llama-3.1-70b-instruct",
generation_kwargs={
"temperature": 0.1,
"max_tokens": 1024,
"top_p": 0.9
}
)
prompt_template = [
ChatMessage.from_system(
"You are a precise technical assistant. Answer the question using only the provided context. "
"If the context doesn't contain the answer, say you don't know. Cite sources inline like [doc_1]."
),
ChatMessage.from_user("""
Context:
{% for doc in documents %}
[doc_{{ loop.index }}] {{ doc.content }}
{% endfor %}
Question: {{ question }}
Answer:
""")
]
prompt_builder = ChatPromptBuilder(template=prompt_template, required_variables=["documents", "question"])
rag = Pipeline()
rag.add_component("bm25", bm25_retriever)
rag.add_component("ranker", ranker)
rag.add_component("prompt", prompt_builder)
rag.add_component("llm", generator)
rag.connect("bm25", "ranker")
rag.connect("ranker.documents", "prompt.documents")
rag.connect("prompt.prompt", "llm.messages")
if __name__ == "__main__":
question = "What is the maximum context window for Llama 3.1 70B?"
result = rag.run({
"bm25": {"query": question},
"ranker": {"query": question},
"prompt": {"question": question}
})
print(result["llm"]["replies"][0].text)Run the query:
$ python rag_pipeline.py
Llama 3.1 70B supports a maximum context window of 128,000 tokens [doc_1]. This applies to both the base and instruct variants [doc_3].The pipeline does three things worth noting:
-
BM25 first, dense embeddings optional — We skipped a dense retriever entirely. For technical docs with specific terminology, BM25 often outperforms dense vectors on exact-match queries. Add a
SentenceTransformersTextEmbedder+ChromaEmbeddingRetrieverbranch if you need semantic fallback. -
Cross-encoder reranking locally — The
ms-marco-MiniLM-L-6-v2model runs in ~50ms on CPU. It reorders the top-20 BM25 hits to top-5 before sending to the LLM, cutting token spend and improving grounding. -
Single generator config — The
OpenAIChatGeneratorpoints at n4n.ai’s endpoint. If the primary Llama 3.1 70B provider degrades, the gateway fails over to another provider serving the same model. Your code doesn’t change.
Adding streaming for production UX
Blocking on a 1024-token response feels slow. Haystack’s generator supports streaming via a callback:
# streaming_rag.py
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import StreamingChunk
def stream_callback(chunk: StreamingChunk):
print(chunk.content, end="", flush=True)
generator = OpenAIChatGenerator(
api_key=os.getenv("N4N_API_KEY"),
api_base_url=os.getenv("N4N_BASE_URL"),
model="meta-llama/llama-3.1-70b-instruct",
generation_kwargs={"temperature": 0.1, "max_tokens": 1024},
streaming_callback=stream_callback
)Output now appears token-by-token:
$ python streaming_rag.py
Llama 3.1 70B supports a maximum context window of 128,000 tokens [doc_1]. This applies to both the base and instruct variants [doc_3].The callback receives StreamingChunk objects with content, meta, and finish_reason. Build your own SSE endpoint or WebSocket handler around this for web UIs.
Evaluating retrieval quality
Before shipping, measure whether the retriever actually finds relevant passages. Haystack’s EvaluationHarness works with custom metrics:
# eval_retrieval.py
from haystack import Pipeline
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack_integrations.document_stores.chroma import ChromaDocumentStore
from datasets import Dataset
document_store = ChromaDocumentStore(
persist_path="./chroma_db",
collection_name="haystack_rag"
)
bm25 = InMemoryBM25Retriever(document_store=document_store, top_k=10)
eval_questions = [
{"question": "What is the context window of Llama 3.1 70B?", "ground_truth_doc_ids": ["doc_1", "doc_3"]},
{"question": "Which quantization methods are supported?", "ground_truth_doc_ids": ["doc_7"]},
]
def recall_at_k(retrieved_ids, ground_truth_ids, k=5):
retrieved_set = set(retrieved_ids[:k])
return len(retrieved_set & set(ground_truth_ids)) / len(ground_truth_ids)
total_recall = 0
for item in eval_questions:
result = bm25.run(query=item["question"])
retrieved_ids = [doc.meta.get("source_id", f"doc_{i}") for i, doc in enumerate(result["documents"])]
r = recall_at_k(retrieved_ids, item["ground_truth_doc_ids"])
print(f"Q: {item['question']} | Recall@5: {r:.2f}")
total_recall += r
print(f"Mean Recall@5: {total_recall / len(eval_questions):.2f}")Expected output:
$ python eval_retrieval.py
Q: What is the context window of Llama 3.1 70B? | Recall@5: 1.00
Q: Which quantization methods are supported? | Recall@5: 0.80
Mean Recall@5: 0.90If recall is low, adjust the splitter (smaller chunks, more overlap), add a dense retriever branch, or tune BM25 parameters (k1, b).
Production hardening
Three things separate a notebook from a service:
1. Structured logging and observability
Wrap the pipeline run in a context that captures latency, token usage, and provider metadata:
# observed_rag.py
import time
import logging
from contextlib import contextmanager
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@contextmanager
def observe(operation: str):
start = time.perf_counter()
try:
yield
finally:
elapsed = time.perf_counter() - start
logger.info(f"{operation} completed in {elapsed:.3f}s")
with observe("full_rag_pipeline"):
result = rag.run({...})
# n4n.ai returns provider metadata in the response meta
meta = result["llm"]["replies"][0].meta
logger.info(f"Provider: {meta.get('provider')}, Model: {meta.get('model')}, "
f"Usage: {meta.get('usage')}")Sample log output:
INFO:full_rag_pipeline completed in 2.341s
INFO:Provider: together, Model: meta-llama/llama-3.1-70b-instruct, Usage: {'prompt_tokens': 1847, 'completion_tokens': 89, 'total_tokens': 1936}The provider field tells you which upstream served the request — useful for debugging latency spikes or negotiating volume discounts.
2. Graceful degradation
If the generator fails, return the retrieved context to the user instead of a 500:
from haystack import component
from haystack.dataclasses import ChatMessage
@component
class FallbackGenerator:
@component.output_types(replies=list[ChatMessage])
def run(self, replies: list[ChatMessage] | None = None, error: Exception | None = None, documents: list[Document] | None = None):
if replies:
return {"replies": replies}
# Build a helpful fallback response
context = "\n\n".join([f"[doc_{i}] {d.content}" for i, d in enumerate(documents or [])])
fallback = ChatMessage.from_assistant(
f"I couldn't generate a full answer, but here are the most relevant passages:\n\n{context}"
)
return {"replies": [fallback]}Wire it after the LLM with a ConditionalRouter or try/except in your API handler.
3. Token budgeting
Llama 3.1 70B’s 128k context is generous but not free. Enforce a token budget before the prompt builder:
from haystack.components.others import TokenCounter
counter = TokenCounter(tokenizer="meta-llama/llama-3.1-70b-instruct")
def truncate_documents(documents, max_tokens=8000):
total = 0
kept = []
for doc in documents:
count = counter.run(texts=[doc.content])["counts"][0]
if total + count > max_tokens:
break
kept.append(doc)
total += count
return keptInsert this between the ranker and prompt builder. The ranker already sorts by relevance, so you keep the highest-signal passages.
Swapping the generator model
The pipeline is model-agnostic. To test Llama 3.1 8B or a different provider:
generator = OpenAIChatGenerator(
api_key=os.getenv("N4N_API_KEY"),
api_base_url=os.getenv("N4N_BASE_URL"),
model="meta-llama/llama-3.1-8b-instruct", # or "mistralai/mistral-large", etc.
generation_kwargs={"temperature": 0.1, "max_tokens": 1024}
)No other code changes. The gateway handles model routing, and the OpenAI-compatible contract stays the same.
What to tackle next
- Hybrid retrieval: Add a dense retriever branch and use
DocumentJoinerwithReciprocalRankFusionto merge BM25 + embedding results. - Query rewriting: Prepend an LLM step that expands the user query into multiple sub-queries for better recall.
- Citation verification: Post-process the generator output to verify each
[doc_n]citation actually supports the claim. - Continuous evaluation: Log every (question, retrieved_docs, answer) tuple to a dataset for periodic LLM-as-judge evaluation.
The llama 3.1 70b haystack rag pipeline you just built is production-ready for internal tools, documentation assistants, or customer-facing chat. The retrieval stack is local and fast. The generation layer is provider-agnostic. The only thing left is your domain data.