FastAPI দিয়ে CV Model Serving

REST endpoint তৈরি।

🎬 গল্প দিয়ে শুরু

Notebook-এ model কাজ করে — কিন্তু client/app সেটা কীভাবে ব্যবহার করবে? উত্তর — REST API। FastAPI দিয়ে কয়েক লাইনে CV model-কে production endpoint বানানো।

কেন FastAPI?

Async — high concurrency (uvicorn/gunicorn)।
Pydantic — automatic request validation।
Auto Swagger UI — /docs-এ live test।
Type hint native — IDE-friendly।

মিনিমাল CV API

bash

pip install fastapi uvicorn[standard] python-multipart pillow ultralytics

python

main.py

from fastapi import FastAPI, UploadFile, File
from fastapi.responses import JSONResponse
from ultralytics import YOLO
from PIL import Image
import io

app   = FastAPI(title="CV Inference API")
model = YOLO("yolov8n.pt")   # startup-এ একবার load

@app.post("/detect")
async def detect(file: UploadFile = File(...)):
    img = Image.open(io.BytesIO(await file.read())).convert("RGB")
    res = model(img, verbose=False)[0]
    out = [{
        "cls":  model.names[int(b.cls)],
        "conf": float(b.conf),
        "xyxy": b.xyxy[0].tolist(),
    } for b in res.boxes]
    return JSONResponse({"objects": out})

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

bash

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 2
# Swagger:  http://localhost:8000/docs

Production-grade pattern

Model startup-এ load — request-এ নয় (warm start)।
Pydantic schema দিয়ে response type lock করুন।
Background task / queue (Celery, RQ) — heavy job-এ।
Rate limiting (slowapi) ও API key auth।
CORS middleware frontend-এর জন্য।
Logging — structlog/loguru, request_id সহ।

python

from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(CORSMiddleware,
    allow_origins=["https://your-frontend.com"],
    allow_methods=["*"], allow_headers=["*"])

Batching ও GPU efficiency

Micro-batching

একাধিক request আসলে কয়েক ms wait করে batch বানান — GPU throughput কয়েকগুণ বাড়ে। Ray Serve / Triton Inference Server এটি built-in দেয়।

Streaming response (large output)

python

from fastapi.responses import StreamingResponse
@app.get("/video")
def video():
    def gen():
        for frame in pipeline():           # bytes generator
            yield (b"--f\r\nContent-Type: image/jpeg\r\n\r\n" + frame + b"\r\n")
    return StreamingResponse(gen(),
        media_type="multipart/x-mixed-replace; boundary=f")

প্র্যাকটিস টাস্ক

YOLOv8 detection API বানান, /docs থেকে ছবি upload করে test করুন।
/classify endpoint যোগ করুন — Top-5 class + confidence return।
API key middleware বানান, invalid key-তে 401 দিন।