Your first Haystack pipeline with n4n.ai and Qwen3

You can wire a haystack pipeline n4n.ai qwen3 integration in about fifty lines of code. This tutorial walks through a complete retrieval-augmented generation setup: a document store, a sparse retriever, and a generator pointed at n4n.ai’s unified endpoint. You’ll see the exact component configuration, the pipeline definition, and the expected output at each verification step.

Prerequisites

• Python 3.10+
• An n4n.ai API key (set as N4N_API_KEY in your environment)
• Haystack 2.6+ installed: pip install haystack-ai
• Optional: pip install rank-bm25 for the sparse retriever used below

The code assumes you have a .env file or shell export for N4N_API_KEY. No other secrets are required.

Install and verify imports

# verify_imports.py
import os
import haystack
from haystack import Pipeline, Document
from haystack.components.generators import OpenAIGenerator
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.builders import PromptBuilder

print(f"Haystack version: {haystack.__version__}")
print(f"API key present: {'N4N_API_KEY' in os.environ}")

Run it:

python verify_imports.py

Expected output:

Haystack version: 2.6.0
API key present: True

Configure the document store and retriever

Haystack’s InMemoryDocumentStore is fine for prototypes and small corpora. For production you’d swap in PostgreSQL with pgvector, Weaviate, or Qdrant — the pipeline definition stays the same.

# build_store.py
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack import Document

doc_store = InMemoryDocumentStore()

docs = [
    Document(content="Qwen3 is a family of large language models developed by Alibaba Cloud.", meta={"source": "alibaba-announcement"}),
    Document(content="The Qwen3 series includes base and instruction-tuned variants ranging from 0.5B to 235B parameters.", meta={"source": "technical-report"}),
    Document(content="Qwen3 models support 128K context length and are trained on multilingual data covering 100+ languages.", meta={"source": "technical-report"}),
    Document(content="n4n.ai provides an OpenAI-compatible endpoint that routes to 240+ models including Qwen3 variants.", meta={"source": "n4n-docs"}),
    Document(content="Automatic fallback activates when a provider is rate-limited or degraded, preserving uptime.", meta={"source": "n4n-docs"}),
]

doc_store.write_documents(docs)
retriever = InMemoryBM25Retriever(document_store=doc_store, top_k=3)

# Quick sanity check
results = retriever.run(query="What is Qwen3 context length?")
for doc in results["documents"]:
    print(f"[{doc.score:.3f}] {doc.content[:80]}...")

Run it:

python build_store.py

Expected output (scores will vary slightly):

[0.842] Qwen3 models support 128K context length and are trained on multilingual data covering 100+ languages.
[0.421] The Qwen3 series includes base and instruction-tuned variants ranging from 0.5B to 235B parameters.
[0.210] Qwen3 is a family of large language models developed by Alibaba Cloud.

Wire the generator to n4n.ai

The OpenAIGenerator component works with any OpenAI-compatible endpoint. Point base_url at n4n.ai and pass the model identifier you want — here qwen3-235b-a22b-instruct-2507. The component honors standard OpenAI parameters (temperature, max_tokens, stream) and forwards provider cache-control hints when present.

# generator_setup.py
import os
from haystack.components.generators import OpenAIGenerator

generator = OpenAIGenerator(
    api_key=os.environ["N4N_API_KEY"],
    base_url="https://api.n4n.ai/v1",
    model="qwen3-235b-a22b-instruct-2507",
    generation_kwargs={
        "temperature": 0.2,
        "max_tokens": 512,
    },
)

# Smoke test
response = generator.run(prompt="Reply with exactly: OK")
print(response["replies"][0])

Run it:

python generator_setup.py

Expected output:

OK

If you see a 401, verify the API key. If you see a 404 on the model, check the exact model slug in the n4n.ai model catalog — model IDs are case-sensitive.

Build the prompt template

Haystack’s PromptBuilder uses Jinja2. Keep the template strict: instruct the model to cite only the provided context, and to say “I don’t know” when the answer isn’t there. This reduces hallucination without needing a separate guardrail component.

# prompt_template.py
from haystack.components.builders import PromptBuilder

template = """
You are a precise technical assistant. Answer the question using only the provided context.
If the context does not contain the answer, reply exactly: I don't know.

Context:
{% for doc in documents %}
[{{ loop.index }}] {{ doc.content }}
{% endfor %}

Question: {{ query }}

Answer:
"""

prompt_builder = PromptBuilder(template=template)

# Test the rendered prompt
rendered = prompt_builder.run(
    query="What is Qwen3 context length?",
    documents=[
        type("Doc", (), {"content": "Qwen3 models support 128K context length."})(),
    ],
)
print(rendered["prompt"])

