Haystack n4n.ai setup: switching models without code changes

If you’re building with Haystack and want to swap models without touching pipeline code, the haystack n4n.ai switch models no code pattern lets you route every request through a single OpenAI-compatible endpoint. You define the model at runtime via environment variable or request header, and the gateway handles provider fallback, caching hints, and per-token metering automatically. This walkthrough shows the complete setup from empty directory to a verified multi-model pipeline.

Step 1: Set up the project environment

Create a clean virtual environment and install Haystack with the OpenAI generator integration. The OpenAI generator is what speaks the compatible protocol.

python -m venv .venv
source .venv/bin/activate
pip install --upgrade pip
pip install "haystack-ai[openai]" python-dotenv

Verify the imports work:

# verify_imports.py
from haystack import Pipeline
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import PromptBuilder
print("Imports OK")

Run it:

python verify_imports.py

You should see Imports OK with no errors.

Step 2: Configure n4n.ai credentials

The gateway uses a single API key. Store it in .env so it never hits source control.

# .env
N4N_API_KEY=sk-your-key-here
N4N_BASE_URL=https://api.n4n.ai/v1
DEFAULT_MODEL=meta-llama/llama-3.1-8b-instruct

Load it in your application bootstrap:

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

N4N_API_KEY = os.getenv("N4N_API_KEY")
N4N_BASE_URL = os.getenv("N4N_BASE_URL", "https://api.n4n.ai/v1")
DEFAULT_MODEL = os.getenv("DEFAULT_MODEL", "meta-llama/llama-3.1-8b-instruct")

if not N4N_API_KEY:
    raise RuntimeError("N4N_API_KEY not set in environment")

Step 3: Build the reusable generator component

Create a thin wrapper around OpenAIGenerator that reads the model from an environment variable at initialization time. This is the only place the model name lives in your code.

# generator.py
from haystack.components.generators import OpenAIGenerator
from config import N4N_API_KEY, N4N_BASE_URL, DEFAULT_MODEL

def build_generator(model: str | None = None) -> OpenAIGenerator:
    """
    Returns an OpenAIGenerator pointed at the n4n.ai gateway.
    The model can be overridden per-call via generation_kwargs.
    """
    return OpenAIGenerator(
        api_key=N4N_API_KEY,
        api_base_url=N4N_BASE_URL,
        model=model or DEFAULT_MODEL,
        generation_kwargs={
            "temperature": 0.2,
            "max_tokens": 512,
        },
    )

Note: generation_kwargs are defaults. You can still override model per request by passing generation_kwargs={"model": "other/model"} when you run() the pipeline — this is how you achieve zero-code model switching.

Step 4: Assemble the pipeline

A minimal RAG-style pipeline: prompt builder → generator. Save this so you can import it from tests and your application.

# pipeline.py
from haystack import Pipeline
from haystack.components.builders import PromptBuilder
from generator import build_generator

