Haystack n4n.ai setup: configuring the OpenAIGenerator

Haystack’s OpenAIGenerator is the standard way to call chat and completion models in a pipeline, but pointing it at a gateway instead of OpenAI directly requires a few deliberate configuration choices. This guide walks through the complete haystack openaigenerator n4n.ai config — from dependency installation through verified pipeline execution — so you can swap providers without rewriting prompt logic or error handling.

Step 1: install the required packages

Haystack’s OpenAI integration lives in a separate package. You need both the core library and the generator component.

pip install --upgrade haystack-ai haystack-openai

If you’re on an older Haystack 1.x codebase, the import paths differ — this guide assumes Haystack 2.x (haystack-ai >= 2.0). Verify your version:

import haystack
print(haystack.__version__)  # should be 2.x

Step 2: set up authentication and base URL

n4n.ai exposes an OpenAI-compatible endpoint at https://api.n4n.ai/v1. Treat it like any other OpenAI-compatible API: pass your gateway key as the api_key and override base_url. Keep secrets out of source control — load from environment variables or a secrets manager.

import os
from haystack.components.generators import OpenAIGenerator

N4N_API_KEY = os.getenv("N4N_API_KEY")  # set in your environment
BASE_URL = "https://api.n4n.ai/v1"

generator = OpenAIGenerator(
    api_key=N4N_API_KEY,
    base_url=BASE_URL,
    model="gpt-4o-mini",  # any model the gateway serves
    generation_kwargs={
        "temperature": 0.2,
        "max_tokens": 1024,
    },
)

The model parameter accepts any identifier the gateway recognizes — OpenAI, Anthropic, Mistral, or open-weight models. Because the gateway normalizes the request format, your Haystack code stays identical regardless of which backend model you target.

Step 3: configure model routing and fallback behavior

One reason to use a gateway is automatic fallback when a provider is rate-limited or degraded. n4n.ai honors client routing directives passed via the model field or extra headers. The simplest pattern: specify a primary model and let the gateway handle fallback.

generator = OpenAIGenerator(
    api_key=N4N_API_KEY,
    base_url=BASE_URL,
    model="gpt-4o-mini",  # primary; gateway falls back per its policy
    generation_kwargs={
        "temperature": 0.2,
        "max_tokens": 1024,
        "top_p": 0.9,
    },
)

If you need explicit control, pass a routing directive in generation_kwargs using the gateway’s x-n4n-routing header (forwarded via extra_headers):

generator = OpenAIGenerator(
    api_key=N4N_API_KEY,
    base_url=BASE_URL,
    model="gpt-4o-mini",
    generation_kwargs={
        "temperature": 0.2,
        "max_tokens": 1024,
        "extra_headers": {
            "x-n4n-routing": "priority:latency;fallback:claude-3-haiku,llama-3-70b"
        },
    },
)

This tells the gateway to optimize for latency first, then fall back through the listed models in order. The header is forwarded transparently; Haystack treats it as a standard OpenAI extra_headers entry.

Step 4: enable usage metering and response metadata

The gateway returns per-token usage in the standard OpenAI usage object. OpenAIGenerator surfaces this in the meta field of each Generation object. Capture it for cost tracking or observability.

from haystack import Pipeline
from haystack.components.builders import ChatPromptBuilder
from haystack.dataclasses import ChatMessage

pipe = Pipeline()
pipe.add_component("prompt", ChatPromptBuilder(template=[
    ChatMessage.from_system("You are a concise technical assistant."),
    ChatMessage.from_user("{{question}}"),
]))
pipe.add_component("llm", generator)
pipe.connect("prompt.prompt", "llm.messages")

result = pipe.run({
    "prompt": {"question": "Explain the difference between a mutex and a semaphore in three sentences."}
})

for gen in result["llm"]["replies"]:
    print(gen.text)
    print("Usage:", gen.meta.get("usage"))

Typical usage payload:

{
  "prompt_tokens": 42,
  "completion_tokens": 87,
  "total_tokens": 129,
  "model": "gpt-4o-mini",
  "provider": "openai"
}

The provider field tells you which upstream model actually served the request — useful when fallback occurs.

Step 5: handle streaming responses

OpenAIGenerator supports streaming via the streaming_callback parameter. This works unchanged with the gateway because the SSE format is identical.

def print_token(token: str) -> None:
    print(token, end="", flush=True)

streaming_generator = OpenAIGenerator(
    api_key=N4N_API_KEY,
    base_url=BASE_URL,
    model="gpt-4o-mini",
    streaming_callback=print_token,
    generation_kwargs={"temperature": 0.2, "max_tokens": 512},
)

pipe.replace_component("llm", streaming_generator)
pipe.run({"prompt": {"question": "Write a haiku about distributed systems."}})

Streaming is especially valuable when fallback triggers — the gateway switches providers mid-stream without breaking the SSE connection, so your callback continues receiving tokens uninterrupted.

Step 6: verify the integration end to end

Run a minimal self-test that exercises authentication, routing, usage capture, and error handling. Save this as verify_n4n.py:

#!/usr/bin/env python3
"""Verify haystack openaigenerator n4n.ai config works end to end."""
import os
import sys
from haystack import Pipeline
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import ChatPromptBuilder
from haystack.dataclasses import ChatMessage

