একটা ML মডেল সার্ভ করা

mkdir -p ~/projects/iris-api
cd ~/projects/iris-api
uv init --bare -q
uv add -q "fastapi[standard]" pydantic-settings scikit-learn joblib
uv add -q --dev pytest httpx2
# with pip: python -m venv .venv && source .venv/bin/activate && pip install -q "fastapi[standard]" pydantic-settings scikit-learn joblib pytest httpx2
mkdir -p app tests
touch app/__init__.py
cat >> pyproject.toml <<'EOF'

[tool.fastapi]
entrypoint = "app.main:app"

[tool.pytest.ini_options]
pythonpath = ["."]
EOF

# file: train.py
"""Train two small classifiers and save them as versioned artifacts."""
import sys
from pathlib import Path

import joblib
import sklearn
from sklearn.datasets import load_iris
from sklearn.ensemble import RandomForestClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler

FEATURES = ["sepal_length", "sepal_width", "petal_length", "petal_width"]
CLASSES = ["setosa", "versicolor", "virginica"]


def build_bundle(version, estimator, X, y):
    """Fit scaler + estimator as ONE pipeline, so preprocessing can never be forgotten."""
    pipeline = make_pipeline(StandardScaler(), estimator)
    pipeline.fit(X, y)
    return {"version": version, "features": FEATURES, "classes": CLASSES,
            "sklearn": sklearn.__version__, "pipeline": pipeline}


def main(out_dir="models"):
    iris = load_iris()
    X_train, X_test, y_train, y_test = train_test_split(
        iris.data, iris.target, test_size=0.25, random_state=42, stratify=iris.target)
    candidates = {
        "v1": LogisticRegression(max_iter=200),
        "v2": RandomForestClassifier(n_estimators=50, random_state=42),
    }
    Path(out_dir).mkdir(exist_ok=True)
    for version, estimator in candidates.items():
        bundle = build_bundle(version, estimator, X_train, y_train)
        accuracy = bundle["pipeline"].score(X_test, y_test)
        joblib.dump(bundle, Path(out_dir) / f"iris-{version}.joblib")
        print(f"{version}: {type(estimator).__name__}, test accuracy {accuracy:.3f}")


if __name__ == "__main__":
    main(*sys.argv[1:])

python train.py
ls models

v1: LogisticRegression, test accuracy 0.921
v2: RandomForestClassifier, test accuracy 0.921
iris-v1.joblib
iris-v2.joblib

# file: app/settings.py
from pathlib import Path

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="IRIS_", extra="ignore")

    model_dir: Path = Path("models")
    default_version: str = "v1"
    predict_timeout_s: float = Field(default=2.0, gt=0)     # give up on one prediction after this
    max_concurrent: int = Field(default=4, ge=1)            # predictions running at once

# file: app/schemas.py
from typing import Annotated

from pydantic import BaseModel, ConfigDict, Field

MAX_BATCH = 100

Centimetres = Annotated[float, Field(gt=0, le=20, allow_inf_nan=False)]


class Flower(BaseModel):
    model_config = ConfigDict(extra="forbid")       # a misspelt feature is an error, not ignored

    sepal_length: Centimetres
    sepal_width: Centimetres
    petal_length: Centimetres
    petal_width: Centimetres


class BatchIn(BaseModel):
    items: list[Flower] = Field(min_length=1, max_length=MAX_BATCH)


class Result(BaseModel):
    label: str
    probabilities: dict[str, float]


class Prediction(Result):
    model_version: str


class BatchOut(BaseModel):
    model_version: str
    results: list[Result]

# file: app/registry.py
from pathlib import Path

import joblib
import numpy as np

from app.schemas import Flower

WARMUP_ROW = [[5.1, 3.5, 1.4, 0.2]]


def predict_rows(bundle: dict, rows: list[list[float]]) -> list[dict]:
    """Plain blocking function: one call for any number of rows."""
    proba = bundle["pipeline"].predict_proba(np.array(rows))
    results = []
    for row in proba:
        best = int(row.argmax())
        results.append({
            "label": bundle["classes"][best],
            "probabilities": {name: round(float(p), 3) for name, p in zip(bundle["classes"], row)},
        })
    return results


def load_registry(model_dir: Path) -> dict[str, dict]:
    """Load every iris-*.joblib file, keyed by version. Fail loudly, at startup."""
    registry = {}
    for path in sorted(model_dir.glob("iris-*.joblib")):
        bundle = joblib.load(path)
        if bundle["features"] != list(Flower.model_fields):
            raise RuntimeError(f"{path.name}: features {bundle['features']} do not match the API schema")
        predict_rows(bundle, WARMUP_ROW)                 # warm-up: result thrown away
        registry[bundle["version"]] = bundle
    if not registry:
        raise RuntimeError(f"No model files found in {model_dir}")
    return registry

