Panduan Deployment CoreML: Dari Pelatihan ke Produksi di Mac Mini M4

Mengapa CoreML di Mac Mini M4?

CoreML adalah framework machine learning native dari Apple, dirancang khusus untuk mengekstrak performa maksimal dari Apple Silicon. Berbeda dari framework ML generik yang memperlakukan GPU sebagai satu perangkat komputasi tunggal, CoreML secara cerdas mendistribusikan beban kerja ke CPU, GPU, dan Neural Engine 16-core khusus, sering menjalankan bagian model yang berbeda pada unit komputasi yang berbeda secara bersamaan.

Konversi Model PyTorch ke CoreML

Library coremltools dari Apple menyediakan jalur konversi langsung dari model PyTorch ke format .mlpackage CoreML. Konversi ini menelusuri model Anda dengan input sampel dan menerjemahkan setiap operasi ke representasi internal CoreML.

Langkah 1: Instal Dependensi

# Create a virtual environment
python3 -m venv ~/coreml-env
source ~/coreml-env/bin/activate

# Install coremltools and PyTorch
pip install coremltools torch torchvision

# Verify installation
python3 -c "import coremltools as ct; print(ct.__version__)"
# 9.0

Langkah 2: Konversi Pengklasifikasi Gambar PyTorch

import torch
import torchvision
import coremltools as ct

# Load a pretrained ResNet50 model
model = torchvision.models.resnet50(weights=torchvision.models.ResNet50_Weights.DEFAULT)
model.eval()

# Create a sample input (batch=1, channels=3, height=224, width=224)
example_input = torch.randn(1, 3, 224, 224)

# Trace the model with TorchScript
traced_model = torch.jit.trace(model, example_input)

# Convert to CoreML
coreml_model = ct.convert(
    traced_model,
    inputs=[ct.ImageType(
        name="image",
        shape=(1, 3, 224, 224),
        scale=1.0 / (255.0 * 0.226),
        bias=[-0.485 / 0.226, -0.456 / 0.226, -0.406 / 0.226],
        color_layout=ct.colorlayout.RGB
    )],
    classifier_config=ct.ClassifierConfig("imagenet_classes.txt"),
    compute_units=ct.ComputeUnit.ALL,  # Use Neural Engine + GPU + CPU
    minimum_deployment_target=ct.target.macOS15,
)

# Save the model
coreml_model.save("ResNet50.mlpackage")
print("Model saved: ResNet50.mlpackage")

Langkah 3: Konversi Model PyTorch Kustom

import torch
import torch.nn as nn
import coremltools as ct

# Define a custom text embedding model
class TextEncoder(nn.Module):
    def __init__(self, vocab_size=30000, embed_dim=512, num_heads=8, num_layers=6):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.pos_encoding = nn.Parameter(torch.randn(1, 512, embed_dim))
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=embed_dim, nhead=num_heads, batch_first=True
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)
        self.output_proj = nn.Linear(embed_dim, 256)

    def forward(self, input_ids):
        x = self.embedding(input_ids) + self.pos_encoding[:, :input_ids.shape[1], :]
        x = self.transformer(x)
        x = x.mean(dim=1)  # Global average pooling
        return self.output_proj(x)

# Initialize and load your trained weights
model = TextEncoder()
# model.load_state_dict(torch.load("text_encoder_weights.pt"))
model.eval()

# Trace with example input
example_input = torch.randint(0, 30000, (1, 128))
traced_model = torch.jit.trace(model, example_input)

# Convert to CoreML
coreml_model = ct.convert(
    traced_model,
    inputs=[ct.TensorType(name="input_ids", shape=(1, 128), dtype=int)],
    outputs=[ct.TensorType(name="embedding")],
    compute_units=ct.ComputeUnit.ALL,
    minimum_deployment_target=ct.target.macOS15,
)

# Add metadata
coreml_model.author = "My Remote Mac"
coreml_model.short_description = "Text embedding model for semantic search"
coreml_model.version = "1.0.0"

