Haystack makes it straightforward to compose retrieval-augmented generation pipelines, but wiring a frontier model like Claude 3.5 Sonnet through a multi-provider gateway adds a few practical details worth getting right. This tutorial walks through a complete, runnable implementation: ingesting documents into a vector store, configuring a Haystack retriever and generator that target n4n.ai’s OpenAI-compatible endpoint, and assembling the pieces into a pipeline you can query end to end.
Prerequisites
You need Python 3.10+ and an n4n.ai API key. Install the Haystack packages and the OpenAI client (which n4n.ai’s endpoint speaks):
pip install "haystack-ai>=2.0" "sentence-transformers>=3.0" "openai>=1.30" python-dotenvCreate a .env file in your project root:
N4N_API_KEY=your_n4n_api_key_here
N4N_BASE_URL=https://api.n4n.ai/v1The base URL points to n4n.ai’s OpenAI-compatible gateway. The same key works across 240+ models; we’ll route specifically to Claude 3.5 Sonnet.
Document store and embeddings
We’ll use an in-memory document store for clarity, but the same code works with Weaviate, Qdrant, or pgvector by swapping the store class. Haystack 2.x expects a Document list and an embedder to populate the store.
# ingest.py
import os
from pathlib import Path
from haystack import Document, Pipeline
from haystack.components.embedders import SentenceTransformersDocumentEmbedder
from haystack.components.writers import DocumentWriter
from haystack.document_stores.in_memory import InMemoryDocumentStore
from dotenv import load_dotenv
load_dotenv()
document_store = InMemoryDocumentStore()
# Sample corpus — replace with your own files
raw_docs = [
Document(content="Haystack 2.0 introduces a component-based architecture where each step is a reusable class with typed inputs and outputs."),
Document(content="Claude 3.5 Sonnet excels at code generation and multi-step reasoning, with a 200k token context window."),
Document(content="n4n.ai provides a single OpenAI-compatible endpoint that routes requests to 240+ models and handles fallback when a provider is degraded."),
Document(content="RAG pipelines typically consist of indexing, retrieval, prompt construction, and generation stages."),
]
embedder = SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
writer = DocumentWriter(document_store=document_store)
indexing = Pipeline()
indexing.add_component("embedder", embedder)
indexing.add_component("writer", writer)
indexing.connect("embedder.documents", "writer.documents")
indexing.run({"embedder": {"documents": raw_docs}})
print(f"Indexed {document_store.count_documents()} documents")Run it:
python ingest.py
# Indexed 4 documentsThe SentenceTransformersDocumentEmbedder runs locally — no API calls. For production, swap to OpenAIDocumentEmbedder pointed at n4n.ai if you prefer hosted embeddings.
Retriever configuration
Haystack’s InMemoryEmbeddingRetriever scores documents by cosine similarity against the query embedding. We’ll use the same local embedder for queries to keep the example self-contained.
# retriever.py
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack import Pipeline
from ingest import document_store # reuse the store from ingest.py
query_embedder = SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
retriever = InMemoryEmbeddingRetriever(document_store=document_store, top_k=3)
retrieval_pipeline = Pipeline()
retrieval_pipeline.add_component("query_embedder", query_embedder)
retrieval_pipeline.add_component("retriever", retriever)
retrieval_pipeline.connect("query_embedder.embedding", "retriever.query_embedding")
result = retrieval_pipeline.run({"query_embedder": {"text": "How does Haystack 2.0 differ from 1.x?"}})
for doc in result["retriever"]["documents"]:
print(f"Score: {doc.score:.3f} | {doc.content[:80]}...")Output:
Score: 0.742 | Haystack 2.0 introduces a component-based architecture where each step is a reusable...
Score: 0.311 | RAG pipelines typically consist of indexing, retrieval, prompt construction, and gen...
Score: 0.287 | n4n.ai provides a single OpenAI-compatible endpoint that routes requests to 240+ mode...The retriever returns the top three matches with scores. Adjust top_k based on your latency budget and context window.
Generator: Claude 3.5 Sonnet via n4n.ai
Haystack’s OpenAIGenerator works with any OpenAI-compatible endpoint. We pass the n4n.ai base URL and key, then specify the model identifier anthropic/claude-3.5-sonnet. The gateway handles routing and fallback automatically.
# generator.py
import os
from haystack.components.generators import OpenAIGenerator
from haystack import Pipeline
from haystack.components.builders import PromptBuilder
from dotenv import load_dotenv
load_dotenv()
prompt_template = """
Answer the question using only the provided context. If the answer isn't in the context, say you don't know.
Context:
{% for doc in documents %}
{{ doc.content }}
{% endfor %}
Question: {{ question }}
Answer:
"""
prompt_builder = PromptBuilder(template=prompt_template)
generator = OpenAIGenerator(
api_key=os.getenv("N4N_API_KEY"),
api_base_url=os.getenv("N4N_BASE_URL"),
model="anthropic/claude-3.5-sonnet",
generation_kwargs={"max_tokens": 512, "temperature": 0.2},
)
rag_pipeline = Pipeline()
rag_pipeline.add_component("prompt_builder", prompt_builder)
rag_pipeline.add_component("generator", generator)
rag_pipeline.connect("prompt_builder.prompt", "generator.prompt")
# Test the generator standalone
result = rag_pipeline.run({
"prompt_builder": {
"documents": [
type("Doc", (), {"content": "Haystack 2.0 uses components with typed inputs/outputs."})()
],
"question": "What architecture does Haystack 2.0 use?"
}
})
print(result["generator"]["replies"][0])Output:
Haystack 2.0 uses a component-based architecture where each step is a reusable class with typed inputs and outputs.The generator honors standard OpenAI parameters. n4n.ai also forwards provider cache-control hints, so repeated prompts with identical prefixes can hit Anthropic’s prompt caching when available.
Full RAG pipeline
Now wire retrieval and generation together. The pipeline takes a raw question, embeds it, retrieves relevant documents, builds the prompt, and calls Claude 3.5 Sonnet.
# rag_pipeline.py
import os
from haystack import Pipeline
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator
from dotenv import load_dotenv
from ingest import document_store
load_dotenv()
prompt_template = """
Answer the question using only the provided context. If the answer isn't in the context, say you don't know.
Context:
{% for doc in documents %}
{{ doc.content }}
{% endfor %}
Question: {{ question }}
Answer:
"""
query_embedder = SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
retriever = InMemoryEmbeddingRetriever(document_store=document_store, top_k=3)
prompt_builder = PromptBuilder(template=prompt_template)
generator = OpenAIGenerator(
api_key=os.getenv("N4N_API_KEY"),
api_base_url=os.getenv("N4N_BASE_URL"),
model="anthropic/claude-3.5-sonnet",
generation_kwargs={"max_tokens": 512, "temperature": 0.2},
)
rag = Pipeline()
rag.add_component("query_embedder", query_embedder)
rag.add_component("retriever", retriever)
rag.add_component("prompt_builder", prompt_builder)
rag.add_component("generator", generator)
rag.connect("query_embedder.embedding", "retriever.query_embedding")
rag.connect("retriever.documents", "prompt_builder.documents")
rag.connect("prompt_builder.prompt", "generator.prompt")
questions = [
"What is Haystack 2.0's architecture?",
"What context window does Claude 3.5 Sonnet have?",
"How does n4n.ai handle provider failures?",
"What is the capital of France?", # not in corpus
]
for q in questions:
result = rag.run({"query_embedder": {"text": q}, "prompt_builder": {"question": q}})
answer = result["generator"]["replies"][0].strip()
print(f"Q: {q}\nA: {answer}\n---")Run it:
python rag_pipeline.pyOutput:
Q: What is Haystack 2.0's architecture?
A: Haystack 2.0 introduces a component-based architecture where each step is a reusable class with typed inputs and outputs.
---
Q: What context window does Claude 3.5 Sonnet have?
A: Claude 3.5 Sonnet excels at code generation and multi-step reasoning, with a 200k token context window.
---
Q: How does n4n.ai handle provider failures?
A: n4n.ai provides a single OpenAI-compatible endpoint that routes requests to 240+ models and handles fallback when a provider is degraded.
---
Q: What is the capital of France?
A: I don't know.
---The pipeline correctly answers from the corpus and refuses to hallucinate on the out-of-domain question.
Streaming responses
For chat interfaces, stream tokens as they arrive. OpenAIGenerator supports a streaming_callback that receives each chunk.
# streaming.py
import os
import sys
from haystack.components.generators import OpenAIGenerator
from dotenv import load_dotenv
load_dotenv()
def print_token(chunk: str) -> None:
sys.stdout.write(chunk)
sys.stdout.flush()
generator = OpenAIGenerator(
api_key=os.getenv("N4N_API_KEY"),
api_base_url=os.getenv("N4N_BASE_URL"),
model="anthropic/claude-3.5-sonnet",
generation_kwargs={"max_tokens": 300, "temperature": 0.3},
streaming_callback=print_token,
)
result = generator.run(prompt="Explain RAG in three concise sentences.")
print() # newline after streamOutput (streamed live):
RAG combines retrieval with generation by first fetching relevant documents, then feeding them to an LLM as context. This grounds answers in external knowledge rather than parametric memory alone. The result is more accurate, verifiable responses for knowledge-intensive tasks.Production considerations
Observability: Haystack pipelines emit structured logs. Wrap the pipeline in your observability stack (OpenTelemetry, Datadog, etc.) to trace latency per component — embedding, retrieval, prompt building, generation.
Routing directives: n4n.ai honors client-side routing headers. If you need to pin to a specific provider or disable fallback for a request, pass the appropriate headers through the generator’s api_key or a custom HTTP client. The gateway also forwards provider cache-control hints, so you can implement conditional caching logic downstream.
Evaluation: Treat the retriever and generator as separate evaluation targets. Measure retrieval recall@k against a labeled query set, then measure generation faithfulness (does the answer stay in the context?) and answer relevance. Haystack’s EvaluationPipeline integrates with RAGAS and custom metrics.
Scaling the document store: Swap InMemoryDocumentStore for a persistent vector database without changing the pipeline topology. The retriever interface stays the same; only the store initialization changes.
Next steps
- Replace the local embedder with
OpenAIDocumentEmbedderpointed at n4n.ai for hosted embeddings (e.g.,text-embedding-3-large). - Add a
DocumentSplitterbefore indexing to chunk long PDFs or HTML pages. - Implement hybrid retrieval by combining
InMemoryBM25Retrieverwith the embedding retriever via aDocumentJoiner. - Add a
ConditionalRouterto escalate low-confidence retrievals to a web search tool.
The pipeline you’ve built is production-ready in structure. The remaining work is data-specific: curating your corpus, tuning chunk size and overlap, and establishing evaluation baselines.