#สร้าง MLOps Pipeline และ Machine Learning API ด้วย FastAPI

MLOps (Machine Learning Operations) คือแนวทางนำหลักการ DevOps มาประยุกต์กับวงจรชีวิตของ Machine Learning เพื่อให้การพัฒนา ทดสอบ จัดเก็บเวอร์ชัน Deploy และติดตามโมเดลทำได้อย่างเป็นระบบ

บทความนี้สร้าง Workshop ขนาดเล็กที่ครบเส้นทาง:

Dataset
   ↓
Train Model
   ↓
Evaluate
   ↓
MLflow Tracking
   ↓
Save / Register Model
   ↓
FastAPI
   ↓
API Test
   ↓
Docker
   ↓
GitHub Actions CI
   ↓
Deploy / Monitor

#1. Technology Stack

  • Python 3.12+
  • uv สำหรับจัดการ Python project/dependencies
  • scikit-learn สำหรับ Machine Learning
  • MLflow สำหรับ Experiment Tracking
  • FastAPI สำหรับ Model Serving
  • Pydantic สำหรับ Data Validation
  • pytest สำหรับ Automated Test
  • Docker สำหรับ Container
  • GitHub Actions สำหรับ CI/CD

#2. สร้างโปรเจกต์

ติดตั้ง uv แล้วสร้างโปรเจกต์:

#macOS / Linux

curl -LsSf https://astral.sh/uv/install.sh | sh
mkdir mlops-fastapi
cd mlops-fastapi
uv init

#Windows PowerShell

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
mkdir mlops-fastapi
cd mlops-fastapi
uv init

ติดตั้ง dependencies:

uv add fastapi "uvicorn[standard]" scikit-learn pandas joblib mlflow
uv add --dev pytest httpx

เมื่อใช้ uv ไม่จำเป็นต้องสร้าง virtual environment ด้วย python -m venv .venv เอง เพราะ uv สามารถจัดการ environment ให้โปรเจกต์ได้

#3. โครงสร้างโปรเจกต์

mlops-fastapi/
├── app/
│   ├── __init__.py
│   └── main.py
├── src/
│   ├── __init__.py
│   └── train.py
├── models/
│   └── model.joblib
├── tests/
│   └── test_api.py
├── .github/
│   └── workflows/
│       └── ci.yml
├── Dockerfile
├── .dockerignore
└── pyproject.toml

สร้าง directory:

mkdir -p app src models tests
touch app/__init__.py src/__init__.py

บน Windows สามารถสร้างโฟลเดอร์/ไฟล์ด้วย File Explorer หรือ PowerShell ได้

#4. Train Machine Learning Model

ตัวอย่างใช้ Iris Dataset และ Random Forest เพื่อให้ Workshop รันได้ทันทีโดยไม่ต้องดาวน์โหลด dataset เพิ่ม

สร้าง src/train.py

from pathlib import Path

import joblib
import mlflow
import mlflow.sklearn

from sklearn.datasets import load_iris
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score
from sklearn.model_selection import train_test_split


MODEL_DIR = Path("models")
MODEL_PATH = MODEL_DIR / "model.joblib"


def train():
    iris = load_iris()

    X_train, X_test, y_train, y_test = train_test_split(
        iris.data,
        iris.target,
        test_size=0.2,
        random_state=42,
        stratify=iris.target,
    )

    params = {
        "n_estimators": 100,
        "max_depth": 5,
        "random_state": 42,
    }

    mlflow.set_experiment("iris-classification")

    with mlflow.start_run():
        model = RandomForestClassifier(**params)
        model.fit(X_train, y_train)

        prediction = model.predict(X_test)
        accuracy = accuracy_score(y_test, prediction)

        mlflow.log_params(params)
        mlflow.log_metric("accuracy", accuracy)

        mlflow.sklearn.log_model(
            sk_model=model,
            name="iris-model",
        )

        MODEL_DIR.mkdir(exist_ok=True)
        joblib.dump(model, MODEL_PATH)

        print(f"Accuracy: {accuracy:.4f}")
        print(f"Saved: {MODEL_PATH}")


if __name__ == "__main__":
    train()

Train:

uv run python src/train.py

ผลลัพธ์จะได้ไฟล์:

models/model.joblib

#5. เปิด MLflow UI

หลังจาก Train แล้ว สามารถดู experiment ได้ด้วย:

uv run mlflow server --host 0.0.0.0 --port 5000

เปิด:

http://localhost:5000

