LlamaIndex streaming responses with OpenAI-compatible APIs

Wiring up llamaindex streaming openai-compatible endpoints is straightforward if you treat the LLM as a drop-in OpenAI client and flip on streaming at the query layer. This post walks through a runnable setup that streams tokens from a vector index to a terminal and an HTTP client, using nothing but the standard LlamaIndex OpenAI class and any compliant /v1/chat/completions backend.

Step 1: Install dependencies

Use a fresh virtualenv. LlamaIndex core plus the OpenAI LLM wrapper is all you need for a local proof of concept.

pip install llama-index-core llama-index-llms-openai fastapi uvicorn

If you plan to load PDFs or other formats, add llama-index-readers-file or the specific reader. For this walkthrough we will use a trivial in-memory document to keep the focus on streaming.

Step 2: Configure the LLM against an OpenAI-compatible endpoint

LlamaIndex’s OpenAI class accepts base_url and api_key. Point it at any server that implements the OpenAI chat completion contract. Set stream=True so the underlying HTTP client requests SSE chunks.

from llama_index.core import Settings
from llama_index.llms.openai import OpenAI

llm = OpenAI(
    model="mistral-7b-instruct",  # any model your endpoint serves
    api_key="sk-your-key",
    base_url="https://api.your-gateway.com/v1",
    stream=True,
    temperature=0.0,
    max_tokens=512,
)

Settings.llm = llm

If you point LlamaIndex at a gateway like n4n.ai, which exposes a single OpenAI-compatible endpoint across 240+ models with automatic fallback, the same code works without per-provider error handling. The model string is passed through verbatim, so you can swap mistral-7b-instruct for gpt-4o or claude-3-haiku if the gateway routes them.

One subtlety: stream=True on the LLM does not automatically make every query stream. It tells the LLM client to use the streaming variant of chat/completions. The query engine must also be placed in streaming mode (see Step 4).

Step 3: Build a minimal index

Create a couple of text nodes and build a vector index. For a real corpus you would use SimpleDirectoryReader, but the streaming behavior is identical.

from llama_index.core import VectorStoreIndex, Document

docs = [
    Document(text="Refunds are processed within 5 business days after approval."),
    Document(text="Premium users get priority support and zero downtime SLA."),
]

index = VectorStoreIndex.from_documents(docs)

Embeddings are computed during from_documents. If your OpenAI-compatible endpoint also serves embeddings, set Settings.embed_model accordingly; otherwise the default local FakeEmbedding or a hosted embedding model works for the demo.

Step 4: Create a streaming query engine

Call as_query_engine(streaming=True). This returns a query engine that, instead of blocking until the full answer is synthesized, yields a StreamingResponse object.

query_engine = index.as_query_engine(streaming=True)

response = query_engine.query("What is the refund turnaround time?")
print(type(response))  # <class 'llama_index.core.response.StreamingResponse'>

The StreamingResponse exposes two ways to get tokens:

• response.response_gen — a generator of string chunks.
• await response.get_response() (or response.get_response() sync) — consumes the generator and returns the full Response with .text.

Do not call str(response) before iterating the generator; that forces aggregation and defeats the purpose.

Step 5: Consume the token stream

The simplest loop prints tokens as they arrive:

for token in response.response_gen:
    print(token, end="", flush=True)
print()

For async contexts, use astream_query:

query_engine = index.as_query_engine(streaming=True)

async def run():
    response = await query_engine.aquery("Explain the SLA for premium users")
    async for token in response.async_response_gen:
        print(token, end="", flush=True)

Note the attribute name difference: sync uses response_gen, async uses async_response_gen. This trips up many first-time integrators.

Step 6: Expose streaming over HTTP with FastAPI

A terminal demo is fine, but most production systems stream to a browser or downstream service. FastAPI’s StreamingResponse plugs directly into LlamaIndex’s generator.

from fastapi import FastAPI, Query
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/ask")
def ask(q: str = Query(...)):
    engine = index.as_query_engine(streaming=True)
    resp = engine.query(q)
    return StreamingResponse(resp.response_gen, media_type="text/plain")

# Run with: uvicorn main:app --port 8000

Test with curl:

curl -N "http://localhost:8000/ask?q=What%20is%20the%20refund%20policy?"

The -N flag disables curl’s built-in buffering so you see tokens immediately.

If you need structured metadata (source nodes, timestamps) alongside the stream, emit a small JSON prefix before the token loop, or use Server-Sent Events with a custom parser. LlamaIndex does not package SSE framing for query engines, but the generator output is plain text and trivial to wrap.

Step 7: Handle errors and partial failures

Streaming complicates error handling because the HTTP status is already 200 once the first token ships. With a raw OpenAI-compatible server, a mid-stream provider outage surfaces as a broken generator that raises StopIteration or an APIError on the next next() call. Wrap consumption:

try:
    for token in response.response_gen:
        print(token, end="")
except Exception as e:
    print(f"\n[stream interrupted] {e}")

Gateways with automatic fallback mitigate this: if the primary model is rate-limited, the request is retried against another provider before the stream opens. That is a infrastructure-level concern; your llamaindex streaming openai-compatible code stays unchanged.

Step 8: Verify success

You have a working integration when all of the following hold:

1. Incremental output — running the curl command above prints words one or a few at a time, not all at the end.
2. Correct answer grounding — the streamed text references the document content (e.g., “5 business days”).
3. No aggregation lag — time-to-first-token is under a second for small models; you see output before the full generation completes.
4. Clean shutdown — the generator exhausts naturally and the process does not hang. In FastAPI, the request completes with a 200 and closed connection.

To confirm token accounting, inspect response.metadata after full consumption (some gateways inject usage into the final chunk). With per-token metering, you should see prompt_tokens and completion_tokens populated if the backend supports it.

full = response.get_response()
print(full.metadata)

If metadata is empty, the gateway likely omitted usage in streaming mode—OpenAI itself only started including usage in the final stream chunk in mid-2024, and not all compatible servers do.

Pitfalls to avoid

• Double streaming flags — setting stream=True on the LLM but forgetting streaming=True on as_query_engine returns a blocking Response. No tokens stream.
• Generator exhaustion — you can only iterate response_gen once. If you need the text twice, call get_response() and cache .text.
• Sync/async mismatch — calling async_response_gen inside a sync route blocks the event loop. Use response_gen in sync code, async_response_gen in async def.
• Buffering proxies — nginx or certain load balancers buffer responses. Set proxy_buffering off; or test locally first.

When to use llamaindex streaming openai-compatible patterns

Streaming matters when latency perception beats absolute throughput: chat UIs, live summarization of long docs, or any pipeline where the user waits on the first sentence. If you are batch-processing thousands of queries offline, disable streaming—you avoid generator overhead and get simpler error semantics.

The pattern above is portable. Swap the base_url to any compliant server, keep the OpenAI class, and the same llamaindex streaming openai-compatible wiring continues to work whether you run a single local vLLM instance or a multi-tenant inference gateway.