coreml_model.save("TextEncoder.mlpackage")
print("Model saved: TextEncoder.mlpackage")

Langkah 4: Verifikasi Model yang Dikonversi

import coremltools as ct
import numpy as np

# Load the converted model
model = ct.models.MLModel("ResNet50.mlpackage")

# Inspect model metadata
spec = model.get_spec()
print(f"Model type: {spec.WhichOneof('Type')}")
print(f"Inputs: {[inp.name for inp in spec.description.input]}")
print(f"Outputs: {[out.name for out in spec.description.output]}")

# Run a test prediction
from PIL import Image
img = Image.open("test_image.jpg").resize((224, 224))
prediction = model.predict({"image": img})
print(f"Top prediction: {prediction}")

# Check which compute units are available
print(f"Compute units: {model.compute_unit}")

Konversi Model TensorFlow ke CoreML

CoreML mengonversi model TensorFlow: direktori SavedModel dan model Keras (model tf.keras atau file .h5). File TensorFlow Lite .tflite tidak dapat dikonversi: mulailah dari model TensorFlow aslinya.

Konversi TensorFlow SavedModel

import coremltools as ct
import tensorflow as tf

# Load a TensorFlow SavedModel (e.g., EfficientNet trained on your data)
tf_model = tf.keras.applications.EfficientNetV2S(
    weights="imagenet",
    input_shape=(384, 384, 3)
)

# Convert to CoreML
coreml_model = ct.convert(
    tf_model,
    inputs=[ct.ImageType(
        name="image",
        shape=(1, 384, 384, 3),
        scale=1.0 / 255.0,
        color_layout=ct.colorlayout.RGB
    )],
    compute_units=ct.ComputeUnit.ALL,
    minimum_deployment_target=ct.target.macOS15,
)

coreml_model.save("EfficientNetV2S.mlpackage")
print("Saved EfficientNetV2S.mlpackage")

Konversi Model Keras H5

import coremltools as ct
import tensorflow as tf

# Load your custom Keras model
model = tf.keras.models.load_model("my_custom_model.h5")

# Print model summary to understand input/output shapes
model.summary()

# Convert with explicit input/output specifications
coreml_model = ct.convert(
    model,
    inputs=[ct.TensorType(name="features", shape=(1, 128))],
    outputs=[ct.TensorType(name="prediction")],
    compute_units=ct.ComputeUnit.ALL,
    minimum_deployment_target=ct.target.macOS15,
)

# Add model metadata
coreml_model.author = "ML Team"
coreml_model.license = "Proprietary"
coreml_model.short_description = "Customer churn prediction model v2.1"
coreml_model.version = "2.1.0"

coreml_model.save("ChurnPredictor.mlpackage")

Model ONNX: Konversi dari Framework Asal

coremltools tidak lagi mengonversi file ONNX: konverter onnx-coreml-nya dibekukan, dan ct.convert hanya menerima model TensorFlow dan PyTorch (path .onnx gagal dengan "Unable to determine the type of the model"). Jika model Anda tersedia dalam format ONNX, konversikan dari framework yang menghasilkannya. Dua kasus umum:

Ekspor YOLOv8 Langsung ke CoreML

# Ultralytics exports its YOLO models to CoreML itself (it runs coremltools for you)
pip install ultralytics

# Writes yolov8n.mlpackage, with non-maximum suppression built in
yolo export model=yolov8n.pt format=coreml imgsz=640 nms=True

Konversi Model PyTorch Asli, Bukan Ekspor ONNX-nya

import torch
import torchvision
import coremltools as ct

# Start from the PyTorch model you would have exported to ONNX
model = torchvision.models.mobilenet_v3_large(
    weights=torchvision.models.MobileNet_V3_Large_Weights.DEFAULT
).eval()
example_input = torch.rand(1, 3, 224, 224)

