Bài 6: Endpoint inference — nhận data, trả kết quả prediction

Sau bài này bạn sẽ:

• Tổ chức project FastAPI inference theo cấu trúc tách biệt schema / model logic / dependency / router.
• Viết POST /predict hoàn chỉnh cho sklearn classifier: nhận JSON, trả class name + probabilities.
• Viết POST /classify hoàn chỉnh cho PyTorch image model: nhận file upload, trả top-5 predictions.
• Tránh được 5 lỗi serialization và inference phổ biến nhất.

Bài này không lặp lại lý thuyết lifespan (bài 5), Pydantic validation (bài 3), hay quyết định async/sync (bài 4) — chỉ tham chiếu ngắn khi cần nhắc nhở.

app/
  main.py        # FastAPI app + lifespan + router
  schemas.py     # Pydantic request/response models
  models.py      # load model + inference functions
  deps.py        # Depends() helpers lấy từ app.state
requirements.txt

Quy tắc phân chia:

• schemas.py không import gì từ FastAPI — thuần Pydantic, dễ test độc lập.
• models.py không biết FastAPI tồn tại — chỉ là Python functions nhận/trả Python types.
• deps.py là cầu nối: lấy object từ app.state và cung cấp cho handler qua Depends().
• main.py chịu trách nhiệm lifecycle (lifespan) và định nghĩa router.

Cấu trúc này đủ cho service có 1–10 endpoint. Nếu lớn hơn, tách router vào thư mục routers/ riêng — nhưng ở đây không cần.

fastapi>=0.110.0
uvicorn[standard]>=0.29.0
pydantic>=2.6.0
scikit-learn>=1.4.0
joblib>=1.3.0
numpy>=1.26.0
torch>=2.2.0
torchvision>=0.17.0
Pillow>=10.2.0
python-multipart>=0.0.9

python-multipart bắt buộc để FastAPI xử lý UploadFile (form/multipart). Thiếu thư viện này, endpoint nhận file sẽ raise ImportError khi server start.

Cài đặt:

pip install -r requirements.txt

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


# --- Iris (sklearn) ---

class IrisRequest(BaseModel):
    """Request body cho POST /predict."""
    features: Annotated[
        list[float],
        Field(
            min_length=4,
            max_length=4,
            description="4 features: sepal_length, sepal_width, petal_length, petal_width",
            examples=[[5.1, 3.5, 1.4, 0.2]],
        )
    ]

class IrisResponse(BaseModel):
    """Response cho POST /predict."""
    class_id: int
    class_name: str
    probas: list[float]   # xác suất cho từng class, sum = 1.0


# --- Image classification (PyTorch) ---

class ClassPred(BaseModel):
    """1 mục trong top-k."""
    rank: int           # 1-based
    class_id: int
    label: str
    score: float        # softmax score, đã round 4 chữ số

class ClassifyResponse(BaseModel):
    """Response cho POST /classify."""
    top5: list[ClassPred]

Điểm cần chú ý:

• IrisRequest.features dùng min_length=4, max_length=4 thay vì validator thủ công — Pydantic v2 validate list length trực tiếp trên type (bài 3 đã giải thích chi tiết). Client gửi thiếu/thừa features sẽ nhận 422 ngay.
• ClassifyResponse không có request schema cho image — phần upload dùng UploadFile của FastAPI (multipart/form-data), không phải JSON body. Bài 7 đào sâu phần này.
• Tất cả field trả về là Python native type (int, str, float, list) — không có np.int64 hay torch.Tensor. Lý do ở phần pitfalls.

File này có 2 phần: load model và hàm inference. Hai phần tách nhau để dễ test từng hàm độc lập.

Load function

# app/models.py
import io
import numpy as np
import joblib
import torch
import torch.nn.functional as F
from torchvision import models, transforms
from torchvision.models import ResNet18_Weights
from sklearn.linear_model import LogisticRegression
from PIL import Image


DEVICE = "cuda" if torch.cuda.is_available() else "cpu"

# Tên class của Iris dataset (thứ tự tương ứng với class_id 0,1,2)
IRIS_CLASS_NAMES = ["setosa", "versicolor", "virginica"]