# file: app/main.py
import asyncio
import json
import logging
import time
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Query, Request

from app.registry import load_registry, predict_rows
from app.schemas import BatchIn, BatchOut, Flower, Prediction, Result
from app.settings import Settings

log = logging.getLogger("iris_api.predictions")


class Metrics:
    """In-memory counters. In production export these to Prometheus, Datadog or your platform."""

    def __init__(self) -> None:
        self.requests = self.errors = self.slow = 0
        self.total_ms = self.max_ms = 0.0

    def record(self, ms: float, *, ok: bool) -> None:
        self.requests += 1
        self.errors += 0 if ok else 1
        self.slow += 1 if ms > 1000 else 0
        self.total_ms += ms
        self.max_ms = max(self.max_ms, ms)

    def snapshot(self) -> dict:
        average = self.total_ms / self.requests if self.requests else 0.0
        return {"requests": self.requests, "errors": self.errors, "slow_over_1s": self.slow,
                "avg_ms": round(average, 2), "max_ms": round(self.max_ms, 2)}


def create_app(settings: Settings | None = None) -> FastAPI:
    settings = settings or Settings()

    @asynccontextmanager
    async def lifespan(app: FastAPI):
        logging.basicConfig(level="INFO", format="%(asctime)s %(levelname)s %(name)s %(message)s")
        app.state.settings = settings
        app.state.registry = load_registry(settings.model_dir)     # blocks startup, not requests
        if settings.default_version not in app.state.registry:
            raise RuntimeError(f"Default model {settings.default_version!r} was not loaded")
        app.state.predict = predict_rows
        app.state.metrics = Metrics()
        app.state.slots = asyncio.Semaphore(settings.max_concurrent)
        yield
        app.state.registry.clear()                                  # release the memory

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

    def get_bundle(request: Request, version: Annotated[str | None, Query(pattern=r"^[a-z0-9.-]{1,20}$")] = None) -> dict:
        registry = request.app.state.registry
        wanted = version or request.app.state.settings.default_version
        if wanted not in registry:
            raise HTTPException(404, f"Unknown model version {wanted!r}; available: {sorted(registry)}")
        return registry[wanted]

    BundleDep = Annotated[dict, Depends(get_bundle)]

    async def infer(request: Request, bundle: dict, flowers: list[Flower]) -> list[dict]:
        state = request.app.state
        rows = [[getattr(flower, name) for name in bundle["features"]] for flower in flowers]
        started = time.perf_counter()
        ok = False
        try:
            async with state.slots:                                 # at most N predictions at once
                work = asyncio.to_thread(state.predict, bundle, rows)     # off the event loop
                results = await asyncio.wait_for(work, state.settings.predict_timeout_s)
            ok = True
        except TimeoutError:
            raise HTTPException(504, "The model took too long to answer") from None
        except Exception:
            log.exception("inference failed")                       # details stay in the logs
            raise HTTPException(500, "Prediction failed") from None
        finally:
            duration_ms = (time.perf_counter() - started) * 1000
            state.metrics.record(duration_ms, ok=ok)
        fields = {"model_version": bundle["version"], "n_items": len(rows), "ok": ok,
                  "labels": [r["label"] for r in results], "duration_ms": round(duration_ms, 1)}
        log.info("prediction %s", json.dumps(fields))
        return results

    @app.get("/health")
    def health() -> dict:
        return {"status": "ok", "versions": sorted(app.state.registry)}

    @app.get("/v1/model")
    def model_info(bundle: BundleDep) -> dict:
        return {"version": bundle["version"], "features": bundle["features"],
                "classes": bundle["classes"], "estimator": type(bundle["pipeline"][-1]).__name__}

    @app.post("/v1/predict")
    async def predict(flower: Flower, request: Request, bundle: BundleDep) -> Prediction:
        (result,) = await infer(request, bundle, [flower])
        return Prediction(model_version=bundle["version"], **result)

    @app.post("/v1/predict/batch")
    async def predict_batch(data: BatchIn, request: Request, bundle: BundleDep) -> BatchOut:
        results = await infer(request, bundle, data.items)
        return BatchOut(model_version=bundle["version"], results=[Result(**r) for r in results])

    @app.get("/v1/metrics")
    def metrics() -> dict:
        return app.state.metrics.snapshot()

    return app


app = create_app()

