Exécuter sur Vertex AI Pipelines

Le pipeline tourne sur votre poste ; l'équipe veut qu'il tourne sans votre poste, déclenché par la CI à chaque modification de prompt, avec un lien de console à coller dans la demande de fusion. C'est le rôle de Vertex AI Pipelines. Le passage au cloud ajoute trois sujets que kfp.local cache : où vivent les fichiers, sous quelle identité tournent les étapes, et quelles machines elles consomment.

Le mécanisme

sequenceDiagram
  participant D as Poste ou CI
  participant GCS as Cloud Storage (bucket du lab)
  participant VP as Vertex AI Pipelines
  participant CJ as CustomJob (une par tâche)
  participant G as Gemini (Vertex AI)
  D->>GCS: make cloud-up : données + registre
  D->>VP: make e2e : YAML compilé + paramètres + labels
  VP->>CJ: une machine par tâche, quota « Custom model training CPUs »
  CJ->>GCS: lecture /gcs/<bucket>/data, écriture /gcs/<bucket>/root
  CJ->>G: appels d'évaluation (compte de service de l'exécution)
  VP-->>D: état des tâches, artefacts dans ML Metadata

Trois points à retenir :

• Fichiers. Vertex monte les buckets du projet dans chaque conteneur sous /gcs/<bucket>/ (Cloud Storage FUSE). Les composants lisent et écrivent donc des chemins de fichiers ordinaires, ce qui permet de les tester en local sans client Cloud Storage.
• Identité. Sans service_account, les étapes tournent sous le compte de service Compute Engine par défaut du projet. Il lui faut l'accès au bucket et le rôle roles/aiplatform.user pour appeler Gemini. En production, passez un compte dédié à job.submit(service_account=...).
• Machines. Chaque tâche est un CustomJob. Les limites set_cpu_limit("1") et set_memory_limit("2G") orientent le choix de la machine, mais la machine attribuée compte plus d'un vCPU (par défaut une e2-standard-4, 4 vCPU, d'après la documentation). Deux tâches lancées en parallèle (prepare-dataset et load-production) demandent donc au moins 8 vCPU de quota.

La mise en œuvre

make cloud-up PROJECT=mon-projet   # bucket <projet>-lab-vertex-ai-pipelines, label formation-lab, données, registre v1
make e2e PROJECT=mon-projet        # compile, soumet, attend, affiche l'état des tâches et le tableau Experiments

Le bucket est créé avec l'accès uniforme et la prévention d'accès public, puis étiqueté. make cloud-up est relançable : il ne recrée pas le bucket et n'écrase pas un registre existant.

aiplatform.init(project=project, location=region, staging_bucket=f"gs://{b}")
aiplatform.Experiment.get_or_create(EXPERIMENT, description="Portes de promotion des prompts")
job = aiplatform.PipelineJob(
    display_name=PIPELINE_NAME, template_path=template, pipeline_root=f"gs://{b}/root",
    # Vertex monte les buckets dans les conteneurs sous /gcs/<bucket>/
    parameter_values=params(f"/gcs/{b}/data/reference.jsonl", f"/gcs/{b}/registry",
                            cfg("MODEL", "gemini-2.5-flash"), project),
    enable_caching=cfg("ENABLE_CACHING", "true") == "true",
    labels=LAB_LABEL,
)
job.submit(experiment=EXPERIMENT)

submit(experiment=...) rattache l'exécution à l'expérience lab-vertex-ai-pipelines : ses paramètres et les métriques de ses artefacts Metrics deviennent une ligne du tableau Experiments.

Le mode local publie aussi ses mesures dans cette expérience quand il tourne avec Vertex AI (GOOGLE_GENAI_USE_VERTEXAI=TRUE), une ligne par version évaluée, avec experiment_tensorboard=False pour ne pas créer de TensorBoard facturé.

Résultat obtenu (2026-10-16)

La soumission réelle a échoué, et l'échec vaut d'être lu :

RuntimeError: Job failed with:
code: 9
message: " The DAG failed because some tasks failed. The failed tasks are: [prepare-dataset, load-production]."
prepare-dataset FAILED com.google.cloud.ai.platform.common.errors.AiPlatformException:
  code=RESOURCE_EXHAUSTED, message=The following quota metrics exceed quota limits:
  aiplatform.googleapis.com/custom_model_training_cpus

Le quota custom_model_training_cpus du projet de test vaut 1 en europe-west1, europe-west4 et us-central1. Aucune machine n'a démarré : le coût se limite au plus aux frais d'exécution de pipeline (0,03 $). Nous n'avons pas demandé d'augmentation de quota pour ce lab ; l'exécution complète du même DAG (cinq étapes, promotion comprise) a été faite avec kfp.local et Gemini sur Vertex AI (leçon 3), et ses mesures ont été publiées dans Vertex AI Experiments (extrait de aiplatform.get_experiment_df, colonnes choisies) :

run_name                                         param.version  metric.accuracy  metric.latency_p50_ms  metric.total_tokens
lab-vertex-ai-pipelines-local-v1-20261016160747  v1             0.9474           3300.3                 8428.0
lab-vertex-ai-pipelines-local-v2-20261016160747  v2             1.0000           2705.8                 10946.0
lab-vertex-ai-pipelines-prompt-gate-20261016151726  (exécution Vertex : paramètres seuls, aucune métrique)

Ce qui casse en production

• Quota de vCPU d'entraînement. Le cas ci-dessus. Symptôme : tâches en FAILED en quelques secondes, RESOURCE_EXHAUSTED dans task_details[].error, jamais dans le message principal du job. Diagnostic : gcloud alpha services quota list --service=aiplatform.googleapis.com --consumer=projects/<projet> --filter="metric:custom_model_training_cpus". Parade : demander 8 vCPU ou plus dans la région, ou sérialiser les deux premières tâches (prod.after(prep)) pour tenir dans 4.
• Le compte de service par défaut ne voit pas le bucket. Symptôme : FileNotFoundError sur /gcs/... alors que l'objet existe. Parade : rôle roles/storage.objectUser sur le bucket pour le compte de l'exécution, ou compte dédié.
• Région du modèle. gemini-3.5-flash n'est pas servi par Vertex AI en europe-west1 (404) ; gemini-2.5-flash l'est. Parade : MODEL explicite par environnement, testé par make run en mode Vertex avant la soumission.
• Lire le message principal seulement. « The DAG failed because some tasks failed » ne dit rien. Parade : run.py affiche l'état de chaque tâche ; en cas d'échec, lisez t.error.message de chaque tâche en FAILED.

La règle de décision

Itérez avec kfp.local, soumettez à Vertex AI quand le pipeline doit tourner sans vous, sur un déclencheur (CI, planification) ou pour profiter du cache et du lignage. Avant la première soumission dans un projet : quota de vCPU d'entraînement, compte de service, modèle disponible dans la région.

Pratique

Exercice. Ajoutez à submit() l'affichage du message d'erreur de chaque tâche en échec, même quand job.wait() lève une exception.

try:
    job.wait()
finally:
    for t in sorted(job.task_details, key=lambda t: t.start_time or t.create_time):
        msg = t.error.message.split(";")[0] if t.error.message else ""
        print(f"  {t.task_name:<24} {t.state.name:<14} {msg[:160]}")

• Je prépare le bucket et le registre avec make cloud-up, relançable.
• Je vérifie quota de vCPU, compte de service et modèle régional avant la première soumission.
• Je lis l'erreur au niveau des tâches, pas seulement celle du job.
• Je rattache chaque exécution à une expérience Vertex AI.