Nasazení ML modelů s FastAPI 2026: Produkční serving v Pythonu

Aktualizováno: 18. června 2026

Nasazení ML modelu s FastAPI v produkci znamená zabalit natrénovaný model (typicky scikit-learn, PyTorch nebo ONNX) do asynchronního HTTP API s validací přes Pydantic v2, spustit ho přes Uvicorn za Gunicornem ve workeru s předem načteným modelem a zajistit pozorovatelnost přes health checky a Prometheus metriky. V tomto průvodci pro rok 2026 ukážu konkrétní kód, latenční rozpočty, které jsem nasbíral u modelů v reálném provozu, a chyby, na které jsem narazil, když jsme tlačili p99 pod 50 ms.

• FastAPI 0.115+ s Pydantic v2 dává 2–3× rychlejší serializaci oproti Pydantic v1 a je v roce 2026 standardem pro Python ML serving.
• Model musí být načtený jednou na worker přes lifespan, nikoli per-request. Jinak GIL a I/O sežerou váš latency budget.
• Pro CPU-bound predikce běžte synchronně v def endpointu (FastAPI je pustí do threadpoolu); pro async I/O (databáze, vector store) použijte async def.
• ONNX Runtime sníží inferenční latenci u sklearn/PyTorch modelů typicky o 30–70 % a zbaví vás závislosti na trénovacím frameworku v runtime image.
• Batching požadavků na úrovni serveru (dynamic batching) zvedne propustnost o 5–10×, ale přidá tail latency. Vždy měřte p99, nikoli průměr.
• Bez Prometheus metrik (latence, error rate, model version) nemáte produkční nasazení, ale demo.

Proč FastAPI pro ML serving v roce 2026

FastAPI se mezi roky 2022 a 2026 stal de-facto standardem pro Python ML API, a má k tomu důvody, které se v produkci skutečně počítají. Postavený nad Starlette a Pydantic v2 dostáváte ASGI, asynchronní zpracování, automatickou validaci vstupů, OpenAPI dokumentaci zdarma a typovou bezpečnost přes Python type hinty. Pro tým, který už používá scikit-learn nebo PyTorch, je vstupní cena minimální.

Z mé zkušenosti s nasazováním tabulárních klasifikátorů a embeddingových služeb pro doporučovací systém se FastAPI vyplatí ze tří konkrétních důvodů. Za prvé, Pydantic v2 (psaný v Rustu) v praxi vyřídí serializaci/deserializaci 50–80 mikrosekund na request místo 200+ mikrosekund Flasku s ručním JSON validátorem, což je při p99 cíli 50 ms nezanedbatelné. Za druhé, podpora async umožňuje volat vector store nebo Redis bez blokování workeru. Za třetí, ekosystém kolem (Uvicorn, Gunicorn, sentry-sdk, prometheus-fastapi-instrumentator) je dospělý a žádné z toho už nemusíte psát ručně.

Co FastAPI není: GPU inference server pro deep learning batchování (na to je Triton Inference Server od NVIDIA nebo Ray Serve), serverless edge runtime (tam Cloudflare Workers s ONNX Web), ani nástroj pro správu experimentů (to je scikit-learn pipeline v kombinaci s MLflow). FastAPI je HTTP vrstva pro váš model. Vše ostatní si přinesete.

Příprava modelu pro produkci

Než vůbec napíšete první endpoint, váš model musí splňovat čtyři podmínky, které v notebooku obvykle nesplňuje. Jakmile je porušíte, latence a stabilita se rozsypou v produkci, ne během testů.

Deterministický featurizér. Pokud se input pre-processing liší mezi tréninkem a inferencí, model bude tiše predikovat blbosti. Reprodukovatelný preprocessing pipeline obvykle řeším přes sklearn.pipeline.Pipeline, nebo když potřebuju vlastní transformace, přes ColumnTransformer. Více kontextu k feature pipeline najdete v článku o feature engineeringu s Pandas a scikit-learn.

Verzování artefaktu. Model uložte s metadaty (verze knihovny, hash trénovacích dat, datum, F1 score na holdout setu). Já používám joblib pro sklearn modely a vždy přidám sidecar JSON:

import joblib, json, hashlib
from datetime import datetime, timezone

joblib.dump(pipeline, "model_v3.joblib")
meta = {
    "model_version": "3.0.1",
    "sklearn_version": "1.6.1",
    "trained_at": datetime.now(timezone.utc).isoformat(),
    "train_hash": hashlib.sha256(open("train.parquet","rb").read()).hexdigest()[:16],
    "f1_holdout": 0.873,
}
with open("model_v3.meta.json", "w") as f:
    json.dump(meta, f, indent=2)