fastapi run --port 8000 > server.log 2>&1 &
export SERVER_PID=$!
export BASE=http://localhost:8000
curl -s --retry 20 --retry-connrefused --retry-delay 0 $BASE/health
echo
curl -s $BASE/v1/model

{"status":"ok","versions":["v1","v2"]}
{"version":"v1","features":["sepal_length","sepal_width","petal_length","petal_width"],"classes":["setosa","versicolor","virginica"],"estimator":"LogisticRegression"}

curl -s $BASE/v1/predict -H "Content-Type: application/json" \
  -d '{"sepal_length": 6.3, "sepal_width": 2.9, "petal_length": 5.6, "petal_width": 1.8}'
echo
curl -s "$BASE/v1/predict?version=v2" -H "Content-Type: application/json" \
  -d '{"sepal_length": 6.3, "sepal_width": 2.9, "petal_length": 5.6, "petal_width": 1.8}'
echo
curl -s "$BASE/v1/predict?version=v9" -H "Content-Type: application/json" \
  -d '{"sepal_length": 6.3, "sepal_width": 2.9, "petal_length": 5.6, "petal_width": 1.8}'

{"label":"virginica","probabilities":{"setosa":0.001,"versicolor":0.195,"virginica":0.804},"model_version":"v1"}
{"label":"virginica","probabilities":{"setosa":0.0,"versicolor":0.0,"virginica":1.0},"model_version":"v2"}
{"detail":"Unknown model version 'v9'; available: ['v1', 'v2']"}

import os

from fastapi.testclient import TestClient

from app.main import create_app
from app.settings import Settings

app = create_app(Settings())
client = TestClient(app).__enter__()          # runs the lifespan, like a real startup
good = {"sepal_length": 6.3, "sepal_width": 2.9, "petal_length": 5.6, "petal_width": 1.8}

bad_inputs = {
    "negative": {**good, "petal_length": -1},
    "missing": {k: v for k, v in good.items() if k != "petal_width"},
    "text": {**good, "sepal_width": "wide"},
    "misspelt": {**good, "petal_lenght": 5.6},
    "huge": {**good, "sepal_length": 9999},
}
for name, body in bad_inputs.items():
    response = client.post("/v1/predict", json=body)
    problem = response.json()["detail"][0]
    print(f"{name:9} {response.status_code} {problem['loc'][-1]}: {problem['msg']}")

negative  422 petal_length: Input should be greater than 0
missing   422 petal_width: Field required
text      422 sepal_width: Input should be a valid number, unable to parse string as a number
misspelt  422 petal_lenght: Extra inputs are not permitted
huge      422 sepal_length: Input should be less than or equal to 20

import time

import numpy as np

from app.registry import predict_rows

bundle = app.state.registry["v1"]
rows = np.tile([[6.3, 2.9, 5.6, 1.8]], (100, 1)).tolist()

start = time.perf_counter()
for row in rows:
    predict_rows(bundle, [row])
one_by_one = time.perf_counter() - start

start = time.perf_counter()
results = predict_rows(bundle, rows)
as_batch = time.perf_counter() - start

print(len(results), "results; batch faster than 100 single calls:", as_batch < one_by_one)
response = client.post("/v1/predict/batch", json={"items": [good, {**good, "petal_length": 1.4, "petal_width": 0.2}]})
print(response.json())
print(client.post("/v1/predict/batch", json={"items": []}).status_code,
      client.post("/v1/predict/batch", json={"items": [good] * 101}).status_code)

100 results; batch faster than 100 single calls: True
{'model_version': 'v1', 'results': [{'label': 'virginica', 'probabilities': {'setosa': 0.001, 'versicolor': 0.195, 'virginica': 0.804}}, {'label': 'versicolor', 'probabilities': {'setosa': 0.373, 'versicolor': 0.627, 'virginica': 0.0}}]}
422 422

def slow_predict(bundle, rows):
    time.sleep(0.5)
    return predict_rows(bundle, rows)


tight = create_app(Settings(predict_timeout_s=0.1))
with TestClient(tight) as tight_client:
    tight.state.predict = slow_predict
    response = tight_client.post("/v1/predict", json=good)
    print(response.status_code, response.json())
    tight.state.predict = predict_rows
    print(tight_client.post("/v1/predict", json=good).status_code)
    print(tight_client.get("/v1/metrics").json()["errors"], "error counted")

504 {'detail': 'The model took too long to answer'}
200
1 error counted

import json
import logging

lines = []


class Capture(logging.Handler):
    def emit(self, record):
        lines.append(record.getMessage())


logging.getLogger("iris_api").addHandler(Capture())
logging.getLogger("iris_api").setLevel(logging.INFO)

