تقديم النماذج مع BentoML

BentoML يُبسط تحويل نماذج ML إلى APIs جاهزة للإنتاج. يتعامل مع التغليف والحاويات والنشر بأقل كود معياري.

التثبيت

pip install bentoml

# تحقق من التثبيت
bentoml --version

حفظ النماذج في BentoML

حفظ من التدريب

import bentoml
from sklearn.ensemble import RandomForestClassifier

# درّب النموذج
model = RandomForestClassifier(n_estimators=100)
model.fit(X_train, y_train)

# احفظ في مخزن نماذج BentoML
saved_model = bentoml.sklearn.save_model(
    "fraud_detector",
    model,
    signatures={"predict": {"batchable": True}},
    labels={"team": "risk", "version": "1.0"},
    metadata={"accuracy": 0.95}
)

print(f"حُفظ: {saved_model.tag}")
# المخرج: fraud_detector:abc123xyz

دعم الأطر

# PyTorch
import bentoml
import torch

bentoml.pytorch.save_model("my_pytorch_model", pytorch_model)

# TensorFlow/Keras
bentoml.keras.save_model("my_keras_model", keras_model)

# XGBoost
bentoml.xgboost.save_model("my_xgb_model", xgb_model)

# ONNX
bentoml.onnx.save_model("my_onnx_model", onnx_model)

# نماذج مخصصة
bentoml.picklable_model.save_model("custom_model", my_model)

إنشاء خدمة

تعريف خدمة أساسية

# service.py
import bentoml
import numpy as np

@bentoml.service(
    resources={"cpu": "2", "memory": "4Gi"},
    traffic={"timeout": 30}
)
class FraudDetector:

    def __init__(self):
        # حمّل النموذج من مخزن BentoML
        self.model = bentoml.sklearn.load_model("fraud_detector:latest")

    @bentoml.api
    def predict(self, features: np.ndarray) -> np.ndarray:
        """توقع احتمالية الاحتيال."""
        return self.model.predict_proba(features)

    @bentoml.api
    def classify(self, features: np.ndarray) -> list[int]:
        """صنّف كاحتيال أو لا."""
        return self.model.predict(features).tolist()

مع التحقق بـ Pydantic

# service.py
import bentoml
from pydantic import BaseModel
import numpy as np

class TransactionInput(BaseModel):
    amount: float
    merchant_category: int
    hour_of_day: int
    is_international: bool

class PredictionOutput(BaseModel):
    is_fraud: bool
    confidence: float

@bentoml.service(
    resources={"cpu": "2", "memory": "4Gi"}
)
class FraudDetector:

    def __init__(self):
        self.model = bentoml.sklearn.load_model("fraud_detector:latest")

    @bentoml.api
    def predict(self, transaction: TransactionInput) -> PredictionOutput:
        features = np.array([[
            transaction.amount,
            transaction.merchant_category,
            transaction.hour_of_day,
            int(transaction.is_international)
        ]])

        proba = self.model.predict_proba(features)[0]
        is_fraud = proba[1] > 0.5

        return PredictionOutput(
            is_fraud=is_fraud,
            confidence=float(proba[1]) if is_fraud else float(proba[0])
        )

تشغيل الخدمة

وضع التطوير

# ابدأ خادم التطوير
bentoml serve service:FraudDetector --reload

# الوصول للـ API
curl -X POST http://localhost:3000/predict \
  -H "Content-Type: application/json" \
  -d '{"amount": 1500.0, "merchant_category": 5, "hour_of_day": 3, "is_international": true}'

توثيق API

BentoML يولّد توثيق OpenAPI تلقائياً في http://localhost:3000/docs.

بناء Bento

إنشاء bentofile.yaml

# bentofile.yaml
service: "service:FraudDetector"
labels:
  team: risk
  environment: production
include:
  - "*.py"
python:
  packages:
    - scikit-learn==1.4.0
    - numpy>=1.24.0
    - pydantic>=2.0.0

بناء الـ Bento