MLflow ช่วยดู Parameters, Metrics, Runs และ Model artifacts ทำให้สามารถเปรียบเทียบการทดลองหลายรอบได้

#6. สร้าง Machine Learning API ด้วย FastAPI

สร้าง app/main.py

from contextlib import asynccontextmanager
from pathlib import Path

import joblib
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field


MODEL_PATH = Path("models/model.joblib")
model = None

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


class IrisRequest(BaseModel):
    sepal_length: float = Field(gt=0)
    sepal_width: float = Field(gt=0)
    petal_length: float = Field(gt=0)
    petal_width: float = Field(gt=0)


class PredictionResponse(BaseModel):
    prediction: int
    class_name: str


@asynccontextmanager
async def lifespan(app: FastAPI):
    global model

    if not MODEL_PATH.exists():
        raise RuntimeError(
            "Model not found. Run training before starting API."
        )

    model = joblib.load(MODEL_PATH)
    yield
    model = None


app = FastAPI(
    title="Iris ML API",
    version="1.0.0",
    lifespan=lifespan,
)


@app.get("/")
def root():
    return {
        "message": "MLOps FastAPI is running"
    }


@app.get("/health")
def health():
    return {
        "status": "healthy",
        "model_loaded": model is not None,
    }


@app.post(
    "/predict",
    response_model=PredictionResponse,
)
def predict(data: IrisRequest):
    if model is None:
        raise HTTPException(
            status_code=503,
            detail="Model is not available",
        )

    features = [[
        data.sepal_length,
        data.sepal_width,
        data.petal_length,
        data.petal_width,
    ]]

    prediction = int(model.predict(features)[0])

    return {
        "prediction": prediction,
        "class_name": CLASS_NAMES[prediction],
    }

#7. รัน FastAPI

ก่อนรัน API ต้องมี model:

uv run python src/train.py

จากนั้น:

uv run uvicorn app.main:app --reload

เปิด API:

http://localhost:8000

Swagger UI:

http://localhost:8000/docs

ReDoc:

http://localhost:8000/redoc

#8. ทดลอง Prediction

ส่ง POST /predict

{
  "sepal_length": 5.1,
  "sepal_width": 3.5,
  "petal_length": 1.4,
  "petal_width": 0.2
}

ตัวอย่างผลลัพธ์:

{
  "prediction": 0,
  "class_name": "setosa"
}

ทดลองด้วย curl:

curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -d '{
    "sepal_length": 5.1,
    "sepal_width": 3.5,
    "petal_length": 1.4,
    "petal_width": 0.2
  }'

#9. Automated API Testing

สร้าง tests/test_api.py

from fastapi.testclient import TestClient

from app.main import app


def test_predict():
    with TestClient(app) as client:
        response = client.post(
            "/predict",
            json={
                "sepal_length": 5.1,
                "sepal_width": 3.5,
                "petal_length": 1.4,
                "petal_width": 0.2,
            },
        )

    assert response.status_code == 200

    data = response.json()

    assert "prediction" in data
    assert "class_name" in data

Train ก่อน test:

uv run python src/train.py
uv run pytest -v

นี่เป็นจุดสำคัญของ MLOps เพราะโมเดลไม่ควรถูก Deploy เพียงเพราะ Train สำเร็จ แต่ควรผ่าน automated validation/test ก่อน

#10. Dockerize FastAPI

สร้าง Dockerfile

FROM python:3.12-slim

WORKDIR /app

COPY . .

RUN pip install --no-cache-dir uv \
    && uv sync --frozen --no-dev

RUN uv run python src/train.py

EXPOSE 8000

CMD [
  "uv",
  "run",
  "uvicorn",
  "app.main:app",
  "--host",
  "0.0.0.0",
  "--port",
  "8000"
]

สร้าง .dockerignore

.git
.github
.venv
__pycache__
.pytest_cache
mlruns
*.pyc

Build:

docker build -t mlops-fastapi .

Run:

docker run --rm -p 8000:8000 mlops-fastapi

ตรวจสอบ:

http://localhost:8000/docs

#11. CI ด้วย GitHub Actions

สร้าง .github/workflows/ci.yml

name: MLOps CI

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest

    steps:
      - name: Checkout
        uses: actions/checkout@v4

      - name: Install uv
        uses: astral-sh/setup-uv@v6
        with:
          enable-cache: true

      - name: Install Python
        run: uv python install 3.12

      - name: Install dependencies
        run: uv sync --locked

      - name: Train model
        run: uv run python src/train.py

      - name: Run tests
        run: uv run pytest -v

      - name: Build Docker image
        run: docker build -t mlops-fastapi:${{ github.sha }} .

