Le workflow : étapes, parallélisme, réessais, erreurs

Le job sait traiter un fichier. Il reste à décider, chaque nuit : quels fichiers, combien à la fois, que faire si un job échoue, comment produire la synthèse et qui prévenir. Écrit en Python dans un conteneur, ce chef d'orchestre devrait lui-même tourner quelque part, attendre des heures, survivre à un redémarrage et réessayer proprement. Cloud Workflows fournit exactement cela : une exécution durable, décrite en YAML, dont chaque étape est journalisée et dont l'état survit à n'importe quelle panne d'infrastructure.

Le mécanisme

Un workflow est une liste d'étapes nommées. Les briques utilisées ici :

flowchart TB
  A["init + date_cible"] --> B{"termine.json ?"}
  B -- "oui" --> R0["return deja_traite"]
  B -- "non (404)" --> L["lister entrant/date/"]
  L --> P["parallel for, 2 à la fois"]
  subgraph branche["une branche par fichier"]
    J["jobs.run (retry : job_reessayable, 30 s, x2)"] --> S["lire les stats.json"]
  end
  P --> branche
  branche --> T["totaux"] --> G["generateContent (retry 429, x4)"]
  G --> RP["rapport.json"] --> N["publish ok"] --> M["termine.json"]
  L -. "toute erreur" .-> X["publish echec, puis raise"]
  G -. "toute erreur" .-> X

Les connecteurs évitent d'écrire des appels HTTP à la main : ils gèrent l'authentification, les réessais réseau élémentaires et, pour les opérations longues (lancer un job, par exemple), l'attente de la fin. Le connecteur googleapis.run.v2.projects.locations.jobs.run ne rend la main que quand l'exécution du job est terminée, ou lève une erreur si elle a échoué.

La mise en œuvre

La branche parallèle, avec la variable partagée où chaque branche ajoute ses statistiques :

            - par_fichier:
                parallel:
                  shared: [stats]
                  concurrency_limit: 2  # quota Gemini partagé : 2 jobs x N tâches en même temps, pas plus
                  for:
                    value: obj
                    in: ${entrees.items}
                    steps:
                      ...
                      - lancer_job:
                          try:
                            call: googleapis.run.v2.projects.locations.jobs.run
                            args:
                              name: ${"projects/" + project + "/locations/" + region + "/jobs/" + job}
                              body:
                                overrides:
                                  taskCount: ${tasks_per_file}
                                  containerOverrides:
                                    - env:
                                        - name: INPUT_URI
                                          value: ${"gs://" + bucket + "/" + obj.name}
                                        - name: OUTPUT_PREFIX
                                          value: ${"gs://" + bucket + "/" + sortie}
                            result: execution
                          retry:
                            predicate: ${job_reessayable}
                            max_retries: 2
                            backoff:
                              initial_delay: 30
                              max_delay: 120
                              multiplier: 2

Le prédicat décide quelles erreurs méritent un réessai. Un job qui échoue (tâches en échec, quota) est réessayé, puisqu'il est idempotent ; une erreur de configuration ne l'est pas :

job_reessayable:
  params: [e]
  steps:
    - sans_code_http:
        switch:
          - condition: ${get_type(e) != "map"}
            return: true
          - condition: ${not("code" in e)}
            return: true
    - erreur_client:
        switch:
          - condition: ${e.code == 400 or e.code == 403 or e.code == 404}
            return: false
    - sinon:
        return: true

L'appel direct à Vertex AI, sans conteneur, pour la synthèse :

            - synthese_gemini:
                try:
                  call: googleapis.aiplatform.v1.projects.locations.endpoints.generateContent
                  args:
                    model: ${"projects/" + project + "/locations/" + region + "/publishers/google/models/" + model}
                    region: ${region}
                    body:
                      contents:
                        role: user
                        parts:
                          - text: ${consigne + json.encode_to_string(volumes) + "\n\nPar tranche :\n" + json.encode_to_string(stats)}
                      generationConfig:
                        temperature: 0.2
                        maxOutputTokens: 2048
                        thinkingConfig:
                          thinkingBudget: 0
                  result: gemini
                retry:
                  predicate: ${http.default_retry_predicate}  # 429, 502, 503, 504, coupures réseau
                  max_retries: 4

volumes contient les totaux par fichier et par thème, calculés juste avant par des boucles for du workflow (avis_par_theme[th]: ${default(map.get(avis_par_theme, th), 0) + s.par_theme[th]}) : le modèle rédige, il ne compte pas. La leçon 5 montre pourquoi.