def load_iris_model() -> LogisticRegression:
    """
    Load sklearn LogisticRegression từ file.
    Dùng joblib — đây là format joblib.dump() chuẩn của sklearn.
    Nếu chưa có file, train nhanh và dump (xem phần train bên dưới).
    """
    clf: LogisticRegression = joblib.load("iris_clf.pkl")
    return clf


def load_resnet() -> tuple[torch.nn.Module, transforms.Compose]:
    """
    Load ResNet18 pretrained (ImageNet-1k).
    Dùng weights enum thay vì pretrained=True (deprecated từ torchvision 0.13).
    Trả về (model, transform) để tái dùng transform ở mọi request.
    """
    weights = ResNet18_Weights.IMAGENET1K_V1
    model = models.resnet18(weights=weights)
    model.to(DEVICE)
    model.eval()   # tắt Dropout / BatchNorm training mode
    transform = weights.transforms()   # transform đi kèm với weights này
    return model, transform

Để train và lưu file Iris model (chạy 1 lần trước khi start server):

# train_iris.py — chạy 1 lần để tạo iris_clf.pkl
from sklearn.datasets import load_iris
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
import joblib

X, y = load_iris(return_X_y=True)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

clf = LogisticRegression(max_iter=200)
clf.fit(X_train, y_train)
print(f"Test accuracy: {clf.score(X_test, y_test):.4f}")

joblib.dump(clf, "iris_clf.pkl")
print("Saved iris_clf.pkl")

Inference function

# (tiếp theo app/models.py)

def predict_iris(clf: LogisticRegression, features: list[float]) -> dict:
    """
    Chạy inference sklearn.
    Trả dict với Python native types (int, str, list[float]) — không trả np.ndarray.
    """
    # sklearn yêu cầu input 2D: shape (n_samples, n_features)
    x = np.array([features])           # shape (1, 4)
    class_id = int(clf.predict(x)[0])  # np.int64 → int để JSON serializable
    probas = clf.predict_proba(x)[0]   # shape (3,) ndarray

    return {
        "class_id": class_id,
        "class_name": IRIS_CLASS_NAMES[class_id],
        # .tolist() chuyển ndarray → list[float]
        "probas": [round(p, 4) for p in probas.tolist()],
    }


def classify_image(
    model: torch.nn.Module,
    img_bytes: bytes,
    transform: transforms.Compose,
    top_k: int = 5,
) -> dict:
    """
    Chạy inference PyTorch ResNet18 trên ảnh.
    img_bytes: raw bytes của ảnh (JPEG, PNG, ...)
    Trả dict với top_k predictions.
    """
    # Đọc ảnh từ bytes, convert sang RGB (loại bỏ alpha channel nếu PNG có)
    img = Image.open(io.BytesIO(img_bytes)).convert("RGB")
    # Áp dụng transform của ResNet18 (resize 256, center crop 224, normalize)
    tensor = transform(img).unsqueeze(0).to(DEVICE)  # shape (1, 3, 224, 224)

    with torch.no_grad():
        logits = model(tensor)                         # shape (1, 1000)
        probs = F.softmax(logits, dim=1)[0]            # shape (1000,)

    # Lấy top-k
    scores, indices = torch.topk(probs, k=top_k)

    # Labels của ImageNet-1k từ weights metadata
    labels = ResNet18_Weights.IMAGENET1K_V1.meta["categories"]

    top5 = []
    for rank, (score, idx) in enumerate(zip(scores.tolist(), indices.tolist()), start=1):
        top5.append({
            "rank": rank,
            "class_id": idx,              # int
            "label": labels[idx],
            "score": round(score, 4),     # float, 4 chữ số
        })

    return {"top5": top5}

Các quyết định kỹ thuật cần giải thích:

