Modèles d'IA et progrès algorithmiques

Dans tout système MLOps de production que je construis, la gouvernance des modèles est primordiale. Il ne s'agit pas seulement de suivre la lignée, mais aussi d'appliquer la conformité et la sécurité.

• Contrôle de version : Imposer Git pour tout le code de pipeline et utiliser le Registre de modèles pour le versionnement des binaires de modèles et de leurs métadonnées associées.
• Journalisation d'audit : Tous les événements du cycle de vie — exécutions d'entraînement, déploiements, changements d'étape, journaux de prédiction — doivent être capturés. GCP et Azure fournissent une journalisation étendue et non répudiable à cette fin.
• Contrôle d'accès : Utiliser des politiques IAM strictes pour définir qui peut entraîner, enregistrer, déployer ou interroger des modèles. Cela suit le principe du moindre privilège.
• Analyse de modèles : Une pratique en évolution mais critique est l'intégration d'analyses de sécurité pour les artefacts de modèles afin de détecter les vulnérabilités ou les données sensibles intégrées. Cela devrait faire partie de votre pipeline CI/CD.

Voici un aperçu pratique de la façon dont une seule étape d'entraînement devient un actif codifié et versionné dans un pipeline Vertex AI à l'aide du SDK Kubeflow Pipelines (kfp).

# pipeline_components.py
from kfp.v2.dsl import pipeline, component
from kfp.compiler import Compiler

# Define a custom component for model training
@component(packages_to_install=['scikit-learn', 'pandas', 'google-cloud-aiplatform'])
def train_model_component(
    dataset_uri: str,
    model_output_uri: str,
    project_id: str,
    region: str = 'europe-west1'
):
    import pandas as pd
    from sklearn.model_selection import train_test_split
    from sklearn.linear_model import LogisticRegression
    from sklearn.metrics import accuracy_score
    import google.cloud.aiplatform as aiplatform
    import logging

    logging.basicConfig(level=logging.INFO)
    aiplatform.init(project=project_id, location=region)

    logging.info(f"Loading dataset from: {dataset_uri}")
    # In a real scenario, this would involve reading from GCS or BigQuery
    # For simplicity, we'll simulate a dataset
    data = pd.DataFrame({
        'feature1': [i for i in range(100)],
        'feature2': [i * 2 for i in range(100)],
        'target': [0 if i < 50 else 1 for i in range(100)]
    })

    X = data[['feature1', 'feature2']]
    y = data['target']
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

    logging.info("Training Logistic Regression model...")
    model = LogisticRegression()
    model.fit(X_train, y_train)
    predictions = model.predict(X_test)
    accuracy = accuracy_score(y_test, predictions)
    logging.info(f"Model trained with accuracy: {accuracy}")

    # Simulate saving model to GCS for later registration
    # In a real scenario, you'd use job.log_artifact or save to GCS
    # For this example, we just 'log' the output URI
    logging.info(f"Model artifact would be saved to: {model_output_uri}/model.pkl")

# This component would then be used inside a larger pipeline definition.
# For example:
# @pipeline(name='simple-mlops-pipeline', pipeline_root='gs://your-bucket/pipeline_root')
# def ml_pipeline(project_id: str):
#    train_task = train_model_component(dataset_uri='gs://my-data/dataset.csv',
#                                       model_output_uri='gs://my-models',
#                                       project_id=project_id)
# Compiler().compile(pipeline_func=ml_pipeline, package_path='simple_mlops_pipeline.json')

Ce composant simple encapsule la logique d'entraînement, en faisant une étape réutilisable et versionnable dans un flux de travail automatisé plus large.

Tout d'abord, créez un compte de service dédié aux pipelines MLOps pour appliquer le principe du moindre privilège.

# Set environment variables. Replace 'your-gcp-project-id' with your actual project ID.
export GCP_PROJECT_ID="your-gcp-project-id"
export GCP_REGION="europe-west1"

# Enable the Vertex AI API
gcloud services enable aiplatform.googleapis.com --project=$GCP_PROJECT_ID

# Create a service account for your MLOps pipeline
export SERVICE_ACCOUNT_NAME="mlops-pipeline-sa"
export SERVICE_ACCOUNT_EMAIL="${SERVICE_ACCOUNT_NAME}@${GCP_PROJECT_ID}.iam.gserviceaccount.com"

