Розгортання ML-моделей на FastAPI у 2026: від scikit-learn до продакшн-API

Оновлено: 4 серпня 2026

Розгортання ML-моделей на FastAPI у 2026 році, це найкоротший шлях від навченого scikit-learn pipeline до продакшн-API з валідацією, документацією OpenAPI та async-обробкою запитів. Я пройшов через півдесятка перенесень моделей з Jupyter у продакшн і скажу прямо: якщо ваша модель важить менше кількох гігабайт і не потребує GPU-серверу вагових коефіцієнтів, то стек за замовчуванням це FastAPI + Pydantic v2 + Uvicorn. У цьому посібнику покажу мінімальний робочий приклад, розберу lifespan-події, батчинг, Docker-збірку та моніторинг Prometheus.

• FastAPI 0.115+ використовує lifespan async-контекстний менеджер замість застарілого @app.on_event, тож модель завантажується один раз при старті сервера.
• Pydantic v2 з Rust-ядром робить валідацію запитів у 5–50 разів швидшою, ніж Pydantic v1. Це критично для ML-ендпоінтів з великими JSON-пейлоадами.
• Для CPU-bound inference (класичні моделі scikit-learn, XGBoost) використовуйте синхронні def-ендпоінти, і FastAPI автоматично виконає їх у threadpool.
• Батчинг запитів через asyncio.Queue або fastapi-batch збільшує throughput у 3–10 разів для векторних моделей.
• У продакшн запускайте gunicorn з uvicorn.workers.UvicornWorker, по одному воркеру на CPU-ядро. Модель дублюється, але throughput масштабується лінійно.
• Мінімум, який очікує будь-який Kubernetes-кластер: health-чек /livez, /readyz та Prometheus-метрики через prometheus-fastapi-instrumentator.

Чому FastAPI: стандарт для розгортання ML-моделей у 2026

FastAPI став де-факто стандартом для ML-сервісів через три речі: Pydantic-валідацію з коробки, автогенеровану OpenAPI-документацію та async ASGI-архітектуру на Starlette. Чесно кажучи, коли я вперше витягував модель кредитного скорингу з Flask-сервісу, найбільш болючою частиною була не сама модель. Це був хендкодинг валідаторів для 40 фіч, які клієнтський фронтенд періодично надсилав то як float, то як string. У FastAPI ця робота зникає: описали BaseModel, отримали і валідацію, і документацію, і типізовані підказки в IDE.

За даними опитування Stack Overflow Developer Survey 2025, FastAPI обігнав Flask серед професійних Python-розробників. Для ML-сервісів причина проста. Бібліотеки scikit-learn, xgboost, lightgbm, catboost усі повертають CPU-bound передбачення, а FastAPI дозволяє тримати одну модель у пам'яті та ділити її між воркерами через gunicorn. Це дає передбачувану латентність (5–50 мс на запит для більшості табличних моделей) і чіткий шлях до Kubernetes через Docker.

Якщо ви серверите LLM або дифузійні моделі, FastAPI сам по собі не підійде. Подивіться натомість наш посібник з vLLM для продакшн-сервінгу LLM у 2026, який спеціалізується на генеративних моделях з continuous batching.

Мінімальний FastAPI-сервіс: від scikit-learn до /predict за 15 хвилин

Почнемо з навчання та збереження моделі. Використаю простий класифікатор iris, тому що фокус тут на сервінгу, а не на ML. Той самий шаблон працює для градієнтного бустингу з нашого порівняння XGBoost, LightGBM та CatBoost.

# train.py
from sklearn.datasets import load_iris
from sklearn.ensemble import RandomForestClassifier
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
import joblib

X, y = load_iris(return_X_y=True)
pipeline = Pipeline([
    ("scaler", StandardScaler()),
    ("clf", RandomForestClassifier(n_estimators=100, random_state=42)),
])
pipeline.fit(X, y)
joblib.dump(pipeline, "iris_model.joblib")
print(f"scikit-learn version: {joblib.__version__}")

Тепер сам FastAPI-сервіс. Це повний робочий приклад з lifespan-подіями, Pydantic-схемою та health-чеком:

# app.py
from contextlib import asynccontextmanager
from typing import Annotated
import joblib
import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