Run it:

python prompt_template.py

Expected output (whitespace normalized):

You are a precise technical assistant. Answer the question using only the provided context.
If the context does not contain the answer, reply exactly: I don't know.

Context:
[1] Qwen3 models support 128K context length.

Question: What is Qwen3 context length?

Answer:

Assemble the pipeline

Haystack 2.x pipelines are directed acyclic graphs. Connect retriever.documents → prompt_builder.documents, and prompt_builder.prompt → generator.prompt. The query input fans out to both retriever and prompt builder.

# pipeline.py
import os
from haystack import 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.generators import OpenAIGenerator
from haystack import Document

# --- Document store ---
doc_store = InMemoryDocumentStore()
docs = [
    Document(content="Qwen3 is a family of large language models developed by Alibaba Cloud.", meta={"source": "alibaba-announcement"}),
    Document(content="The Qwen3 series includes base and instruction-tuned variants ranging from 0.5B to 235B parameters.", meta={"source": "technical-report"}),
    Document(content="Qwen3 models support 128K context length and are trained on multilingual data covering 100+ languages.", meta={"source": "technical-report"}),
    Document(content="n4n.ai provides an OpenAI-compatible endpoint that routes to 240+ models including Qwen3 variants.", meta={"source": "n4n-docs"}),
    Document(content="Automatic fallback activates when a provider is rate-limited or degraded, preserving uptime.", meta={"source": "n4n-docs"}),
]
doc_store.write_documents(docs)

# --- Components ---
retriever = InMemoryBM25Retriever(document_store=doc_store, top_k=3)

template = """
You are a precise technical assistant. Answer the question using only the provided context.
If the context does not contain the answer, reply exactly: I don't know.

Context:
{% for doc in documents %}
[{{ loop.index }}] {{ doc.content }}
{% endfor %}

Question: {{ query }}

Answer:
"""
prompt_builder = PromptBuilder(template=template)

generator = OpenAIGenerator(
    api_key=os.environ["N4N_API_KEY"],
    base_url="https://api.n4n.ai/v1",
    model="qwen3-235b-a22b-instruct-2507",
    generation_kwargs={"temperature": 0.2, "max_tokens": 512},
)

# --- Pipeline ---
rag = Pipeline()
rag.add_component("retriever", retriever)
rag.add_component("prompt_builder", prompt_builder)
rag.add_component("generator", generator)

rag.connect("retriever.documents", "prompt_builder.documents")
rag.connect("prompt_builder.prompt", "generator.prompt")

# --- Run ---
question = "What is the context length of Qwen3 models?"
result = rag.run({"retriever": {"query": question}, "prompt_builder": {"query": question}})

print("=== Retrieved documents ===")
for i, doc in enumerate(result["retriever"]["documents"], 1):
    print(f"[{i}] {doc.content}")

print("\n=== Generated answer ===")
print(result["generator"]["replies"][0])

Run it:

python pipeline.py

Expected output:

=== Retrieved documents ===
[1] Qwen3 models support 128K context length and are trained on multilingual data covering 100+ languages.
[2] The Qwen3 series includes base and instruction-tuned variants ranging from 0.5B to 235B parameters.
[3] Qwen3 is a family of large language models developed by Alibaba Cloud.

=== Generated answer ===
Qwen3 models support 128K context length.

Add streaming for lower perceived latency

For chat interfaces, stream tokens as they arrive. OpenAIGenerator supports streaming_callback — pass a callable that receives each chunk. The pipeline returns the full concatenated reply in replies[0] regardless.

# streaming.py
import os
import sys
from haystack import 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.generators import OpenAIGenerator
from haystack import Document

doc_store = InMemoryDocumentStore()
doc_store.write_documents([
    Document(content="Qwen3 models support 128K context length and are trained on multilingual data covering 100+ languages."),
])

retriever = InMemoryBM25Retriever(document_store=doc_store, top_k=1)

template = """
Answer using only the context. If unknown, say: I don't know.

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

Question: {{ query }}
Answer:
"""
prompt_builder = PromptBuilder(template=template)

def print_token(chunk: str) -> None:
    sys.stdout.write(chunk)
    sys.stdout.flush()

generator = OpenAIGenerator(
    api_key=os.environ["N4N_API_KEY"],
    base_url="https://api.n4n.ai/v1",
    model="qwen3-235b-a22b-instruct-2507",
    generation_kwargs={"temperature": 0.2, "max_tokens": 256},
    streaming_callback=print_token,
)

rag = Pipeline()
rag.add_component("retriever", retriever)
rag.add_component("prompt_builder", prompt_builder)
rag.add_component("generator", generator)
rag.connect("retriever.documents", "prompt_builder.documents")
rag.connect("prompt_builder.prompt", "generator.prompt")

