FastAPI Serving Scikit-Learn di Production 2026: Latency, Batching, dan Best Practices

Diperbarui: 2 Juli 2026

Serving model scikit-learn dengan FastAPI di production butuh tiga hal: endpoint dengan Pydantic V2 untuk validasi request, worker Uvicorn di belakang Gunicorn untuk paralelisme, dan strategi caching model supaya tidak di-load ulang setiap request. Kombinasi ini rutin mencapai p99 latency di bawah 30 ms untuk model klasik seperti gradient boosting atau logistic regression di instance CPU kecil. Panduan ini adalah playbook production yang saya pakai untuk memindahkan model dari notebook ke endpoint yang bisa dijaga tim on-call, lengkap dengan batching, monitoring, dan pitfalls yang tidak muncul di tutorial "hello world".

• FastAPI 0.115+ dengan Pydantic V2 memberikan validasi request 5-10x lebih cepat dibanding V1, dan itu sangat cocok untuk endpoint inference latency-sensitive.
• Load model sekali di lifespan handler. Jangan pernah panggil joblib.load() di dalam fungsi endpoint karena bakal menambah 50-500 ms per request.
• Untuk model scikit-learn CPU-bound, jalankan endpoint sebagai def biasa (bukan async def) supaya FastAPI otomatis memindahkan ke thread pool dan tidak memblokir event loop.
• Micro-batching request dengan window 5-20 ms bisa menaikkan throughput 3-8x untuk model tree-based tanpa merusak SLA latency.
• Konversi ke ONNX Runtime memberikan speedup 2-4x untuk model scikit-learn dan menghilangkan dependency Python di container serving.
• Prometheus metrics untuk latency percentile (p50, p95, p99), error rate, dan model version wajib dipasang sebelum production, bukan setelah insiden pertama.

Kenapa FastAPI untuk ML Serving

FastAPI menang untuk serving model machine learning karena tiga alasan konkret: dukungan async native yang cocok untuk I/O-bound preprocessing, integrasi Pydantic V2 untuk validasi input yang ketat, dan overhead framework yang cukup kecil supaya tidak jadi bottleneck. Di production saya sendiri, migrasi dari Flask ke FastAPI menurunkan p99 latency dari 85 ms menjadi 42 ms untuk endpoint klasifikasi churn scikit-learn. Sebagian besar berkat Pydantic V2 dan penggunaan thread pool otomatis untuk endpoint CPU-bound.

Tapi FastAPI bukan silver bullet. Kalau model kamu berat (deep learning, transformer besar) atau butuh dynamic batching cross-request yang canggih, framework seperti BentoML, Ray Serve, atau NVIDIA Triton akan lebih tepat. Untuk 80% kasus scikit-learn (logistic regression, gradient boosting, random forest), FastAPI dengan setup yang benar sudah lebih dari cukup dan memberikan cost-per-prediction yang sangat kompetitif di instance c7i.large atau setara.

Yang sering dilupakan: FastAPI itu framework, bukan model server. Kamu tetap bertanggung jawab atas warm-up model, health check, graceful shutdown, dan integrasi dengan model registry. Bagusnya, kesederhanaan ini bikin kamu bisa audit setiap milidetik latency. Tidak ada black box seperti di beberapa serving framework demo-only.

Arsitektur Endpoint Production yang Benar

Sebelum menulis kode, pikirkan arsitektur end-to-end dulu. Endpoint ML production yang sehat punya lima lapisan: (1) load balancer atau API gateway, (2) proses Uvicorn di belakang Gunicorn dengan beberapa worker, (3) app FastAPI dengan model yang di-cache di memory, (4) preprocessing pipeline yang deterministik, (5) inference engine (scikit-learn native atau ONNX Runtime). Setiap lapisan punya failure mode berbeda, dan monitoring harus meliput semuanya.

Struktur direktori project yang saya rekomendasikan:

ml-service/
├── app/
│   ├── main.py           # FastAPI app + lifespan
│   ├── schemas.py        # Pydantic models
│   ├── inference.py      # Wrapper model + preprocessing
│   ├── batching.py       # Micro-batcher
│   └── metrics.py        # Prometheus metrics
├── models/
│   └── churn_v3.joblib   # Model artifacts
├── tests/
│   └── test_inference.py
├── Dockerfile
├── gunicorn_conf.py
└── requirements.txt