# Trace it (ct.convert also accepts a torch.export ExportedProgram)
traced_model = torch.jit.trace(model, example_input)

# Add your model's input normalization as scale/bias, as in Step 2
coreml_model = ct.convert(
    traced_model,
    inputs=[ct.ImageType(name="image", shape=(1, 3, 224, 224), scale=1.0 / 255.0)],
    compute_units=ct.ComputeUnit.ALL,
    minimum_deployment_target=ct.target.macOS15,
)

coreml_model.save("MobileNetV3.mlpackage")
print("Saved MobileNetV3.mlpackage")

Optimalkan untuk Neural Engine

Neural Engine menghadirkan performa puncak dengan model terkuantisasi. Menerapkan kuantisasi pasca-pelatihan, palettization, dan pruning dapat mengurangi ukuran model sebesar 4-8x dan meningkatkan throughput Neural Engine sebesar 2-4x, sering dengan kehilangan akurasi yang dapat diabaikan.

Float16: Sudah Menjadi Presisi Bawaan ML Program

import os
import torch
import torchvision
import coremltools as ct

# ML Programs (.mlpackage) are typed in float16 by default: the ResNet50 from
# Step 2 already stores float16 weights, so no extra quantization pass is needed.
# To see what float16 saves, convert the same model once in float32:
model = torchvision.models.resnet50(weights=torchvision.models.ResNet50_Weights.DEFAULT).eval()
traced_model = torch.jit.trace(model, torch.randn(1, 3, 224, 224))

model_fp32 = ct.convert(
    traced_model,
    inputs=[ct.ImageType(name="image", shape=(1, 3, 224, 224))],
    compute_precision=ct.precision.FLOAT32,
    minimum_deployment_target=ct.target.macOS15,
)
model_fp32.save("ResNet50_fp32.mlpackage")

# Check file sizes
def size_on_disk(path):
    return sum(
        os.path.getsize(os.path.join(dp, f))
        for dp, dn, fn in os.walk(path) for f in fn
    )

fp32_size = size_on_disk("ResNet50_fp32.mlpackage")
fp16_size = size_on_disk("ResNet50.mlpackage")
print(f"Float32: {fp32_size / 1e6:.1f} MB")
print(f"Float16 (default): {fp16_size / 1e6:.1f} MB")
print(f"Reduction: {(1 - fp16_size / fp32_size) * 100:.1f}%")

Kuantisasi Pasca-Pelatihan Int8

import coremltools as ct
import coremltools.optimize as cto
from pathlib import Path
from PIL import Image

# Load the model
model = ct.models.MLModel("ResNet50.mlpackage")

# Configure linear (int8) quantization
op_config = cto.coreml.OpLinearQuantizerConfig(
    mode="linear_symmetric",
    dtype="int8",
    granularity="per_channel"
)

config = cto.coreml.OptimizationConfig(global_config=op_config)

# Weight-only int8 quantization: no calibration data needed
model_int8 = cto.coreml.linear_quantize_weights(model, config=config)
model_int8.save("ResNet50_int8.mlpackage")
print("Saved int8 quantized model")

# Optional: quantize activations too (W8A8), calibrated on real samples.
# Samples use the same format as model.predict(): here, the "image" input.
def load_calibration_data():
    """Load 100-200 representative images for calibration."""
    return [
        {"image": Image.open(path).convert("RGB").resize((224, 224))}
        for path in sorted(Path("calibration_images").glob("*.jpg"))[:200]
    ]

activation_config = cto.coreml.OptimizationConfig(
    global_config=cto.coreml.OpLinearQuantizerConfig(mode="linear_symmetric")
)
model_a8 = cto.coreml.linear_quantize_activations(
    model, activation_config, load_calibration_data()
)
model_w8a8 = cto.coreml.linear_quantize_weights(model_a8, config=config)
model_w8a8.save("ResNet50_w8a8.mlpackage")
print("Saved W8A8 quantized model")

Palettization (Pengelompokan Bobot)

