Parallélisme, reprise sur échec, journal

Cinq sous-agents de 3 à 5 secondes chacun : 18 secondes en file, 4 à 6 en parallèle. Le gain est réel, mais il arrive avec trois questions que le code séquentiel ne posait pas. Que se passe-t-il quand la branche st-2 reçoit un 503 au milieu de son travail ? Comment empêcher cinq branches de déclencher ensemble les quotas de l'API ? Et, quand le dossier est faux ou trop cher, comment retrouver quelle branche a fait quoi, quand, et pour combien ? Cette leçon répond aux trois : un sémaphore, une politique de reprise, un journal en arbre.

Le mécanisme

flowchart TB
  G["asyncio.gather(une(st) pour chaque sous-tâche)"] --> S["Semaphore(parallelisme)"]
  S --> T1["essai 1 : périmètre neuf, compteur neuf"]
  T1 -->|"ok"| OK["constat enregistré"]
  T1 -->|"exception, budget, wait_for"| T2["essai 2 : périmètre neuf, compteur neuf"]
  T2 -->|"ok"| OK
  T2 -->|"échec"| D["constats[st] = None, erreur gardée"]
  OK --> Y["synthèse"]
  D --> Y
  Y --> J["statut racine : ok ou degrade"]

asyncio.gather lance une coroutine par sous-tâche. Aucune exception n'en sort : chaque branche rattrape ses propres erreurs, sinon la première panne annulerait l'attente des autres et ferait perdre leurs résultats.
asyncio.Semaphore(parallelisme) plafonne le nombre de branches actives. parallelisme=1 donne la version séquentielle, avec exactement le même code : c'est ce qui rend la comparaison de la leçon 7 honnête.
Deux plafonds de temps : la vérification souple du Compteur entre deux appels, et asyncio.wait_for qui annule la branche si elle dépasse max_secondes, même bloquée dans un appel réseau.
Reprise : une seconde tentative repart de zéro (périmètre et compteur neufs). Si elle échoue aussi, la branche est marquée en échec et l'orchestrateur continue en mode dégradé : la synthèse reçoit « NON INSTRUITE » pour cette sous-tâche et doit le dire.
Journal : chaque étape ouvre un span (un intervalle nommé avec parent, début, fin, statut, tokens, coût). L'arbre des spans est la trajectoire.

La mise en œuvre

files/orchestre/orchestrateur.py (branches)

sem = asyncio.Semaphore(cfg.parallelisme)

async def une(st):
    async with sem:
        for essai in range(1, cfg.essais + 1):
            span = journal.ouvrir(f"{st.id} (essai {essai})", "sous-agent", racine,
                                  question=st.question, perimetre=st.doc_ids)
            perimetre = Perimetre(docs, st.doc_ids)   # périmètre neuf à chaque tentative
            m = _ModeleEnPanne(modele) if essai == 1 and st.id in cfg.pannes else modele
            compteur = Compteur(cfg.budget)
            try:
                c = await asyncio.wait_for(
                    executer_sous_tache(m, cfg.modele, st, perimetre, compteur, journal, span),
                    timeout=cfg.budget.max_secondes)
                journal.fermer(span, appels=compteur.appels, lus=perimetre.lus)
                constats[st.id] = c
                lus.update(perimetre.lus)
                return
            except Exception as e:  # une sous-tâche ne doit jamais faire tomber les autres
                msg = f"{type(e).__name__}: {e}"[:200]
                journal.fermer(span, "echec", erreur=msg, appels=compteur.appels)
                erreurs[st.id] = msg
        constats[st.id] = None

await asyncio.gather(*(une(st) for st in plan.sous_taches))

_ModeleEnPanne sert à la démonstration : il laisse passer le premier appel de la branche puis lève un 503 simulé, comme une panne réelle au milieu d'une sous-tâche. Les vraies erreurs transitoires (429, 500, 503, 504) sont d'abord absorbées plus bas, par le SDK :

files/orchestre/llm.py (extrait)

retry = types.HttpRetryOptions(attempts=4, initial_delay=2, http_status_codes=[429, 500, 503, 504])
self.client = genai.Client(http_options=types.HttpOptions(retry_options=retry, timeout=120_000))

Deux niveaux, deux rôles : le SDK réessaie un appel avec attente exponentielle ; l'orchestrateur reprend une sous-tâche entière quand l'appel a définitivement échoué, quand le budget est dépassé ou quand le constat est invalide. Le client est créé une fois et gardé : un genai.Client() temporaire non conservé se ferme et les appels suivants échouent avec « client has been closed ».

Le journal tient en quelques lignes (journal.py) ; rapport.py en fait une page HTML autonome (frise positionnée en pourcentage de la durée totale, cumul des tokens et du coût de chaque nœud, tableau des sous-agents), sans bibliothèque JavaScript.

files/orchestre/journal.py (extrait)

def appel(self, parent, modele, t_in, t_out, debut, **details):
    sid = self.ouvrir(f"appel {modele}", "appel", parent, **details)
    s = self.spans[sid - 1]
    s.debut = debut - self.t0
    s.tokens_in, s.tokens_out, s.cout = t_in, t_out, cout_usd(modele, t_in, t_out)
    self.fermer(sid)

