LlamaIndex chat engine with streaming responses

Streaming responses transform a chat engine from a batch processor into a responsive interface. This llamaindex chat engine streaming tutorial walks through a production-ready setup: a CondensePlusContextChatEngine backed by a vector index, with token-level streaming, conversation memory, and a minimal FastAPI endpoint you can hit with curl. You’ll finish with a working service and a clear mental model of where latency hides.

Step 1: Install dependencies and configure the environment

Pin versions. LlamaIndex moves fast; unpinned installs break tutorials.

pip install "llama-index==0.10.56" "llama-index-llms-openai==0.1.15" \
  "llama-index-embeddings-openai==0.1.8" "fastapi==0.110.1" "uvicorn==0.29.0" \
  "python-dotenv==1.0.1" "pydantic==2.7.1"

Create a .env file with your OpenAI key. If you use a different provider, swap the LLM/embedding classes in Step 2 — the engine logic stays the same.

# .env
OPENAI_API_KEY=sk-...

Step 2: Build the index and chat engine in a reusable module

Keep the engine construction separate from the HTTP layer. This makes unit testing and REPL debugging trivial.

# engine.py
import os
from pathlib import Path
from llama_index.core import (
    VectorStoreIndex,
    SimpleDirectoryReader,
    StorageContext,
    load_index_from_storage,
    Settings,
)
from llama_index.core.chat_engine import CondensePlusContextChatEngine
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

# Global settings — single source of truth for model params
Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0.2)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

PERSIST_DIR = Path("./storage")
DATA_DIR = Path("./data")  # put .txt/.md/.pdf files here

def get_chat_engine() -> CondensePlusContextChatEngine:
    """Build or load index, then wrap in a streaming chat engine with memory."""
    if PERSIST_DIR.exists():
        storage_context = StorageContext.from_defaults(persist_dir=PERSIST_DIR)
        index = load_index_from_storage(storage_context)
    else:
        DATA_DIR.mkdir(exist_ok=True)
        documents = SimpleDirectoryReader(DATA_DIR).load_data()
        if not documents:
            raise RuntimeError(f"No documents found in {DATA_DIR}. Add files and re-run.")
        index = VectorStoreIndex.from_documents(documents)
        index.storage_context.persist(persist_dir=PERSIST_DIR)

    # Memory buffer: keeps last N token-equivalents of conversation
    memory = ChatMemoryBuffer.from_defaults(token_limit=3000)

    # CondensePlusContextChatEngine:
    # 1. Condenses follow-up + history into a standalone query
    # 2. Retrieves relevant nodes
    # 3. Synthesizes answer with context + history
    return CondensePlusContextChatEngine.from_defaults(
        index=index,
        memory=memory,
        system_prompt=(
            "You are a precise technical assistant. "
            "Cite source doc_ids inline like [doc_3] when you use retrieved context. "
            "If the answer isn't in the context, say you don't know."
        ),
        verbose=True,  # logs condensed query + retrieved nodes to stdout
    )

Why CondensePlusContextChatEngine? It handles multi-turn context correctly: the follow-up question gets rewritten using chat history before retrieval, so you retrieve for the actual intent, not the literal phrasing. ContextChatEngine skips the rewrite step; SimpleChatEngine skips retrieval entirely.

Step 3: Add a streaming generator

LlamaIndex’s stream_chat returns a StreamingAgentChatResponse with a response_gen iterator. Wrap it to yield clean SSE-compatible chunks.

# streaming.py
from typing import AsyncGenerator
from llama_index.core.chat_engine.types import StreamingAgentChatResponse
from engine import get_chat_engine

chat_engine = get_chat_engine()

async def stream_response(message: str) -> AsyncGenerator[str, None]:
    """Yield token deltas as they arrive from the LLM."""
    streaming_response: StreamingAgentChatResponse = chat_engine.stream_chat(message)
    for token in streaming_response.response_gen:
        # token is already a string delta; no need to decode
        yield token
    
    # Optional: expose source nodes after stream completes
    # Access via streaming_response.source_nodes

Critical detail: stream_chat is synchronous but returns a generator. We wrap it in an async function so FastAPI can await each yield without blocking the event loop. If you call the synchronous generator directly in an async route, you’ll starve the loop.

Step 4: Expose a FastAPI endpoint with Server-Sent Events

SSE is simpler than WebSockets for unidirectional streaming and works through every proxy.

# main.py
from fastapi import FastAPI, Query
from fastapi.responses import StreamingResponse
from streaming import stream_response

app = FastAPI(title="LlamaIndex Streaming Chat")

