How to Deploy FastAPI with LLM in a concise way

Deploying LLMs with FastAPI: A Practical Guide

From prototype to production by serving language models the right way

Large Language Models are everywhere. But there’s a big gap between running a model in a Jupyter notebook and serving it reliably to real users. Over the past year, I’ve built and shipped several LLM-backed APIs, and FastAPI has consistently been my framework of choice for the application layer.

In this post, I will walk through how to deploy an LLM behind a FastAPI service from a minimal working example to production-grade patterns like streaming, GPU-aware Docker builds, and high-throughput inference with vLLM.

Why FastAPI for LLMs?

FastAPI hits a sweet spot for LLM serving:

  • Async-first — critical when you’re waiting on tokens or coordinating with inference engines
  • Pydantic validation — chat messages, sampling parameters, and token limits get type-checked for free
  • Native streaming — StreamingResponse makes server-sent events (SSE) trivial, which is exactly what you need for token-by-token output
  • OpenAPI docs — auto-generated /docs is a huge win when other teams need to integrate
  • Ecosystem — integrates cleanly with Redis, Prometheus, Docker, and Kubernetes

It’s not an inference engine, though. FastAPI handles the API contract; you still need something like vLLM, TGI, or llama.cpp underneath for real throughput.

The Architecture

Here’s the mental model I use:

Client → FastAPI (auth, validation, routing)
           ↓
       Inference Backend (vLLM / transformers / OpenAI)
           ↓
       GPU or Remote API

FastAPI is the application layer. It should never be the bottleneck.

Project Layout

I keep things simple and predictable:

llm-api/
├── app/
│   ├── main.py           # FastAPI app
│   ├── config.py         # Settings via pydantic-settings
│   ├── schemas.py        # Request/response models
│   └── llm_service.py    # Inference backend wrapper
├── requirements.txt
├── Dockerfile
└── docker-compose.yml

Requirements

fastapi==0.115.0
uvicorn[standard]==0.32.0
pydantic==2.9.0
pydantic-settings==2.5.0
transformers==4.45.0
torch==2.4.0
accelerate==1.0.0
vllm==0.6.0          # for high-throughput serving
openai==1.50.0       # for API-based LLMs
httpx==0.27.0
sse-starlette==2.1.3 # for streaming
redis==5.0.0         # optional caching

Let’s walk through what each one is doing.

The web layer

fastapi — the framework itself: routing, validation, auto-generated docs.

uvicorn[standard] — the ASGI server that runs FastAPI. The [standard] extra pulls in uvloop and httptools for a meaningfully faster server.

Validation & config

pydantic — powers every request/response model; v2 is the fast Rust rewrite.

pydantic-settings — loads config from env vars and .env files (split out of Pydantic in v2).

Local inference

transformers — Hugging Face's library for loading models and tokenizers

torch — PyTorch, the deep learning backbone.

accelerate — makes device_map="auto" work, handling GPU/CPU placement.

High-throughput inference (alternative backend)

vllm — PagedAttention + continuous batching for 5–20× the throughput. Swap it in for transformers in production.

Remote API clients

openai — official SDK; useful when proxying to OpenAI or any compatible endpoint.

httpx — async HTTP client, used by FastAPI's test client and for proxying to a vLLM sidecar.

Streaming

sse-starlette — proper Server-Sent Events helper, handles formatting and reconnects so you don't hand-roll data.

Step 1: Configuration

Start with centralized settings so nothing is hardcoded.

# app/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    model_name: str = "meta-llama/Llama-3.2-1B-Instruct"
    max_new_tokens: int = 512
    temperature: float = 0.7
    device: str = "cuda"
    use_vllm: bool = False
    api_key: str = "your-secret-key"

    class Config:
        env_file = ".env"

settings = Settings()

Step 2: Define the Contract

Start with schemas. Everything else follows from what your API promises.

# app/schemas.py
from pydantic import BaseModel, Field
from typing import List, Literal

class Message(BaseModel):
    role: Literal["system", "user", "assistant"]
    content: str

class ChatRequest(BaseModel):
    messages: List[Message]
    max_tokens: int = Field(default=512, le=4096)
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)
    stream: bool = False

class ChatResponse(BaseModel):
    content: str
    model: str
    usage: dict

Notice the constraints: max_tokens capped, temperature bounded. These guards save you from a client accidentally asking for 100k tokens and pinning your GPU.

Step 3: Wrap the Inference Backend

Keep model code isolated. This lets you swap transformers for vLLM without touching your routes.

# app/llm_service.py
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer
from threading import Thread
from app.config import settings