ml_models: dict = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Завантажуємо модель ОДИН РАЗ при старті сервера
    ml_models["iris"] = joblib.load("iris_model.joblib")
    yield
    # Очищуємо ресурси при зупинці
    ml_models.clear()

app = FastAPI(title="Iris Classifier API", lifespan=lifespan)

class IrisFeatures(BaseModel):
    sepal_length: Annotated[float, Field(gt=0, lt=10)]
    sepal_width: Annotated[float, Field(gt=0, lt=10)]
    petal_length: Annotated[float, Field(gt=0, lt=10)]
    petal_width: Annotated[float, Field(gt=0, lt=10)]

class Prediction(BaseModel):
    class_id: int
    class_name: str
    probability: float

CLASSES = ["setosa", "versicolor", "virginica"]

@app.get("/livez")
def liveness():
    return {"status": "ok"}

@app.get("/readyz")
def readiness():
    if "iris" not in ml_models:
        raise HTTPException(status_code=503, detail="model not loaded")
    return {"status": "ready"}

@app.post("/predict", response_model=Prediction)
def predict(features: IrisFeatures):
    X = np.array([[
        features.sepal_length, features.sepal_width,
        features.petal_length, features.petal_width,
    ]])
    model = ml_models["iris"]
    proba = model.predict_proba(X)[0]
    class_id = int(np.argmax(proba))
    return Prediction(
        class_id=class_id,
        class_name=CLASSES[class_id],
        probability=float(proba[class_id]),
    )

Запустити локально: uvicorn app:app --reload --port 8000. Відкрийте http://localhost:8000/docs, і FastAPI сам згенерує інтерактивну Swagger UI з формою для тестування /predict. Це та частина роботи, яку у Flask довелося б писати руками через flask-restx або flasgger.

Валідація вхідних даних з Pydantic v2 та Rust-ядром

Pydantic v2, який FastAPI використовує за замовчуванням з версії 0.100, переписаний на Rust через pydantic-core. Це не косметичне поліпшення: валідація моделей стала у 5–50 разів швидшою залежно від структури даних. Для ML-сервісу, який приймає JSON з 200+ ознаками (типовий фіче-вектор для табличної моделі), різниця між v1 і v2 це різниця між 8 мс і 0.3 мс на валідацію одного запиту. При 1000 RPS це різниця між 8 CPU-секундами та 0.3 CPU-секунди щосекунди.

Практичний приклад, модель, яка приймає масив фіч довільної довжини з обмеженнями:

from pydantic import BaseModel, Field, field_validator
from typing import Annotated

class PredictionRequest(BaseModel):
    features: Annotated[list[float], Field(min_length=10, max_length=200)]
    model_version: str = "v1"
    request_id: str | None = None

    @field_validator("features")
    @classmethod
    def check_no_nan(cls, v: list[float]) -> list[float]:
        if any(x != x for x in v):  # NaN != NaN
            raise ValueError("features contain NaN, clean data before sending")
        return v

Три речі, які варто застосовувати завжди: Field(gt=0) замість ручних перевірок if x <= 0, field_validator для кастомних правил (наприклад, заборона NaN, типова причина падінь inference), і response_model у декораторі ендпоінту, бо Pydantic автоматично серіалізує та валідує вихід. Детальніше про модельний код у нашому посібнику з автоматизованого очищення даних з pandas та pandera.

Порада: якщо на профайлингу JSON-серіалізація стає вузьким місцем, встановіть orjson і додайте default_response_class=ORJSONResponse. Це дає ще 2–3x прискорення на великих відповідях.

Як завантажити ML-модель у FastAPI: lifespan-події та lazy loading

Найпоширеніша помилка новачків, це завантажувати модель на верхньому рівні модуля або в кожному запиті. Перше зламає --reload у розробці і подвоїть RAM при запуску Gunicorn з preload; друге вб'є латентність, бо joblib.load для 500 МБ моделі це 2–5 секунд дискового I/O. Правильний шлях, це lifespan-контекстний менеджер, який FastAPI офіційно рекомендує з версії 0.93 (офіційна документація Lifespan Events).

