الرئيسية › قاعدة المعرفة › أسس الذكاء الاصطناعي › نشر نموذج ML في المصنع: من Jupyter إلى الإنتاج الحقيقي
أسس الذكاء الاصطناعي

نشر نموذج ML في المصنع: من Jupyter إلى الإنتاج الحقيقي

9 دقائق للقراءة أسس الذكاء الاصطناعي

من التجربة إلى الإنتاج: التحدي الحقيقي

بناء نموذج في Jupyter Notebook هو 10% من المشروع. الـ 90% الباقية هي نشره في بيئة إنتاج حقيقية حيث يعمل 24/7 ويتعامل مع بيانات حقيقية:

تحديات النشر الصناعي

  • الموثوقية: النموذج يجب أن يعمل بلا توقف
  • السرعة: التنبؤ يجب أن يتم خلال ميلي ثوانٍ
  • المراقبة: يجب كشف انحراف الأداء مبكراً
  • التحديث: يجب تحديث النموذج بدون إيقاف الخدمة
# مثال: بناء نموذج جاهز للنشر
import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split

# بيانات فحص جودة
np.random.seed(42)
n = 3000
X = pd.DataFrame({
    'vibration': np.random.normal(2.5, 0.5, n),
    'temperature': np.random.normal(70, 5, n),
    'pressure': np.random.normal(100, 3, n),
    'current': np.random.normal(15, 2, n)
})
y = ((X['vibration'] > 3.2) | (X['temperature'] > 78)).astype(int)

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Pipeline: تطبيع + نموذج في كائن واحد
pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('classifier', RandomForestClassifier(n_estimators=100, max_depth=6, random_state=42))
])

pipeline.fit(X_train, y_train)
print(f"دقة النموذج: {pipeline.score(X_test, y_test):.4f}")

تصدير النموذج: pickle و ONNX

حفظ بـ pickle (الطريقة الأبسط)

import pickle
import os

# حفظ Pipeline كاملاً (يشمل التطبيع والنموذج)
model_path = 'quality_inspector_v1.pkl'
with open(model_path, 'wb') as f:
    pickle.dump(pipeline, f)

file_size = os.path.getsize(model_path) / 1024
print(f"تم حفظ النموذج: {model_path} ({file_size:.1f} KB)")

# تحميل واستخدام النموذج
with open(model_path, 'rb') as f:
    loaded_model = pickle.load(f)

# اختبار سريع
sample = pd.DataFrame({
    'vibration': [3.5], 'temperature': [72],
    'pressure': [101], 'current': [16]
})
prediction = loaded_model.predict(sample)[0]
probability = loaded_model.predict_proba(sample)[0]
print(f"التنبؤ: {'معيب' if prediction == 1 else 'سليم'}")
print(f"الاحتمالات: سليم={probability[0]:.2%}, معيب={probability[1]:.2%}")

حفظ بصيغة ONNX (للتوافق بين الأنظمة)

# ONNX يسمح بتشغيل النموذج في أي لغة (C++, Rust, JavaScript)
try:
    from skl2onnx import convert_sklearn
    from skl2onnx.common.data_types import FloatTensorType

    initial_type = [('features', FloatTensorType([None, 4]))]
    onnx_model = convert_sklearn(pipeline, initial_types=initial_type)

    with open('quality_inspector_v1.onnx', 'wb') as f:
        f.write(onnx_model.SerializeToString())
    print("تم تصدير النموذج بصيغة ONNX")
except ImportError:
    print("لتصدير ONNX: pip install skl2onnx")

بناء واجهة API بـ Flask أو FastAPI

FastAPI (الخيار الموصى به)

