Streaming responses in Haystack RAG pipelines with n4n.ai

Streaming responses in a Haystack RAG pipeline lets you show users partial answers while the model is still generating, cutting perceived latency dramatically. This guide walks through wiring a streaming responses haystack rag pipeline end to end, using n4n.ai as the LLM backend. You’ll get runnable code, a verification checklist, and the error-handling patterns that keep production traffic healthy.

Step 1: Install the right dependencies

Haystack 2.x moved streaming into the core Generator interface. You need the OpenAI generator (n4n.ai speaks the OpenAI API) and a document store. Pin versions to avoid surprise breakage.

pip install --upgrade "haystack-ai>=2.6.0" "openai>=1.30.0" "python-dotenv>=1.0.0"

If you prefer a vector store other than the in-memory default, add its package now (weaviate-haystack, qdrant-haystack, pgvector-haystack, etc.). The code below uses InMemoryDocumentStore so you can run it without extra infrastructure.

Step 2: Configure n4n.ai as the LLM provider

n4n.ai exposes a single OpenAI-compatible endpoint. Treat it like any other OpenAI base URL — just point the generator at it and pass your n4n.ai API key. Keep secrets out of source control.

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

N4N_API_KEY = os.getenv("N4N_API_KEY")          # your n4n.ai key
N4N_BASE_URL = os.getenv("N4N_BASE_URL", "https://api.n4n.ai/v1")
MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o-mini")  # any of the 240+ models n4n.ai routes to

# llm.py
from haystack.components.generators import OpenAIGenerator
from config import N4N_API_KEY, N4N_BASE_URL, MODEL_NAME

def build_streaming_generator() -> OpenAIGenerator:
    """
    Returns a generator that streams tokens from n4n.ai.
    The `streaming_callback` is wired in Step 4.
    """
    return OpenAIGenerator(
        api_key=N4N_API_KEY,
        api_base_url=N4N_BASE_URL,
        model=MODEL_NAME,
        generation_kwargs={
            "temperature": 0.2,
            "max_tokens": 1024,
            "stream": True,                 # critical: enables server-sent events
        },
    )

The stream=True flag tells the OpenAI client to return an async iterator instead of blocking on the full completion. n4n.ai honors this and forwards provider cache-control hints so you can reason about latency per hop.

Step 3: Build the RAG pipeline with streaming components

A minimal streaming RAG pipeline has four nodes: Retriever, PromptBuilder, Generator, and an OutputAdapter that shapes the generator’s stream into your application’s format. Wire them in a Pipeline and declare the input/output sockets explicitly — this makes debugging easier when you add branches later.

# pipeline.py
from haystack import Pipeline, Document
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.components.builders import PromptBuilder
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.embedders import SentenceTransformersTextEmbedder
from llm import build_streaming_generator

# 1. Document store + sample data (replace with your real index)
document_store = InMemoryDocumentStore()
document_store.write_documents([
    Document(content="Haystack 2.x introduces native streaming support via the Generator interface."),
    Document(content="n4n.ai provides a single OpenAI-compatible endpoint for 240+ models with automatic fallback."),
    Document(content="Streaming reduces time-to-first-token to under 200 ms in typical deployments."),
])

# 2. Embedder for the query
query_embedder = SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")

# 3. Retriever
retriever = InMemoryEmbeddingRetriever(document_store=document_store, top_k=3)