Le workflow reçoit ses paramètres par des variables d'environnement de déploiement (--set-env-vars=BUCKET=…,JOB=…), lues par sys.get_env. Le même YAML se déploie ainsi dans plusieurs projets sans modification.

Résultat obtenu (2026-10-16)

Le premier déploiement a été refusé par l'analyseur de Workflows, pour deux raisons qu'aucun test local ne voyait :

ERROR: (gcloud.workflows.deploy) [INVALID_ARGUMENT] main.yaml:16:23: parse error: in workflow 'main',
  step 'aucun_argument': cannot assign to a reserved variable 'args'
main.yaml:155:35: parse error: in workflow 'main', step 'synthese_gemini':
  maximum length for an expression is 400 characters, the provided expression has a length of 429

D'où la variable entree (copie de args, qui est en lecture seule) et la consigne de synthèse déclarée comme une chaîne littérale dans un assign, hors expression. Seconde exécution réelle, refusée cette fois à l'exécution :

UnhandledBranchError: One or more branches or iterations encountered an unhandled runtime error
  in step "lancer_job": HTTP 403 Permission 'run.operations.get' denied on resource
  'projects/<projet>/locations/europe-west1/operations/874151e5-…'

Le rôle roles/run.jobsExecutorWithOverrides, accordé sur le job, permet de le lancer ; mais le connecteur attend la fin en lisant l'opération longue, permission vérifiée au niveau du projet. Les trois jobs avaient bien été lancés, et ont réussi : gcloud run jobs executions list les montrait 2 / 2. C'est le prédicat qui a évité trois réessais inutiles : un 403 n'est pas réessayé. Correction dans cloud-up.sh : roles/run.viewer (lecture seule de Cloud Run) sur le projet pour le compte du workflow. L'exécution échouée a duré 6 s ; la suivante, après correction, a réussi en 144 s.

Ce qui casse en production

• Variable partagée mal déclarée. Sans shared: [stats], chaque branche écrit dans sa propre copie de stats, et la liste est vide après l'étape parallèle : la synthèse est rédigée sur zéro avis, sans erreur. Parade : le test test_parallelisme_borne_et_variable_partagee.
• Taille des variables. La mémoire d'une exécution est limitée (quelques centaines de Ko pour l'ensemble des variables). Lire les résultats complets des tâches dans le workflow ferait échouer l'exécution dès quelques milliers d'avis. Parade : le workflow ne lit que les stats.json (des compteurs) ; les résultats restent dans Cloud Storage.
• Réessayer ce qui ne passera jamais. Un retry: ${http.default_retry} sur le lancement du job réessaierait aussi les échecs de tâches ; sans idempotence, chaque réessai repaierait Gemini. Avec un prédicat trop large, un 403 de configuration ferait patienter l'exécution 3 fois 30 s à 120 s avant d'échouer. Parade : un prédicat écrit pour l'appel, et des jobs idempotents.
• Erreur avalée par except. Un bloc except qui notifie sans raise termine l'exécution en SUCCEEDED : la supervision ne voit rien. Parade : raise: ${e} en dernière étape du bloc, vérifié par un test.
• Ordre des étapes de fin. Écrire termine.json avant de publier la notification perd la notification si la publication échoue (la relance s'arrête sur deja_traite). Parade : rapport, notification, marqueur, dans cet ordre ; une relance peut alors renvoyer une notification en double, que le destinataire doit tolérer.

La règle de décision

Workflows orchestre, il ne calcule pas : appels d'API par connecteurs, données lourdes dans Cloud Storage, compteurs dans les variables. Chaque appel qui peut échouer a son retry avec un prédicat qui distingue passager et définitif. Tout except notifie puis relance l'erreur. Validez le YAML par un vrai gcloud workflows deploy dans la CI ou en début de cloud-up : l'analyseur de Google est la seule référence.

Pratique

Exercice. Ajoutez une étape qui fait échouer l'exécution (avec notification) si plus de 10 % des lignes d'entrée ont été rejetées, avant la synthèse. Écrivez le YAML.

            - controle_qualite:
                switch:
                  - condition: ${rejetees * 10 > avis + rejetees}
                    raise: ${"qualité des entrées : " + string(rejetees) + " lignes rejetées sur " + string(avis + rejetees)}

• J'écris un workflow avec étapes, branches parallèles bornées et variable partagée.
• J'appelle Cloud Run jobs et Vertex AI par connecteurs, et je sais lesquels attendent une opération longue.
• J'écris un prédicat de réessai qui distingue erreur passagère et erreur de configuration.
• Je termine chaque except par une notification et un raise.