# ملف: app.py
"""
from fastapi import FastAPI
from pydantic import BaseModel
import pickle
import numpy as np

app = FastAPI(title="نظام فحص الجودة", version="1.0")

# تحميل النموذج عند بدء التطبيق
with open('quality_inspector_v1.pkl', 'rb') as f:
    model = pickle.load(f)

class SensorReading(BaseModel):
    vibration: float
    temperature: float
    pressure: float
    current: float

class PredictionResult(BaseModel):
    status: str
    confidence: float
    probabilities: dict

@app.post("/predict", response_model=PredictionResult)
async def predict_quality(reading: SensorReading):
    features = np.array([[
        reading.vibration, reading.temperature,
        reading.pressure, reading.current
    ]])
    prediction = model.predict(features)[0]
    probabilities = model.predict_proba(features)[0]

    return PredictionResult(
        status="معيب" if prediction == 1 else "سليم",
        confidence=float(max(probabilities)),
        probabilities={
            "سليم": float(probabilities[0]),
            "معيب": float(probabilities[1])
        }
    )

@app.get("/health")
async def health_check():
    return {"status": "running", "model_version": "v1.0"}
"""
print("ملف app.py جاهز. لتشغيله:")
print("  pip install fastapi uvicorn")
print("  uvicorn app:app --host 0.0.0.0 --port 8000")

اختبار الواجهة

# اختبار محلي باستخدام requests
"""
import requests

response = requests.post(
    "http://localhost:8000/predict",
    json={
        "vibration": 3.5,
        "temperature": 72.0,
        "pressure": 101.0,
        "current": 16.0
    }
)
result = response.json()
print(f"الحالة: {result['status']}")
print(f"الثقة: {result['confidence']:.2%}")
"""
print("بعد تشغيل الخادم، استخدم requests أو curl للاختبار")

حاوية Docker للنموذج

Docker يضمن أن النموذج يعمل بنفس الطريقة على أي خادم:

ملف Dockerfile

dockerfile_content = """
# Dockerfile للنموذج
FROM python:3.11-slim

WORKDIR /app

# تثبيت المتطلبات
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# نسخ النموذج والكود
COPY quality_inspector_v1.pkl .
COPY app.py .

# تشغيل الخادم
EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]
"""

requirements_content = """fastapi==0.109.0
uvicorn==0.27.0
scikit-learn==1.4.0
numpy==1.26.3
pydantic==2.5.3
"""

print("Dockerfile:")
print(dockerfile_content)
print("requirements.txt:")
print(requirements_content)
print("أوامر Docker:")
print("  docker build -t quality-inspector:v1 .")
print("  docker run -p 8000:8000 quality-inspector:v1")

المراقبة المستمرة: انحراف النموذج

بعد النشر، أداء النموذج قد ينخفض مع الوقت بسبب تغير ظروف التشغيل:

import numpy as np
import pandas as pd
from scipy.stats import ks_2samp

# محاكاة: مقارنة توزيع بيانات التدريب مع بيانات الإنتاج
np.random.seed(42)

# بيانات التدريب (الأصلية)
train_vibration = np.random.normal(2.5, 0.5, 1000)

# بيانات الإنتاج: الشهر الأول (مشابهة)
month1_vibration = np.random.normal(2.5, 0.5, 500)

# بيانات الإنتاج: الشهر السادس (انحراف!)
month6_vibration = np.random.normal(3.0, 0.7, 500)

# اختبار KS لكشف انحراف التوزيع
stat1, p1 = ks_2samp(train_vibration, month1_vibration)
stat6, p6 = ks_2samp(train_vibration, month6_vibration)

print("كشف انحراف البيانات:")
print(f"  الشهر 1: KS={stat1:.4f}, p={p1:.4f} {'(لا انحراف)' if p1 > 0.05 else '(انحراف!)'}")
print(f"  الشهر 6: KS={stat6:.4f}, p={p6:.4f} {'(لا انحراف)' if p6 > 0.05 else '(انحراف!)'}")

نظام مراقبة بسيط

class ModelMonitor:
    def __init__(self, training_data, alert_threshold=0.05):
        self.training_stats = {
            col: {'mean': training_data[col].mean(), 'std': training_data[col].std()}
            for col in training_data.columns
        }
        self.alert_threshold = alert_threshold
        self.prediction_log = []

    def check_drift(self, new_data):
        alerts = []
        for col in new_data.columns:
            if col in self.training_stats:
                ref = self.training_stats[col]
                new_mean = new_data[col].mean()
                drift = abs(new_mean - ref['mean']) / ref['std']
                if drift > 2.0:
                    alerts.append(f"انحراف في {col}: {drift:.2f} sigma")
        return alerts

    def log_prediction(self, features, prediction, actual=None):
        self.prediction_log.append({
            'timestamp': pd.Timestamp.now(),
            'prediction': prediction,
            'actual': actual
        })