gcloud iam service-accounts create $SERVICE_ACCOUNT_NAME \
    --display-name="Service Account for MLOps Pipelines" \
    --project=$GCP_PROJECT_ID

# Grant necessary roles. For production, use more granular roles than storage.objectAdmin.
gcloud projects add-iam-policy-binding $GCP_PROJECT_ID \
    --member="serviceAccount:${SERVICE_ACCOUNT_EMAIL}" \
    --role="roles/aiplatform.user"

gcloud projects add-iam-policy-binding $GCP_PROJECT_ID \
    --member="serviceAccount:${SERVICE_ACCOUNT_EMAIL}" \
    --role="roles/storage.objectAdmin"

Cette configuration unique crée l'identité et les autorisations que notre flux de travail automatisé utilisera.

Ce script simule une exécution d'entraînement locale, puis télécharge l'artefact résultant vers Cloud Storage et l'enregistre dans le Registre de modèles Vertex AI. Il est conçu pour être idempotent, gérant les cas où des ressources comme les buckets GCS existent déjà.

# train_and_register.py
import os
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score
import joblib

import google.cloud.aiplatform as aiplatform
from google.cloud import storage
from google.api_core import exceptions

# --- Configuration ---
# These are read from environment variables set in your shell.
PROJECT_ID = os.environ.get('GCP_PROJECT_ID')
REGION = os.environ.get('GCP_REGION', 'europe-west1')

if not PROJECT_ID:
    raise ValueError("GCP_PROJECT_ID environment variable not set.")

BUCKET_NAME = f"{PROJECT_ID}-mlops-models-eu"
MODEL_DISPLAY_NAME = "MyFraudDetectionModel"
MODEL_DESCRIPTION = "A simple RandomForestClassifier for fraud detection"

# Initialize Vertex AI SDK
aiplatform.init(project=PROJECT_ID, location=REGION)

# --- Simulate Data ---
# In a real project, you'd fetch data from BigQuery or GCS.
data = pd.DataFrame({
    'transaction_amount': [100, 200, 50, 1000, 75, 500, 120, 800, 30, 600],
    'num_items': [1, 2, 1, 5, 1, 3, 1, 4, 1, 2],
    'is_fraud': [0, 0, 0, 1, 0, 1, 0, 1, 0, 1]
})
X = data[['transaction_amount', 'num_items']]
y = data['is_fraud']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# --- Train Model ---
print("Training RandomForestClassifier...")
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)
predictions = model.predict(X_test)
accuracy = accuracy_score(y_test, predictions)
print(f"Model trained with accuracy: {accuracy:.2f}")

# --- Save Model Artifact Locally ---
local_model_path = "model.joblib"
joblib.dump(model, local_model_path)
print(f"Model saved locally to {local_model_path}")

# --- Upload Model to GCS ---
storage_client = storage.Client(project=PROJECT_ID)
bucket_uri = f"gs://{BUCKET_NAME}"
try:
    bucket = storage_client.create_bucket(BUCKET_NAME, location=REGION)
    print(f"Created GCS bucket: {bucket_uri}")
except exceptions.Conflict:
    print(f"GCS bucket {bucket_uri} already exists.")
    bucket = storage_client.get_bucket(BUCKET_NAME)

model_filename = "model.joblib"
artifact_directory = f"{MODEL_DISPLAY_NAME}"
blob = bucket.blob(f"{artifact_directory}/{model_filename}")
blob.upload_from_filename(local_model_path)
gcs_artifact_path = f"{bucket_uri}/{artifact_directory}"
print(f"Model artifact uploaded to GCS: {gcs_artifact_path}")

# --- Register Model in Vertex AI Model Registry ---
# Using a pre-built scikit-learn container for serving.
SERVING_CONTAINER_IMAGE = f"{REGION}-docker.pkg.dev/cloud-aiplatform/prediction/sklearn-cpu.1-0:latest"

# Check if model already exists to upload a new version.
existing_models = aiplatform.Model.list(filter=f'display_name="{MODEL_DISPLAY_NAME}"')
parent_model_resource = None
if existing_models:
    print(f"Model '{MODEL_DISPLAY_NAME}' already exists. Creating a new version.")
    parent_model_resource = existing_models[0].resource_name
else:
    print(f"Creating new model '{MODEL_DISPLAY_NAME}' and registering version 1.")