# 4. Prompt template — keep it tight for streaming
prompt_template = """
Answer the question using only the provided context.
Context:
{% for doc in documents %}
  {{ doc.content }}
{% endfor %}

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

# 5. Generator (streaming)
generator = build_streaming_generator()

# 6. Assemble pipeline
rag_pipeline = Pipeline()
rag_pipeline.add_component("query_embedder", query_embedder)
rag_pipeline.add_component("retriever", retriever)
rag_pipeline.add_component("prompt_builder", prompt_builder)
rag_pipeline.add_component("generator", generator)

rag_pipeline.connect("query_embedder.embedding", "retriever.query_embedding")
rag_pipeline.connect("retriever.documents", "prompt_builder.documents")
rag_pipeline.connect("prompt_builder.prompt", "generator.prompt")

# Expose a clean run interface
def run_streaming_rag(question: str, streaming_callback):
    """
    Executes the pipeline and yields tokens via streaming_callback.
    Returns the full accumulated answer for logging/archival.
    """
    # The generator expects the callback at runtime, not construction time.
    # We inject it through generation_kwargs override.
    result = rag_pipeline.run(
        data={
            "query_embedder": {"text": question},
            "generator": {"generation_kwargs": {"streaming_callback": streaming_callback}},
        },
        include_outputs_from={"generator"},
    )
    return result["generator"]["replies"][0]

Note the include_outputs_from={"generator"} — without it, the pipeline only returns the last component’s output, and you lose access to the generator’s metadata (finish reason, token counts, etc.).

Step 4: Wire up the streaming callback

Haystack’s OpenAIGenerator accepts a streaming_callback callable with signature Callable[[str], None]. It fires once per token chunk. In a CLI you can print directly; in a web server you’d push to a WebSocket or SSE response.

# callbacks.py
import sys
from typing import Callable

def cli_streaming_callback(chunk: str) -> None:
    """Simple callback that writes tokens to stdout without buffering."""
    sys.stdout.write(chunk)
    sys.stdout.flush()

def sse_streaming_callback(chunk: str) -> Callable[[], None]:
    """
    Returns a callback suitable for Starlette/FastAPI SSE streaming.
    Usage:
        async def event_generator():
            callback = sse_streaming_callback("")
            # ... run pipeline with callback ...
        return EventSourceResponse(event_generator())
    """
    # In practice you'd close over a queue or async iterator.
    # This skeleton shows the shape; adapt to your framework.
    def _inner(token: str) -> None:
        # push token to your transport layer here
        pass
    return _inner

For a quick CLI verification, the cli_streaming_callback is enough. In production, wrap the callback so it can handle backpressure (see Step 6).

Step 5: Run and verify

Create a tiny entry point that exercises the whole path. This is your smoke test — run it every deploy.

# main.py
from pipeline import run_streaming_rag
from callbacks import cli_streaming_callback

if __name__ == "__main__":
    question = "How does Haystack 2.x handle streaming?"
    print(f"Q: {question}\nA: ", end="", flush=True)
    full_answer = run_streaming_rag(question, cli_streaming_callback)
    print(f"\n\n[Done] Full answer length: {len(full_answer)} chars")

Run it:

python main.py

Verification checklist

• Tokens appear incrementally, not in one burst after a long pause.
• The final printed answer matches the accumulated full_answer variable.
• No openai.BadRequestError about stream parameter — confirms n4n.ai accepted the flag.
• Latency: time-to-first-token should be well under 500 ms on a warm connection.

If you see a single block after a delay, double-check generation_kwargs={"stream": True} on the generator and that your n4n.ai key has access to the requested model.

Step 6: Handle errors and backpressure

Streaming introduces two failure modes you don’t see with blocking calls: partial output on upstream failure, and client disconnect mid-stream. Handle both.

6.1 Wrap the pipeline run in a try/except that yields a sentinel

# resilient.py
from pipeline import rag_pipeline
from callbacks import cli_streaming_callback

def run_streaming_rag_safe(question: str, streaming_callback) -> str:
    """
    Runs the pipeline, guarantees the callback receives either tokens
    or a single error sentinel, never a half-written traceback.
    """
    accumulated = []
    def wrapped_callback(chunk: str) -> None:
        accumulated.append(chunk)
        streaming_callback(chunk)

    try:
        rag_pipeline.run(
            data={
                "query_embedder": {"text": question},
                "generator": {"generation_kwargs": {"streaming_callback": wrapped_callback}},
            },
            include_outputs_from={"generator"},
        )
    except Exception as e:
        # Send a machine-parseable error token so the frontend can render a toast
        error_token = f"\n[ERROR: {type(e).__name__}: {e}]"
        streaming_callback(error_token)
        accumulated.append(error_token)
    return "".join(accumulated)

6.2 Backpressure for web clients

If the client disappears (browser tab closed, mobile signal lost), the generator keeps producing tokens into a dead socket. Detect this by checking the transport layer inside your callback.

# backpressure.py
import asyncio
from starlette.requests import Request

async def sse_event_generator(request: Request, question: str):
    queue: asyncio.Queue[str] = asyncio.Queue()
    done = asyncio.Event()

    def producer_callback(token: str) -> None:
        # If client disconnected, stop queuing
        if request.is_disconnected:
            done.set()
            return
        try:
            queue.put_nowait(token)
        except asyncio.QueueFull:
            # Apply backpressure: drop or slow down
            pass

    # Run pipeline in a thread so we don't block the event loop
    asyncio.create_task(asyncio.to_thread(run_streaming_rag_safe, question, producer_callback))

    while not done.is_set():
        try:
            token = await asyncio.wait_for(queue.get(), timeout=0.5)
            yield f"data: {token}\n\n"
        except asyncio.TimeoutError:
            continue
        except asyncio.CancelledError:
            break

The request.is_disconnected check is the standard Starlette/FastAPI pattern. Adapt the same logic for WebSocket websocket.client_state.

Step 7: Observability — log token counts and latency

You can’t optimize what you don’t measure. Hook the generator’s metadata (returned in replies and meta) into your logging pipeline.

# observability.py
import time
import structlog

log = structlog.get_logger()

def run_streaming_rag_observed(question: str, streaming_callback) -> str:
    start = time.perf_counter()
    full_answer = run_streaming_rag_safe(question, streaming_callback)
    duration = time.perf_counter() - start

    # The generator meta is available if you capture the pipeline result directly.
    # For brevity we log what we have; extend to capture token counts from meta.
    log.info(
        "rag_stream_complete",
        question=question[:80],
        answer_chars=len(full_answer),
        duration_ms=round(duration * 1000, 1),
    )
    return full_answer

When you need per-token latency, wrap the callback and timestamp each chunk. That data drives your SLO dashboards and alerts on tail latency regressions.

Troubleshooting quick reference

What to tackle next

• Hybrid retrieval: Add a BM25Retriever in parallel with the embedding retriever and a DocumentJoiner before the prompt builder.
• Caching: Hash the prompt and cache full responses in Redis; serve cached answers instantly, stream only on cache miss.
• Structured output: Use OpenAIGenerator with response_format={"type": "json_object"} and a Pydantic model — streaming still works, but you’ll need a JSON parser that handles partial chunks.
• Multi-turn: Pass conversation history through PromptBuilder and keep the same generator instance; n4n.ai honors cache-control hints so repeat prefixes hit the provider cache.

You now have a streaming responses haystack rag pipeline that runs against n4n.ai, survives client disconnects, and emits the observability signals you need to run it in production. Ship it, measure it, iterate.