import coremltools as ct
import coremltools.optimize as cto

# Palettization clusters weights into a small lookup table
# 4-bit palettization = 16 unique weight values per tensor
# Achieves ~4x compression with minimal accuracy loss

model = ct.models.MLModel("ResNet50.mlpackage")

# Configure palettization
op_config = cto.coreml.OpPalettizerConfig(
    mode="kmeans",
    nbits=4,              # 4-bit = 16 clusters, 2-bit = 4 clusters
    granularity="per_tensor"
)

config = cto.coreml.OptimizationConfig(global_config=op_config)

# Apply palettization
model_palettized = cto.coreml.palettize_weights(model, config=config)
model_palettized.save("ResNet50_palettized_4bit.mlpackage")

print("4-bit palettized model saved")
print("This model runs optimally on the Neural Engine")

Pruning (Sparsitas)

import coremltools as ct
import coremltools.optimize as cto

# Pruning sets small weights to zero, enabling sparse computation
# The Neural Engine can skip zero-weight operations for speed gains

model = ct.models.MLModel("ResNet50.mlpackage")

# Configure magnitude-based pruning
op_config = cto.coreml.OpMagnitudePrunerConfig(
    target_sparsity=0.75,              # Remove 75% of smallest weights
    block_size=None                     # Unstructured pruning
)

config = cto.coreml.OptimizationConfig(global_config=op_config)

# Apply pruning
model_pruned = cto.coreml.prune_weights(model, config=config)
model_pruned.save("ResNet50_pruned_75.mlpackage")

print("75% sparse model saved")

Pipeline Optimasi Gabungan

import coremltools as ct
import coremltools.optimize as cto

# For maximum optimization, combine pruning + palettization + quantization
# This can achieve 8-16x compression with 1-2% accuracy loss

model = ct.models.MLModel("ResNet50.mlpackage")

# Step 1: Prune (set small weights to zero)
prune_config = cto.coreml.OptimizationConfig(
    global_config=cto.coreml.OpMagnitudePrunerConfig(target_sparsity=0.5)
)
model = cto.coreml.prune_weights(model, config=prune_config)
print("Step 1: Pruning complete (50% sparsity)")

# Step 2: Palettize (cluster remaining weights)
palette_config = cto.coreml.OptimizationConfig(
    global_config=cto.coreml.OpPalettizerConfig(mode="kmeans", nbits=4)
)
model = cto.coreml.palettize_weights(model, config=palette_config)
print("Step 2: Palettization complete (4-bit)")

# Save the fully optimized model
model.save("ResNet50_optimized.mlpackage")
print("Fully optimized model saved -- ready for Neural Engine deployment")

Bangun REST API untuk Inferensi CoreML

Membungkus model CoreML Anda dalam REST API membuatnya dapat diakses oleh klien mana pun: aplikasi web, aplikasi seluler, microservice, atau pipeline pemrosesan batch. Berikut contoh siap produksi menggunakan Flask dan FastAPI.

Opsi A: Server API Flask

# flask_coreml_server.py
# pip install flask pillow coremltools gunicorn

import io
import time
import coremltools as ct
from flask import Flask, request, jsonify
from PIL import Image

app = Flask(__name__)

# Load the CoreML model at startup (runs on Neural Engine)
print("Loading CoreML model...")
model = ct.models.MLModel(
    "ResNet50_optimized.mlpackage",
    compute_units=ct.ComputeUnit.ALL
)
print("Model loaded successfully")

@app.route("/health", methods=["GET"])
def health():
    return jsonify({"status": "healthy", "model": "ResNet50"})

@app.route("/predict", methods=["POST"])
def predict():
    if "image" not in request.files:
        return jsonify({"error": "No image file provided"}), 400

    # Read and preprocess the image
    image_file = request.files["image"]
    image = Image.open(io.BytesIO(image_file.read())).resize((224, 224))

    # Run inference with timing
    start = time.perf_counter()
    prediction = model.predict({"image": image})
    latency_ms = (time.perf_counter() - start) * 1000

    return jsonify({
        "prediction": prediction,
        "latency_ms": round(latency_ms, 2),
        "compute_unit": "neural_engine+gpu+cpu"
    })