Žádné notebookové importy. Pokud váš train.ipynb definoval custom třídu transformeru, joblib.load selže, protože pickle hledá tu třídu v __main__. Přesuňte definice do regulérního Python modulu (features.py) ještě před tréninkem. Jinak strávíte půl dne debugováním AttributeError: Can't get attribute 'MyTransformer' on <module '__main__'>.

Latence změřená off-line. Než nasadíte, změřte predict() na jednom requestu a na batchi 100. Pokud single-request inference trvá 30 ms, vaše HTTP p99 nebude nikdy pod 40 ms bez ONNX optimalizace nebo batchingu.

Minimální FastAPI ML API krok za krokem

Tady je minimální, ale produkční-grade kostra, ze které začínám každý nový ML serving projekt. Předpokládá Python 3.12+, FastAPI 0.115+ a Pydantic v2.

# requirements.txt
fastapi==0.115.6
uvicorn[standard]==0.32.1
gunicorn==23.0.0
pydantic==2.10.4
scikit-learn==1.6.1
joblib==1.4.2
numpy==2.2.1
prometheus-fastapi-instrumentator==7.0.2

Pydantic schémata pro vstup a výstup definují kontrakt API. Konstrainty jako ge, le a Literal chrání před nesmyslnými inputy a vrátí 422 dřív, než se model vůbec spustí.

from pydantic import BaseModel, Field
from typing import Literal

class PredictRequest(BaseModel):
    age: int = Field(ge=18, le=120)
    income: float = Field(ge=0)
    employment: Literal["full_time", "part_time", "self_employed", "unemployed"]
    credit_score: int = Field(ge=300, le=850)

class PredictResponse(BaseModel):
    probability: float
    label: Literal["approve", "deny"]
    model_version: str

A endpoint, který tahá model ze sdíleného stavu:

from fastapi import FastAPI, Request
import numpy as np

app = FastAPI(title="Credit Scoring API", version="3.0.1")

@app.post("/v1/predict", response_model=PredictResponse)
def predict(req: PredictRequest, request: Request) -> PredictResponse:
    model = request.app.state.model
    meta = request.app.state.meta
    employment_map = {"full_time": 0, "part_time": 1, "self_employed": 2, "unemployed": 3}
    x = np.array([[req.age, req.income, employment_map[req.employment], req.credit_score]])
    proba = float(model.predict_proba(x)[0, 1])
    return PredictResponse(
        probability=proba,
        label="approve" if proba >= 0.5 else "deny",
        model_version=meta["model_version"],
    )

Načtení modelu přes lifespan a sdílený stav

Načítat model na začátku každého requestu je nejčastější chyba začátečníků v ML serving. Soubor o 200 MB jednoduše nepřečtete v sub-100 ms latence. A i kdyby ano, GIL by vám trhal CPU jádra. Model patří načíst jednou na worker při startu a sdílet přes app.state. K tomu slouží FastAPI lifespan context manager (nahradil deprecated @app.on_event("startup")).

from contextlib import asynccontextmanager
import joblib, json

@asynccontextmanager
async def lifespan(app: FastAPI):
    print("Loading model...")
    app.state.model = joblib.load("model_v3.joblib")
    with open("model_v3.meta.json") as f:
        app.state.meta = json.load(f)
    print(f"Loaded model {app.state.meta['model_version']}")
    yield
    print("Shutting down")

app = FastAPI(title="Credit Scoring API", lifespan=lifespan)

Async vs sync endpointy pro inference

Tady padají kariéry: FastAPI vás nechá deklarovat endpoint jako async def i def, ale pro ML má každá varianta jiný význam pro výkon. Pravidlo, které dodržuji bez výjimky: CPU-bound predikce patří do def, async I/O patří do async def.

def endpoint FastAPI automaticky pustí do threadpoolu (default 40 vláken). Worker tak dokáže paralelně vyřídit více requestů, dokud čeká na CPU. Naopak async def endpoint běží přímo v event loopu. Pokud zavoláte model.predict() (synchronní, CPU-bound), zablokujete celou event loop a všechny ostatní requesty na tom workeru zamrznou.

# Spatne, blokuje event loop
@app.post("/v1/predict")
async def predict(req: PredictRequest):
    return model.predict(...)  # CPU-bound, blokuje VŠECHNY requesty

# Správně pro CPU-bound inference
@app.post("/v1/predict")
def predict(req: PredictRequest):
    return model.predict(...)  # běží v threadpoolu