# استخدام نظام المراقبة
monitor = ModelMonitor(X_train)

# محاكاة بيانات جديدة بها انحراف
new_batch = X_test.copy()
new_batch['vibration'] += 1.0  # محاكاة تغير

alerts = monitor.check_drift(new_batch)
if alerts:
    print("تنبيهات انحراف النموذج:")
    for alert in alerts:
        print(f"  {alert}")
    print("\nالتوصية: إعادة تدريب النموذج على بيانات حديثة")
else:
    print("لا يوجد انحراف ملحوظ")

مثال عملي: نشر نموذج كشف عيوب كخدمة REST

سنجمع كل المكونات في مشروع نشر متكامل:

import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
import pickle
import json

# 1. بناء وتدريب النموذج النهائي
np.random.seed(42)
n = 5000
X = pd.DataFrame({
    'vibration_rms': np.random.normal(2.5, 0.5, n),
    'temperature': np.random.normal(70, 5, n),
    'current_draw': np.random.normal(15, 2, n),
    'sound_level': np.random.normal(75, 3, n),
    'oil_pressure': np.random.normal(45, 3, n)
})
y = ((X['vibration_rms'] > 3.5) | (X['temperature'] > 80) |
     (X['oil_pressure'] < 38)).astype(int)

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('model', RandomForestClassifier(n_estimators=150, max_depth=8, random_state=42))
])
pipeline.fit(X_train, y_train)

# 2. تقييم نهائي
y_pred = pipeline.predict(X_test)
print("التقييم النهائي قبل النشر:")
print(classification_report(y_test, y_pred, target_names=['سليم', 'معيب']))

# 3. حفظ النموذج مع البيانات الوصفية
model_package = {
    'model': pipeline,
    'feature_names': list(X.columns),
    'version': '1.0.0',
    'training_date': '2025-03-15',
    'training_samples': len(X_train),
    'test_metrics': {
        'accuracy': float(pipeline.score(X_test, y_test)),
        'feature_ranges': {
            col: {'min': float(X[col].min()), 'max': float(X[col].max())}
            for col in X.columns
        }
    }
}

with open('defect_detector_v1.pkl', 'wb') as f:
    pickle.dump(model_package, f)

# 4. محاكاة استخدام في الإنتاج
print("\n--- محاكاة الاستخدام في الإنتاج ---")
with open('defect_detector_v1.pkl', 'rb') as f:
    package = pickle.load(f)

model = package['model']
print(f"النموذج: الإصدار {package['version']}")
print(f"تاريخ التدريب: {package['training_date']}")

# تنبؤ على قراءة جديدة
new_reading = pd.DataFrame([{
    'vibration_rms': 4.1, 'temperature': 72,
    'current_draw': 16.5, 'sound_level': 78, 'oil_pressure': 43
}])

result = model.predict(new_reading)[0]
proba = model.predict_proba(new_reading)[0]
print(f"\nقراءة جديدة -> الحكم: {'معيب - يتطلب فحص' if result else 'سليم'}")
print(f"مستوى الثقة: {max(proba)*100:.1f}%")

الخلاصة

نشر نماذج التعلم الآلي في البيئة الصناعية يتطلب أكثر من مجرد كود تدريب. تعلمنا تصدير النماذج، وبناء واجهات API، وتغليفها في حاويات Docker، ومراقبة أدائها بعد النشر. هذا الدرس يختتم السلسلة التي أخذتنا من أساسيات Python إلى نشر نماذج تعلم آلي صناعية حقيقية. الخطوة التالية هي تطبيق ما تعلمته على بيانات مصنعك الحقيقية.

deployment API model-export monitoring MLOps production النشر واجهة البرمجة تصدير النموذج المراقبة الإنتاج التشغيل المستمر
التالي تقييم النماذج: كيف تعرف أن نموذجك يعمل فعلاً؟