Патерн, який я використовую у продакшн, тримає моделі в замиканні модуля, підтримує lazy warm-up та коректно очищує GPU-пам'ять при shutdown:

from contextlib import asynccontextmanager
from pathlib import Path
import joblib
import logging

log = logging.getLogger(__name__)

class ModelRegistry:
    def __init__(self) -> None:
        self._models: dict[str, object] = {}

    def load(self, name: str, path: Path) -> None:
        log.info("loading model %s from %s", name, path)
        self._models[name] = joblib.load(path)

    def get(self, name: str):
        if name not in self._models:
            raise KeyError(f"model {name} not loaded")
        return self._models[name]

    def clear(self) -> None:
        self._models.clear()

registry = ModelRegistry()

@asynccontextmanager
async def lifespan(app):
    registry.load("iris_v1", Path("iris_model.joblib"))
    registry.load("iris_v2", Path("iris_model_v2.joblib"))
    # warm-up: перший predict завжди повільніший через lazy imports
    import numpy as np
    for name in ("iris_v1", "iris_v2"):
        registry.get(name).predict(np.zeros((1, 4)))
    yield
    registry.clear()

Warm-up виклик, це трюк, який рятує від "перший запит клієнта завжди повільніший на 500 мс". Причина в тому, що scikit-learn ліниво імпортує BLAS-бекенди при першому виклику .predict. Один прогін нульового масиву при старті прибирає цей джиттер.

async чи sync: коли async корисний для ML-ендпоінтів, а коли шкідливий

Це найбільш неправильно зрозуміла тема у FastAPI-сервісах. Правило просте: async def варто використовувати для I/O-bound операцій (виклики БД, HTTP до інших сервісів, читання файлів через aiofiles), а звичайний def для CPU-bound роботи. Inference класичних моделей (scikit-learn, XGBoost, LightGBM), це CPU-bound. Якщо ви напишете async def predict і виконаєте всередині model.predict(X), ви заблокуєте event loop, і всі інші запити застрягнуть, поки цей predict не завершиться.

Ось як не треба:

# НЕПРАВИЛЬНО: блокує event loop
@app.post("/predict")
async def predict_bad(features: IrisFeatures):
    X = np.array([features.model_dump().values()]).reshape(1, -1)
    return {"pred": int(model.predict(X)[0])}  # блокуючий виклик!

А ось як правильно, просто прибрати async:

# ПРАВИЛЬНО: FastAPI сам виконає у threadpool
@app.post("/predict")
def predict_good(features: IrisFeatures):
    X = np.array([features.model_dump().values()]).reshape(1, -1)
    return {"pred": int(model.predict(X)[0])}

FastAPI автоматично виконує синхронні ендпоінти у threadpool через anyio, тому кілька predict-запитів справді виконуються паралельно (обмежені GIL для чистого Python, але numpy/scikit-learn відпускають GIL у C-коді). Використовуйте async def, якщо всередині є справжні await-виклики: наприклад, підвантаження фіч з Redis через aioredis, запит до фіче-стору через httpx.AsyncClient, або запис аудитного логу в PostgreSQL через asyncpg.

Batch inference: як пакетна обробка збільшує throughput у 10 разів

Для векторних моделей (нейромереж, ансамблів дерев, ембеддінг-моделей) виклик model.predict(X) на масиві з 100 записів у 10–50 разів швидший, ніж 100 окремих викликів на одному записі. Це і є ключова причина, чому vLLM, TorchServe і Triton обов'язково мають dynamic batching. Для класичних моделей я реалізую власний батчер на asyncio.Queue, який групує запити з вікна ~10–50 мс.

import asyncio
import numpy as np
from fastapi import FastAPI

app = FastAPI()

BATCH_QUEUE: asyncio.Queue = asyncio.Queue()
MAX_BATCH = 32
MAX_WAIT_MS = 20

async def batch_worker():
    while True:
        items = []
        try:
            first = await BATCH_QUEUE.get()
            items.append(first)
            deadline = asyncio.get_event_loop().time() + MAX_WAIT_MS / 1000
            while len(items) < MAX_BATCH:
                timeout = deadline - asyncio.get_event_loop().time()
                if timeout <= 0:
                    break
                try:
                    items.append(await asyncio.wait_for(
                        BATCH_QUEUE.get(), timeout=timeout))
                except asyncio.TimeoutError:
                    break
            # виконуємо предикт над батчем
            X = np.vstack([it["features"] for it in items])
            preds = model.predict(X)
            for it, pred in zip(items, preds):
                it["future"].set_result(int(pred))
        except Exception as e:
            for it in items:
                it["future"].set_exception(e)