@app.route("/predict/batch", methods=["POST"])
def predict_batch():
    """Process multiple images in a single request."""
    if "images" not in request.files:
        return jsonify({"error": "No image files provided"}), 400

    results = []
    files = request.files.getlist("images")

    start = time.perf_counter()
    for image_file in files:
        image = Image.open(io.BytesIO(image_file.read())).resize((224, 224))
        prediction = model.predict({"image": image})
        results.append(prediction)
    total_ms = (time.perf_counter() - start) * 1000

    return jsonify({
        "predictions": results,
        "total_latency_ms": round(total_ms, 2),
        "images_processed": len(results),
        "avg_latency_ms": round(total_ms / len(results), 2)
    })

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

# Production: gunicorn flask_coreml_server:app -w 2 -b 0.0.0.0:5000 --timeout 120

Opsi B: Server FastAPI (Async + Dokumentasi OpenAPI)

# fastapi_coreml_server.py
# pip install fastapi uvicorn python-multipart pillow coremltools

import io
import time
import asyncio
from concurrent.futures import ThreadPoolExecutor
import coremltools as ct
from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import JSONResponse
from PIL import Image
from typing import List

app = FastAPI(
    title="CoreML Inference API",
    description="Production CoreML model serving on Mac Mini M4",
    version="1.0.0"
)

# Load model at startup
model = ct.models.MLModel(
    "ResNet50_optimized.mlpackage",
    compute_units=ct.ComputeUnit.ALL
)

# Thread pool for blocking CoreML calls
executor = ThreadPoolExecutor(max_workers=4)

def run_prediction(image_bytes: bytes) -> dict:
    """Run CoreML prediction in a thread (blocking call)."""
    image = Image.open(io.BytesIO(image_bytes)).resize((224, 224))
    start = time.perf_counter()
    result = model.predict({"image": image})
    latency = (time.perf_counter() - start) * 1000
    return {"prediction": result, "latency_ms": round(latency, 2)}

@app.get("/health")
async def health():
    return {"status": "healthy", "model": "ResNet50_optimized", "engine": "CoreML"}

@app.post("/predict")
async def predict(image: UploadFile = File(...)):
    if not image.content_type.startswith("image/"):
        raise HTTPException(status_code=400, detail="File must be an image")

    image_bytes = await image.read()
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(executor, run_prediction, image_bytes)
    return result

@app.post("/predict/batch")
async def predict_batch(images: List[UploadFile] = File(...)):
    loop = asyncio.get_event_loop()
    tasks = []
    for img in images:
        image_bytes = await img.read()
        tasks.append(loop.run_in_executor(executor, run_prediction, image_bytes))

    results = await asyncio.gather(*tasks)
    return {
        "predictions": list(results),
        "total_images": len(results)
    }

# Run: uvicorn fastapi_coreml_server:app --host 0.0.0.0 --port 8000 --workers 2

Uji API

# Test single prediction
curl -X POST http://localhost:8000/predict \
  -F "image=@test_image.jpg"

# Test batch prediction
curl -X POST http://localhost:8000/predict/batch \
  -F "images=@image1.jpg" \
  -F "images=@image2.jpg" \
  -F "images=@image3.jpg"

# Health check
curl http://localhost:8000/health

# Python client example
import requests

with open("test_image.jpg", "rb") as f:
    response = requests.post(
        "http://your-mac-mini:8000/predict",
        files={"image": f}
    )
print(response.json())
# {"prediction": {"classLabel": "golden_retriever", "confidence": 0.94}, "latency_ms": 3.2}

Layanan Gaya Systemd dengan launchd