PROMPT_TEMPLATE = """
Answer the question using only the context below.
If the context is insufficient, say you don't know.

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

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

def build_pipeline(model: str | None = None) -> Pipeline:
    pipe = Pipeline()
    pipe.add_component("prompt", PromptBuilder(template=PROMPT_TEMPLATE))
    pipe.add_component("llm", build_generator(model))
    pipe.connect("prompt", "llm")
    return pipe

Step 5: Switch models without code changes

There are three ways to change the model — pick the one that fits your deployment model.

Option A: Environment variable (deploy-time)

Change DEFAULT_MODEL in .env and restart the process. No Python changes.

# .env
DEFAULT_MODEL=mistralai/mixtral-8x7b-instruct

Option B: Per-request override (runtime)

Pass the model in generation_kwargs when you invoke the pipeline. This is useful for A/B testing or per-tenant model selection.

# run_with_override.py
from pipeline import build_pipeline

pipe = build_pipeline()  # uses DEFAULT_MODEL from env

result = pipe.run({
    "prompt": {
        "documents": [{"content": "Paris is the capital of France."}],
        "question": "What is the capital of France?"
    },
    "llm": {
        "generation_kwargs": {"model": "anthropic/claude-3.5-sonnet"}
    }
})

print(result["llm"]["replies"][0])

Option C: Programmatic default (config-driven)

Read the model from a config file or feature flag at startup.

# app.py
import yaml
from pipeline import build_pipeline

with open("models.yaml") as f:
    config = yaml.safe_load(f)

pipe = build_pipeline(model=config["active_model"])

models.yaml:

active_model: "google/gemma-2-9b-it"

All three approaches require zero changes to the pipeline definition. The generator stays the same; only the model identifier changes.

Step 6: Verify the setup end to end

Create a test script that exercises both the default model and an override.

# test_switching.py
import os
from pipeline import build_pipeline

def test_default_model():
    pipe = build_pipeline()
    out = pipe.run({
        "prompt": {
            "documents": [{"content": "Water boils at 100°C at sea level."}],
            "question": "At what temperature does water boil at sea level?"
        }
    })
    reply = out["llm"]["replies"][0]
    assert "100" in reply or "100°C" in reply
    print(f"[default] {reply[:80]}...")

def test_override_model():
    pipe = build_pipeline()
    out = pipe.run({
        "prompt": {
            "documents": [{"content": "The speed of light is ~299,792 km/s."}],
            "question": "What is the speed of light?"
        },
        "llm": {
            "generation_kwargs": {"model": os.getenv("OVERRIDE_MODEL", "mistralai/mixtral-8x7b-instruct")}
        }
    })
    reply = out["llm"]["replies"][0]
    assert "299" in reply or "300" in reply
    print(f"[override] {reply[:80]}...")

if __name__ == "__main__":
    test_default_model()
    test_override_model()
    print("All verification tests passed")

Run it:

OVERRIDE_MODEL=mistralai/mixtral-8x7b-instruct python test_switching.py

Expected output (truncated):

[default] Water boils at 100°C at sea level...
[override] The speed of light is approximately 299,792 kilometers per second...
All verification tests passed

If you see both replies, the haystack n4n.ai switch models no code flow is working: the same pipeline object produced answers from two different models.

Step 7: Observe routing and fallback in production

The gateway returns headers that tell you which provider actually served the request. Capture them for observability.

# logging_middleware.py
import logging
from haystack import component
from haystack.dataclasses import StreamingChunk

@component
class GenerationLogger:
    @component.output_types(replies=list[str])
    def run(self, replies: list[str], meta: list[dict] | None = None):
        if meta:
            for m in meta:
                provider = m.get("provider", "unknown")
                model = m.get("model", "unknown")
                latency_ms = m.get("latency_ms", 0)
                logging.info(f"provider={provider} model={model} latency_ms={latency_ms}")
        return {"replies": replies}

Insert it after the generator:

# pipeline_with_logging.py
from haystack import Pipeline
from haystack.components.builders import PromptBuilder
from generator import build_generator
from logging_middleware import GenerationLogger

def build_pipeline(model: str | None = None) -> Pipeline:
    pipe = Pipeline()
    pipe.add_component("prompt", PromptBuilder(template=PROMPT_TEMPLATE))
    pipe.add_component("llm", build_generator(model))
    pipe.add_component("logger", GenerationLogger())
    pipe.connect("prompt", "llm")
    pipe.connect("llm.replies", "logger.replies")
    pipe.connect("llm.meta", "logger.meta")
    return pipe

The meta field includes provider-level details the gateway forwards — useful for dashboards and alerting on fallback events.

Step 8: Handle streaming if your UI needs it

OpenAIGenerator supports streaming out of the box. Enable it in generation_kwargs and iterate chunks.

# stream_example.py
from pipeline import build_pipeline

pipe = build_pipeline()

for chunk in pipe.run({
    "prompt": {
        "documents": [{"content": "Streaming works chunk by chunk."}],
        "question": "Explain streaming in one sentence."
    },
    "llm": {
        "generation_kwargs": {"stream": True, "model": "meta-llama/llama-3.1-8b-instruct"}
    }
}, include_outputs_from=["llm"]):
    if "llm" in chunk and "replies" in chunk["llm"]:
        for reply in chunk["llm"]["replies"]:
            if isinstance(reply, str):
                print(reply, end="", flush=True)
            elif hasattr(reply, "content"):
                print(reply.content, end="", flush=True)
print()

Streaming works identically regardless of which model the gateway routes to.

Step 9: Pin model families for consistent behavior

Different model families have different instruction formats and context windows. If your prompt templates assume a specific style (e.g., chat vs. completion), constrain the allowed models to a compatible set.

# model_policy.py
ALLOWED_MODELS = {
    "chat": [
        "meta-llama/llama-3.1-8b-instruct",
        "meta-llama/llama-3.1-70b-instruct",
        "mistralai/mixtral-8x7b-instruct",
        "google/gemma-2-9b-it",
    ],
    "code": [
        "deepseek/deepseek-coder-v2-instruct",
        "qwen/qwen2.5-coder-32b-instruct",
    ],
}

def validate_model(model: str, family: str = "chat") -> str:
    if model not in ALLOWED_MODELS[family]:
        raise ValueError(f"Model {model} not allowed for {family} family. Allowed: {ALLOWED_MODELS[family]}")
    return model

Call validate_model before passing the model to the pipeline. This prevents silent degradation when someone points the gateway at a base model that doesn’t follow chat formatting.

Step 10: Deploy with confidence

Containerize the service. The only runtime configuration is the .env file (or your orchestrator’s secret store).

# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
ENV PYTHONUNBUFFERED=1
CMD ["python", "app.py"]

requirements.txt:

haystack-ai[openai]==2.6.0
python-dotenv==1.0.1
pyyaml==6.0.1

Deploy the same image to staging and production. Swap models by updating the DEFAULT_MODEL env var in your platform (Kubernetes ConfigMap, ECS task definition, Fly.io secrets, etc.) and rolling the deployment. No code rebuild, no pipeline migration.

Quick reference: what changes when you switch models

Switch method	Where the model name lives	Restart required?
`.env` `DEFAULT_MODEL`	Environment file	Yes (process restart)
`generation_kwargs={"model": "..."}`	Request payload	No
Config file read at startup	YAML/JSON/feature flag	Yes (process restart)

All three use the same pipeline code. The gateway handles provider auth, rate-limit fallback, and cache-control forwarding — your Haystack components stay oblivious.

Troubleshooting checklist

401 Unauthorized: Verify N4N_API_KEY is set and valid.
404 Model not found: The model identifier must match the gateway’s catalog exactly (e.g., meta-llama/llama-3.1-8b-instruct, not llama3.1).
Empty replies: Check meta for finish_reason == "length" — increase max_tokens.
Streaming hangs: Ensure your ASGI server (uvicorn, gunicorn+uvicorn workers) has proxy_read_timeout high enough for long generations.
Fallback not triggering: The gateway falls back on provider errors (5xx, timeouts). Client-side 4xx (bad request, invalid model) do not trigger fallback.

You now have a Haystack pipeline that treats the model as a configuration parameter, not a code dependency. Change the model in .env, in a request header, or in a feature flag — the pipeline definition never moves. That’s the haystack n4n.ai switch models no code pattern in production.