def main() -> int:
    api_key = os.getenv("N4N_API_KEY")
    if not api_key:
        print("ERROR: N4N_API_KEY not set", file=sys.stderr)
        return 1

    generator = OpenAIGenerator(
        api_key=api_key,
        base_url="https://api.n4n.ai/v1",
        model="gpt-4o-mini",
        generation_kwargs={"temperature": 0.1, "max_tokens": 64},
    )

    pipe = Pipeline()
    pipe.add_component("prompt", ChatPromptBuilder(template=[
        ChatMessage.from_system("Reply with only the word 'ok'."),
        ChatMessage.from_user("test"),
    ]))
    pipe.add_component("llm", generator)
    pipe.connect("prompt.prompt", "llm.messages")

    try:
        result = pipe.run({})
    except Exception as e:
        print(f"ERROR: pipeline failed: {e}", file=sys.stderr)
        return 2

    replies = result["llm"]["replies"]
    if not replies:
        print("ERROR: no generations returned", file=sys.stderr)
        return 3

    gen = replies[0]
    print(f"Response: {gen.text.strip()}")
    usage = gen.meta.get("usage")
    if usage:
        print(f"Tokens: {usage['total_tokens']} (prompt={usage['prompt_tokens']}, completion={usage['completion_tokens']})")
        print(f"Provider: {usage.get('provider', 'unknown')}")
    else:
        print("WARNING: usage metadata missing")

    return 0

if __name__ == "__main__":
    sys.exit(main())

Run it:

export N4N_API_KEY="your-gateway-key"
python verify_n4n.py

Expected output:

Response: ok
Tokens: 18 (prompt=12, completion=6)
Provider: openai

If you see a different provider (e.g., anthropic or together), fallback activated — the gateway served the request from an alternate backend. That’s the integration working as designed.

Step 7: common pitfalls and fixes

Wrong base URL path

The gateway expects /v1 in the base URL. Omitting it produces 404s on /chat/completions.

# Wrong
base_url="https://api.n4n.ai"

# Correct
base_url="https://api.n4n.ai/v1"

Missing extra_headers for routing directives

If you pass routing instructions in generation_kwargs without extra_headers, they’re sent as JSON body fields and ignored by the gateway. Always nest under extra_headers.

# Wrong
generation_kwargs={"x-n4n-routing": "..."}

# Correct
generation_kwargs={"extra_headers": {"x-n4n-routing": "..."}}

Model name not recognized

The gateway validates model identifiers against its catalog. If you request a model the gateway doesn’t serve, you’ll get a 400 with a list of available models. Query the catalog programmatically if you need dynamic selection:

curl -H "Authorization: Bearer $N4N_API_KEY" https://api.n4n.ai/v1/models

Streaming callback signature mismatch

Haystack 2.x passes a single string token to the callback. A callback expecting ChatChunk or a dict will raise. Keep it simple:

def on_token(token: str) -> None:
    print(token, end="", flush=True)

Step 8: production hardening

Retries and timeouts

OpenAIGenerator uses the OpenAI Python SDK under the hood, which respects timeout and max_retries in generation_kwargs. Set both explicitly for gateway traffic:

generator = OpenAIGenerator(
    api_key=N4N_API_KEY,
    base_url=BASE_URL,
    model="gpt-4o-mini",
    generation_kwargs={
        "temperature": 0.2,
        "max_tokens": 1024,
        "timeout": 30.0,          # seconds
        "max_retries": 3,
    },
)

The gateway’s own fallback is faster than client-side retries for provider outages, but retries still help with transient network blips.

Structured logging

Wrap the generator in a component that logs request/response metadata without duplicating pipeline logic:

import logging
from haystack import component
from haystack.dataclasses import ChatMessage
from typing import List

logger = logging.getLogger("n4n.generator")

@component
class LoggedGenerator:
    def __init__(self, generator: OpenAIGenerator):
        self.generator = generator

    @component.output_types(replies=List[ChatMessage], meta=List[dict])
    def run(self, messages: List[ChatMessage], generation_kwargs: dict | None = None):
        logger.info("request", extra={"model": self.generator.model, "msg_count": len(messages)})
        result = self.generator.run(messages, generation_kwargs)
        for gen in result["replies"]:
            usage = gen.meta.get("usage")
            if usage:
                logger.info("response", extra={
                    "provider": usage.get("provider"),
                    "total_tokens": usage.get("total_tokens"),
                    "model": usage.get("model"),
                })
        return result

Drop this into any pipeline in place of the raw generator — same interface, observable output.

Step 9: switching models without code changes

Because the gateway normalizes model identifiers, you can externalize model selection entirely. Store the model name in config or feature flags:

import json

with open("model_config.json") as f:
    config = json.load(f)

generator = OpenAIGenerator(
    api_key=N4N_API_KEY,
    base_url=BASE_URL,
    model=config["default_model"],  # e.g., "gpt-4o-mini" or "claude-3-haiku"
    generation_kwargs={"temperature": config.get("temperature", 0.2)},
)

Changing model_config.json from "gpt-4o-mini" to "claude-3-haiku" reroutes all traffic — no pipeline edits, no redeploys. The gateway handles the provider translation.

Verification checklist

Before considering the integration done, confirm each item:

• verify_n4n.py exits 0 and prints usage metadata
• Streaming callback receives tokens without error
• Fallback triggers cleanly (simulate by requesting a model the gateway maps to a secondary provider)
• Retries fire on induced network latency (tc qdisc add dev lo root netem delay 200ms)
• Structured logs capture provider and token counts per request
• Model switch via config file works without code change

That’s the complete haystack openaigenerator n4n.ai config. The gateway becomes a transparent routing layer — your pipelines stay pure Haystack, and provider decisions move to configuration where they belong.