Aller au contenu principal

Module 9 — Vertex AI Pipelines

Jusqu'ici, chaque module a été piloté par un carnet et un humain : lancer l'entraînement, attendre, uploader le modèle, déployer, planifier le lot. À la longue, cela n'est ni reproductible ni auditable. Vertex AI Pipelines transforme cette suite de gestes en un graphe déclaratif, versionnable, planifiable, où chaque étape trace ce qu'elle a consommé et ce qu'elle a produit.

Kubeflow Pipelines, sur infrastructure Google​

Vertex AI Pipelines exécute des pipelines écrits avec kfp (Kubeflow Pipelines), un SDK Python. On décrit les étapes, on compile en JSON, on soumet à Vertex qui provisionne les ressources, exécute et journalise. Aucune infrastructure Kubeflow à gérer.

Deux concepts fondent le cadre :

  • Composant : une unité de traitement, avec des entrées et des sorties typées. Un composant tourne dans son propre conteneur.
  • Pipeline : un graphe de composants, avec les dépendances de données déduites automatiquement du flux des sorties vers les entrées.

Un premier composant​

Un composant Python léger se déclare avec un décorateur :

from kfp import dsl
from kfp.dsl import Dataset, Model, Output, Input

@dsl.component(
    base_image="europe-west1-docker.pkg.dev/paiement-fraude-prod/vertex/fraude-xgb:2026-09-06",
)
def entrainer(
    donnees: Input[Dataset],
    n_estimators: int,
    modele_sortie: Output[Model],
) -> float:
    import xgboost as xgb, pandas as pd, joblib, os
    from sklearn.metrics import roc_auc_score

    df = pd.read_parquet(donnees.path + "/train.parquet")
    y = df.pop("est_fraude")
    modele = xgb.XGBClassifier(n_estimators=n_estimators, tree_method="hist").fit(df, y)

    os.makedirs(modele_sortie.path, exist_ok=True)
    joblib.dump(modele, modele_sortie.path + "/model.joblib")
    return roc_auc_score(y, modele.predict_proba(df)[:, 1])

Input[Dataset] et Output[Model] ne sont pas de simples annotations : ce sont les types d'artefacts que le pipeline tracera. Chaque artefact reçoit un URI GCS attribué automatiquement, un hash pour la déduplication, et une position dans le graphe de lignage.

Le pipeline​

On assemble les composants dans un pipeline, également décoré :

from kfp import dsl, compiler

@dsl.pipeline(
    name="fraude-hebdo",
    description="Extraction, entraînement, évaluation et déploiement conditionnel.",
    pipeline_root="gs://paiement-fraude-modeles/pipelines/",
)
def pipeline_fraude(
    date_debut: str = "2026-06-01",
    date_fin: str = "2026-08-31",
    seuil_aucpr: float = 0.70,
):
    extraction_op = extraire_donnees(date_debut=date_debut, date_fin=date_fin)
    entrainement_op = entrainer(
        donnees=extraction_op.outputs["donnees"],
        n_estimators=800,
    )
    evaluation_op = evaluer(modele=entrainement_op.outputs["modele_sortie"])

    with dsl.If(evaluation_op.outputs["aucpr"] > seuil_aucpr, name="deploiement-conditionnel"):
        upload_op = uploader_modele(
            modele=entrainement_op.outputs["modele_sortie"],
            evaluation=evaluation_op.outputs["metriques"],
        )
        deployer_endpoint(modele=upload_op.outputs["model_resource"], trafic=10)

compiler.Compiler().compile(pipeline_fraude, "fraude-hebdo.json")

Trois choses valent d'être relevées. Le graphe des dépendances se déduit du flux .outputs["..."] d'un composant vers l'entrée d'un autre — on ne l'écrit pas explicitement. La dsl.If(...) n'exécute la branche que si la condition est vraie au moment de l'exécution, pas à la compilation : la sortie aucpr est un artefact, sa valeur n'existe qu'après l'exécution du composant evaluer. Enfin, la compilation produit un JSON auto-suffisant : versionnable dans Git, envoyable à Vertex sans le code Python.

Exécuter et planifier​

from google.cloud import aiplatform

job = aiplatform.PipelineJob(
    display_name="fraude-hebdo-2026-09-06",
    template_path="fraude-hebdo.json",
    pipeline_root="gs://paiement-fraude-modeles/pipelines/",
    parameter_values={"date_debut": "2026-06-01", "date_fin": "2026-08-31"},
    enable_caching=True,
)