@app.on_event("startup")
async def start_worker():
    asyncio.create_task(batch_worker())

@app.post("/predict")
async def predict(features: IrisFeatures):
    fut: asyncio.Future = asyncio.get_event_loop().create_future()
    await BATCH_QUEUE.put({
        "features": np.array([[features.sepal_length, features.sepal_width,
                               features.petal_length, features.petal_width]]),
        "future": fut,
    })
    return {"pred": await fut}

Компроміс тут очевидний: латентність окремого запиту зростає на MAX_WAIT_MS, але throughput сервісу зростає у 5–10 разів на векторних моделях. Для трафіку >100 RPS це майже завжди вигідна угода. Для трафіку <10 RPS батчинг зайвий, краще залишити прямий predict.

Docker, Uvicorn, Gunicorn: як розгорнути FastAPI ML-сервіс у продакшн

У розробці uvicorn --reload це нормально. У продакшн потрібен процес-менеджер, який перезапустить воркер при OOM або сегфолті. Стандарт індустрії, це gunicorn з класом воркерів uvicorn.workers.UvicornWorker. Кількість воркерів = кількість CPU-ядер (кожен тримає повну копію моделі, тож пам'ять множиться).

Мінімальний Dockerfile, який я використовую:

FROM python:3.12-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

FROM python:3.12-slim
WORKDIR /app
COPY --from=builder /root/.local /root/.local
ENV PATH=/root/.local/bin:$PATH
COPY app.py iris_model.joblib ./

# Читаємо CPU з cgroup, не з /proc/cpuinfo, важливо для Kubernetes
ENV WEB_CONCURRENCY=2
EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=5s CMD curl -f http://localhost:8000/livez || exit 1

CMD ["gunicorn", "app:app", \
     "-k", "uvicorn.workers.UvicornWorker", \
     "-b", "0.0.0.0:8000", \
     "--workers", "2", \
     "--timeout", "60", \
     "--graceful-timeout", "30", \
     "--access-logfile", "-"]

Файл requirements.txt має бути з піннутими версіями. Я бачив забагато продакшн-падінь через "неочікуване оновлення scikit-learn зламало pickle". Приклад: fastapi==0.115.6, uvicorn[standard]==0.32.1, gunicorn==23.0.0, scikit-learn==1.8.0, pydantic==2.9.2. Ті ж принципи ми використовували у нашому посібнику з ML-пайплайнів на scikit-learn 1.8.

Моніторинг ML-API: Prometheus-метрики, health-чеки та model drift

ML-сервіс без метрик, це чорна скринька, яка з часом дрейфує. Мінімум, що варто експортувати: латентність (histogram), request rate (counter), розподіл вхідних фіч (histogram) та розподіл передбачень (counter по класах). Найшвидший шлях, це prometheus-fastapi-instrumentator:

from prometheus_fastapi_instrumentator import Instrumentator
from prometheus_client import Counter, Histogram

predictions_counter = Counter(
    "ml_predictions_total",
    "Total predictions by class",
    ["class_name"],
)
prediction_confidence = Histogram(
    "ml_prediction_confidence",
    "Model output probability",
    buckets=[0.1, 0.3, 0.5, 0.7, 0.9, 0.95, 0.99],
)

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

@app.post("/predict", response_model=Prediction)
def predict(features: IrisFeatures):
    X = np.array([[features.sepal_length, features.sepal_width,
                   features.petal_length, features.petal_width]])
    proba = registry.get("iris_v1").predict_proba(X)[0]
    class_id = int(np.argmax(proba))
    predictions_counter.labels(class_name=CLASSES[class_id]).inc()
    prediction_confidence.observe(float(proba[class_id]))
    return Prediction(class_id=class_id, class_name=CLASSES[class_id],
                      probability=float(proba[class_id]))

На дашборді Grafana три речі варто дивитись щодня: p95 latency (регрес зазвичай означає деградацію моделі або GC-паузи), розподіл класів у передбаченнях (різкий зсув = data drift), розподіл confidence (падіння середньої впевненості = модель бачить дані, яких не бачила у тренуванні). Для більш серйозного моніторингу підключіть evidently або whylogs, вони автоматично рахують метрики дрейфу порівняно з референсним датасетом.

FastAPI vs Flask vs BentoML vs Ray Serve: що обрати у 2026

Вибір залежить від масштабу та типу моделі. Ось порівняльна таблиця, яку я тримаю у голові при виборі стеку:

Критерій	FastAPI	Flask	BentoML	Ray Serve
Async-підтримка	Нативна (ASGI)	Обмежена	Так	Так
Валідація запитів	Pydantic v2	Ручна	Pydantic	Ручна
OpenAPI-документація	Автоматична	Через flasgger	Автоматична	Немає
Dynamic batching	Ручний	Ручний	Вбудований	Вбудований
Мульти-модельний деплой	Ручний	Ручний	Вбудований	Вбудований
GPU-масштабування	Через K8s	Через K8s	Через Yatai	Нативне
Крива навчання	Низька	Дуже низька	Середня	Висока
Найкраще для	Малі/середні API	Legacy-код	MLOps-платформи	Розподілений inference

Мій практичний висновок: FastAPI за замовчуванням для 90% табличних та легких deep-learning моделей. BentoML, коли треба ML-платформа з реєстром моделей, батчингом, canary-деплоями (він фактично оболонка над FastAPI + Runner-архітектура). Ray Serve для розподілених inference-пайплайнів з кількома моделями та автоскейлінгом на GPU. Flask, лише якщо у вас вже є Flask-моноліт і додавання окремого сервісу дорожче за розширення існуючого.

Часті запитання

Як завантажити ML-модель у FastAPI лише один раз?

Використовуйте lifespan async-контекстний менеджер, переданий у FastAPI(lifespan=...). Код до yield виконається один раз при старті сервера, після yield, при shutdown. Це замінило застарілий @app.on_event("startup") починаючи з FastAPI 0.93.

Чим FastAPI кращий за Flask для розгортання ML-моделей?

Три головні переваги: автоматична валідація вхідних даних через Pydantic v2 (з Rust-ядром, у 5–50 разів швидша за Flask-схеми), автогенерована OpenAPI-документація зі Swagger UI без додаткових пакетів, і нативна async-підтримка через ASGI/Starlette. Для табличних моделей FastAPI також показує вищу пропускну здатність через async-request handling.

Чи потрібно писати async def для ML-ендпоінтів?

Ні. Для CPU-bound inference (scikit-learn, XGBoost, LightGBM) використовуйте звичайний def, і FastAPI автоматично виконає його у threadpool через anyio. Виклик async def потрібен лише коли всередині є справжні await-виклики: запити до async-БД, HTTP через httpx.AsyncClient, читання через aiofiles.

Як батчити запити у FastAPI для ML-моделей?

Створіть asyncio.Queue як буфер, фоновий воркер який збирає запити з вікна 10–50 мс, виконує model.predict(X_batch) над зібраним батчем, і повертає результати через asyncio.Future. Для готового рішення дивіться fastapi-batch або BentoML Runner, який реалізує dynamic batching з коробки.

Скільки Gunicorn-воркерів запускати для FastAPI ML-сервісу?

Стандартна рекомендація, це 2 * CPU_cores + 1, але для ML-сервісів з великими моделями (>500 МБ) кожен воркер тримає власну копію моделі у пам'яті. Тому починайте з workers = CPU_cores, замірюйте RAM, і за потреби використовуйте preload_app = True для copy-on-write економії пам'яті.

Як моніторити model drift у продакшн FastAPI-сервісі?

Мінімум: експортуйте розподіл вхідних фіч (Prometheus Histogram) та розподіл передбачень (Counter по класах), різкі зсуви між релізами сигналізують дрейф. Для повноцінного моніторингу підключіть evidently або whylogs: вони автоматично рахують PSI, KL-divergence та статистичні тести проти референсного датасету, і мають готові дашборди для Grafana.