Lancez la démo : version à la main, panne injectée sur st-2.

make run
xdg-open sorties/main-demo/trajectoire.html     # macOS : open

Résultat obtenu (2026-10-15)

$ make run
WARNING orchestre.orchestrateur : st-2 essai 1 : PanneInjectee: 503 UNAVAILABLE (panne injectée)
main-demo  31.7 s (plan 4.3 / sous-agents 3.9 / synthèse 23.6)  11 appels  17275 in  7384 out  0.0306 $
           qualité 80.8/100 (faits 11/12, contradictions 2/4, citations 25/25)  échecs -  arbitrages suspects 0
Trajectoire : sorties/main-demo/trajectoire.html  Dossier : sorties/main-demo/dossier.md

Trajectoire de make run : st-2 échoue après 1 s (en rouge), sa reprise démarre aussitôt pendant que st-1, st-3 et st-4 travaillent ; la synthèse occupe les trois quarts de la frise

La reprise de st-2 a démarré pendant que les autres branches travaillaient : la phase des sous-agents a duré 3,9 s malgré la panne. La frise montre aussi ce que les chiffres agrégés cachent : sur 31,7 s, la synthèse en prend 23,6. Paralléliser davantage les sous-agents ne changera presque rien au temps total de ce dossier.

Ce qui casse en production

Reprendre une erreur déterministe. Un BudgetDepasse ou une ValidationError qui vient d'un périmètre trop large se reproduira à l'identique : la reprise double la dépense pour rien. Parade : ne reprendre à l'identique que les erreurs transitoires ; pour un budget dépassé, reprendre avec un périmètre réduit ou accepter le constat partiel de l'atterrissage forcé.
Le parallélisme qui déclenche les quotas. Cinq branches qui démarrent ensemble envoient cinq requêtes dans la même seconde ; sur un projet partagé, les 429 arrivent par rafales et les réessais du SDK se synchronisent. Parade : parallelisme réglé sous le quota, gigue (jitter) sur les réessais, et le journal pour voir les appels qui ont duré anormalement.
wait_for n'arrête pas la facture. Annuler la coroutine coupe l'attente côté client ; la génération déjà lancée côté serveur peut être facturée. Le plafond de temps protège la latence du dossier, pas le budget en dollars. Parade : le plafond de tokens par appel (max_output_tokens) et le compteur de tokens.
Un état partagé modifié par les branches. Ici, constats, erreurs et lus sont partagés, mais chaque branche n'écrit que sa clé et asyncio n'interrompt une coroutine qu'à un await : pas de course. Avec des threads ou des processus, il faudrait renvoyer les résultats au lieu d'écrire dans des dictionnaires communs.
Un journal sans parent. Des lignes de log à plat ne permettent pas de dire quel appel appartient à quelle tentative. Parade : un identifiant de parent sur chaque span, dès le départ.

La règle de décision

Parallélisez les branches indépendantes, plafonnez avec un sémaphore réglé sur votre quota, et gardez parallelisme=1 disponible pour mesurer. Une seule reprise par sous-tâche suffit dans la plupart des cas ; au-delà, l'échec est presque toujours déterministe. Acceptez le mode dégradé quand l'axe manquant n'est pas bloquant pour la décision, et faites-le apparaître dans le dossier. Regardez la frise avant d'optimiser : le goulot n'est pas toujours là où on l'attend.

Pratique

La frise montre : planificateur 4 s, sous-agents 4 s en parallèle, synthèse 24 s. On vous demande de réduire le temps total. Quelle action a le plus d'effet ?

La branche st-3 échoue deux fois avec BudgetDepasse (tokens). Les autres réussissent. Que doit produire l'orchestrateur ?

Exercice. Modifiez une() pour qu'une seconde tentative après BudgetDepasse reparte avec un budget doublé, et qu'une ValidationError ne soit pas reprise du tout.

Voir la solution

from pydantic import ValidationError
from orchestre.budget import BudgetDepasse

async def une(st):
    async with sem:
        budget = cfg.budget
        for essai in range(1, cfg.essais + 1):
            ...
            compteur = Compteur(budget)
            try:
                ...
                return
            except ValidationError as e:
                journal.fermer(span, "echec", erreur=str(e)[:200])
                erreurs[st.id] = "constat invalide"
                break                                   # déterministe : on n'insiste pas
            except BudgetDepasse as e:
                journal.fermer(span, "echec", erreur=str(e))
                budget = replace(budget, max_appels=budget.max_appels * 2,
                                 max_tokens=budget.max_tokens * 2)
            except Exception as e:
                journal.fermer(span, "echec", erreur=str(e)[:200])
        constats[st.id] = None

Ajoutez un test avec un faux modèle qui ne conclut jamais avant le 7e appel : il doit échouer à l'essai 1 et réussir à l'essai 2. Le doublement doit rester borné (ici, une seule fois puisque essais=2), sinon une branche folle consomme le budget de tout le dossier.

Je sais lancer N branches avec gather sans qu'une panne annule les autres
Je sais distinguer réessai d'appel (SDK) et reprise de sous-tâche (orchestrateur)
Je sais produire un dossier dégradé qui dit ce qui manque
Je sais lire une frise de trajectoire pour trouver le goulot
make run produit sorties/main-demo/trajectoire.html