job.submit(service_account="vertex-fraude-runner@paiement-fraude-prod.iam.gserviceaccount.com")

enable_caching=True est le levier d'économie principal. Si un composant s'exécute avec des entrées identiques (mêmes valeurs, mêmes artefacts d'entrée par hash), Vertex renvoie directement le résultat cache sans le rejouer. Sur un développement où l'on modifie seulement la dernière étape, seuls les composants aval se rejouent — quelques minutes au lieu d'une heure.

Pour la production, on planifie :

schedule = job.create_schedule(
    display_name="fraude-hebdo-lundi-3h",
    cron="0 3 * * MON",
    max_concurrent_run_count=1,
    max_run_count=None,
)

Chaque lundi 3 h, un nouveau PipelineRun démarre. max_concurrent_run_count=1 empêche qu'un second lundi lance un run pendant que le précédent tourne encore (rare mais possible).

Le lignage, sans effort​

C'est ici que Vertex Pipelines gagne son prix. Chaque artefact produit par un composant est enregistré dans Vertex ML Metadata avec :

  • son URI GCS et son hash de contenu,
  • le run et le composant qui l'ont produit,
  • les paramètres passés (date_debut, n_estimators),
  • les artefacts consommés en amont.

Ouvrir le graphe de lignage revient à ouvrir l'onglet « Lineage » de la console Vertex. À gauche, la table BigQuery source ; à droite, le modèle déployé. Entre les deux, chaque composant est un nœud cliquable qui expose ses paramètres et l'ID du conteneur qui l'a exécuté.

En audit, la question « à partir de quelles données ce modèle a-t-il été formé le 12 février 2026 ? » se répond en quinze secondes : on ouvre le run de cette date, on clique sur le composant extraire_donnees, on lit la requête BigQuery exacte qui a été exécutée. Ce niveau de traçabilité est ce qui distingue un pipeline Vertex d'un cron sur une VM.

Composants prédéfinis Google​

Google fournit une bibliothèque de composants prêts à l'emploi dans google-cloud-pipeline-components : CustomTrainingJobOp, ModelUploadOp, EndpointCreateOp, ModelDeployOp, BatchPredictionJobOp. Notre pipeline de production les utilise plutôt que des composants maison — moins de code, une meilleure intégration avec le lignage.

from google_cloud_pipeline_components.v1.model import ModelUploadOp

ModelUploadOp(
    project="paiement-fraude-prod",
    display_name="fraude-xgb-europe",
    parent_model=parent_model_arg,
    artifact_uri=entrainement_op.outputs["modele_sortie"].uri,
    serving_container_image_uri="europe-docker.pkg.dev/vertex-ai/prediction/sklearn-cpu.1-3:latest",
)

Déboguer un pipeline​

Deux règles évitent le plus gros des heures perdues.

D'abord, tester chaque composant en local avant de l'assembler. Un composant @dsl.component reste une fonction Python appelable : entrainer(donnees=obj_local, n_estimators=100) s'exécute sans Vertex. Beaucoup d'erreurs (dépendance manquante, mauvaise clé) sortent alors en trente secondes.

Ensuite, lire les logs par composant dans la console Vertex Pipelines. Chaque nœud du graphe ouvre les journaux du conteneur qui l'a exécuté. Une erreur du type « ImportError : No module named xgboost » indique que l'image base_image du composant ne contient pas ce paquet — pas que le pipeline est cassé.

Une paramétrisation par le run, pas par le code

Toute valeur qui change d'un run à l'autre — plage de dates, seuil, hyperparamètre — doit être un paramètre du pipeline, pas une constante dans le code. Cela permet de rejouer un run avec parameter_values modifiés (« refais l'entraînement sur juillet uniquement ») sans recompiler le JSON, sans revalider en revue de code. La discipline coûte trois lignes ; l'absence de discipline coûte des jours quand un incident demande de rejouer.

En résumé​

  • Un composant kfp est une fonction typée qui tourne dans un conteneur ; un pipeline est un graphe de composants.
  • Le graphe est déduit du flux .outputs ; la compilation produit un JSON versionnable dans Git.
  • enable_caching=True rejoue seulement ce qui a changé, gain massif en développement.
  • dsl.If(...) conditionne le déploiement à l'évaluation, sans écrire de branche « on n'a pas déployé » ailleurs.
  • Chaque artefact est enregistré dans Vertex ML Metadata ; le lignage complet est disponible sans code supplémentaire.

Module suivant : nous quittons les modèles tabulaires classiques pour convoquer un modèle de fondation du Model Garden sur les commentaires de litige.