client.post("/v1/predict?version=v2", json=good)
entry = json.loads(lines[-1].removeprefix("prediction "))
print({key: value for key, value in entry.items() if key != "duration_ms"})
print("duration under 1 s:", entry["duration_ms"] < 1000)

stats = client.get("/v1/metrics").json()
print(stats["requests"] > 0, stats["errors"], stats["slow_over_1s"], "average under 1 s:", stats["avg_ms"] < 1000)

{'model_version': 'v2', 'n_items': 1, 'ok': True, 'labels': ['virginica']}
duration under 1 s: True
True 0 0 average under 1 s: True

kill $SERVER_PID
wait $SERVER_PID 2>/dev/null
echo stopped

stopped

# file: tests/conftest.py
import joblib
import pytest
from fastapi.testclient import TestClient
from sklearn.datasets import load_iris
from sklearn.linear_model import LogisticRegression

from app.main import create_app
from app.settings import Settings
from train import build_bundle

GOOD = {"sepal_length": 5.1, "sepal_width": 3.5, "petal_length": 1.4, "petal_width": 0.2}   # a setosa


@pytest.fixture
def client(tmp_path):
    iris = load_iris()
    X, y = iris.data[::5], iris.target[::5]                  # 30 rows: a tiny model
    joblib.dump(build_bundle("test", LogisticRegression(max_iter=200), X, y), tmp_path / "iris-test.joblib")
    app = create_app(Settings(model_dir=tmp_path, default_version="test", predict_timeout_s=0.2))
    with TestClient(app) as test_client:
        yield test_client

# file: tests/test_api.py
import time

import pytest

from app.registry import predict_rows
from tests.conftest import GOOD


def test_prediction_shape_and_version(client):
    body = client.post("/v1/predict", json=GOOD).json()
    assert body["model_version"] == "test"
    assert body["label"] == "setosa"
    assert abs(sum(body["probabilities"].values()) - 1) < 0.01


def test_batch_matches_single(client):
    single = client.post("/v1/predict", json=GOOD).json()["label"]
    batch = client.post("/v1/predict/batch", json={"items": [GOOD, GOOD]}).json()
    assert [r["label"] for r in batch["results"]] == [single, single]


@pytest.mark.parametrize("field,value", [("petal_length", -1), ("sepal_width", "wide"), ("sepal_length", 9999)])
def test_invalid_features_are_rejected(client, field, value):
    assert client.post("/v1/predict", json={**GOOD, field: value}).status_code == 422


def test_unknown_version_is_404(client):
    assert client.post("/v1/predict?version=nope", json=GOOD).status_code == 404


def test_slow_model_times_out_and_service_recovers(client):
    def slow(bundle, rows):
        time.sleep(0.5)
        return predict_rows(bundle, rows)

    client.app.state.predict = slow
    assert client.post("/v1/predict", json=GOOD).status_code == 504
    client.app.state.predict = predict_rows
    assert client.post("/v1/predict", json=GOOD).status_code == 200


def test_model_crash_is_a_generic_500(client):
    def broken(bundle, rows):
        raise ValueError("secret internal detail")

    client.app.state.predict = broken
    response = client.post("/v1/predict", json=GOOD)
    assert response.status_code == 500
    assert "secret" not in response.text

pytest -q

........                                                                                     [100%]
8 passed in 0.97s

# 1. Easy
import joblib

from app.registry import predict_rows

loaded = joblib.load("models/iris-v1.joblib")
print(predict_rows(loaded, [[5.1, 3.5, 1.4, 0.2]]))


# 2. Medium (run it as a pytest test: add it to tests/test_api.py)
def test_batch_limit(client):
    from tests.conftest import GOOD
    assert len(client.post("/v1/predict/batch", json={"items": [GOOD] * 100}).json()["results"]) == 100
    assert client.post("/v1/predict/batch", json={"items": [GOOD] * 101}).status_code == 422


# 3. Hard: the check can live in one small function that works on a result
def with_confidence(result: dict) -> dict:
    confidence = max(result["probabilities"].values())
    return {**result, "confidence": confidence, "low_confidence": confidence < 0.6}


between = predict_rows(loaded, [[6.0, 2.7, 4.9, 1.7]])[0]
print(between["probabilities"])
print(with_confidence(between)["low_confidence"], with_confidence(predict_rows(loaded, [[5.1, 3.5, 1.4, 0.2]])[0])["low_confidence"])

[{'label': 'setosa', 'probabilities': {'setosa': 0.981, 'versicolor': 0.019, 'virginica': 0.0}}]
{'setosa': 0.003, 'versicolor': 0.455, 'virginica': 0.542}
True False