uploaded_model = aiplatform.Model.upload(
    display_name=MODEL_DISPLAY_NAME,
    artifact_uri=gcs_artifact_path,
    serving_container_image_uri=SERVING_CONTAINER_IMAGE,
    description=MODEL_DESCRIPTION,
    parent_model=parent_model_resource,
    sync=True
)

print(f"Model '{uploaded_model.display_name}' version '{uploaded_model.version_id}' registered.")
print(f"Model resource name: {uploaded_model.resource_name}")

Notez le paramètre parent_model — il est crucial pour créer de nouvelles versions d'un modèle existant, constituant le cœur de votre lignage de modèles.

Une fois enregistré, le déploiement du modèle vers un point de terminaison géré le rend disponible pour des prédictions en temps réel.

# deploy_model.py
import os
import google.cloud.aiplatform as aiplatform

PROJECT_ID = os.environ.get('GCP_PROJECT_ID')
REGION = os.environ.get('GCP_REGION', 'europe-west1')
MODEL_DISPLAY_NAME = "MyFraudDetectionModel"
ENDPOINT_DISPLAY_NAME = "fraud-detection-endpoint"

aiplatform.init(project=PROJECT_ID, location=REGION)

# Retrieve the latest version of the model from the registry.
models = aiplatform.Model.list(filter=f'display_name="{MODEL_DISPLAY_NAME}"', order_by="create_time desc")
if not models:
    raise ValueError(f"Model '{MODEL_DISPLAY_NAME}' not found in registry.")

model_to_deploy = models[0]
print(f"Found model '{model_to_deploy.display_name}' version '{model_to_deploy.version_id}'.")

# Create or get existing endpoint.
endpoints = aiplatform.Endpoint.list(filter=f'display_name="{ENDPOINT_DISPLAY_NAME}"')
if endpoints:
    endpoint = endpoints[0]
    print(f"Using existing endpoint: {endpoint.display_name}")
else:
    print(f"Creating new endpoint: {ENDPOINT_DISPLAY_NAME}")
    endpoint = aiplatform.Endpoint.create(
        display_name=ENDPOINT_DISPLAY_NAME,
        project=PROJECT_ID,
        location=REGION
    )

# Deploy the model. I start with min_replica_count=1 for cost efficiency.
TRAFFIC_SPLIT = {"0": 100} # 100% traffic to the new model version.

endpoint.deploy(
    model=model_to_deploy,
    deployed_model_display_name=f"{MODEL_DISPLAY_NAME}-v{model_to_deploy.version_id}",
    machine_type="n1-standard-2", # Choose a machine type appropriate for your workload.
    min_replica_count=1,
    max_replica_count=2,
    traffic_split=TRAFFIC_SPLIT,
    sync=True
)

print(f"Model '{model_to_deploy.display_name}' version '{model_to_deploy.version_id}' deployed to endpoint '{endpoint.display_name}'.")

Ce script masque l'infrastructure sous-jacente, vous offrant un service d'inférence évolutif et hautement disponible. Le paramètre traffic_split est la façon dont j'implémente les déploiements canary et les tests A/B en production.

Une fois le modèle déployé, vous pouvez maintenant lui envoyer des requêtes d'inférence.

# predict_online.py
import os
import google.cloud.aiplatform as aiplatform

PROJECT_ID = os.environ.get('GCP_PROJECT_ID')
REGION = os.environ.get('GCP_REGION', 'europe-west1')
ENDPOINT_DISPLAY_NAME = "fraud-detection-endpoint"

aiplatform.init(project=PROJECT_ID, location=REGION)

# Get the endpoint object.
endpoints = aiplatform.Endpoint.list(filter=f'display_name="{ENDPOINT_DISPLAY_NAME}"')
if not endpoints:
    raise ValueError(f"Endpoint '{ENDPOINT_DISPLAY_NAME}' not found.")
endpoint = endpoints[0]

# Prepare instances for prediction. The format must match what the model expects.
instances = [
    {"transaction_amount": 150.0, "num_items": 1},
    {"transaction_amount": 900.0, "num_items": 4},
]

print(f"Sending prediction request to endpoint: {endpoint.display_name}")
prediction_response = endpoint.predict(instances=instances)

print("Prediction response:")
for prediction in prediction_response.predictions:
    print(f"  {prediction}")

Portez une attention particulière au format d'entrée. Toute non-concordance entre les fonctionnalités envoyées ici et celles utilisées lors de l'entraînement est une source d'erreur courante.

