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 で受け取ったモデルは、それを作成した元のフレームワークから変換してください。よくあるケースは次の 2 つです。

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 でラップすると、Web アプリ、モバイルアプリ、マイクロサービス、バッチ処理パイプラインなど、あらゆるクライアントからアクセスできるようになります。以下に、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 サーバー（非同期 + 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

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

複数モデルによるスケーリング

本番デプロイでは、複数のモデルを提供したり、高い同時実行数を処理したりする必要が多くあります。nginx をリバースプロキシおよびロードバランサーとして使い、複数の Mac Mini M4 インスタンスに負荷を分散させることも、1 台のマシンから複数のモデルを提供することもできます。

マルチモデルサーバー

# 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 システムでは、推論レイテンシ、スループット、エラー率、システムリソースの使用状況を監視する必要があります。ここでは、Prometheus のメトリクスとシステムレベルの監視を使って CoreML API を計装する方法を紹介します。

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

よくある質問