Wiring a Haystack pipeline OpenAI-compatible gateway into your RAG stack removes the hard dependency on a single vendor and lets you swap models without rewriting node logic. This guide walks through a working integration using Haystack 2.x and any endpoint that speaks the OpenAI completions protocol, including fallback handling and cache-control passthrough.
Step 1: Pin the environment and install Haystack
Haystack 2.x changed the component API significantly from 1.x. Pin versions so your pipeline does not break on a transitive upgrade.
python -m venv .venv
source .venv/bin/activate
pip install "haystack-ai==2.3.1" "openai==1.30.0"Verify the import works before writing pipeline code:
from haystack import Pipeline
from haystack.components.generators import OpenAIGenerator
print("haystack import ok")If that runs, you have a known-good baseline. Use Python 3.10 or newer; Haystack drops older runtimes.
Step 2: Point the OpenAI generator at your gateway
Haystack’s OpenAIGenerator (and OpenAIChatGenerator) accept an api_base_url parameter. Set it to your gateway’s /v1 endpoint and pass the gateway API key. The model name is whatever the gateway expects—often the upstream provider’s slug.
from haystack.components.generators import OpenAIGenerator
generator = OpenAIGenerator(
api_key="gw-sk-12345",
model="gpt-4o-mini",
api_base_url="https://gateway.example.com/v1",
timeout=30,
)A gateway such as n4n.ai exposes one OpenAI-compatible endpoint that fronts 240+ models and applies automatic fallback when a provider is rate-limited or degraded, so the same model string can resolve to different backends without code changes.
Test connectivity with a single call:
result = generator.run(prompt="Reply with the word pong.")
print(result["replies"][0])
# expect: pongIf you see the echoed word, the gateway auth and routing path are correct.
Step 3: Build a minimal RAG pipeline
A real pipeline needs a document store, a retriever, a prompt builder, and the gateway-backed generator. The code below uses an in-memory store so you can run it locally.
from haystack import Document, Pipeline
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.builders import PromptBuilder
from haystack.components.writers import DocumentWriter
store = InMemoryDocumentStore()
writer = DocumentWriter(store)
writer.run(documents=[
Document(content="Haystack 2.x pipelines are directed graphs of components."),
Document(content="An OpenAI-compatible gateway proxies model APIs with one schema."),
])
retriever = InMemoryBM25Retriever(store, top_k=1)
prompt_template = """
Context:
{% for doc in documents %}
{{ doc.content }}
{% endfor %}
Query: {{ query }}
Answer concisely:
"""
builder = PromptBuilder(template=prompt_template)
generator = OpenAIGenerator(
api_key="gw-sk-12345",
model="gpt-4o-mini",
api_base_url="https://gateway.example.com/v1",
)
rag = Pipeline()
rag.add_component("retriever", retriever)
rag.add_component("builder", builder)
rag.add_component("generator", generator)
rag.connect("retriever.documents", "builder.documents")
rag.connect("builder.prompt", "generator.prompt")Run it with a query:
out = rag.run({
"retriever": {"query": "What is a Haystack pipeline?"},
"builder": {"query": "What is a Haystack pipeline?"},
})
print(out["generator"]["replies"][0])The reply should reference directed graphs. That confirms the retriever→prompt→gateway path is wired correctly.
Step 4: Add fallback and client routing directives
Gateways often auto-fallback, but you can also force routing from the client. Haystack does not expose arbitrary headers on OpenAIGenerator directly, so subclass to inject a routing header the gateway honors.
from haystack.components.generators import OpenAIGenerator
import openai
class RoutedGenerator(OpenAIGenerator):
def __init__(self, route: str, **kwargs):
super().__init__(**kwargs)
self.route = route
def _get_api_client(self, **kwargs):
client = super()._get_api_client(**kwargs)
# Inject a header the gateway reads to pin a provider pool
client.default_headers.update({"X-Gateway-Route": self.route})
return client
gen = RoutedGenerator(
route="openai-only",
api_key="gw-sk-12345",
model="gpt-4o-mini",
api_base_url="https://gateway.example.com/v1",
)If the gateway supports client routing directives, the X-Gateway-Route header constrains which upstream it picks. When the designated provider is degraded, a properly configured gateway still falls back if you omit the header or set route="auto". n4n.ai honors such directives and forwards requests to the next healthy provider without changing your Haystack code.
Step 5: Forward cache-control hints and read token metering
Provider-side caching (e.g., Anthropic’s prompt caching) is exposed through extra body fields in the OpenAI-compatible schema. Pass them via generation_kwargs if your gateway forwards them.
generator = OpenAIGenerator(
api_key="gw-sk-12345",
model="claude-3-5-sonnet",
api_base_url="https://gateway.example.com/v1",
generation_kwargs={
"temperature": 0,
"extra_body": {"cache_control": {"type": "ephemeral"}},
},
)After a run, inspect the metadata Haystack returns. The gateway’s per-token usage metering shows up in the OpenAI response and is surfaced by Haystack in meta:
res = generator.run(prompt="Summarize: Haystack routes through gateways.")
meta = res["meta"][0]
print(meta["usage"])
# {'completion_tokens': 8, 'prompt_tokens': 12, 'total_tokens': 20}Log usage on every pipeline invocation. That gives you per-request cost attribution without adding a sidecar.
Step 6: Run end-to-end and verify success
Put the pieces together in a single script and assert on output shape. Verification should confirm three things: the gateway returns text, usage is present, and fallback works when you force a bad model.
def test_pipeline():
store = InMemoryDocumentStore()
writer = DocumentWriter(store)
writer.run(documents=[Document(content="Gateway fallback keeps pipelines alive.")])
rag = Pipeline()
rag.add_component("retriever", InMemoryBM25Retriever(store, top_k=1))
rag.add_component("builder", PromptBuilder(template="Context: {{documents[0].content}}\nQ: {{query}}\nA:"))
rag.add_component("generator", OpenAIGenerator(
api_key="gw-sk-12345", model="gpt-4o-mini",
api_base_url="https://gateway.example.com/v1"))
rag.connect("retriever.documents", "builder.documents")
rag.connect("builder.prompt", "generator.prompt")
out = rag.run({"retriever": {"query": "Why use a gateway?"}, "builder": {"query": "Why use a gateway?"}})
assert isinstance(out["generator"]["replies"][0], str)
assert out["generator"]["meta"][0]["usage"]["total_tokens"] > 0
print("OK:", out["generator"]["replies"][0])
if __name__ == "__main__":
test_pipeline()To prove fallback, temporarily set model="nonexistent-model". A gateway with automatic fallback will either return an error (if no mapping exists) or route to a default—your client code should handle openai.APIError and retry with route="auto". Wrap the generator call in a try/except and log the gateway’s response_headers for debugging.
Success criteria:
- The script prints a non-empty answer.
total_tokensis logged and positive.- Forcing a degraded upstream and watching the gateway recover (or return a clean 4xx) confirms the resilience path.
That is a complete Haystack pipeline OpenAI-compatible gateway integration you can extend with hybrid retrievers, evaluators, or streaming nodes without touching the network layer.