La surveillance des modèles est non négociable pour l'IA en production. Bien qu'une configuration complète dépasse le cadre de cet article, il est crucial de comprendre le concept. Vous définissez ce qu'il faut surveiller (par exemple, le biais des données, la dérive conceptuelle) et fournissez un ensemble de données de référence (vos données d'entraînement) pour comparaison. Vertex AI enregistre ensuite automatiquement les requêtes de prédiction et vous alerte lorsque les performances dévient.

# configure_monitoring.py (Conceptual)
import os
import google.cloud.aiplatform as aiplatform

PROJECT_ID = os.environ.get('GCP_PROJECT_ID')
REGION = os.environ.get('GCP_REGION', 'europe-west1')
ENDPOINT_DISPLAY_NAME = "fraud-detection-endpoint"

aiplatform.init(project=PROJECT_ID, location=REGION)

endpoints = aiplatform.Endpoint.list(filter=f'display_name="{ENDPOINT_DISPLAY_NAME}"')
if not endpoints:
    raise ValueError(f"Endpoint '{ENDPOINT_DISPLAY_NAME}' not found.")
endpoint = endpoints[0]

# This is a conceptual representation. A real implementation requires:
# 1. A BigQuery table with training data for the baseline schema and distribution.
# 2. Skew and drift detection configurations with specified thresholds.
# 3. Alerting configurations (e.g., email notifications).

# from google.cloud.aiplatform import model_monitoring
# from google.cloud.aiplatform.model_monitoring import
#     SkewDetectionConfig, DriftDetectionConfig, ObjectiveConfig

# skew_config = SkewDetectionConfig(data_source=..., skew_thresholds=...)
# objective_config = ObjectiveConfig(skew_detection_config=skew_config, ...)
# endpoint.update_monitoring(objective_configs=objective_config, ...)

print("Conceptual monitoring configuration initiated.")
print("For a full implementation, refer to the Vertex AI Model Monitoring documentation.")

Cette boucle de rétroaction continue est ce qui rend la Quatrième couche de l'IA robuste et résiliente. Pour une analyse approfondie, je renvoie mes clients à la documentation officielle de Vertex AI Model Monitoring.

# Verify registered models
gcloud ai models list --project=$GCP_PROJECT_ID --region=$GCP_REGION --filter="displayName=$MODEL_DISPLAY_NAME"

# Verify deployed endpoints
gcloud ai endpoints list --project=$GCP_PROJECT_ID --region=$GCP_REGION --filter="displayName=$ENDPOINT_DISPLAY_NAME"

# Describe an endpoint to see its deployed models
# First, get the endpoint ID
ENDPOINT_ID=$(gcloud ai endpoints list --project=$GCP_PROJECT_ID --region=$GCP_REGION --filter="displayName=$ENDPOINT_DISPLAY_NAME" --format="value(name)")
gcloud ai endpoints describe $ENDPOINT_ID --project=$GCP_PROJECT_ID --region=$GCP_REGION

1. Erreur : 403 Permission denied

   • Cause : Presque toujours un problème IAM. Le compte de service (ou votre utilisateur) ne dispose pas des autorisations nécessaires comme roles/aiplatform.user ou roles/storage.objectAdmin.
   • Solution : Vérifiez les rôles accordés à l'entité exécutant le code. Assurez-vous que l'API aiplatform.googleapis.com est activée dans votre projet.

2. Erreur : BucketAlreadyOwnedByYou lors de la configuration

   • Cause : Le bucket GCS que vous tentez de créer existe déjà suite à une exécution précédente.
   • Solution : Ceci est attendu dans les scripts idempotents. Mon exemple train_and_register.py inclut un bloc try-except pour gérer cela gracieusement.

3. Erreur : The model's serving container failed to start lors du déploiement

   • Cause : Une non-concordance entre l'artefact du modèle et le conteneur de service. Les problèmes courants incluent un artifact_uri incorrect (il doit pointer vers le répertoire contenant le modèle), l'utilisation du mauvais conteneur pour votre framework ML (par exemple, un modèle TensorFlow avec un conteneur Scikit-learn), ou un machine_type trop petit.
   • Solution : Vérifiez les journaux du conteneur pour le point de terminaison dans Google Cloud Logging. Les messages d'erreur y sont généralement spécifiques et vous orienteront vers la cause profonde.

