Deploying Production LLM APIs with FastAPI

+-----------------------------------------------------------------------------+
|                     STREAMING SERVER-SENT EVENTS (SSE)                      |
+-----------------------------------------------------------------------------+
|  Client (Browser/App) -- HTTP POST Request -----------------> FastAPI Server|
|  Client <--- SSE Token 1 ("The") <--- HTTP Stream Chunk 1 --- FastAPI Server|
|  Client <--- SSE Token 2 (" capital") <--- HTTP Stream 2 ---- FastAPI Server|
|  Client <--- SSE Token 3 (" of") <--- HTTP Stream 3 --------- FastAPI Server|
|  Client <--- SSE [DONE] <------------------------------------ FastAPI Server|
+-----------------------------------------------------------------------------+

Why Streaming (SSE) is Essential

Generating a 500-token response from an LLM can take 3 to 10 seconds. If an API waits until the entire generation completes, users experience a freeze (high Time-To-First-Token).

Using Server-Sent Events (SSE) allows streaming tokens to the client the millisecond they are generated by the model.

Production Streaming FastAPI Implementation

import os
import json
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from openai import AsyncOpenAI

app = FastAPI(title="Production LLM Service")
client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

class ChatRequest(BaseModel):
    query: str
    temperature: float = 0.2

async def event_generator(prompt: str, temperature: float):
    try:
        stream = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "You are an expert Data Engineering AI assistant."},
                {"role": "user", "content": prompt}
            ],
            stream=True,
            temperature=temperature
        )
        async for chunk in stream:
            content = chunk.choices[0].delta.content
            if content:
                # Format as Server-Sent Event (SSE)
                data = json.dumps({"token": content})
                yield f"data: {data}

"
        yield "data: [DONE]

"
    except Exception as e:
        yield f"data: {json.dumps({'error': str(e)})}

"

@app.post("/v1/chat/stream")
async def stream_chat(req: ChatRequest):
    return StreamingResponse(
        event_generator(req.query, req.temperature),
        media_type="text/event-stream"
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)