# Create a LaunchAgent: the API starts with your user session and restarts
# if it crashes (enable automatic login so it also returns after a reboot)
mkdir -p /Users/admin/coreml-api/logs
cat <<EOF > ~/Library/LaunchAgents/com.coreml.api.plist
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
  "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <key>Label</key>
    <string>com.coreml.api</string>
    <key>ProgramArguments</key>
    <array>
        <string>/Users/admin/coreml-env/bin/uvicorn</string>
        <string>fastapi_coreml_server:app</string>
        <string>--host</string>
        <string>0.0.0.0</string>
        <string>--port</string>
        <string>8000</string>
        <string>--workers</string>
        <string>2</string>
    </array>
    <key>WorkingDirectory</key>
    <string>/Users/admin/coreml-api</string>
    <key>RunAtLoad</key>
    <true/>
    <key>KeepAlive</key>
    <true/>
    <key>StandardOutPath</key>
    <string>/Users/admin/coreml-api/logs/coreml-api.log</string>
    <key>StandardErrorPath</key>
    <string>/Users/admin/coreml-api/logs/coreml-api-error.log</string>
</dict>
</plist>
EOF

# Load the service
launchctl load ~/Library/LaunchAgents/com.coreml.api.plist

# Verify
curl http://localhost:8000/health

Benchmark Performa

Benchmark ini membandingkan inferensi CoreML di Mac Mini M4 dengan PyTorch MPS (Metal Performance Shaders) dan eksekusi hanya-CPU. Semua pengujian menggunakan inferensi gambar tunggal dengan batch size 1.

Klasifikasi Gambar (ResNet50, input 224x224)

Deteksi Objek (YOLOv8n, input 640x640)

Jalankan Benchmark Anda Sendiri

import coremltools as ct
import numpy as np
import time

model = ct.models.MLModel("ResNet50_optimized.mlpackage", compute_units=ct.ComputeUnit.ALL)

# Warmup (first inference compiles the model for the Neural Engine)
from PIL import Image
dummy = Image.new("RGB", (224, 224))
for _ in range(10):
    model.predict({"image": dummy})

# Benchmark
latencies = []
for _ in range(1000):
    start = time.perf_counter()
    model.predict({"image": dummy})
    latencies.append((time.perf_counter() - start) * 1000)

latencies = np.array(latencies)
print(f"Mean latency:   {latencies.mean():.2f} ms")
print(f"Median latency: {np.median(latencies):.2f} ms")
print(f"P95 latency:    {np.percentile(latencies, 95):.2f} ms")
print(f"P99 latency:    {np.percentile(latencies, 99):.2f} ms")
print(f"Throughput:     {1000 / latencies.mean():.0f} images/sec")

Penskalaan dengan Banyak Model

Deployment produksi sering kali memerlukan penyajian banyak model atau menangani konkurensi tinggi. Anda dapat menggunakan nginx sebagai reverse proxy dan load balancer di beberapa instance Mac Mini M4, atau menyajikan banyak model dari satu mesin.

Server Multi-Model

# multi_model_server.py
import io
import time
import coremltools as ct
from fastapi import FastAPI, File, UploadFile, HTTPException
from PIL import Image

app = FastAPI(title="Multi-Model CoreML Server")

# Load multiple models at startup
models = {}

@app.on_event("startup")
async def load_models():
    print("Loading models...")
    models["resnet50"] = ct.models.MLModel(
        "ResNet50_optimized.mlpackage", compute_units=ct.ComputeUnit.ALL
    )
    models["yolov8"] = ct.models.MLModel(
        "YOLOv8n.mlpackage", compute_units=ct.ComputeUnit.ALL
    )
    models["efficientnet"] = ct.models.MLModel(
        "EfficientNetV2S.mlpackage", compute_units=ct.ComputeUnit.ALL
    )
    print(f"Loaded {len(models)} models: {list(models.keys())}")

@app.get("/models")
async def list_models():
    return {"models": list(models.keys())}