J'utilise un script d'orchestration simple comme celui-ci pour tester l'ensemble du pipeline localement.

#!/bin/bash

# Exit immediately if a command exits with a non-zero status.
set -e

# --- Configuration ---
# Make sure this is set to your actual GCP project ID.
export GCP_PROJECT_ID="your-gcp-project-id"
export GCP_REGION="europe-west1"
export MODEL_DISPLAY_NAME="MyFraudDetectionModel"
export ENDPOINT_DISPLAY_NAME="fraud-detection-endpoint"

if [[ -z "$GCP_PROJECT_ID" || "$GCP_PROJECT_ID" == "your-gcp-project-id" ]]; then
    echo "Error: GCP_PROJECT_ID is not set. Please edit the script and set it to your GCP project ID."
    exit 1
fi

# Step 1: Initial setup of service account is manual (see guide).

# Step 2: Train and Register Model
echo "\n--- Running Model Training and Registration ---"
python3.12 train_and_register.py

# Step 3: Deploy Model
echo "\n--- Running Model Deployment ---"
python3.12 deploy_model.py

# Step 4: Perform Online Prediction
echo "\n--- Running Online Prediction ---"
python3.12 predict_online.py

_endpoint_id=$(gcloud ai endpoints list --project=$GCP_PROJECT_ID --region=$GCP_REGION --filter="displayName=$ENDPOINT_DISPLAY_NAME" --format="value(name)")
_model_id=$(gcloud ai models list --project=$GCP_PROJECT_ID --region=$GCP_REGION --filter="displayName=$MODEL_DISPLAY_NAME" --format="value(name)" | head -n 1)

echo "\n--- MLOps Pipeline Execution Complete ---"

# Cleanup (optional - uncomment to undeploy and delete resources after testing)
# echo "\n--- Cleaning up deployed model and endpoint ---"
# DEPLOYED_MODEL_ID=$(gcloud ai endpoints describe ${_endpoint_id} --project=$GCP_PROJECT_ID --region=$GCP_REGION --format='value(deployedModels[0].id)')
# gcloud ai endpoints undeploy-model ${_endpoint_id} --deployed-model-id=${DEPLOYED_MODEL_ID} --project=$GCP_PROJECT_ID --region=$GCP_REGION --quiet
# gcloud ai endpoints delete ${_endpoint_id} --project=$GCP_PROJECT_ID --region=$GCP_REGION --quiet
# echo "--- Cleaning up registered model --- Succeeded"
# gcloud ai models delete ${_model_id} --project=$GCP_PROJECT_ID --region=$GCP_REGION --quiet

• MLOps est fondamental : Traitez les modèles comme des logiciels, avec un cycle de vie automatisé, reproductible et gouvernable.
• Utilisez des plateformes gérées : Tirez parti des services cloud-natifs comme Vertex AI ou Azure Machine Learning pour leurs registres de modèles, pipelines et outils de surveillance intégrés.
• Versionnez et surveillez tout : Chaque modèle, configuration et déploiement doit être suivi. La surveillance continue de la dérive est non négociable.
• Sécurisez dès le départ : Mettez en œuvre un IAM strict et intégrez l'analyse de sécurité dans vos pipelines de déploiement.

Pour faire mûrir votre Quatrième couche de l'IA, je vous recommande de vous concentrer sur les domaines suivants :

1. CI/CD automatisé pour le ML : Intégrez ces scripts à Cloud Build, GitHub Actions ou Azure DevOps pour créer un véritable flux de travail GitOps pour vos modèles.
2. Surveillance avancée des modèles : Allez au-delà de la détection de dérive de base. Explorez les outils d'explicabilité (XAI) et créez des déclencheurs de réentraînement automatisés. Commencez par Vertex AI Explainable AI.
3. Optimisation des coûts et FinOps : Analysez le coût de votre calcul d'entraînement et d'inférence. Utilisez des types de machines appropriés et explorez des options comme les instances réservées pour les charges de travail prévisibles.
4. Intégration d'un Feature Store : Pour les opérations à l'échelle de l'entreprise, l'intégration d'un feature store est la prochaine étape logique pour assurer la cohérence et éliminer les efforts d'ingénierie des données redondants.

En adoptant ces principes MLOps, vous transformez la Quatrième couche de l'IA, souvent opaque, en un moteur transparent, efficace et puissant pour l'innovation de votre organisation.