Introduction to Streaming Responses with LlamaIndex
When building LLM-powered applications, latency is one of the most critical factors affecting user experience. Traditional request-response patterns force users to wait for the entire response to be generated before seeing any output. With large language models, this can mean waiting 10, 20, or even 60 seconds for a complex answer. Streaming responses solves this problem by delivering tokens to the client as they are generated, dramatically improving perceived performance.
LlamaIndex, one of the most popular frameworks for building LLM applications, provides robust streaming support across its query engines, agents, and chat interfaces. In this guide, we'll cover everything you need to know to deploy streaming responses in a production environment, including architecture decisions, implementation patterns, error handling, and deployment best practices.
Why Streaming Matters in Production
Before diving into implementation, it's worth understanding why streaming is so important for production applications:
- Perceived latency reduction: Users see the first token within 1-2 seconds instead of waiting for the full response.
- Higher engagement: Studies show that visible progress keeps users engaged and reduces abandonment rates.
- Better UX for long responses: For tasks like document summarization or code generation, streaming prevents the "frozen screen" effect.
- Early cancellation: Users can stop generation if the response is going in the wrong direction, saving API costs.
- Competitive parity: Modern AI products like ChatGPT set user expectations for streaming behavior.
In production specifically, streaming also enables better resource utilization. Instead of holding a request open with a large buffer, you can process and forward tokens incrementally, reducing memory pressure on your servers.
Understanding LlamaIndex's Streaming Architecture
LlamaIndex provides streaming support at multiple levels. Understanding the abstraction layers helps you choose the right approach for your use case.
Streaming Response Types
LlamaIndex exposes two main types of streaming responses:
StreamingResponse: A synchronous iterator that yields text chunks as they arrive. Best for simple use cases and synchronous code paths.StreamingResponseAgen: An asynchronous iterator compatible withasync forloops. Essential for high-concurrency production servers.
Both types support the same core operations: iterating over response chunks, accessing the final response object, and handling token-level events.
Where Streaming Happens
Streaming can occur at several points in a LlamaIndex pipeline:
- LLM output streaming: The raw token stream from the model provider (OpenAI, Anthropic, local models, etc.).
- Query engine streaming: LlamaIndex wraps the LLM stream and may inject retrieved context or post-processing.
- Agent streaming: For agentic workflows, streaming includes not just final answers but also intermediate steps, tool calls, and reasoning.
Setting Up Your Environment
Let's start by installing the necessary dependencies. We'll use LlamaIndex with OpenAI as our provider, but the patterns apply to any supported LLM.
pip install llama-index llama-index-llms-openai fastapi uvicorn
Set your API keys as environment variables:
import os
os.environ["OPENAI_API_KEY"] = "sk-your-api-key-here"
For production, always load secrets from a secure vault or environment variable manager rather than hardcoding them.
Basic Streaming with Query Engines
The simplest way to get started with streaming is through a LlamaIndex query engine. Let's build a basic retrieval-augmented generation (RAG) pipeline with streaming enabled.
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.llms.openai import OpenAI
# Configure the LLM with streaming enabled
Settings.llm = OpenAI(model="gpt-4o", temperature=0.7)
# Load documents and build the index
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
# Create a query engine with streaming enabled
query_engine = index.as_query_engine(streaming=True)
# Execute a streaming query
streaming_response = query_engine.query("Explain the key concepts in these documents.")
# Iterate over the response chunks
for text in streaming_response.response_gen:
print(text, end="", flush=True)
# Access the final response object after streaming completes
print("\n\n--- Metadata ---")
print(f"Source nodes: {len(streaming_response.source_nodes)}")
The key parameter here is streaming=True when calling as_query_engine(). This tells LlamaIndex to return a StreamingResponse object instead of a regular Response. The response_gen attribute is a generator that yields text chunks as they arrive from the LLM.
Streaming Chat Engines
For conversational applications, you'll want to use a chat engine instead of a query engine. The streaming interface is nearly identical:
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.llms.openai import OpenAI
Settings.llm = OpenAI(model="gpt-4o", temperature=0.7)
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
# Create a streaming chat engine
chat_engine = index.as_chat_engine(
chat_mode="context",
streaming=True,
)
# First message
response = chat_engine.stream_chat("What are the main topics covered?")
for token in response.response_gen:
print(token, end="", flush=True)
# Follow-up message (context is maintained)
print("\n")
response = chat_engine.stream_chat("Can you elaborate on the second topic?")
for token in response.response_gen:
print(token, end="", flush=True)
Building a Production Streaming API with FastAPI
In production, you'll typically expose your LlamaIndex application through an HTTP API. Server-Sent Events (SSE) is the most common protocol for streaming LLM responses over HTTP. FastAPI makes this straightforward.
Basic SSE Endpoint
Here's a complete FastAPI application that streams LlamaIndex responses to clients:
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.llms.openai import OpenAI
import json
import asyncio
app = FastAPI(title="LlamaIndex Streaming API")
# Initialize on startup
Settings.llm = OpenAI(model="gpt-4o", temperature=0.7)
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine(streaming=True)
class QueryRequest(BaseModel):
query: str
top_k: int = 4
async def stream_response(query: str):
"""Async generator that yields SSE-formatted chunks."""
try:
# Run the synchronous query in a thread pool
response = await asyncio.to_thread(query_engine.query, query)
for chunk in response.response_gen:
if chunk:
data = json.dumps({"type": "token", "content": chunk})
yield f"data: {data}\n\n"
# Send source nodes at the end
sources = []
for node in response.source_nodes:
sources.append({
"text": node.node.text[:200],
"score": node.score,
"metadata": node.node.metadata,
})
final_data = json.dumps({"type": "done", "sources": sources})
yield f"data: {final_data}\n\n"
except Exception as e:
error_data = json.dumps({"type": "error", "message": str(e)})
yield f"data: {error_data}\n\n"
@app.post("/query")
async def query_endpoint(request: QueryRequest):
return StreamingResponse(
stream_response(request.query),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no", # Disable Nginx buffering
},
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Let's break down the key design decisions in this implementation:
- SSE format: Each event is formatted as
data: {json}\n\n, which is the standard SSE protocol that browsers can consume natively via theEventSourceAPI. - Typed events: We distinguish between
token,done, anderrorevents so the client can handle each appropriately. - Thread pool execution: LlamaIndex's synchronous query engine runs in a thread pool via
asyncio.to_thread, preventing it from blocking the async event loop. - Buffering disabled: The
X-Accel-Buffering: noheader tells Nginx (if used as a reverse proxy) not to buffer the response, which would break streaming.
Consuming the Stream on the Client Side
Here's a JavaScript example showing how a frontend client would consume this API:
async function streamQuery(query) {
const response = await fetch("http://localhost:8000/query", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ query: query }),
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split("\n");
buffer = lines.pop(); // Keep incomplete line in buffer
for (const line of lines) {
if (line.startsWith("data: ")) {
const data = JSON.parse(line.slice(6));
if (data.type === "token") {
// Append token to UI
document.getElementById("output").textContent += data.content;
} else if (data.type === "done") {
console.log("Sources:", data.sources);
} else if (data.type === "error") {
console.error("Error:", data.message);
}
}
}
}
}
Async Streaming for High Concurrency
The synchronous approach above works well for moderate traffic, but for high-concurrency production systems, you should use LlamaIndex's async streaming capabilities. This avoids thread pool overhead and allows true concurrent streaming.
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.llms.openai import OpenAI
import json
app = FastAPI()
Settings.llm = OpenAI(model="gpt-4o", temperature=0.7, async_mode=True)
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
class QueryRequest(BaseModel):
query: str
async def stream_response_async(query: str):
"""Fully async streaming generator."""
query_engine = index.as_query_engine(streaming=True)
try:
# Use the async streaming method
response = await query_engine.aquery(query)
async for chunk in response.async_response_gen():
if chunk:
data = json.dumps({"type": "token", "content": chunk})
yield f"data: {data}\n\n"
yield f"data: {json.dumps({'type': 'done'})}\n\n"
except Exception as e:
error_data = json.dumps({"type": "error", "message": str(e)})
yield f"data: {error_data}\n\n"
@app.post("/query")
async def query_endpoint(request: QueryRequest):
return StreamingResponse(
stream_response_async(request.query),
media_type="text/event-stream",
)
The key differences here are: setting async_mode=True on the LLM, using aquery() instead of query(), and iterating with async for over async_response_gen(). This approach scales much better under load because it doesn't consume thread pool resources.
Streaming with Agents
Agents add complexity to streaming because they may perform multiple LLM calls, execute tools, and reason through steps. LlamaIndex provides specialized streaming support for agents that exposes these intermediate steps.
from llama_index.core.agent import ReActAgent
from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI
import json
def search_database(query: str) -> str:
"""Search the internal database for information."""
# Simulated database search
return f"Database results for: {query}"
def calculate(expression: str) -> str:
"""Evaluate a mathematical expression."""
try:
result = eval(expression) # In production, use a safe evaluator
return str(result)
except Exception as e:
return f"Error: {e}"
# Define tools
search_tool = FunctionTool.from_defaults(fn=search_database)
calc_tool = FunctionTool.from_defaults(fn=calculate)
# Create the agent
llm = OpenAI(model="gpt-4o", temperature=0)
agent = ReActAgent.from_tools(
[search_tool, calc_tool],
llm=llm,
streaming=True,
)
async def stream_agent_response(query: str):
"""Stream agent responses including intermediate steps."""
response = await agent.achat(query)
# Stream the final response
async for chunk in response.async_response_gen():
data = json.dumps({
"type": "token",
"content": chunk,
})
yield f"data: {data}\n\n"
# Include source steps if available
if hasattr(response, "sources"):
steps = []
for source in response.sources:
steps.append({
"tool": source.tool_name,
"input": str(source.raw_input),
"output": str(source.raw_output)[:500],
})
yield f"data: {json.dumps({'type': 'steps', 'steps': steps})}\n\n"
yield f"data: {json.dumps({'type': 'done'})}\n\n"
For more granular control over agent streaming, including streaming the reasoning steps as they happen, you can use LlamaIndex's lower-level streaming handlers:
from llama_index.core.agent import ReActAgent
from llama_index.core.callbacks import CallbackManager
import json
class StreamingCallbackHandler:
"""Custom callback handler for streaming agent events."""
def __init__(self):
self.events = []
def on_event_start(self, event_type, payload):
self.events.append({
"type": "start",
"event": event_type,
"payload": str(payload)[:200],
})
def on_event_end(self, event_type, payload):
self.events.append({
"type": "end",
"event": event_type,
"payload": str(payload)[:200],
})
async def stream_agent_with_events(agent, query: str):
"""Stream agent execution with detailed event tracking."""
handler = agent.stream_chat(query)
# Stream the main response
async for chunk in handler.async_response_gen():
if chunk:
yield f"data: {json.dumps({'type': 'token', 'content': chunk})}\n\n"
# Stream intermediate tool calls and reasoning
async for event in handler.stream_events():
event_data = {
"type": "event",
"event_type": event.type,
"content": str(event.payload)[:500],
}
yield f"data: {json.dumps(event_data)}\n\n"
yield f"data: {json.dumps({'type': 'done'})}\n\n"
Error Handling and Resilience
Production streaming systems need robust error handling. Unlike traditional APIs where a single error response suffices, streaming responses may fail mid-stream. You need to handle partial failures gracefully.
Comprehensive Error Handling Pattern
import asyncio
import json
import logging
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
logger = logging.getLogger(__name__)
app = FastAPI()
class StreamingErrorHandler:
"""Wraps a streaming generator with comprehensive error handling."""
def __init__(self, max_retries=2, timeout=60):
self.max_retries = max_retries
self.timeout = timeout
async def wrap_stream(self, generator_factory, query: str):
"""Wrap a generator factory with retry and timeout logic."""
last_error = None
for attempt in range(self.max_retries + 1):
try:
async for chunk in self._stream_with_timeout(
generator_factory(query), attempt
):
yield chunk
return # Success, exit retry loop
except asyncio.TimeoutError:
logger.warning(f"Stream timeout on attempt {attempt + 1}")
last_error = "Request timed out"
yield f"data: {json.dumps({'type': 'retry', 'attempt': attempt + 1})}\n\n"
except Exception as e:
logger.error(f"Stream error on attempt {attempt + 1}: {e}")
last_error = str(e)
yield f"data: {json.dumps({'type': 'retry', 'attempt': attempt + 1})}\n\n"
# Wait before retry
if attempt < self.max_retries:
await asyncio.sleep(2 ** attempt)
# All retries exhausted
yield f"data: {json.dumps({'type': 'error', 'message': last_error})}\n\n"
async def _stream_with_timeout(self, generator, attempt):
"""Apply a timeout to each chunk retrieval."""
async for chunk in generator:
yield chunk
# Usage in an endpoint
@app.post("/query")
async def query_endpoint(query: str):
handler = StreamingErrorHandler(max_retries=2, timeout=60)
async def generate():
async for chunk in handler.wrap_stream(stream_response_async, query):
yield chunk
return StreamingResponse(
generate(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)
Handling Client Disconnections
In production, clients will disconnect mid-stream. Your server needs to detect this and clean up resources, especially to avoid unnecessary LLM token costs:
from fastapi import Request
@app.post("/query")
async def query_endpoint(request: Request, query: str):
async def generate():
try:
response = await query_engine.aquery(query)
async for chunk in response.async_response_gen():
# Check if client is still connected
if await request.is_disconnected():
logger.info("Client disconnected, stopping generation")
break
yield f"data: {json.dumps({'type': 'token', 'content': chunk})}\n\n"
yield f"data: {json.dumps({'type': 'done'})}\n\n"
except asyncio.CancelledError:
logger.info("Stream cancelled by client")
raise
return StreamingResponse(
generate(),
media_type="text/event-stream",
)
Production Deployment Considerations
Reverse Proxy Configuration (Nginx)
If you're using Nginx as a reverse proxy, you must configure it to support SSE streaming. Without proper configuration, Nginx will buffer responses and break the streaming behavior:
server {
listen 80;
server_name api.example.com;
location /query {
proxy_pass http://127.0.0.1:8000;
proxy_http_version 1.1;
proxy_set_header Connection "";
proxy_buffering off;
proxy_cache off;
proxy_read_timeout 300s;
chunked_transfer_encoding on;
}
}
The critical settings are proxy_buffering off and proxy_cache off, which ensure Nginx forwards chunks immediately rather than buffering them.
Running with Gunicorn and Uvicorn Workers
For production, run your FastAPI app with Gunicorn managing multiple Uvicorn workers:
gunicorn main:app \
--workers 4 \
--worker-class uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000 \
--timeout 120 \
--keep-alive 5 \
--max-requests 1000 \
--max-requests-jitter 100
The --timeout 120 is important because streaming requests can take longer than the default 30-second timeout. The --max-requests setting causes workers to restart periodically, which helps prevent memory leaks from long-running processes.
Connection Pooling and Rate Limiting
Streaming connections are long-lived, which means they consume server resources for longer than typical requests. Implement rate limiting to prevent abuse:
from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
@app.post("/query")
@limiter.limit("10/minute")
async def query_endpoint(request: Request, query: str):
# ... streaming logic
pass
You should also implement a maximum concurrent connections limit to protect your LLM API budget:
import asyncio
MAX_CONCURRENT_STREAMS = 50
stream_semaphore = asyncio.Semaphore(MAX_CONCURRENT_STREAMS)
@app.post("/query")
async def query_endpoint(request: Request, query: str):
if stream_semaphore.locked():
return JSONResponse(
status_code=429,
content={"error": "Too many concurrent streams. Please try again later."}
)
async with stream_semaphore:
# ... streaming logic
pass
Monitoring and Observability
Streaming applications require specialized monitoring. Key metrics to track include time-to-first-token (TTFT), total generation time, stream completion rate, and error rates.
import time
from prometheus_client import Counter, Histogram, Gauge
# Define metrics
STREAM_REQUESTS = Counter(
"llamaindex_stream_requests_total",
"Total streaming requests",
["status"]
)
TIME_TO_FIRST_TOKEN = Histogram(
"llamaindex_ttft_seconds",
"Time to first token in seconds"
)
STREAM_DURATION = Histogram(
"llamaindex_stream_duration_seconds",
"Total stream duration in seconds"
)
ACTIVE_STREAMS = Gauge(
"llamaindex_active_streams",
"Currently active streaming connections"
)
async def monitored_stream(query: str):
start_time = time.time()
first_token_time = None
ACTIVE_STREAMS.inc()
try:
response = await query_engine.aquery(query)
async for chunk in response.async_response_gen():
if first_token_time is None:
first_token_time = time.time()
TIME_TO_FIRST_TOKEN.observe(first_token_time - start_time)
yield f"data: {json.dumps({'type': 'token', 'content': chunk})}\n\n"
STREAM_DURATION.observe(time.time() - start_time)
STREAM_REQUESTS.labels(status="success").inc()
yield f"data: {json.dumps({'type': 'done'})}\n\n"
except Exception as e:
STREAM_REQUESTS.labels(status="error").inc()
yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
finally:
ACTIVE_STREAMS.dec()
Best Practices Summary
- Always use async streaming in production: The synchronous API is fine for prototyping, but
aquery()andasync_response_gen()scale far better under concurrent load. - Implement proper SSE formatting: Use
data: {json}\n\nformat and include event types so clients can distinguish tokens, metadata, and errors. - Handle client disconnections: Check
request.is_disconnected()regularly to avoid wasting LLM API credits on abandoned streams. - Disable proxy buffering: Configure Nginx (or any reverse proxy) with
proxy_buffering offto ensure chunks are forwarded immediately. - Set appropriate timeouts: Streaming requests take longer than typical API calls. Set generous timeouts at every layer (Gunicorn, Nginx, load balancer).
- Implement rate limiting and concurrency caps: Streaming connections are resource-intensive. Protect your infrastructure and your LLM budget with limits.
- Monitor time-to-first-token: TTFT is the most important user-facing metric for streaming applications. Alert if it exceeds 3-5 seconds.
- Use typed events in your stream protocol: Distinguish between
token,tool_call,source,error, anddoneevents so clients can render rich UIs. - Test with slow networks: Streaming behavior can differ significantly under poor network conditions. Test with throttled connections.
- Consider WebSocket as an alternative: For bidirectional communication (e.g., real-time agent interactions), WebSockets may be more appropriate than SSE.
Conclusion
Streaming responses are essential for any production LLM application that values user experience. LlamaIndex provides flexible streaming primitives at every level of its abstraction stack, from raw LLM token streams to agent execution events. By combining LlamaIndex's streaming capabilities with FastAPI's SSE support, proper error handling, and production-grade infrastructure configuration, you can build responsive, scalable AI applications that match the UX standards set by leading AI products. The key is to treat streaming as a first-class architectural concern from the start, designing your error handling, monitoring, and rate limiting around the realities of long-lived, token-by-token communication rather than retrofitting streaming onto a traditional request-response design.