• np.array([features]): sklearn predict() và predict_proba() yêu cầu input 2D. Nếu truyền np.array(features) (1D shape (4,)), sklearn vẫn chạy nhưng xuất hiện cảnh báo DataConversionWarning và behavior có thể không như mong đợi khi dùng pipeline.
• int(clf.predict(x)[0]): predict() trả np.ndarray với element type np.int64. FastAPI dùng json.dumps() để serialize response — Python json module không biết np.int64, raise TypeError. Cast sang int thuần trước khi trả.
• model.eval(): gọi trong load_resnet() lúc load — một lần duy nhất. Không cần gọi lại mỗi request.
• torch.no_grad(): bọc toàn bộ forward pass để PyTorch không tính gradient và không giữ computation graph. Tiết kiệm ~30–50% RAM so với chạy không có context manager này.
• weights.transforms(): ResNet18_Weights.IMAGENET1K_V1.transforms() trả về transform chuẩn đi kèm với weights. Dùng cách này thay vì tự viết transform để đảm bảo mean/std normalization đúng với weights đã train.
• scores.tolist() và indices.tolist(): tensor → list[float]/list[int] native Python trước khi đưa vào dict.

# app/deps.py
from fastapi import Request
from sklearn.linear_model import LogisticRegression
from torchvision import transforms
import torch.nn as nn


def get_iris_model(request: Request) -> LogisticRegression:
    """Lấy sklearn classifier từ app.state."""
    return request.app.state.iris_model


def get_resnet(request: Request) -> nn.Module:
    """Lấy PyTorch ResNet18 từ app.state."""
    return request.app.state.resnet


def get_transform(request: Request) -> transforms.Compose:
    """Lấy torchvision transform từ app.state."""
    return request.app.state.resnet_transform

Ba hàm này là dependency function (xem bài 5 — Dependency injection). Lợi ích chính: trong test, gọi app.dependency_overrides[get_iris_model] = lambda: mock_clf để thay thế model thật bằng mock mà không cần sửa handler.

Lưu ý: các hàm này là def thường, không phải async def — chúng chỉ đọc attribute từ app.state, không có I/O.

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends, UploadFile, File, HTTPException
from sklearn.linear_model import LogisticRegression
import torch.nn as nn
from torchvision import transforms

from .schemas import IrisRequest, IrisResponse, ClassifyResponse
from .models import load_iris_model, load_resnet, predict_iris, classify_image
from .deps import get_iris_model, get_resnet, get_transform


@asynccontextmanager
async def lifespan(app: FastAPI):
    """
    Load cả 2 model khi server start.
    Code sau yield chạy khi server shutdown.
    Chi tiết lifespan pattern: bài 5.
    """
    # Sklearn model — đọc file .pkl từ disk, ~ms
    app.state.iris_model = load_iris_model()

    # PyTorch ResNet18 — download weights lần đầu (~45 MB), cache sau
    model, transform = load_resnet()
    app.state.resnet = model
    app.state.resnet_transform = transform

    print("Models loaded. Server ready.")
    yield

    # Giải phóng reference — quan trọng khi dùng --reload
    del app.state.iris_model
    del app.state.resnet
    del app.state.resnet_transform


app = FastAPI(
    title="Inference API",
    description="POST /predict — Iris classifier | POST /classify — ImageNet ResNet18",
    version="1.0.0",
    lifespan=lifespan,
)


# --- Endpoint 1: Iris prediction ---

@app.post(
    "/predict",
    response_model=IrisResponse,
    summary="Iris species classifier (LogisticRegression)",
)
def predict(
    body: IrisRequest,
    clf: LogisticRegression = Depends(get_iris_model),
) -> IrisResponse:
    """
    Nhận 4 float features của hoa Iris, trả class và probability.
    Dùng def (sync) — sklearn predict là CPU-bound, FastAPI tự offload thread pool.
    """
    result = predict_iris(clf, body.features)
    return IrisResponse(**result)


# --- Endpoint 2: Image classification ---