@app.post("/predict/{model_name}")
async def predict(model_name: str, image: UploadFile = File(...)):
    if model_name not in models:
        raise HTTPException(404, f"Model '{model_name}' not found. Available: {list(models.keys())}")

    image_data = Image.open(io.BytesIO(await image.read()))

    # Resize based on model requirements
    input_sizes = {"resnet50": (224, 224), "yolov8": (640, 640), "efficientnet": (384, 384)}
    image_data = image_data.resize(input_sizes.get(model_name, (224, 224)))

    start = time.perf_counter()
    result = models[model_name].predict({"image": image_data})
    latency = (time.perf_counter() - start) * 1000

    return {"model": model_name, "prediction": result, "latency_ms": round(latency, 2)}

# Run: uvicorn multi_model_server:app --host 0.0.0.0 --port 8000

Nginx Load Balancer di Beberapa Mac Mini

# /opt/homebrew/etc/nginx/servers/coreml.conf
# Install nginx: brew install nginx
# (Homebrew's nginx.conf loads every file in servers/ inside its http block)

# Rate limiting zone (defined at http level)
limit_req_zone $binary_remote_addr zone=api:10m rate=100r/s;

upstream coreml_backend {
    # Round-robin across multiple Mac Mini M4 instances
    server 10.0.1.10:8000 weight=1;   # Mac Mini M4 #1
    server 10.0.1.11:8000 weight=1;   # Mac Mini M4 #2
    server 10.0.1.12:8000 weight=1;   # Mac Mini M4 #3

    # Health check: remove unhealthy backends
    keepalive 32;
}

server {
    listen 80;
    server_name api.yourdomain.com;

    location / {
        limit_req zone=api burst=50 nodelay;

        proxy_pass http://coreml_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

        # Timeout settings for ML inference
        proxy_connect_timeout 10s;
        proxy_send_timeout 30s;
        proxy_read_timeout 30s;

        # Enable keepalive to backend
        proxy_http_version 1.1;
        proxy_set_header Connection "";
    }

    location /health {
        proxy_pass http://coreml_backend;
        access_log off;
    }
}

# Test config and start
# nginx -t
# nginx

Docker-Compose untuk Pengembangan Lokal

# docker-compose.yml
# Note: CoreML requires macOS -- Docker containers run CPU-only inference
# For production, use launchd services directly on macOS

version: "3.8"
services:
  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      # coreml.conf from above; for local runs, point its upstream at coreml-api:8000
      - ./coreml.conf:/etc/nginx/conf.d/default.conf:ro
    depends_on:
      - coreml-api

  coreml-api:
    build: .
    # No host port: nginx reaches both replicas on the compose network
    volumes:
      - ./models:/app/models
    environment:
      - MODEL_PATH=/app/models/ResNet50_optimized.mlpackage
      - WORKERS=2
    deploy:
      replicas: 2

Pemantauan & Observabilitas

Sistem ML produksi memerlukan pemantauan untuk latensi inferensi, throughput, tingkat kesalahan, dan penggunaan sumber daya sistem. Berikut cara menginstrumentasi API CoreML Anda dengan metrik Prometheus dan pemantauan tingkat sistem.

Tambahkan Metrik Prometheus ke FastAPI

# pip install prometheus-client prometheus-fastapi-instrumentator

import io
import time
import coremltools as ct
from fastapi import FastAPI, File, UploadFile
from PIL import Image
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from starlette.responses import Response

app = FastAPI(title="CoreML API with Monitoring")

# Prometheus metrics
PREDICTIONS_TOTAL = Counter(
    "coreml_predictions_total",
    "Total number of predictions",
    ["model", "status"]
)
PREDICTION_LATENCY = Histogram(
    "coreml_prediction_latency_seconds",
    "Prediction latency in seconds",
    ["model"],
    buckets=[0.001, 0.002, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0]
)
MODEL_LOAD_TIME = Gauge(
    "coreml_model_load_time_seconds",
    "Time taken to load the model",
    ["model"]
)
ACTIVE_REQUESTS = Gauge(
    "coreml_active_requests",
    "Number of currently active requests"
)