Pisahkan logic inference dari route handler. Ini bukan sekadar clean code. Ini bikin model bisa dipakai ulang untuk batch job, testing tanpa HTTP layer, dan migrasi ke serving framework lain di masa depan tanpa refactor besar-besaran.

Load Model dengan Lifespan Handler

Kesalahan nomor satu yang saya lihat di kode kandidat: memuat model di dalam fungsi endpoint. Ini menambah 50-500 ms per request tergantung ukuran model, dan lebih parah lagi, setiap worker Uvicorn akan me-load ulang model dari disk secara berulang. Solusinya adalah lifespan handler FastAPI yang menjalankan setup sekali saat startup dan cleanup saat shutdown.

from contextlib import asynccontextmanager
from pathlib import Path
import joblib
from fastapi import FastAPI

MODEL_PATH = Path("models/churn_v3.joblib")
ml_state = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: load model sekali per worker
    model = joblib.load(MODEL_PATH)
    ml_state["model"] = model
    ml_state["model_version"] = "churn_v3"
    # Warm-up: jalankan 1 dummy prediction supaya cache CPU siap
    import numpy as np
    _ = model.predict(np.zeros((1, model.n_features_in_)))
    yield
    # Shutdown
    ml_state.clear()

app = FastAPI(title="Churn Prediction Service", lifespan=lifespan)

Perhatikan langkah warm-up prediction. Model tree-based seperti XGBoost dan LightGBM sering punya cold-start penalty 10-50 ms di request pertama karena JIT compilation dan cache CPU. Menjalankan satu inference dummy saat startup memastikan request user pertama tidak jadi outlier di grafik latency. Saya pertama kali kena hal ini saat launch produk baru dan dashboard menunjukkan spike aneh di 5 menit pertama tiap kali deploy. Ternyata cuma soal warm-up.

Validasi Request dengan Pydantic V2

Pydantic V2 di-release akhir 2023 dan sekarang jadi standar untuk FastAPI 0.100+. Core validation-nya ditulis dalam Rust, memberikan speedup 5-17x dibanding V1 di benchmark Pydantic official. Untuk endpoint ML dengan throughput tinggi, ini bukan optimasi mikro. Validasi request rutin memakan 20-30% dari total latency di serving Python.

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

class ChurnRequest(BaseModel):
    tenure_months: int = Field(ge=0, le=600, description="Lama berlangganan dalam bulan")
    monthly_charges: float = Field(ge=0, le=10000)
    total_charges: float = Field(ge=0)
    contract_type: Literal["month-to-month", "one-year", "two-year"]
    payment_method: Literal["electronic-check", "mailed-check", "bank-transfer", "credit-card"]
    internet_service: Literal["dsl", "fiber", "none"]

    @field_validator("total_charges")
    @classmethod
    def validate_totals(cls, v, info):
        # Cross-field validation: total tidak boleh negatif atau inkonsisten
        if v < 0:
            raise ValueError("total_charges tidak boleh negatif")
        return v

class ChurnResponse(BaseModel):
    churn_probability: float = Field(ge=0.0, le=1.0)
    prediction: Literal["churn", "retain"]
    model_version: str
    latency_ms: float

Perhatikan tiga hal. Pertama, pakai Literal untuk field kategorikal supaya input invalid ditolak sebelum sampai ke model. Kedua, Field(ge=..., le=...) memaksa range check yang deterministik. Ketiga, field_validator hanya untuk validasi yang tidak bisa diekspresikan sebagai constraint sederhana. Semakin banyak validasi di boundary, semakin sedikit surprise di grafik prediction distribution kamu.

Menulis Endpoint Inference yang Benar

Ini bagian yang paling sering salah dipahami: kapan pakai async def dan kapan pakai def. Aturannya simpel. Kalau handler kamu murni CPU-bound (memanggil model.predict() sinkron), pakai def. FastAPI otomatis memindahkan eksekusinya ke thread pool sehingga tidak memblokir event loop. Kalau pakai async def tapi memanggil kode blocking di dalamnya, kamu bakal menghambat semua request lain di worker yang sama.

import time
import numpy as np
from fastapi import FastAPI, HTTPException