print("Streaming answer: ", end="", flush=True)
result = rag.run({"retriever": {"query": "Context length?"}, "prompt_builder": {"query": "Context length?"}})
print(f"\n\nFull reply captured: {result['generator']['replies'][0][:60]}...")

Run it:

python streaming.py

Expected output (tokens appear incrementally):

Streaming answer: Qwen3 models support 128K context length.

Full reply captured: Qwen3 models support 128K context length.

Swap the retriever for dense retrieval (optional)

BM25 works well for keyword overlap. For semantic matching, replace InMemoryBM25Retriever with InMemoryEmbeddingRetriever backed by a sentence-transformers model. The pipeline connections don’t change.

# dense_retriever.py
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.components.embedders import SentenceTransformersDocumentEmbedder, SentenceTransformersTextEmbedder
from haystack import Document, Pipeline

doc_store = InMemoryDocumentStore(embedding_similarity_function="cosine")

# Embed documents at index time
doc_embedder = SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
docs = [Document(content=t) for t in [
    "Qwen3 models support 128K context length.",
    "Qwen3 was released by Alibaba Cloud in 2024.",
]]
docs_with_emb = doc_embedder.run(docs)["documents"]
doc_store.write_documents(docs_with_emb)

# Pipeline with query embedder + dense retriever
query_embedder = SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
retriever = InMemoryEmbeddingRetriever(document_store=doc_store, top_k=2)

p = Pipeline()
p.add_component("query_embedder", query_embedder)
p.add_component("retriever", retriever)
p.connect("query_embedder.embedding", "retriever.query_embedding")

result = p.run({"query_embedder": {"text": "What is the context window?"}})
for d in result["retriever"]["documents"]:
    print(f"[{d.score:.3f}] {d.content}")

Run it:

python dense_retriever.py

Expected output:

[0.712] Qwen3 models support 128K context length.
[0.341] Qwen3 was released by Alibaba Cloud in 2024.

Observability: capture usage and routing metadata

n4n.ai returns standard OpenAI usage fields (prompt_tokens, completion_tokens, total_tokens) plus provider-level metadata in the response headers. Haystack surfaces the raw response in generator.meta. Log this for cost tracking and fallback debugging.

# usage_logging.py
import os
from haystack import 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.generators import OpenAIGenerator
from haystack import Document

doc_store = InMemoryDocumentStore()
doc_store.write_documents([Document(content="Qwen3 supports 128K context.")])

retriever = InMemoryBM25Retriever(document_store=doc_store, top_k=1)
prompt_builder = PromptBuilder(template="Context: {{ documents[0].content }}\nQ: {{ query }}\nA:")
generator = OpenAIGenerator(
    api_key=os.environ["N4N_API_KEY"],
    base_url="https://api.n4n.ai/v1",
    model="qwen3-235b-a22b-instruct-2507",
)

rag = Pipeline()
rag.add_component("retriever", retriever)
rag.add_component("prompt_builder", prompt_builder)
rag.add_component("generator", generator)
rag.connect("retriever.documents", "prompt_builder.documents")
rag.connect("prompt_builder.prompt", "generator.prompt")

result = rag.run({"retriever": {"query": "context length"}, "prompt_builder": {"query": "context length"}})

meta = result["generator"]["meta"][0]
usage = meta.get("usage", {})
print(f"Prompt tokens: {usage.get('prompt_tokens')}")
print(f"Completion tokens: {usage.get('completion_tokens')}")
print(f"Total tokens: {usage.get('total_tokens')}")
print(f"Model used: {meta.get('model')}")
print(f"Provider: {meta.get('provider', 'unknown')}")

Run it:

python usage_logging.py

Expected output:

Prompt tokens: 142
Completion tokens: 18
Total tokens: 160
Model used: qwen3-235b-a22b-instruct-2507
Provider: together

The provider field tells you which upstream served the request — useful when automatic fallback kicks in.

Common failure modes and fixes

Next steps

• Persist the document store: swap InMemoryDocumentStore for PgvectorDocumentStore or WeaviateDocumentStore — only the store instantiation changes.
• Add a DocumentCleaner and DocumentSplitter before indexing to handle PDFs and long-form content.
• Insert a Ranker (cross-encoder or LLM-based) between retriever and prompt builder for higher precision.
• Wrap the pipeline in a FastAPI or Litestar service; expose /query with request validation and structured logging.
• Implement eval harness: feed a golden QA set, measure retrieval recall@k and answer correctness with an LLM judge.

The haystack pipeline n4n.ai qwen3 pattern scales from notebook to production with minimal structural changes. The component graph stays stable; you swap implementations at the edges.