# Load model with timing
load_start = time.perf_counter()
model = ct.models.MLModel("ResNet50_optimized.mlpackage", compute_units=ct.ComputeUnit.ALL)
MODEL_LOAD_TIME.labels(model="resnet50").set(time.perf_counter() - load_start)

@app.get("/metrics")
async def metrics():
    return Response(content=generate_latest(), media_type="text/plain")

@app.post("/predict")
async def predict(image: UploadFile = File(...)):
    ACTIVE_REQUESTS.inc()
    try:
        img = Image.open(io.BytesIO(await image.read())).resize((224, 224))

        start = time.perf_counter()
        result = model.predict({"image": img})
        latency = time.perf_counter() - start

        PREDICTION_LATENCY.labels(model="resnet50").observe(latency)
        PREDICTIONS_TOTAL.labels(model="resnet50", status="success").inc()

        return {"prediction": result, "latency_ms": round(latency * 1000, 2)}
    except Exception as e:
        PREDICTIONS_TOTAL.labels(model="resnet50", status="error").inc()
        raise
    finally:
        ACTIVE_REQUESTS.dec()

# Run: uvicorn monitored_server:app --host 0.0.0.0 --port 8000

Konfigurasi Prometheus

# prometheus.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: "coreml-api"
    static_configs:
      - targets:
        - "10.0.1.10:8000"   # Mac Mini #1
        - "10.0.1.11:8000"   # Mac Mini #2
        - "10.0.1.12:8000"   # Mac Mini #3
    metrics_path: /metrics
    scrape_interval: 5s

  - job_name: "node-exporter"
    static_configs:
      - targets:
        - "10.0.1.10:9100"
        - "10.0.1.11:9100"
        - "10.0.1.12:9100"

Skrip Pemantauan Tingkat Sistem

#!/bin/bash
# monitor_coreml.sh -- System health monitoring for CoreML inference servers
# Run with: ./monitor_coreml.sh

echo "=== CoreML Server Health Monitor ==="
echo "$(date)"
echo ""

# Memory usage (critical for CoreML model loading)
echo "--- Memory Usage ---"
vm_stat | head -10
echo ""
memory_pressure
echo ""

# CPU and GPU utilization
echo "--- CPU Usage ---"
top -l 1 -n 5 -stats pid,command,cpu,mem | head -10
echo ""

# GPU/Neural Engine power (indicates compute unit activity)
echo "--- GPU/Neural Engine Power ---"
sudo powermetrics --samplers gpu_power,ane_power -n 1 -i 2000 2>/dev/null | grep -E "(GPU|ANE|Neural)"
echo ""

# Disk usage (model files can be large)
echo "--- Disk Usage ---"
df -h / | tail -1
echo ""

# Network connections to API
echo "--- Active API Connections ---"
netstat -an | grep ":8000" | wc -l | xargs echo "Active connections on port 8000:"
echo ""

# API health check
echo "--- API Health Check ---"
curl -s -w "\nHTTP Status: %{http_code}\nResponse Time: %{time_total}s\n" \
  http://localhost:8000/health 2>/dev/null || echo "API is DOWN"

Kueri Dasbor Grafana

# Useful PromQL queries for your Grafana dashboard:

# Average prediction latency (last 5 minutes)
rate(coreml_prediction_latency_seconds_sum[5m]) / rate(coreml_prediction_latency_seconds_count[5m])

# Predictions per second
rate(coreml_predictions_total[1m])

# P99 latency
histogram_quantile(0.99, rate(coreml_prediction_latency_seconds_bucket[5m]))

# Error rate percentage
rate(coreml_predictions_total{status="error"}[5m]) / rate(coreml_predictions_total[5m]) * 100

# Active concurrent requests
coreml_active_requests

Pertanyaan yang Sering Diajukan