@app.post("/predict", response_model=ChurnResponse)
def predict(request: ChurnRequest) -> ChurnResponse:
    start = time.perf_counter()
    try:
        # Preprocessing: konversi ke feature vector
        features = np.array([[
            request.tenure_months,
            request.monthly_charges,
            request.total_charges,
            request.contract_type,
            request.payment_method,
            request.internet_service,
        ]], dtype=object)

        model = ml_state["model"]
        proba = float(model.predict_proba(features)[0, 1])
        label = "churn" if proba >= 0.5 else "retain"

        latency_ms = (time.perf_counter() - start) * 1000
        return ChurnResponse(
            churn_probability=proba,
            prediction=label,
            model_version=ml_state["model_version"],
            latency_ms=round(latency_ms, 2),
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Inference error: {type(e).__name__}")

Kembalikan latency_ms di response untuk debugging. Ini sangat berguna saat menelusuri masalah p99 spike dari log. Untuk endpoint public-facing, hilangkan detail exception dari response body dan log ke observability platform terpisah.

Micro-Batching untuk Throughput 3-8x Lebih Tinggi

Untuk model tree-based seperti gradient boosting, cost prediksi tidak linear terhadap batch size. Memprediksi 32 sampel sekaligus hanya sekitar 1.5-2x lebih mahal dibanding 1 sampel karena overhead Python dan cache CPU. Micro-batching menangkap keuntungan ini dengan menampung request yang datang dalam window pendek (5-20 ms) lalu memprosesnya sebagai satu batch. Baca panduan Polars vs Pandas untuk pemrosesan batch untuk konteks bagaimana operasi vektorisasi mempercepat throughput data pipeline.

import asyncio
from dataclasses import dataclass, field

@dataclass
class BatchItem:
    features: np.ndarray
    future: asyncio.Future = field(default_factory=asyncio.Future)

class MicroBatcher:
    def __init__(self, max_batch: int = 32, max_wait_ms: int = 10):
        self.max_batch = max_batch
        self.max_wait_ms = max_wait_ms
        self.queue = []
        self.lock = asyncio.Lock()
        self._task = None

    async def submit(self, features):
        item = BatchItem(features=features)
        async with self.lock:
            self.queue.append(item)
            if len(self.queue) >= self.max_batch:
                self._flush()
            elif self._task is None or self._task.done():
                self._task = asyncio.create_task(self._flush_after_delay())
        return await item.future

    async def _flush_after_delay(self):
        await asyncio.sleep(self.max_wait_ms / 1000)
        async with self.lock:
            self._flush()

    def _flush(self):
        if not self.queue:
            return
        batch, self.queue = self.queue, []
        stacked = np.vstack([item.features for item in batch])
        preds = ml_state["model"].predict_proba(stacked)[:, 1]
        for item, pred in zip(batch, preds):
            item.future.set_result(pred)

Trade-off yang harus kamu terima: latency p50 bakal sedikit naik (misalnya dari 8 ms jadi 12 ms) karena wait window, tapi throughput naik signifikan. Pastikan max_wait_ms tetap di bawah SLA latency budget kamu. Untuk endpoint dengan SLA 100 ms p99, window 10-15 ms sangat aman.

Optimasi Latency dengan ONNX Runtime

Salah satu quick win terbesar untuk model scikit-learn di production adalah konversi ke ONNX Runtime. Speedup 2-4x sangat umum untuk tree-based model, dan kamu bisa menghilangkan seluruh stack scikit-learn dari container serving. Yang kamu butuhkan cuma onnxruntime. Ini menurunkan ukuran Docker image dari sekitar 1.5 GB ke 200 MB.

# Konversi (dilakukan sekali, biasanya di CI pipeline)
from skl2onnx import to_onnx
import numpy as np

X_sample = np.zeros((1, model.n_features_in_), dtype=np.float32)
onnx_model = to_onnx(model, X_sample, target_opset=17)
with open("models/churn_v3.onnx", "wb") as f:
    f.write(onnx_model.SerializeToString())

# Serving
import onnxruntime as ort

session_options = ort.SessionOptions()
session_options.intra_op_num_threads = 1  # Penting untuk high-concurrency serving
session_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL

sess = ort.InferenceSession(
    "models/churn_v3.onnx",
    sess_options=session_options,
    providers=["CPUExecutionProvider"],
)

def predict_onnx(features):
    output = sess.run(None, {"input": features.astype(np.float32)})
    return float(output[1][0][1])  # probability class 1

Setting intra_op_num_threads=1 kelihatan counterintuitive tapi kritikal. Di serving multi-worker, kamu ingin setiap request pakai satu core saja supaya tidak ada kontensi CPU antar-worker. ONNX Runtime default akan pakai semua core, dan itu justru menurunkan throughput saat concurrency tinggi.

Deployment: Uvicorn + Gunicorn

Untuk production, jalankan Uvicorn workers di belakang Gunicorn process manager. Gunicorn menangani worker lifecycle, graceful reload, dan restart otomatis kalau worker crash. Kalau kamu punya pipeline data yang berat untuk feature engineering, panduan feature engineering dengan scikit-learn Pipeline membantu memastikan preprocessing kamu deterministik antara training dan serving.

# gunicorn_conf.py
import multiprocessing

workers = min(4, multiprocessing.cpu_count())
worker_class = "uvicorn.workers.UvicornWorker"
bind = "0.0.0.0:8000"
timeout = 60
graceful_timeout = 30
keepalive = 5
max_requests = 10000
max_requests_jitter = 500  # Restart worker berkala untuk cegah memory leak
preload_app = False  # HARUS False supaya lifespan handler jalan per worker

Rumus jumlah worker untuk ML serving: min(4, CPU cores). Jangan pakai formula generic (2 * cores) + 1 dari Gunicorn docs, itu untuk workload I/O-bound tradisional. Untuk CPU-bound inference, terlalu banyak worker justru menyebabkan context switching dan cache thrashing.

Dockerfile minimal untuk production:

FROM python:3.12-slim

WORKDIR /app

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

COPY app/ ./app/
COPY models/ ./models/
COPY gunicorn_conf.py .

# Non-root user untuk security
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser

EXPOSE 8000
HEALTHCHECK --interval=15s --timeout=3s CMD curl -f http://localhost:8000/health || exit 1

CMD ["gunicorn", "app.main:app", "-c", "gunicorn_conf.py"]

Monitoring dengan Prometheus

Endpoint tanpa monitoring itu utang teknis. Minimal metrics yang harus kamu ekspos: request count, latency histogram (untuk p50/p95/p99), error rate per exception type, model version, dan prediction distribution. Prometheus dengan library prometheus-client adalah opsi standar dan gratis.

from prometheus_client import Counter, Histogram, make_asgi_app

INFERENCE_LATENCY = Histogram(
    "ml_inference_latency_seconds",
    "Latency inference model dalam detik",
    labelnames=["model_version", "endpoint"],
    buckets=(0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5),
)

INFERENCE_REQUESTS = Counter(
    "ml_inference_requests_total",
    "Total request inference",
    labelnames=["model_version", "endpoint", "status"],
)

PREDICTION_DISTRIBUTION = Histogram(
    "ml_prediction_probability",
    "Distribusi probability prediksi",
    labelnames=["model_version"],
    buckets=(0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0),
)

# Mount /metrics endpoint
app.mount("/metrics", make_asgi_app())

Prediction distribution adalah metrics yang paling sering dilupakan, tapi paling berharga. Kalau distribusi probability tiba-tiba shift (misalnya rata-rata naik dari 0.3 ke 0.6 dalam sehari), itu sinyal kuat ada data drift atau bug preprocessing, jauh sebelum kamu dapat komplain dari stakeholder. Jujur, metrics ini yang menyelamatkan saya dari insiden bad-deploy di project sebelumnya.

FastAPI vs Flask vs BentoML: Mana yang Tepat?

Aspek	FastAPI	Flask	BentoML
Validasi Request	Native (Pydantic V2)	Manual / Marshmallow	Native
Async Support	First-class	Terbatas (2.0+)	Native
Dynamic Batching	Custom implementation	Custom implementation	Built-in
Model Registry	Bring your own (MLflow)	Bring your own	Built-in
Overhead per Request	~1-2 ms	~2-4 ms	~2-3 ms
Learning Curve	Rendah	Sangat rendah	Medium
Docker Image Size	~200 MB	~180 MB	~400 MB
Ideal Untuk	Model klasik + custom pipeline	Legacy / prototype	Fleet model dengan variasi tinggi

Rule of thumb yang saya pakai: kalau kamu serving 1-3 model dengan preprocessing kustom yang non-trivial, pilih FastAPI. Kalau kamu punya 10+ model dengan lifecycle berbeda dan butuh model registry terintegrasi, evaluasi BentoML. Flask masih valid untuk maintenance sistem legacy, tapi untuk project baru, keuntungan Pydantic V2 dan async support di FastAPI hampir selalu menang.

Pitfalls yang Sering Terjadi di Production

Beberapa masalah yang saya sudah lihat berulang kali:

Model di-load ulang setiap request

Sudah dibahas di atas, tapi masih sering muncul. Symptomnya: p99 latency 500+ ms dengan p50 di 30 ms. Pemeriksaan cepat, periksa apakah joblib.load() dipanggil di dalam route handler.

Blocking call di async endpoint

Endpoint async def yang memanggil model.predict() langsung. Ini memblokir event loop dan menyebabkan tail latency yang parah saat traffic naik. Solusi: pakai def biasa, atau wrap dengan await run_in_threadpool(model.predict, features).

Preprocessing skew antara training dan serving

Pipeline training pakai pandas, serving pakai numpy manual, hasil encoding sedikit berbeda, dan model berpredikasi ngawur. Solusinya: simpan preprocessing sebagai bagian dari scikit-learn Pipeline, deserialize keseluruhan artifact, jangan pernah tulis ulang preprocessing di serving side. Untuk validasi cepat, kamu bisa pakai DuckDB untuk query batch data serving dan membandingkan feature distribution antara training dan production.

Memory leak dari model registry client

Library seperti MLflow client bisa memleak connection kalau tidak di-close. Kalau kamu melakukan hot-swap model, pastikan tutup client lama sebelum ganti reference.

Serialization mismatch versi scikit-learn

Model di-train dengan scikit-learn 1.4 dan deserialize di container dengan versi 1.6, kadang jalan, kadang segfault. Pin versi scikit-learn di serving container ke versi persis training. Atau, hindari masalah ini sepenuhnya dengan ONNX.

Pertanyaan yang Sering Diajukan

Bagaimana cara deploy model scikit-learn ke production dengan FastAPI?

Simpan model dengan joblib.dump(), load sekali di FastAPI lifespan handler, buat endpoint POST dengan schema Pydantic V2 untuk validasi, dan jalankan dengan Uvicorn workers di belakang Gunicorn. Container-kan dengan Docker dan deploy ke Kubernetes, ECS, atau Cloud Run. Pastikan setup Prometheus metrics dari hari pertama.

Apakah FastAPI lebih cepat dari Flask untuk serving model ML?

Ya, umumnya 30-50% lebih cepat untuk endpoint ML karena Pydantic V2 (Rust-based validation), async support native, dan overhead framework yang lebih kecil. Perbedaan paling terasa untuk endpoint dengan payload kompleks dan request rate tinggi (di atas 100 req/s).

Bagaimana cara mengurangi latency inference model scikit-learn?

Empat quick win: (1) konversi ke ONNX Runtime untuk speedup 2-4x, (2) load model sekali di lifespan handler, (3) implementasi micro-batching untuk throughput lebih tinggi, (4) hindari preprocessing pandas untuk single-row request, pakai numpy langsung. Untuk model yang masih terlalu lambat, evaluasi feature reduction atau distilasi model.

Berapa banyak request per second yang bisa FastAPI handle untuk ML?

Tergantung ukuran model, tapi untuk logistic regression atau gradient boosting kecil di instance c7i.large (2 vCPU), 500-2000 req/s realistis dengan latency p99 di bawah 50 ms. Model tree ensemble besar (XGBoost 1000+ trees) turun ke 100-300 req/s. Batching bisa menaikkan throughput 3-8x tanpa ubah hardware.

Kapan sebaiknya pakai BentoML atau Ray Serve dibanding FastAPI?

Pilih BentoML kalau kamu punya fleet model banyak dengan lifecycle bervariasi dan butuh model registry terintegrasi. Pilih Ray Serve untuk model yang butuh dynamic batching canggih, GPU scheduling, atau pipeline multi-model. Untuk 1-3 model scikit-learn dengan custom preprocessing, FastAPI biasanya cukup dan lebih mudah di-debug saat on-call.

Apakah aman menggunakan pickle atau joblib untuk model production?

Aman kalau file model berasal dari CI pipeline internal yang kamu kontrol penuh. Tidak aman untuk model dari sumber eksternal atau user upload karena pickle bisa mengeksekusi kode arbitrary saat deserialization. Untuk kasus tidak tepercaya, pakai skops atau ONNX yang punya format lebih terbatas dan lebih aman.