@app.post(
    "/classify",
    response_model=ClassifyResponse,
    summary="ImageNet top-5 classifier (ResNet18)",
)
def classify(
    file: UploadFile = File(..., description="File ảnh JPEG hoặc PNG"),
    model: nn.Module = Depends(get_resnet),
    transform: transforms.Compose = Depends(get_transform),
) -> ClassifyResponse:
    """
    Nhận file ảnh qua multipart/form-data.
    Trả top-5 ImageNet predictions với score softmax.
    Upload file chi tiết hơn: bài 7.
    """
    # Kiểm tra content type cơ bản
    if file.content_type not in ("image/jpeg", "image/png", "image/webp"):
        raise HTTPException(
            status_code=415,
            detail=f"Unsupported media type: {file.content_type}. Chỉ nhận JPEG, PNG, WebP.",
        )

    img_bytes = file.file.read()
    result = classify_image(model, img_bytes, transform, top_k=5)
    return ClassifyResponse(**result)

Lý do chọn def cho cả 2 endpoint:

• Cả predict_iris() lẫn classify_image() đều là CPU-bound — không có I/O async. Theo quy tắc bài 4: dùng def, FastAPI offload vào thread pool, event loop không bị chiếm.
• Nếu sau này cần gọi thêm external API (ví dụ log kết quả lên remote), chuyển sang async def + asyncio.to_thread() cho phần inference.

Khởi động server:

# Từ thư mục gốc (chứa app/)
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

Output khi start thành công:

INFO:     Started server process [12345]
INFO:     Waiting for application startup.
Models loaded. Server ready.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

POST /predict — Iris

curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"features": [5.1, 3.5, 1.4, 0.2]}'

Response mẫu:

{
  "class_id": 0,
  "class_name": "setosa",
  "probas": [0.9823, 0.0147, 0.003]
}

Test case với input lỗi (3 features thay vì 4):

curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"features": [5.1, 3.5, 1.4]}'

Response 422 từ Pydantic:

{
  "detail": [
    {
      "type": "too_short",
      "loc": ["body", "features"],
      "msg": "List should have at least 4 items after validation, not 3",
      "input": [5.1, 3.5, 1.4]
    }
  ]
}

POST /classify — Image

curl -X POST http://localhost:8000/classify \
  -F "[email protected]"

Response mẫu:

{
  "top5": [
    {"rank": 1, "class_id": 281, "label": "tabby cat", "score": 0.4521},
    {"rank": 2, "class_id": 285, "label": "Egyptian cat", "score": 0.1834},
    {"rank": 3, "class_id": 282, "label": "tiger cat", "score": 0.1102},
    {"rank": 4, "class_id": 287, "label": "lynx", "score": 0.0348},
    {"rank": 5, "class_id": 292, "label": "lion", "score": 0.0211}
  ]
}

Swagger UI

FastAPI tự sinh /docs (Swagger) và /redoc. Truy cập http://localhost:8000/docs để test cả 2 endpoint trực tiếp trên browser — upload file qua form UI của Swagger cũng hoạt động.

httpx trong Python test

# test_endpoints.py — chạy khi server đang chạy
import httpx

BASE = "http://localhost:8000"

# Test Iris
r = httpx.post(f"{BASE}/predict", json={"features": [5.1, 3.5, 1.4, 0.2]})
print(r.status_code, r.json())

# Test Image
with open("cat.jpg", "rb") as f:
    r = httpx.post(f"{BASE}/classify", files={"file": ("cat.jpg", f, "image/jpeg")})
print(r.status_code, r.json())

1. Trả np.ndarray hoặc np.int64 thẳng → TypeError

# SAI — FastAPI serialize dict này sẽ raise TypeError
return {
    "class_id": clf.predict(x)[0],   # np.int64
    "probas": clf.predict_proba(x)[0]  # np.ndarray
}

# ĐÚNG — cast sang native type
return {
    "class_id": int(clf.predict(x)[0]),
    "probas": clf.predict_proba(x)[0].tolist()
}

Error message: TypeError: Object of type int64 is not JSON serializable hoặc Object of type ndarray is not JSON serializable. Pydantic v2 có thể tự handle một số case nếu dùng response_model, nhưng không nên dựa vào đó — luôn convert tường minh.

2. Quên model.eval() → kết quả inference không ổn định