# Správně pro async I/O + inference
@app.post("/v1/recommend")
async def recommend(req: RecommendRequest):
    embedding = await vector_store.query(req.user_id)  # async I/O
    return await run_in_threadpool(model.predict, embedding)  # CPU mimo loop

Pokud potřebujete async endpoint, který volá synchronní inferenci, použijte from starlette.concurrency import run_in_threadpool. Tím se vyhnete blokování bez nutnosti přepisovat celou trasu.

Batching a propustnost

Jeden request = jedna predikce je plýtvání, pokud váš model umí batch inference (a většina sklearn / PyTorch modelů to umí). Při batchi 32 dostanete na CPU řádově 8–15× lepší propustnost než při single-request inferenci, protože BLAS knihovny (OpenBLAS, MKL) jedou paralelně.

Dvě strategie: klient-side batching (volající posílá pole inputů) je nejjednodušší a často stačí. Server-side dynamic batching akumuluje příchozí requesty po krátkou dobu (např. 5 ms) a pošle je jako batch, což výrazně zvedne propustnost za cenu vyšší p50 latence.

class BatchPredictRequest(BaseModel):
    items: list[PredictRequest] = Field(min_length=1, max_length=128)

@app.post("/v1/predict/batch", response_model=list[PredictResponse])
def predict_batch(req: BatchPredictRequest, request: Request):
    model = request.app.state.model
    meta = request.app.state.meta
    X = np.array([
        [r.age, r.income, employment_map[r.employment], r.credit_score]
        for r in req.items
    ])
    probas = model.predict_proba(X)[:, 1]
    return [
        PredictResponse(
            probability=float(p),
            label="approve" if p >= 0.5 else "deny",
            model_version=meta["model_version"],
        )
        for p in probas
    ]

ONNX Runtime a optimalizace latence

Když potřebujete utáhnout p99 a Pythonové scikit-learn už nemá kam jít, ONNX Runtime je další logický krok. ONNX Runtime je C++ inference engine s Python bindings, který spustí váš model bez závislosti na scikit-learn nebo PyTorch. U tabulárních modelů typicky vidím 30–70% snížení inferenční latence a 5–10× menší runtime image (žádné PyTorch wheely o 800 MB).

Konverze sklearn modelu přes skl2onnx:

from skl2onnx import to_onnx
from skl2onnx.common.data_types import FloatTensorType
import numpy as np

# Konverze
initial_type = [("input", FloatTensorType([None, 4]))]
onnx_model = to_onnx(pipeline, initial_types=initial_type, target_opset=18)
with open("model_v3.onnx", "wb") as f:
    f.write(onnx_model.SerializeToString())

# Inference za runtime
import onnxruntime as ort
session = ort.InferenceSession("model_v3.onnx", providers=["CPUExecutionProvider"])
input_name = session.get_inputs()[0].name
proba = session.run(None, {input_name: X.astype(np.float32)})[0]

Pozor, ne každý sklearn estimator má skl2onnx konvertor, a custom transformer si musíte zaregistrovat sami. U PyTorch modelů použijte torch.onnx.export s dynamo=True (stabilní od PyTorch 2.5).

Pozorovatelnost: health checky a Prometheus

Tři endpointy, které nesmí chybět: /health/live (proces žije), /health/ready (model je načtený a worker je připraven obsluhovat traffic) a /metrics (Prometheus scrape). Kubernetes liveness probe pinguje /live, readiness probe pinguje /ready. Bez toho rolling deploy posílá traffic do workerů, které ještě nedonačetly 500 MB model.

from fastapi import Response, status
from prometheus_fastapi_instrumentator import Instrumentator

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

@app.get("/health/ready")
def ready(request: Request, response: Response):
    if not hasattr(request.app.state, "model"):
        response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
        return {"status": "not_ready"}
    return {"status": "ready", "model_version": request.app.state.meta["model_version"]}

Instrumentator().instrument(app).expose(app, endpoint="/metrics")

Pro ML specifické metriky (predikční distribuce, latence per model version) přidejte vlastní counter/histogram přes prometheus_client. Tracking prediction_probability_bucket vám dá včasné varování před data driftem. Pokud se distribuce posune oproti tréninku, alert vystřelí dřív, než si toho všimne business.

Docker, Gunicorn a horizontální škálování

V produkci nikdy nespouštíte Uvicorn samostatně. Gunicorn s Uvicorn workery dává process management, graceful shutdown a možnost reloadu bez ztráty requestů. Můj Dockerfile pro Python ML API vypadá zhruba takto:

FROM python:3.12-slim-bookworm

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