class LLMService:
    def __init__(self):
        self.tokenizer = AutoTokenizer.from_pretrained(settings.model_name)
        self.model = AutoModelForCausalLM.from_pretrained(
            settings.model_name,
            torch_dtype=torch.float16,
            device_map="auto",
        )
        self.model.eval()

    def _format_prompt(self, messages):
        return self.tokenizer.apply_chat_template(
            messages, tokenize=False, add_generation_prompt=True
        )

    def generate(self, messages, max_tokens, temperature):
        prompt = self._format_prompt(messages)
        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)

        with torch.inference_mode():
            outputs = self.model.generate(
                **inputs,
                max_new_tokens=max_tokens,
                temperature=temperature,
                do_sample=temperature > 0,
                pad_token_id=self.tokenizer.eos_token_id,
            )

        new_tokens = outputs[0][inputs.input_ids.shape[1]:]
        text = self.tokenizer.decode(new_tokens, skip_special_tokens=True)
        return text, len(new_tokens)

    def stream(self, messages, max_tokens, temperature):
        prompt = self._format_prompt(messages)
        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)
        streamer = TextIteratorStreamer(
            self.tokenizer, skip_prompt=True, skip_special_tokens=True
        )

        kwargs = dict(
            **inputs,
            max_new_tokens=max_tokens,
            temperature=temperature,
            do_sample=temperature > 0,
            streamer=streamer,
        )
        Thread(target=self.model.generate, kwargs=kwargs).start()

        for token in streamer:
            yield token


llm_service = None

def get_llm():
    global llm_service
    if llm_service is None:
        llm_service = LLMService()
    return llm_service

Two things worth calling out:

  1. TextIteratorStreamer runs generation on a background thread and yields tokens as they’re produced — this is what makes SSE streaming possible.
  2. The model loads once, at startup. Never reload per request.

Step 4: Wire Up FastAPI

Now the routes. Two endpoints: one for non-streaming, one for streaming.

# app/main.py
import json
from fastapi import FastAPI, HTTPException, Depends, Header
from fastapi.responses import StreamingResponse
from app.schemas import ChatRequest, ChatResponse
from app.llm_service import get_llm
from app.config import settings

app = FastAPI(title="LLM API", version="1.0.0")

@app.on_event("startup")
async def startup():
    get_llm()  # preload model into VRAM

def verify_api_key(authorization: str = Header(...)):
    if authorization != f"Bearer {settings.api_key}":
        raise HTTPException(401, "Invalid API key")

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

@app.post("/v1/chat/completions", response_model=ChatResponse)
async def chat(req: ChatRequest, _=Depends(verify_api_key)):
    llm = get_llm()
    messages = [m.model_dump() for m in req.messages]
    text, tokens = llm.generate(messages, req.max_tokens, req.temperature)

    return ChatResponse(
        content=text,
        model=settings.model_name,
        usage={"completion_tokens": tokens},
    )

@app.post("/v1/chat/stream")
async def chat_stream(req: ChatRequest, _=Depends(verify_api_key)):
    llm = get_llm()
    messages = [m.model_dump() for m in req.messages]

    def event_generator():
        for token in llm.stream(messages, req.max_tokens, req.temperature):
            yield f"data: {json.dumps({'token': token})}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

The streaming endpoint emits tokens in SSE format, which is what EventSource in the browser and the OpenAI Python client expect. This means your API is drop-in compatible with tools that already speak that protocol.

Step 5: Docker for GPU

Here’s the Dockerfile I reach for. It’s CUDA-based, includes only the runtime (not the full toolkit), and caches Hugging Face models in a volume.

FROM nvidia/cuda:12.4.1-cudnn-runtime-ubuntu22.04

