Computer Vision YOLO MLOps Workflow

YOLO MLOps

Computer Vision YOLO MLOps Object Detection YOLOv8 Ultralytics Data Labeling Training Evaluation Deployment Monitoring Real-time GPU Edge Production Pipeline

Training Pipeline

# === YOLOv8 Training Pipeline ===

# pip install ultralytics mlflow wandb

from ultralytics import YOLO
import os

# Data Preparation
# Dataset Structure:
# dataset/
# ├── data.yaml
# ├── train/
# │   ├── images/
# │   │   ├── img001.jpg
# │   │   └── img002.jpg
# │   └── labels/
# │       ├── img001.txt  # class x_center y_center width height
# │       └── img002.txt
# ├── val/
# │   ├── images/
# │   └── labels/
# └── test/
#     ├── images/
#     └── labels/

# data.yaml
# path: /data/dataset
# train: train/images
# val: val/images
# test: test/images
# names:
#   0: person
#   1: car
#   2: truck
#   3: bicycle
#   4: motorcycle

# Training
# model = YOLO("yolov8m.pt")  # Load pretrained
# results = model.train(
#     data="data.yaml",
#     epochs=100,
#     imgsz=640,
#     batch=16,
#     device="0",  # GPU 0
#     workers=8,
#     patience=20,  # Early stopping
#     save=True,
#     project="runs/detect",
#     name="experiment-1",
#     # Augmentation
#     augment=True,
#     hsv_h=0.015,
#     hsv_s=0.7,
#     hsv_v=0.4,
#     degrees=10,
#     translate=0.1,
#     scale=0.5,
#     fliplr=0.5,
#     mosaic=1.0,
#     mixup=0.1,
# )

# Evaluation
# metrics = model.val(data="data.yaml", split="test")
# print(f"mAP50: {metrics.box.map50:.4f}")
# print(f"mAP50-95: {metrics.box.map:.4f}")
# print(f"Precision: {metrics.box.p:.4f}")
# print(f"Recall: {metrics.box.r:.4f}")

# Export
# model.export(format="onnx", imgsz=640, simplify=True)
# model.export(format="engine", imgsz=640, device=0)  # TensorRT

from dataclasses import dataclass

@dataclass
class Experiment:
    name: str
    model: str
    epochs: int
    map50: float
    map50_95: float
    precision: float
    recall: float
    train_time: str

experiments = [
    Experiment("baseline", "yolov8n", 100, 0.823, 0.612, 0.845, 0.798, "2h"),
    Experiment("medium", "yolov8m", 100, 0.891, 0.702, 0.908, 0.865, "8h"),
    Experiment("augmented", "yolov8m", 150, 0.912, 0.735, 0.925, 0.889, "12h"),
    Experiment("large", "yolov8l", 100, 0.925, 0.758, 0.938, 0.901, "16h"),
    Experiment("fine-tuned", "yolov8l", 200, 0.941, 0.782, 0.952, 0.918, "24h"),
]

print("=== Training Experiments ===")
for e in experiments:
    print(f"  [{e.name}] Model: {e.model} | Epochs: {e.epochs}")
    print(f"    mAP50: {e.map50:.3f} | mAP50-95: {e.map50_95:.3f}")
    print(f"    Precision: {e.precision:.3f} | Recall: {e.recall:.3f}")
    print(f"    Training Time: {e.train_time}")

Deployment

# === YOLO Model Deployment ===

# FastAPI Inference Server
# from fastapi import FastAPI, UploadFile
# from ultralytics import YOLO
# import cv2
# import numpy as np
# import io
#
# app = FastAPI()
# model = YOLO("best.pt")
#
# @app.post("/detect")
# async def detect(file: UploadFile):
#     contents = await file.read()
#     nparr = np.frombuffer(contents, np.uint8)
#     img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
#
#     results = model(img, conf=0.5)
#     detections = []
#     for r in results[0].boxes:
#         detections.append({
#             "class": model.names[int(r.cls)],
#             "confidence": float(r.conf),
#             "bbox": r.xyxy[0].tolist(),
#         })
#     return {"detections": detections, "count": len(detections)}

# Dockerfile
# FROM nvidia/cuda:12.1-runtime-ubuntu22.04
# RUN pip install ultralytics fastapi uvicorn python-multipart
# COPY best.pt /app/best.pt
# COPY main.py /app/main.py
# WORKDIR /app
# CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]

# Kubernetes — GPU Deployment
# apiVersion: apps/v1
# kind: Deployment
# spec:
#   replicas: 2
#   template:
#     spec:
#       containers:
#         - name: yolo-api
#           image: yolo-api:v1.0
#           resources:
#             limits:
#               nvidia.com/gpu: 1
#           ports:
#             - containerPort: 8080