@app.get("/chat")
async def chat_endpoint(q: str = Query(..., min_length=1, max_length=2000)):
    """
    Stream tokens as SSE.
    Client usage: curl -N "http://localhost:8000/chat?q=your+question"
    """
    async def event_generator():
        async for token in stream_response(q):
            # SSE format: data: <payload>\n\n
            yield f"data: {token}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",  # disable nginx buffering
        },
    )

@app.get("/health")
async def health():
    return {"status": "ok"}

Run it:

uvicorn main:app --host 0.0.0.0 --port 8000 --reload

Step 5: Verify end-to-end with curl

The -N (no-buffer) flag is mandatory; otherwise curl waits for the full response.

# Seed some data first
mkdir -p data
echo "n4n.ai is an OpenRouter-class LLM inference gateway with 240+ models, automatic fallback, and per-token metering." > data/n4n.txt
echo "LlamaIndex 0.10 introduced CondensePlusContextChatEngine for better multi-turn retrieval." > data/llamaindex.txt

# Restart server to pick up new docs (or call a reload endpoint you add)
# Then test:
curl -N "http://localhost:8000/chat?q=What%20is%20n4n.ai%3F"

Expected output (tokens arrive incrementally):

data: n4n.ai
data:  is
data:  an
data:  OpenRouter-class
data:  LLM
data:  inference
data:  gateway
data:  with
data:  240+
data:  models
data: ,
data:  automatic
data:  fallback
data: ,
data:  and
data:  per-token
data:  metering
data: .
data:  [doc_0]
data: [DONE]

Verify three things:

1. Tokens appear one-by-one (or in small bursts) with no long pause at the start.
2. The final chunk includes a citation like [doc_0] — proves retrieval ran.
3. A follow-up question retains context:

curl -N "http://localhost:8000/chat?q=How%20many%20models%3F"

Should answer “240+” without re-stating the company name, because ChatMemoryBuffer fed the prior turn into the condense step.

Step 6: Handle backpressure and client disconnects

Production traffic includes slow clients and network hiccups. FastAPI’s StreamingResponse cancels the generator on disconnect, but your LLM call may keep running. Add a cancellation check:

# streaming.py (updated)
import asyncio
from typing import AsyncGenerator
from llama_index.core.chat_engine.types import StreamingAgentChatResponse
from engine import get_chat_engine

chat_engine = get_chat_engine()

async def stream_response(message: str) -> AsyncGenerator[str, None]:
    streaming_response: StreamingAgentChatResponse = chat_engine.stream_chat(message)
    
    # Run the synchronous generator in a thread pool so we can await cancellation
    loop = asyncio.get_event_loop()
    gen = streaming_response.response_gen
    
    while True:
        # Check if client disconnected (FastAPI sets this on the request scope)
        # In practice, pass `request: Request` from the route and check `request.is_disconnected()`
        # Here we simulate with a try/except on the executor
        try:
            token = await loop.run_in_executor(None, lambda: next(gen))
        except StopIteration:
            break
        except asyncio.CancelledError:
            # Client hung up — clean up
            break
        
        yield token
    
    yield "[DONE]"

Update the route to pass the request:

# main.py (updated route)
from fastapi import Request

@app.get("/chat")
async def chat_endpoint(request: Request, q: str = Query(..., min_length=1, max_length=2000)):
    async def event_generator():
        async for token in stream_response(q):
            if await request.is_disconnected():
                break
            yield f"data: {token}\n\n"
        yield "data: [DONE]\n\n"
    # ... same StreamingResponse

Step 7: Tune retrieval and memory for latency

Streaming exposes the full pipeline latency: condense LLM call → retrieval → synthesis LLM call. Three levers matter most.

Reduce condense latency

The condense step calls the LLM. Use a smaller model or fewer history turns.

# engine.py — inside get_chat_engine()
from llama_index.core.chat_engine import CondensePlusContextChatEngine

chat_engine = CondensePlusContextChatEngine.from_defaults(
    index=index,
    memory=memory,
    # Limit history fed to condense prompt (default: all)
    condense_prompt_kwargs={"chat_history": memory.get()[-4:]},  # last 4 messages
    # Or swap to a faster model just for condensing
    # condense_llm=OpenAI(model="gpt-3.5-turbo", temperature=0),
)

Reduce retrieval latency

• Use a smaller embedding model (text-embedding-3-small is ~5x faster than large).
• Lower similarity_top_k (default 2). For focused docs, top_k=1 often suffices.
• Enable hybrid search if you have keyword-heavy queries:

# engine.py — when creating index
from llama_index.core.retrievers import VectorIndexRetriever

retriever = VectorIndexRetriever(
    index=index,
    similarity_top_k=2,
    # vector_store_query_mode="hybrid",  # requires vector store support
)

Reduce synthesis latency