เมื่อ Push หรือเปิด Pull Request ระบบจะ:

Checkout
   ↓
Setup Python / uv
   ↓
Install Dependencies
   ↓
Train
   ↓
Test
   ↓
Docker Build

หากขั้นตอนใดล้มเหลว pipeline จะหยุดก่อนนำ artifact ที่มีปัญหาไป Deploy

#12. MLOps กับ CI/CD ต่างจาก Web Application อย่างไร

Web CI/CD โดยทั่วไป:

Code
 ↓
Test
 ↓
Build
 ↓
Deploy

MLOps มีทั้ง Code และ Model/Data:

Code + Data
     ↓
Data Validation
     ↓
Training
     ↓
Model Evaluation
     ↓
Experiment Tracking
     ↓
Model Registry
     ↓
API Test
     ↓
Container
     ↓
Deploy
     ↓
Monitoring

ดังนั้น MLOps ต้องสนใจทั้ง version ของ source code, dataset, parameters, model และ metrics

#13. Production Architecture

ระบบจริงสามารถขยายเป็น:

                 ┌──────────────┐
                 │    GitHub    │
                 └──────┬───────┘
                        │
                        ▼
              ┌─────────────────┐
              │ GitHub Actions  │
              └────────┬────────┘
                       │
          Train → Test → Build
                       │
                       ▼
               ┌───────────────┐
               │ Docker Image  │
               └───────┬───────┘
                       │
                       ▼
              ┌────────────────┐
              │   Deployment   │
              └────────┬───────┘
                       │
                       ▼
Client ───────────► FastAPI
                       │
                       ▼
                 ML Model
                       │
                       ▼
              Prediction Result

MLflow สามารถแยกออกเป็น Tracking Server และเชื่อม artifact/object storage กับ backend database สำหรับระบบ production

#14. Monitoring ที่ควรมี

หลัง Deploy ไม่ควรตรวจสอบแค่ว่า API ยังทำงานหรือไม่ ควรติดตามอย่างน้อย:

  • API latency
  • Request/error rate
  • CPU/RAM
  • Prediction distribution
  • Input feature distribution
  • Data drift
  • Model drift/performance
  • Model version
  • Deployment version

ตัวอย่าง production stack ที่ต่อยอดได้:

FastAPI
   │
   ├── Prometheus
   │       ↓
   │    Grafana
   │
   └── Logs / Traces

#15. Model Promotion

MLOps ที่สมบูรณ์ไม่ควร Deploy โมเดลใหม่ทันทีทุกครั้งที่ Train แต่กำหนด Model Quality Gate เช่น:

New Model
   ↓
accuracy >= threshold ?
   │
   ├── No  → Reject
   │
   └── Yes
         ↓
Compare Champion Model
         ↓
Register / Promote
         ↓
Staging
         ↓
Production

นอกจาก Accuracy อาจใช้ Precision, Recall, F1-score, RMSE, MAE หรือ business metric ตามประเภทปัญหา

#16. แนวทางพัฒนาต่อ

จาก Workshop นี้สามารถขยายเป็น production-grade MLOps โดยเพิ่ม:

  1. Data Versioning ด้วย DVC หรือ object storage
  2. MLflow Tracking Server และ Model Registry
  3. PostgreSQL สำหรับ MLflow backend
  4. S3/MinIO สำหรับ artifacts
  5. Docker Registry
  6. CD ไปยัง VM หรือ Kubernetes
  7. Prometheus + Grafana
  8. Data/Model Drift Monitoring
  9. Scheduled Retraining
  10. Champion/Challenger model deployment

#สรุป

FastAPI เป็นส่วนของ Model Serving ส่วน MLOps ครอบคลุมวงจรที่กว้างกว่า ตั้งแต่ข้อมูล การ Train การประเมิน Experiment Tracking, Model Registry, Testing, Deployment และ Monitoring

Workshop นี้จึงเป็นจุดเริ่มต้นของ pipeline:

Develop
  ↓
Train
  ↓
Track
  ↓
Evaluate
  ↓
Test
  ↓
Package
  ↓
Deploy
  ↓
Monitor
  ↓
Retrain

เมื่อระบบเติบโต สามารถเปลี่ยนจากการเก็บ model.joblib ใน local project ไปเป็น MLflow Model Registry และให้ FastAPI โหลดโมเดลที่ได้รับการ Promote เป็น production โดยตรง ซึ่งทำให้การจัดการเวอร์ชันและ rollback มีความเป็นระบบมากขึ้น