@dataclass
class DeployOption:
    method: str
    hardware: str
    latency: str
    throughput: str
    cost: str
    use_case: str

options = [
    DeployOption("FastAPI + GPU", "NVIDIA T4/A10", "15-30ms", "50-100 FPS", "$$", "Real-time API"),
    DeployOption("TensorRT", "NVIDIA GPU", "5-10ms", "100-300 FPS", "$$", "High Performance"),
    DeployOption("ONNX Runtime", "CPU/GPU", "30-100ms", "10-30 FPS", "$", "Portable"),
    DeployOption("OpenVINO", "Intel CPU/GPU", "20-50ms", "20-60 FPS", "$", "Intel Hardware"),
    DeployOption("Edge (Jetson)", "NVIDIA Jetson", "30-50ms", "20-30 FPS", "$", "Edge IoT"),
    DeployOption("Batch (Video)", "GPU Cluster", "N/A", "100+ videos/hr", "$$$", "Offline Analysis"),
]

print("\n=== Deployment Options ===")
for d in options:
    print(f"  [{d.method}] {d.hardware}")
    print(f"    Latency: {d.latency} | Throughput: {d.throughput}")
    print(f"    Cost: {d.cost} | Use: {d.use_case}")

MLOps Pipeline

# === End-to-End MLOps Pipeline ===

# CI/CD — GitHub Actions
# name: YOLO MLOps Pipeline
# on:
#   push:
#     paths: ['data/**', 'configs/**']
# jobs:
#   train:
#     runs-on: [self-hosted, gpu]
#     steps:
#       - uses: actions/checkout@v4
#       - run: pip install ultralytics mlflow dvc
#       - run: dvc pull  # Get latest dataset
#       - run: python train.py --config configs/production.yaml
#       - run: python evaluate.py --model runs/best.pt
#       - run: |
#           if [ $(python check_metrics.py) == "improved" ]; then
#             python register_model.py
#             docker build -t yolo-api:$SHA .
#             docker push registry/yolo-api:$SHA
#           fi

@dataclass
class PipelineStage:
    stage: str
    tools: str
    trigger: str
    duration: str
    output: str

stages = [
    PipelineStage("Data Collection", "Label Studio CVAT", "Manual/Schedule", "Ongoing", "Labeled Dataset"),
    PipelineStage("Data Versioning", "DVC S3", "On new data", "Minutes", "Dataset Version"),
    PipelineStage("Training", "Ultralytics GPU MLflow", "On data change", "2-24 hours", "Model Weights"),
    PipelineStage("Evaluation", "mAP Precision Recall", "After training", "Minutes", "Metrics Report"),
    PipelineStage("Model Registry", "MLflow W&B", "If improved", "Minutes", "Registered Model"),
    PipelineStage("Deployment", "Docker K8s Helm", "On approval", "Minutes", "Running Service"),
    PipelineStage("Monitoring", "Prometheus Grafana", "Continuous", "Real-time", "Dashboard Alerts"),
    PipelineStage("Retraining", "Auto trigger", "On drift/schedule", "Hours", "New Model"),
]

print("MLOps Pipeline Stages:")
for s in stages:
    print(f"  [{s.stage}]")
    print(f"    Tools: {s.tools} | Trigger: {s.trigger}")
    print(f"    Duration: {s.duration} | Output: {s.output}")

monitoring_metrics = {
    "Inference Latency (p99)": "28ms",
    "Throughput": "85 FPS",
    "GPU Utilization": "72%",
    "Daily Predictions": "2.5M",
    "Confidence > 0.8": "92%",
    "False Positive Rate": "3.2%",
    "Model Version": "v3.2 (fine-tuned)",
    "Data Drift Score": "0.05 (low)",
}

print(f"\n\nProduction Monitoring:")
for k, v in monitoring_metrics.items():
    print(f"  {k}: {v}")

เคล็ดลับ

• Data: Data Quality สำคัญกว่า Model Size Label ให้ดี
• Augmentation: ใช้ Augmentation เพิ่ม Data Diversity
• Export: Export เป็น ONNX/TensorRT สำหรับ Production
• Monitor: ดู Inference Latency และ Accuracy Drift ทุกวัน
• Retrain: Retrain เมื่อ Data Drift หรือ Accuracy ลดลง

YOLO คืออะไร

Real-time Object Detection YOLOv8 Ultralytics Detection Segmentation Classification 30-300+ FPS Autonomous Driving Security Camera Manufacturing QC

สรุป

Computer Vision YOLO MLOps YOLOv8 Object Detection Training Evaluation Deployment ONNX TensorRT Docker Kubernetes Monitoring Data Drift Production Pipeline