RUN apt-get update && apt-get install -y python3.11 python3-pip \
    && rm -rf /var/lib/apt/lists/*

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

COPY app ./app

ENV HF_HOME=/models
EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]

The --workers 1 is not a typo. With local models, each worker loads its own copy into VRAM. Two workers on one GPU usually means an out-of-memory crash. Scale horizontally with more containers instead.

Step 6: Docker Compose with GPU Access

services:
  llm-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - HF_TOKEN=${HF_TOKEN}
      - MODEL_NAME=meta-llama/Llama-3.1-8B-Instruct
    volumes:
      - hf-cache:/models
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]

volumes:
  hf-cache:

Step 7: The vLLM Upgrade

The transformers path is fine for a demo, but it’s slow under real traffic. vLLM uses PagedAttention and continuous batching, and typically delivers 5–20× the throughput on the same hardware.

You have two options.

Option A: Embed vLLM

# app/vllm_service.py
from vllm import LLM, SamplingParams

class VLLMService:
    def __init__(self):
        self.llm = LLM(
            model="meta-llama/Llama-3.1-8B-Instruct",
            tensor_parallel_size=1,
            gpu_memory_utilization=0.9,
            max_model_len=8192,
        )

    def generate(self, messages, max_tokens, temperature):
        tokenizer = self.llm.get_tokenizer()
        prompt = tokenizer.apply_chat_template(
            messages, tokenize=False, add_generation_prompt=True
        )
        params = SamplingParams(temperature=temperature, max_tokens=max_tokens)
        outputs = self.llm.generate([prompt], params)
        return outputs[0].outputs[0].text

Option B: Run vLLM as a sidecar and proxy

This is what I’d pick for anything serious:

vllm serve meta-llama/Llama-3.1-8B-Instruct --port 8000

Then your FastAPI service just becomes a thin gateway — handling auth, rate limiting, and logging — while forwarding to http://vllm:8000/v1. You get vLLM’s OpenAI-compatible server for free, plus all your application concerns in one place.

Production Best Practices

🔒 Security

  • API keys via header, never in URLs
  • Rotate keys, keep them in a vault or env vars
  • Don’t leak model weights or HF tokens in logs

🚦 Concurrency

Use an asyncio.Semaphore to cap in-flight requests — protects your GPU from OOM.

import asyncio
semaphore = asyncio.Semaphore(4)  # tune to GPU memory

@app.post("/v1/chat/completions")
async def chat(req: ChatRequest, _=Depends(verify_api_key)):
    async with semaphore:
        return await run_in_threadpool(llm.generate, ...)

📊 Observability

  • Prometheus metrics via prometheus-fastapi-instrumentator
  • Structured JSON logs with request IDs
  • Track p50/p95 latency per token, not just per request
from prometheus_fastapi_instrumentator import Instrumentator
Instrumentator().instrument(app).expose(app)

⚡ Performance

  • Stream by default — perceived latency drops dramatically
  • Cache identical requests with Redis (hash the message payload)
  • Quantize (GPTQ/AWQ/FP8) if you’re VRAM-constrained
  • Batch when you can
import hashlib, json
import redis.asyncio as redis

r = redis.from_url("redis://localhost:6379")

def cache_key(messages, **kwargs):
    payload = json.dumps({"m": messages, **kwargs}, sort_keys=True)
    return f"llm:{hashlib.sha256(payload.encode()).hexdigest()}"

🛡️ Reliability

  • Timeouts on every route (LLMs can hang)
  • Health check that actually checks model availability
  • Graceful shutdown so in-flight requests finish
from fastapi.middleware.timeout import TimeoutMiddleware

app.add_middleware(TimeoutMiddleware, timeout=120)

🚦 Rate Limiting

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.post("/v1/chat/completions")
@limiter.limit("20/minute")
async def chat(request: Request, req: ChatRequest): ...

Deployment Checklist

Concern Recommendation
Model loading Preload at startup (on_event("startup") or lifespan)
Workers 1 Uvicorn worker per GPU for local models
Scaling Horizontal (containers), not in-process workers
Auth API keys via header, rotate regularly
Quantization Use GPTQ/AWQ/FP8 for smaller VRAM footprint
Batching vLLM/TGI for continuous batching under load
Streaming SSE via StreamingResponse
Health checks /health for k8s liveness + readiness
Observability Prometheus + structured logs
Secrets Env vars / vault, never in code

Choosing Your Stack

Scale Recommended stack
Prototype FastAPI + transformers + Uvicorn
Small production FastAPI + vLLM embedded + Redis
Serious production FastAPI gateway → vLLM/TGI cluster + k8s
API-only FastAPI proxy → OpenAI / Anthropic / Together

Quick Start

pip install -r requirements.txt
export API_KEY=secret
uvicorn app.main:app --reload
curl -X POST http://localhost:8000/v1/chat/completions \
  -H "Authorization: Bearer secret" \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [{"role": "user", "content": "Hello!"}],
    "max_tokens": 100
  }'

The Takeaway

FastAPI is a great application layer for LLM services — but it’s just that: the application layer. Treat it as the gateway, not the engine. Let vLLM or TGI handle batching and memory, keep your routes thin, stream aggressively, and instrument everything.

The stack I keep coming back to for production is FastAPI + vLLM + Docker + Redis, running on Kubernetes with autoscaling. It handles everything from a weekend prototype to real traffic without rewriting the world.


Comments

Leave a comment

What is 2 + 3?
Please solve the arithmetic question above to verify you're human.