# SAI — model đang ở training mode
model = models.resnet18(weights=weights)
# Quên gọi model.eval()
# BatchNorm dùng batch statistics thay vì running mean/var
# Dropout ngẫu nhiên drop neuron → mỗi lần forward cho kết quả khác nhau

# ĐÚNG
model = models.resnet18(weights=weights)
model.eval()   # tắt Dropout và BatchNorm training mode

Với ResNet18 cụ thể, không có Dropout layer, nhưng có BatchNorm. Quên eval() khiến BatchNorm dùng batch statistics của 1 sample (batch size 1 trong inference) thay vì running mean/var tích lũy trong quá trình pretrain — kết quả prediction bị lệch.

3. Quên torch.no_grad() → tốn RAM không cần thiết

# SAI — PyTorch vẫn tính gradient và giữ computation graph
logits = model(tensor)

# ĐÚNG — disable gradient computation
with torch.no_grad():
    logits = model(tensor)

Với ResNet18 forward pass trên input (1, 3, 224, 224): không có no_grad() tốn thêm ~60–80 MB RAM cho intermediate activations và gradient buffers. Khi có 10 concurrent request, con số này tích lũy thành vài trăm MB.

4. Load model trong handler → latency cao mỗi request

# SAI — load model mỗi request, ResNet18 ~200ms mỗi lần
@app.post("/classify")
def classify(file: UploadFile = File(...)):
    model, transform = load_resnet()   # BUG: load lại mỗi request
    ...

# ĐÚNG — load 1 lần trong lifespan, inject qua Depends()
@app.post("/classify")
def classify(
    file: UploadFile = File(...),
    model = Depends(get_resnet),
    transform = Depends(get_transform),
):
    ...

5. Trả float với độ chính xác quá cao → JSON lớn, RTT tăng

# Không round — probas chứa 15-16 chữ số thập phân
"probas": [0.9823456789012345, 0.014567890123456, 0.003086420761309]

# Round 4 chữ số — đủ cho UI demo và log
"probas": [0.9823, 0.0146, 0.0031]

Với top-5 ImageNet, mỗi score là float 64-bit. Không round, JSON response tăng thêm ~100–150 byte/request. Ở throughput cao (10k req/phút), đây là ~1–1.5 MB/phút bandwidth dư thừa.

6. Dùng PIL.Image.open() trực tiếp trên UploadFile

# SAI — UploadFile.file là SpooledTemporaryFile, không phải path
img = Image.open(file)  # có thể raise error tùy Python version

# ĐÚNG — đọc bytes trước, sau đó mở từ BytesIO
img_bytes = file.file.read()
img = Image.open(io.BytesIO(img_bytes))

Cách đọc qua bytes cũng cho phép log kích thước file, hash để detect duplicate, hoặc lưu xuống disk nếu cần — linh hoạt hơn là pass file object trực tiếp.

Các điểm cần nhớ khi xây endpoint inference:

• Tách project theo: schemas.py (Pydantic) / models.py (logic inference) / deps.py (Depends) / main.py (router + lifespan).
• Load cả model lẫn transform trong lifespan, lưu vào app.state, inject qua Depends().
• sklearn: input phải là 2D array. Cast np.int64 → int, gọi .tolist() trước khi trả.
• PyTorch: gọi model.eval() khi load. Bọc forward pass trong torch.no_grad(). Cast tensor sang Python native type trước khi trả.
• Dùng def endpoint cho CPU-bound inference — FastAPI tự offload thread pool.
• Round float về 4 chữ số cho response demo.

Bảng liên kết bài trong Module 1:

Bài	Khái niệm	Dùng ở đâu trong bài 6
Bài 3	Pydantic BaseModel, Field constraints	schemas.py — IrisRequest, IrisResponse, ClassifyResponse
Bài 4	def vs async def cho CPU-bound	Cả 2 endpoint dùng def
Bài 5	lifespan, app.state, Depends()	main.py lifespan + deps.py

Bài 7: Upload file (image, PDF) và xử lý — Đào sâu UploadFile: đọc streaming, giới hạn kích thước, validate content type, xử lý nhiều file cùng lúc, lưu xuống disk hoặc object storage.