The Ultimate 2025 Guide to FastAPI LLM Word Streaming

Step-by-Step: Building Your First Streaming Endpoint

Let's build a practical example. We'll create a FastAPI endpoint that takes a prompt and streams back a response from a mock LLM service.

Project Setup

First, ensure you have the necessary libraries installed.

pip install fastapi uvicorn openai

We'll use the `openai` library as an example, as its API is a common standard, but the principle applies to any LLM provider (Hugging Face, Anthropic, Cohere, etc.) that supports streaming.

Integrating the LLM Client

Let's create a mock function that simulates calling an LLM API and getting a streamed response. In a real application, this would be your actual call to the OpenAI, Anthropic, or other model's API with `stream=True`.

import asyncio

# This is a mock of a real LLM API call
async def get_llm_response_stream(prompt: str):
    """Simulates a streaming response from an LLM."""
    mock_response = f"The answer to your question about '{prompt}' is a stream of tokens. " \
                    f"FastAPI makes this incredibly easy to handle by using an async generator. " \
                    f"Each word you see is being yielded from the server in real-time."
    
    for word in mock_response.split():
        yield word + " "
        await asyncio.sleep(0.1) # Simulate token generation delay

Creating the Async Generator

Our async generator is the core logic. It will call the LLM service and `yield` each piece of data it receives. This function is what we'll pass to `StreamingResponse`.

from typing import AsyncGenerator

async def response_generator(prompt: str) -> AsyncGenerator[str, None]:
    """The generator function that streams the LLM response."""
    try:
        # In a real app, you'd iterate over the response from a library like OpenAI:
        # async for chunk in await client.chat.completions.create(..., stream=True):
        #     content = chunk.choices[0].delta.content or ""
        #     yield content
        
        # For this example, we use our mock function:
        async for chunk in get_llm_response_stream(prompt):
            yield chunk
    except Exception as e:
        print(f"An error occurred during streaming: {e}")
        # You can decide to yield an error message to the client here
        yield "Error: Could not complete the request."
    finally:
        # Any cleanup logic can go here
        print("Stream finished.")

The FastAPI Endpoint

Finally, let's tie it all together in our main `main.py` file. We'll use Pydantic for request body validation.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

# (Include the response_generator and get_llm_response_stream functions from above)

app = FastAPI(
    title="FastAPI LLM Streaming Guide",
    description="A 2025 guide to word streaming with FastAPI."
)

class PromptRequest(BaseModel):
    prompt: str

@app.post("/chat/stream")
async def chat_stream_endpoint(request: PromptRequest):
    """Receives a prompt and streams back the LLM's response."""
    return StreamingResponse(
        response_generator(request.prompt),
        media_type="text/event-stream" # Use text/event-stream for Server-Sent Events
    )

# To run this: uvicorn main:app --reload

Notice we set `media_type="text/event-stream"`. This tells the browser to treat the response as Server-Sent Events (SSE), which is a standard, easy-to-use protocol for server-to-client streaming that has excellent browser support.

Streaming Methods Compared: StreamingResponse vs. SSE vs. WebSockets

FastAPI gives you options. While our `StreamingResponse` example uses SSE, it's helpful to know where it stands against other real-time technologies.

For most LLM response streaming use cases, Server-Sent Events (SSE) via `StreamingResponse` is the optimal choice due to its simplicity and effectiveness.

Advanced Streaming Techniques for 2025

Building a basic streamer is one thing; making it robust and production-ready is another. Here are key considerations for 2025.

Handling Disconnects and Backpressure

What happens if a user closes their browser tab mid-stream? FastAPI handles this gracefully. When the client disconnects, the `await` call within your generator will raise a `ClientDisconnect` exception. You can catch this to stop the LLM generation and clean up resources, saving significant compute costs.

from starlette.requests import Request

async def robust_generator(request: Request):
    try:
        while True:
            # Check for disconnect before doing work
            if await request.is_disconnected():
                print("Client disconnected.")
                break
            # yield data...
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("Stream cancelled by client disconnect.")
    finally:
        # Stop LLM generation, close resources, etc.
        print("Cleanup complete.")

Streaming Structured Data (JSON)

Sometimes you need to stream more than just text. For example, you might want to stream a list of JSON objects as they are generated. The key is to format each chunk correctly. A common technique is to send each JSON object on a new line.

import json

async def json_stream_generator():
    for i in range(5):
        yield json.dumps({"id": i, "message": f"Item {i}"}) + "\n"
        await asyncio.sleep(1)

# In your endpoint:
# return StreamingResponse(json_stream_generator(), media_type="application/x-ndjson")

Using the `application/x-ndjson` (Newline Delimited JSON) media type is a standard way to handle this.

Error Handling Mid-Stream

If an error occurs in the LLM provider or your internal logic, you can't just return a 500 status code, as the headers have already been sent. Instead, you should `yield` a formatted error message. With SSE, you can even send custom event types.

async def generator_with_error_handling():
    try:
        yield "event: message\ndata: Starting generation...\n\n"
        # ... some processing ...
        raise ValueError("Something went wrong!")
    except Exception as e:
        error_message = json.dumps({"error": str(e)})
        yield f"event: error\ndata: {error_message}\n\n"

The client-side JavaScript can then listen for the `error` event and handle it gracefully, rather than the connection just dying.