COPY app/ ./app/
COPY model_v3.joblib model_v3.meta.json ./

EXPOSE 8000
CMD ["gunicorn", "app.main:app", \
     "--workers", "4", \
     "--worker-class", "uvicorn.workers.UvicornWorker", \
     "--bind", "0.0.0.0:8000", \
     "--timeout", "30", \
     "--graceful-timeout", "30", \
     "--access-logfile", "-"]

Počet workerů řiďte podle 2 × CPU + 1 jako výchozí heuristiku, ale měřte. Pro CPU-bound ML inference často méně workerů s vyšším threadpoolem dává lepší výsledky než hodně workerů (modely typicky drží stovky MB paměti a každý worker je další kopie). Na Kubernetu pak škálujte horizontálně přes HPA na základě CPU utilizace nebo p95 latence z Prometheus metrik.

FastAPI vs Flask, BentoML a Ray Serve

Pro tabulární modely a embedding služby v rozsahu stovek až desítek tisíc QPS na uzel je FastAPI správná volba. Pro jiné scénáře existují vhodnější nástroje:

Vlastnost	FastAPI	Flask	BentoML	Ray Serve
Async podpora	Nativní (ASGI)	Pouze přes async addony	Nativní	Nativní
Validace inputů	Pydantic v2 (zdarma)	Manuální	Pydantic	Manuální
Dynamic batching	Manuální	Manuální	Vestavěné	Vestavěné
Multi-model serving	Manuální	Manuální	Vestavěné	Vestavěné
GPU inference	Manuální	Manuální	Vestavěné	Vestavěné
Latence (single request)	Nejnižší	Vyšší	Vyšší (overhead)	Vyšší (overhead)
Kdy zvolit	Tabulární/text NLP, < 50 ms p99	Legacy, jednoduché demo	Vícemodel platforma	Distributed, GPU pool

Flask se v roce 2026 stále drží v legacy stacích, ale pro nové ML projekty není důvod ho volit. FastAPI je rychlejší, type-safe a má lepší DX. BentoML dává smysl, když potřebujete platformu pro mnoho modelů s jednotným deploymentem a vestavěným dynamic batchingem. Ray Serve volte, když máte distribuovaný GPU pool a chcete autoscaling per model. Pro ETL pipeline, které tato API zpravidla doplňují, doporučuji článek o ETL pipeline v Pythonu s Pandas a SQLAlchemy.

Bližší informace o async patternech a deployment best practices najdete v oficiální FastAPI deployment dokumentaci.

Často kladené otázky

Jak nasadit ML model s FastAPI do produkce?

Zabalte natrénovaný model (joblib pro sklearn, ONNX pro framework-agnostické nasazení) do FastAPI aplikace s Pydantic schématy pro vstup, načtěte model v lifespan jednou na worker, spusťte přes Gunicorn s Uvicorn workery v Docker kontejneru a před něj postavte reverse proxy s Prometheus scrape pro metriky.

Je FastAPI rychlejší než Flask pro ML inference?

Ano, typicky 2–4× nižší overhead na request díky ASGI, Starlette a Pydantic v2 (napsaný v Rustu). Pro CPU-bound ML inference je rozdíl menší (model dominuje latenci), ale validace a serializace jsou výrazně rychlejší. Pro nové projekty v roce 2026 nemá Flask oproti FastAPI žádnou technickou výhodu.

Mám použít async def nebo def endpoint pro ML predikci?

Pro čistou CPU-bound predikci použijte def, FastAPI ji pustí v threadpoolu a worker zvládne paralelní requesty. Pro endpoint, který volá async I/O (databáze, vector store, external API), použijte async def a synchronní model.predict() obalte run_in_threadpool, aby neblokoval event loop.

Jak FastAPI ML API škálovat na Kubernetes?

Spusťte Gunicorn s 2–4 Uvicorn workery na pod (podle paměti modelu), nastavte Kubernetes readiness probe na /health/ready, který vrací 503 dokud není model načten, a HorizontalPodAutoscaler škálujte na CPU utilizaci nebo p95 latenci z Prometheus metrik přes KEDA.

Kolik requestů za sekundu FastAPI ML API zvládne?

Záleží primárně na modelu, ne na FastAPI. Pro malý sklearn tabulární model (predict_proba pod 1 ms) běžně vidím 2 000–5 000 RPS na 4-core pod s Gunicornem. Pro ONNX kvantizovaný model klidně 10 000+ RPS. Pro PyTorch transformer model na CPU často jen desítky RPS, kde je řešení GPU, batching nebo distilovaný model.