Stream from the synthesis LLM directly — already done. But ensure token_limit on ChatMemoryBuffer isn’t so large that the context window bloats the prompt. 3000 tokens is a safe default for gpt-4o-mini’s 128k window; drop to 1500 if you hit latency spikes.

Step 8: Add structured logging for observability

verbose=True prints to stdout. Replace with structured JSON logs for production.

# engine.py — add at top
import logging
import json
from llama_index.core.callbacks import CallbackManager, LlamaDebugHandler

debug_handler = LlamaDebugHandler(print_trace_on_end=False)
callback_manager = CallbackManager([debug_handler])

Settings.callback_manager = callback_manager

# After a request, extract timings:
def log_pipeline_timings():
    for event in debug_handler.get_event_pairs("retrieve"):
        duration = event[1].duration
        logging.info(json.dumps({
            "stage": "retrieve",
            "duration_ms": duration * 1000,
            "num_nodes": len(event[1].payload.get("nodes", [])),
        }))
    for event in debug_handler.get_event_pairs("llm"):
        logging.info(json.dumps({
            "stage": "llm",
            "duration_ms": event[1].duration * 1000,
            "prompt_tokens": event[1].payload.get("prompt_token_count"),
            "completion_tokens": event[1].payload.get("completion_token_count"),
        }))

Call log_pipeline_timings() after each /chat request (use a FastAPI dependency or middleware). You’ll see exactly where time goes: condense, retrieve, or synthesize.

Step 9: Test multi-turn memory persistence

The in-memory ChatMemoryBuffer dies on process restart. For real deployments, persist to Redis or Postgres. LlamaIndex provides ChatMemoryBuffer.from_defaults with a custom token_limit but no built-in persistence — you implement get/put yourself.

Minimal Redis-backed memory:

# memory_redis.py
import redis
import json
from llama_index.core.memory import BaseMemory
from llama_index.core.llms import ChatMessage
from typing import List, Optional

class RedisChatMemory(BaseMemory):
    def __init__(self, redis_url: str, session_id: str, token_limit: int = 3000):
        self.client = redis.from_url(redis_url, decode_responses=True)
        self.key = f"chat_memory:{session_id}"
        self.token_limit = token_limit
        self._messages: List[ChatMessage] = []
        self._load()

    def _load(self):
        data = self.client.get(self.key)
        if data:
            self._messages = [ChatMessage(**m) for m in json.loads(data)]

    def _save(self):
        self.client.set(self.key, json.dumps([m.dict() for m in self._messages]))

    def get(self, initial_token_count: int = 0) -> List[ChatMessage]:
        return self._messages

    def put(self, message: ChatMessage) -> None:
        self._messages.append(message)
        # Simple truncation — replace with token-counting for precision
        if len(self._messages) > 20:
            self._messages = self._messages[-20:]
        self._save()

    def reset(self) -> None:
        self._messages = []
        self.client.delete(self.key)

Then in engine.py:

# from memory_redis import RedisChatMemory
# memory = RedisChatMemory("redis://localhost:6379", session_id="user_123")

Pass session_id via header or cookie in your FastAPI route.

Step 10: Package as a Docker image

# Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
# Pre-load index at build time if data is static
RUN mkdir -p data && python -c "from engine import get_chat_engine; get_chat_engine()"

EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# requirements.txt
llama-index==0.10.56
llama-index-llms-openai==0.1.15
llama-index-embeddings-openai==0.1.8
fastapi==0.110.1
uvicorn==0.29.0
python-dotenv==1.0.1
pydantic==2.7.1
redis==5.0.1

Build and run:

docker build -t llamaindex-chat .
docker run -p 8000:8000 --env-file .env llamaindex-chat

Verification checklist

Common failure modes

Empty stream, then full answer at once — You’re buffering. Check: nginx proxy_buffering off, FastAPI X-Accel-Buffering: no, and no middleware wrapping the response in a list.

Condense prompt blows context window — ChatMemoryBuffer.token_limit too high or history unbounded. Lower token_limit or implement sliding-window truncation in your custom memory class.

Retrieval returns zero nodes — Embedding model mismatch between index build and query. Ensure Settings.embed_model is identical at index time and query time. If you switched models, rebuild the index.

Follow-up loses context — You instantiated a new ChatMemoryBuffer per request. The engine must be a singleton (or per-session) with shared memory. In engine.py, get_chat_engine() returns a new engine each call — fix by caching the engine or passing a shared memory instance.

You now have a streaming LlamaIndex chat engine that retrieves, cites, remembers, and survives client disconnects. The same pattern scales: swap the vector store for Pinecone/Weaviate, the LLM for a local model via Ollama, or the transport for WebSockets — the engine interface stays stream_chat(message) -> AsyncGenerator[str].