# ابنِ bento مُغلف
bentoml build

# اعرض bentos
bentoml list

# المخرج:
# Tag                          Size      Created
# fraud_detector:v1_abc123     45.2 MiB  2025-01-15

الحاويات

بناء صورة Docker

# ابنِ حاوية من bento
bentoml containerize fraud_detector:latest

# علّم للسجل
bentoml containerize fraud_detector:latest -t myregistry/fraud-detector:v1

# شغّل الحاوية
docker run -p 3000:3000 fraud_detector:latest

Dockerfile مخصص

# bentofile.yaml
service: "service:FraudDetector"
docker:
  python_version: "3.11"
  system_packages:
    - libgomp1
  env:
    - name: MODEL_TIMEOUT
      value: "30"
python:
  packages:
    - scikit-learn==1.4.0

خيارات النشر

النشر على BentoCloud

# سجّل دخول BentoCloud
bentoml cloud login

# انشر
bentoml deploy fraud_detector:latest

# تحقق من الحالة
bentoml deployment list

نشر Kubernetes

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: fraud-detector
spec:
  replicas: 3
  selector:
    matchLabels:
      app: fraud-detector
  template:
    metadata:
      labels:
        app: fraud-detector
    spec:
      containers:
      - name: fraud-detector
        image: myregistry/fraud-detector:v1
        ports:
        - containerPort: 3000
        resources:
          requests:
            cpu: "1"
            memory: "2Gi"
          limits:
            cpu: "2"
            memory: "4Gi"
---
apiVersion: v1
kind: Service
metadata:
  name: fraud-detector-svc
spec:
  selector:
    app: fraud-detector
  ports:
  - port: 80
    targetPort: 3000
  type: LoadBalancer

kubectl apply -f deployment.yaml

التجميع والأداء

التجميع التكيفي

@bentoml.service(
    traffic={
        "timeout": 60,
        "max_batch_size": 100,
        "max_latency_ms": 500
    }
)
class FraudDetector:

    @bentoml.api(batchable=True)
    def predict_batch(self, features: np.ndarray) -> np.ndarray:
        """معالجة الطلبات المجمعة للكفاءة."""
        return self.model.predict_proba(features)

نقاط نهاية غير متزامنة

import asyncio

@bentoml.service
class AsyncFraudDetector:

    @bentoml.api
    async def predict(self, features: np.ndarray) -> np.ndarray:
        # معالجة غير متزامنة لعمليات I/O
        result = await asyncio.to_thread(
            self.model.predict_proba, features
        )
        return result

خدمات متعددة النماذج

@bentoml.service
class EnsemblePredictor:

    def __init__(self):
        self.rf_model = bentoml.sklearn.load_model("random_forest:latest")
        self.xgb_model = bentoml.xgboost.load_model("xgboost:latest")

    @bentoml.api
    def predict(self, features: np.ndarray) -> dict:
        rf_pred = self.rf_model.predict_proba(features)
        xgb_pred = self.xgb_model.predict_proba(features)

        # متوسط المجموعة
        ensemble = (rf_pred + xgb_pred) / 2

        return {
            "ensemble": ensemble.tolist(),
            "random_forest": rf_pred.tolist(),
            "xgboost": xgb_pred.tolist()
        }

مقارنة: BentoML مقابل البدائل

الميزة	BentoML	FastAPI + Docker	TorchServe
تعقيد الإعداد	منخفض	متوسط	متوسط
دعم الأطر	جميع الرئيسية	يدوي	PyTorch فقط
التجميع	مدمج	يدوي	مدمج
الحاويات	أمر واحد	يدوي	يدوي
نشر سحابي	BentoCloud	افعلها بنفسك	AWS SageMaker

الرؤية الرئيسية: BentoML يسد الفجوة بين تدريب النموذج ونشر الإنتاج بأقل تغييرات في الكود—من model.predict() إلى API قابل للتوسع في دقائق.

التالي، سنستكشف نشر canary واستراتيجيات اختبار A/B. :::