الفصل 3: دورة حياة تعلم الآلة
مقدمة
تشمل دورة حياة تعلم الآلة جميع المراحل من تعريف المشكلة إلى إيقاف النموذج. فهم هذه الدورة هو أساس لتنفيذ ممارسات MLOps الفعّالة.
دورة حياة تعلم الآلة الكاملة
المرحلة 1: تعريف المشكلة
قبل البدء في أي عمل متعلق بتعلم الآلة، يجب تحديد المشكلة بوضوح.
الأسئلة الرئيسية للإجابة عليها
| السؤال | مثال للإجابة |
|---|---|
| ما هي المشكلة التجارية التي نحاول حلها؟ | تقليل فقدان العملاء بنسبة 15% |
| ما نوع مشكلة تعلم الآلة هذه؟ | تصنيف ثنائي |
| ما هو مقياس النجاح؟ | AUC-ROC > 0.85 |
| ما هي متطلبات التأخير؟ | < 100ms للتنبؤات |
| ما هي البيانات المتاحة؟ | سنتان من بيانات العملاء |
وثيقة تعريف المشكلة
# ml_problem_definition.yaml
project_name: Customer Churn Prediction
version: 1.0
business_context:
objective: Predict customers likely to churn within 30 days
impact: $2M annual revenue at risk
stakeholders:
- Customer Success Team
- Marketing Team
- Product Team
ml_formulation:
task_type: binary_classification
target_variable: churned_within_30_days
positive_class_definition: "Customer cancelled subscription"
success_criteria:
primary_metric: auc_roc
threshold: 0.85
secondary_metrics:
- precision_at_10_percent_recall: 0.70
- f1_score: 0.75
constraints:
latency_p99_ms: 100
model_size_mb: 500
interpretability: high # Need to explain predictions
timeline:
mvp: 2024-03-01
production: 2024-04-01
المرحلة 2: جمع البيانات
جمع وفهم البيانات المتاحة لمشروع تعلم الآلة.
جرد البيانات
# data_inventory.py
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime
@dataclass
class DataSource:
name: str
location: str
format: str
owner: str
update_frequency: str
size_gb: float
date_range_start: datetime
date_range_end: datetime
pii_columns: List[str]
quality_score: float # 0-1
# Define available data sources
data_sources = [
DataSource(
name="customer_transactions",
location="s3://data-lake/transactions/",
format="parquet",
owner="data-engineering",
update_frequency="daily",
size_gb=150.0,
date_range_start=datetime(2022, 1, 1),
date_range_end=datetime(2024, 1, 1),
pii_columns=["email", "phone"],
quality_score=0.95
),
DataSource(
name="customer_support_tickets",
location="s3://data-lake/support/",
format="json",
owner="customer-success",
update_frequency="hourly",
size_gb=25.0,
date_range_start=datetime(2022, 6, 1),
date_range_end=datetime(2024, 1, 1),
pii_columns=["customer_email"],
quality_score=0.88
),
]
خط أنابيب جمع البيانات
# data_collection_pipeline.py
from datetime import datetime, timedelta
import pandas as pd
from prefect import flow, task
@task
def extract_transactions(start_date: datetime, end_date: datetime) -> pd.DataFrame:
"""Extract transaction data from data warehouse."""
query = f"""
SELECT
customer_id,
transaction_date,
amount,
product_category
FROM transactions
WHERE transaction_date BETWEEN '{start_date}' AND '{end_date}'
"""
return pd.read_sql(query, connection)
@task
def extract_support_tickets(start_date: datetime, end_date: datetime) -> pd.DataFrame:
"""Extract support ticket data."""
query = f"""
SELECT
customer_id,
ticket_date,
category,
resolution_time_hours,
satisfaction_score
FROM support_tickets
WHERE ticket_date BETWEEN '{start_date}' AND '{end_date}'
"""
return pd.read_sql(query, connection)
@task
def merge_data(transactions: pd.DataFrame, tickets: pd.DataFrame) -> pd.DataFrame:
"""Merge all data sources."""
return transactions.merge(
tickets,
on='customer_id',
how='left'
)
@flow(name="data-collection")
def collect_training_data(lookback_days: int = 365):
end_date = datetime.now()
start_date = end_date - timedelta(days=lookback_days)
transactions = extract_transactions(start_date, end_date)
tickets = extract_support_tickets(start_date, end_date)
merged = merge_data(transactions, tickets)
return merged
المرحلة 3: إعداد البيانات
تنظيف وإعداد البيانات لهندسة الميزات والتدريب.
فحوصات جودة البيانات
# data_quality.py
import great_expectations as gx
def validate_training_data(df):
"""Validate training data quality."""
context = gx.get_context()
# Create expectations
expectations = [
# No null values in critical columns
{"expectation_type": "expect_column_values_to_not_be_null",
"kwargs": {"column": "customer_id"}},
# Values in expected range
{"expectation_type": "expect_column_values_to_be_between",
"kwargs": {"column": "age", "min_value": 18, "max_value": 120}},
# Unique identifiers
{"expectation_type": "expect_column_values_to_be_unique",
"kwargs": {"column": "customer_id"}},
# No future dates
{"expectation_type": "expect_column_values_to_be_between",
"kwargs": {"column": "signup_date",
"max_value": datetime.now().isoformat()}},
# Column exists
{"expectation_type": "expect_column_to_exist",
"kwargs": {"column": "churned"}},
]
# Run validation
results = context.run_validation_operator(
"action_list_operator",
assets_to_validate=[df],
run_id=f"training_data_{datetime.now().isoformat()}"
)
return results.success
خط أنابيب تنظيف البيانات
# data_cleaning.py
import pandas as pd
import numpy as np
def clean_data(df: pd.DataFrame) -> pd.DataFrame:
"""Clean and prepare raw data."""
# Make a copy
df = df.copy()
# Handle missing values
df['age'] = df['age'].fillna(df['age'].median())
df['income'] = df['income'].fillna(df.groupby('segment')['income'].transform('median'))
# Remove duplicates
df = df.drop_duplicates(subset=['customer_id'], keep='last')
# Handle outliers (clip to percentiles)
for col in ['transaction_amount', 'login_frequency']:
lower = df[col].quantile(0.01)
upper = df[col].quantile(0.99)
df[col] = df[col].clip(lower, upper)
# Standardize text columns
df['segment'] = df['segment'].str.lower().str.strip()
# Convert data types
df['signup_date'] = pd.to_datetime(df['signup_date'])
df['churned'] = df['churned'].astype(int)
return df
المرحلة 4: هندسة الميزات
إنشاء ميزات تساعد النموذج على تعلم الأنماط.
خط أنابيب هندسة الميزات
# feature_engineering.py
import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler, LabelEncoder
from datetime import datetime
class FeatureEngineer:
def __init__(self):
self.scaler = StandardScaler()
self.encoders = {}
def create_features(self, df: pd.DataFrame) -> pd.DataFrame:
"""Create all features for the model."""
features = pd.DataFrame()
features['customer_id'] = df['customer_id']
# Temporal features
features['account_age_days'] = (datetime.now() - df['signup_date']).dt.days
features['days_since_last_login'] = (datetime.now() - df['last_login']).dt.days
features['days_since_last_purchase'] = (datetime.now() - df['last_purchase']).dt.days
# Aggregation features
features['total_purchases'] = df['purchase_count']
features['total_spend'] = df['total_spend']
features['avg_order_value'] = df['total_spend'] / df['purchase_count'].clip(1)
# Engagement features
features['login_frequency_30d'] = df['logins_last_30d']
features['support_tickets_90d'] = df['support_tickets_90d']
features['nps_score'] = df['nps_score']
# Ratio features
features['spend_to_income_ratio'] = df['total_spend'] / df['income'].clip(1)
features['engagement_trend'] = (
df['logins_last_30d'] / df['logins_prev_30d'].clip(1)
)
# Categorical encoding
for col in ['segment', 'plan_type', 'channel']:
if col not in self.encoders:
self.encoders[col] = LabelEncoder()
features[f'{col}_encoded'] = self.encoders[col].fit_transform(df[col])
else:
features[f'{col}_encoded'] = self.encoders[col].transform(df[col])
return features
def scale_features(self, features: pd.DataFrame, fit: bool = False) -> pd.DataFrame:
"""Scale numerical features."""
numerical_cols = features.select_dtypes(include=[np.number]).columns
numerical_cols = [c for c in numerical_cols if c != 'customer_id']
if fit:
features[numerical_cols] = self.scaler.fit_transform(features[numerical_cols])
else:
features[numerical_cols] = self.scaler.transform(features[numerical_cols])
return features
المرحلة 5: تدريب النموذج
تدريب وضبط النماذج مع تتبع التجارب بشكل صحيح.
خط أنابيب التدريب باستخدام MLflow
# training_pipeline.py
import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.model_selection import cross_val_score, GridSearchCV
from sklearn.metrics import roc_auc_score, precision_recall_curve
import pandas as pd
import numpy as np
class ModelTrainer:
def __init__(self, experiment_name: str):
mlflow.set_experiment(experiment_name)
self.best_model = None
self.best_score = 0
def train_and_evaluate(
self,
X_train: pd.DataFrame,
y_train: pd.Series,
X_val: pd.DataFrame,
y_val: pd.Series
):
"""Train multiple models and select the best one."""
models = {
'random_forest': {
'model': RandomForestClassifier(random_state=42),
'params': {
'n_estimators': [100, 200],
'max_depth': [10, 20, None],
'min_samples_split': [2, 5]
}
},
'gradient_boosting': {
'model': GradientBoostingClassifier(random_state=42),
'params': {
'n_estimators': [100, 200],
'max_depth': [3, 5, 7],
'learning_rate': [0.01, 0.1]
}
}
}
for model_name, config in models.items():
with mlflow.start_run(run_name=model_name):
# Log parameters
mlflow.log_param("model_type", model_name)
# Grid search for best hyperparameters
grid_search = GridSearchCV(
config['model'],
config['params'],
cv=5,
scoring='roc_auc',
n_jobs=-1
)
grid_search.fit(X_train, y_train)
# Get best model
best_estimator = grid_search.best_estimator_
# Log best parameters
for param, value in grid_search.best_params_.items():
mlflow.log_param(param, value)
# Evaluate on validation set
val_predictions = best_estimator.predict_proba(X_val)[:, 1]
val_auc = roc_auc_score(y_val, val_predictions)
# Log metrics
mlflow.log_metric("val_auc_roc", val_auc)
mlflow.log_metric("cv_score", grid_search.best_score_)
# Log model
mlflow.sklearn.log_model(best_estimator, "model")
# Track best model
if val_auc > self.best_score:
self.best_score = val_auc
self.best_model = best_estimator
print(f"{model_name}: AUC-ROC = {val_auc:.4f}")
return self.best_model
المرحلة 6: تقييم النموذج
تقييم شامل قبل النشر.
إطار التقييم
# model_evaluation.py
import matplotlib.pyplot as plt
from sklearn.metrics import (
roc_auc_score, precision_recall_curve,
confusion_matrix, classification_report
)
import numpy as np
class ModelEvaluator:
def __init__(self, model, X_test, y_test):
self.model = model
self.X_test = X_test
self.y_test = y_test
self.predictions = model.predict(X_test)
self.probabilities = model.predict_proba(X_test)[:, 1]
def generate_report(self) -> dict:
"""Generate comprehensive evaluation report."""
report = {
'auc_roc': roc_auc_score(self.y_test, self.probabilities),
'classification_report': classification_report(
self.y_test,
self.predictions,
output_dict=True
),
'confusion_matrix': confusion_matrix(
self.y_test,
self.predictions
).tolist()
}
# Calculate precision at different recall thresholds
precision, recall, thresholds = precision_recall_curve(
self.y_test,
self.probabilities
)
report['precision_at_recall_80'] = precision[
np.argmin(np.abs(recall - 0.8))
]
return report
def check_deployment_criteria(self, criteria: dict) -> bool:
"""Check if model meets deployment criteria."""
report = self.generate_report()
checks = []
for metric, threshold in criteria.items():
if metric == 'auc_roc':
passed = report['auc_roc'] >= threshold
elif metric == 'precision':
passed = report['classification_report']['1']['precision'] >= threshold
elif metric == 'recall':
passed = report['classification_report']['1']['recall'] >= threshold
else:
passed = True
checks.append(passed)
print(f"{metric}: {'PASS' if passed else 'FAIL'}")
return all(checks)
المرحلة 7: نشر النموذج
نشر النموذج في الإنتاج.
تكوين النشر
# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: churn-model
labels:
app: churn-prediction
spec:
replicas: 3
selector:
matchLabels:
app: churn-prediction
template:
metadata:
labels:
app: churn-prediction
spec:
containers:
- name: model-server
image: mlplatform/churn-model:v2.1.0
ports:
- containerPort: 8080
resources:
requests:
memory: "2Gi"
cpu: "1"
limits:
memory: "4Gi"
cpu: "2"
env:
- name: MODEL_NAME
value: "churn-model"
- name: MODEL_VERSION
value: "2.1.0"
livenessProbe:
httpGet:
path: /health
port: 8080
initialDelaySeconds: 30
periodSeconds: 10
readinessProbe:
httpGet:
path: /ready
port: 8080
initialDelaySeconds: 5
periodSeconds: 5
المرحلة 8: المراقبة
المراقبة المستمرة لأداء النموذج.
مقاييس المراقبة
# monitoring.py
from prometheus_client import Counter, Histogram, Gauge
from datetime import datetime
import numpy as np
# Define metrics
prediction_counter = Counter(
'model_predictions_total',
'Total predictions made',
['model_version', 'prediction_class']
)
prediction_latency = Histogram(
'model_prediction_latency_seconds',
'Prediction latency',
buckets=[0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5]
)
data_drift_gauge = Gauge(
'model_data_drift_score',
'Data drift score by feature',
['feature_name']
)
model_accuracy_gauge = Gauge(
'model_accuracy_rolling',
'Rolling accuracy over last 24 hours'
)
class ModelMonitor:
def __init__(self, model_version: str):
self.model_version = model_version
self.predictions_log = []
def log_prediction(self, input_data: dict, prediction: int, latency: float):
"""Log a prediction for monitoring."""
prediction_counter.labels(
model_version=self.model_version,
prediction_class=str(prediction)
).inc()
prediction_latency.observe(latency)
self.predictions_log.append({
'timestamp': datetime.now(),
'input': input_data,
'prediction': prediction,
'latency': latency
})
def check_drift(self, current_data: np.ndarray, reference_data: np.ndarray):
"""Check for data drift using statistical tests."""
from scipy import stats
drift_scores = {}
for i, feature in enumerate(feature_names):
# Kolmogorov-Smirnov test
statistic, p_value = stats.ks_2samp(
reference_data[:, i],
current_data[:, i]
)
drift_scores[feature] = statistic
data_drift_gauge.labels(feature_name=feature).set(statistic)
return drift_scores
المرحلة 9: إعادة التدريب
تشغيل إعادة تدريب النموذج عند الحاجة.
محفزات إعادة التدريب
# retraining_triggers.py
from dataclasses import dataclass
from enum import Enum
from typing import Optional
from datetime import datetime, timedelta
class TriggerType(Enum):
SCHEDULED = "scheduled"
PERFORMANCE = "performance"
DRIFT = "drift"
DATA_VOLUME = "data_volume"
MANUAL = "manual"
@dataclass
class RetrainingTrigger:
trigger_type: TriggerType
threshold: Optional[float] = None
schedule: Optional[str] = None
def check_retraining_needed(
current_auc: float,
baseline_auc: float,
drift_score: float,
last_training_date: datetime,
new_data_count: int
) -> tuple[bool, TriggerType]:
"""Check if retraining is needed."""
triggers = [
# Performance degradation
RetrainingTrigger(
trigger_type=TriggerType.PERFORMANCE,
threshold=0.05 # 5% AUC drop
),
# Data drift
RetrainingTrigger(
trigger_type=TriggerType.DRIFT,
threshold=0.1 # Drift score > 0.1
),
# Scheduled retraining
RetrainingTrigger(
trigger_type=TriggerType.SCHEDULED,
schedule="30 days"
),
# New data volume
RetrainingTrigger(
trigger_type=TriggerType.DATA_VOLUME,
threshold=100000 # 100k new records
),
]
# Check performance
if baseline_auc - current_auc > 0.05:
return True, TriggerType.PERFORMANCE
# Check drift
if drift_score > 0.1:
return True, TriggerType.DRIFT
# Check schedule
if datetime.now() - last_training_date > timedelta(days=30):
return True, TriggerType.SCHEDULED
# Check data volume
if new_data_count > 100000:
return True, TriggerType.DATA_VOLUME
return False, None
الدروس المستفادة
ملخص
- تعريف المشكلة أمر حاسم - الأهداف غير الواضحة تؤدي إلى فشل المشاريع
- جودة البيانات تحدد جودة النموذج - القمامة في، القمامة خارج
- هندسة الميزات غالبًا ما تكون أكثر أهمية من اختيار النموذج
- التقييم يجب أن يكون شاملاً ويتوافق مع مقاييس العمل
- المراقبة ليست اختيارية - النماذج تتدهور بدونها
- إعادة التدريب تبقي النماذج ذات صلة مع تطور البيانات
قراءة إضافية
التالي: اختبار الوحدة 1