CoreML डिप्लॉयमेंट गाइड: Mac Mini M4 पर ट्रेनिंग से प्रोडक्शन तक

Mac Mini M4 पर CoreML क्यों?

CoreML Apple का नेटिव मशीन लर्निंग फ्रेमवर्क है, जिसे Apple Silicon से अधिकतम प्रदर्शन निकालने के लिए विशेष रूप से बनाया गया है। GPU को एकल कंप्यूट डिवाइस के रूप में मानने वाले सामान्य ML फ्रेमवर्क के विपरीत, CoreML बुद्धिमानी से वर्कलोड को CPU, GPU, और समर्पित 16-कोर Neural Engine में वितरित करता है, अक्सर एक मॉडल के अलग-अलग हिस्सों को अलग-अलग कंप्यूट इकाइयों पर एक साथ चलाता है।

PyTorch मॉडल को CoreML में कन्वर्ट करें

Apple की coremltools लाइब्रेरी PyTorch मॉडल से CoreML के .mlpackage फ़ॉर्मेट तक एक सीधा कन्वर्ज़न पथ प्रदान करती है। कन्वर्ज़न आपके मॉडल को सैंपल इनपुट के साथ ट्रेस करता है और प्रत्येक ऑपरेशन को CoreML के आंतरिक प्रतिनिधित्व में अनुवादित करता है।

चरण 1: डिपेंडेंसी इंस्टॉल करें

# 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

चरण 2: एक 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")

चरण 3: एक कस्टम PyTorch मॉडल कन्वर्ट करें

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")

चरण 4: कन्वर्ट किए गए मॉडल को सत्यापित करें

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}")

TensorFlow मॉडल को CoreML में कन्वर्ट करें

CoreML, TensorFlow मॉडल कन्वर्ट करता है: SavedModel डायरेक्टरी और Keras मॉडल (एक tf.keras मॉडल या .h5 फ़ाइल)। TensorFlow Lite .tflite फ़ाइलें कन्वर्ट नहीं की जा सकतीं: इसके बजाय मूल TensorFlow मॉडल से शुरू करें।

एक 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")

एक 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")

ONNX मॉडल: मूल फ़्रेमवर्क से कन्वर्ट करें

coremltools अब ONNX फ़ाइलें कन्वर्ट नहीं करता: इसका onnx-coreml कन्वर्टर फ़्रीज़ है, और ct.convert केवल TensorFlow और PyTorch मॉडल स्वीकार करता है (.onnx पाथ देने पर "Unable to determine the type of the model" त्रुटि आती है)। अगर कोई मॉडल आपको ONNX में मिलता है, तो उसे उस फ़्रेमवर्क से कन्वर्ट करें जिसमें वह बना था। दो आम स्थितियां:

YOLOv8 को सीधे 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

ONNX एक्सपोर्ट की जगह मूल PyTorch मॉडल कन्वर्ट करें

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")

Neural Engine के लिए ऑप्टिमाइज़ करें

Neural Engine क्वांटाइज़्ड मॉडलों के साथ चरम प्रदर्शन प्रदान करता है। पोस्ट-ट्रेनिंग क्वांटाइज़ेशन, पैलेटाइज़ेशन, और प्रूनिंग लागू करने से मॉडल का आकार 4-8 गुना तक कम हो सकता है और Neural Engine थ्रूपुट 2-4 गुना तक बेहतर हो सकता है, अक्सर नगण्य सटीकता हानि के साथ।

Float16: 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}%")

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")

पैलेटाइज़ेशन (वेट क्लस्टरिंग)

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")

प्रूनिंग (स्पार्सिटी)

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")

संयुक्त ऑप्टिमाइज़ेशन पाइपलाइन

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")

CoreML इन्फ़रेंस के लिए एक REST API बनाएं

अपने CoreML मॉडल को एक REST API में रैप करने से यह किसी भी क्लाइंट के लिए सुलभ हो जाता है: वेब ऐप्स, मोबाइल ऐप्स, माइक्रोसर्विसेज़, या बैच प्रोसेसिंग पाइपलाइन। नीचे Flask और FastAPI दोनों का उपयोग करते हुए प्रोडक्शन-रेडी उदाहरण दिए गए हैं।

विकल्प A: Flask API सर्वर

# 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

विकल्प B: FastAPI सर्वर (Async + OpenAPI Docs)

# 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

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}

launchd के साथ Systemd-शैली सेवा

# 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

प्रदर्शन बेंचमार्क

ये बेंचमार्क Mac Mini M4 पर CoreML इन्फ़रेंस की तुलना PyTorch MPS (Metal Performance Shaders) और केवल-CPU एक्ज़ीक्यूशन से करते हैं। सभी परीक्षण बैच आकार 1 के साथ सिंगल-इमेज इन्फ़रेंस का उपयोग करते हैं।

इमेज क्लासिफ़िकेशन (ResNet50, 224x224 इनपुट)

ऑब्जेक्ट डिटेक्शन (YOLOv8n, 640x640 इनपुट)

अपने स्वयं के बेंचमार्क चलाएं

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")

कई मॉडलों के साथ स्केलिंग

प्रोडक्शन डिप्लॉयमेंट के लिए अक्सर कई मॉडलों की सर्विंग या उच्च समवर्तीता को संभालने की आवश्यकता होती है। आप कई Mac Mini M4 इंस्टेंस में एक रिवर्स प्रॉक्सी और लोड बैलेंसर के रूप में nginx का उपयोग कर सकते हैं, या एक ही मशीन से कई मॉडल सर्व कर सकते हैं।

मल्टी-मॉडल सर्वर

# 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

कई Mac Mini में Nginx लोड बैलेंसर

# /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

# 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

मॉनिटरिंग और ऑब्ज़र्वेबिलिटी

प्रोडक्शन ML सिस्टम को इन्फ़रेंस लेटेंसी, थ्रूपुट, त्रुटि दर, और सिस्टम संसाधन उपयोग के लिए मॉनिटरिंग की आवश्यकता होती है। यहाँ बताया गया है कि अपने CoreML API को Prometheus मेट्रिक्स और सिस्टम-स्तरीय मॉनिटरिंग के साथ कैसे इंस्ट्रुमेंट करें।

FastAPI में Prometheus मेट्रिक्स जोड़ें

# 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

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"

सिस्टम-स्तरीय मॉनिटरिंग स्क्रिप्ट

#!/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"

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

अक्सर पूछे जाने वाले प्रश्न