Sous-agents isolés et budgets

Un sous-agent sans limite se comporte comme un stagiaire sans échéance : il cherche encore une source, relit un document, reformule sa requête, et chaque tour renvoie tout l'historique au modèle. Sur cinq branches en parallèle, une seule qui boucle suffit à doubler la facture du dossier. Et un sous-agent qui « voit un peu plus que sa part » finit par répondre à la question du voisin. Cette leçon construit les deux garde-fous : un périmètre imposé par l'outil, un budget imposé par la boucle.

Le mécanisme

stateDiagram-v2
  [*] --> Verifier: tour suivant
  Verifier --> Echec: appels, tokens ou temps épuisés
  Verifier --> Appel: budget disponible
  Appel --> Conclure: dernier appel autorisé (outils restreints à rendre_constat)
  Appel --> Outils: lire / chercher
  Outils --> Verifier: réponses ajoutées au contexte
  Appel --> Fin: rendre_constat
  Conclure --> Fin: rendre_constat
  Fin --> [*]
  Echec --> [*]

• Périmètre. Perimetre(docs, autorises) ne contient que les documents attribués. lire("doc-10") hors périmètre lève HorsPerimetre, renvoyée au modèle comme une erreur d'outil : il apprend la limite sans la contourner.
• Budget (Budget) : nombre d'appels au modèle, tokens cumulés (entrée + sortie, réflexion comprise), temps. Vérifié avant chaque appel, mis à jour après.
• Atterrissage forcé. Quand il ne reste qu'un appel, on n'autorise plus que rendre_constat (allowed_function_names). Le sous-agent rend ce qu'il a, avec ses lacunes, au lieu de mourir sur BudgetDepasse et de perdre toutes ses lectures.
• Fin explicite. Le constat n'est pas un texte libre analysé après coup : c'est l'appel de l'outil rendre_constat, dont les paramètres suivent le schéma JSON de Constat. Avec mode="ANY", le modèle doit appeler un outil à chaque tour ; il n'y a donc pas de réponse ambiguë « sans outil ni constat ».

La mise en œuvre

@dataclass(frozen=True)
class Budget:
    max_appels: int = 6
    max_tokens: int = 40_000
    max_secondes: float = 90.0

@dataclass
class Compteur:
    budget: Budget
    appels: int = 0
    tokens: int = 0
    debut: float = field(default_factory=time.monotonic)

    def avant_appel(self) -> None:
        b = self.budget
        if self.appels >= b.max_appels:
            raise BudgetDepasse(f"appels : {self.appels}/{b.max_appels}")
        if self.tokens >= b.max_tokens:
            raise BudgetDepasse(f"tokens : {self.tokens}/{b.max_tokens}")
        if time.monotonic() - self.debut >= b.max_secondes:
            raise BudgetDepasse(f"temps : {b.max_secondes:.0f} s écoulées")

    @property
    def dernier_appel(self) -> bool:
        return self.appels == self.budget.max_appels - 1

La boucle du sous-agent (sous_agent.py) :

contents = [types.Content(role="user", parts=[types.Part(text=invite_sous_tache(st, perimetre))])]
while True:
    compteur.avant_appel()
    autorises = ["rendre_constat"] if compteur.dernier_appel else None
    config = types.GenerateContentConfig(
        system_instruction=CONSIGNE_SOUS_AGENT, tools=[OUTILS],
        tool_config=types.ToolConfig(function_calling_config=types.FunctionCallingConfig(
            mode="ANY", allowed_function_names=autorises)),
        automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True))
    t = time.monotonic()
    rep = await modele.generer(nom_modele, contents, config)
    t_in, t_out = tokens(rep.usage_metadata)
    compteur.apres_appel(t_in + t_out)
    journal.appel(span, nom_modele, t_in, t_out, t, force=bool(autorises))

    appels = rep.function_calls or []
    if not appels:
        raise ValueError("le modèle n'a appelé aucun outil malgré le mode ANY")
    contents.append(rep.candidates[0].content)      # garde les signatures de réflexion
    reponses = []
    for fc in appels:
        args = dict(fc.args or {})
        if fc.name == "rendre_constat":
            return Constat.model_validate(args)     # ValidationError = échec de la tentative
        try:
            res = perimetre.lire(args["doc_id"]) if fc.name == "lire" else {"resultats": perimetre.chercher(args.get("requete", ""))}
        except (HorsPerimetre, KeyError) as e:
            res = {"erreur": str(e)}
        reponses.append(types.Part.from_function_response(name=fc.name, response=res))
    contents.append(types.Content(role="user", parts=reponses))

Ce que reçoit le modèle au premier tour, en entier : la consigne système, puis

Question : Quel est le coût global sur 3 ans (licences, support, infrastructure énergétique) ...
Ton périmètre :
doc-09 | 2025-11-18 | blog | Tamaris Inference : la licence Apache 2.0, un atout
doc-10 | 2026-06-02 | annonce-officielle | Tamaris : nouvelle licence à partir de la version 2.6
...

Pas la demande du comité, pas les autres questions. Le test test_isolation_des_contextes le vérifie en inspectant tout ce que le faux modèle a reçu : chaque requête de sous-agent contient exactement une question, aucun identifiant d'un autre périmètre, et jamais le texte de la demande globale. Il tourne sur les deux versions (à la main et ADK).

.venv/bin/python -m pytest -q -k "isolation or budget"

Résultat obtenu (2026-10-15)

$ .venv/bin/python -m pytest -q -k "isolation or budget"
.....                                                                    [100%]
5 passed, 16 deselected in 2.01s

Sur le vrai modèle, la consigne qui autorise plusieurs lire dans la même réponse a changé le profil de dépense. Avant cette phrase, un sous-agent faisait 4 à 6 appels (recherches, lectures une par une) et l'un d'eux a atteint l'atterrissage forcé au 6e appel. Après : 2 appels par sous-agent dans make run (une réponse avec 6 à 8 lire en parallèle, puis rendre_constat), de 0,0025 à 0,0039 $ par sous-agent.

Ce qui casse en production

• Les signatures de réflexion perdues. Si vous reconstruisez le tour du modèle à partir des seuls function_calls au lieu de réinjecter rep.candidates[0].content, les modèles Gemini 3 avec réflexion rejettent ou dégradent le tour suivant. Parade : ajouter le contenu du candidat tel quel.
• Le budget de tokens qui explose en silence. Chaque tour renvoie tout l'historique : le coût d'entrée d'un sous-agent croît comme la somme des tailles successives du contexte. Un sous-agent à 6 tours lit son premier document 6 fois. Parade : compter les tokens cumulés (ce que fait Compteur), et pousser les lectures groupées.
• Le budget de temps qui ne coupe rien. avant_appel ne vérifie le temps qu'entre deux appels ; un appel bloqué 120 s passe. Parade : la vérification souple ici, plus un plafond dur asyncio.wait_for dans l'orchestrateur (leçon 4).
• Un constat mal formé. Le modèle appelle rendre_constat avec un champ manquant ; model_validate lève une ValidationError. Ici, cela fait échouer la tentative et déclenche la reprise. Variante plus économe : renvoyer l'erreur de validation au modèle comme réponse d'outil et lui laisser un tour pour corriger.
• Un outil qui lève une exception non prévue. Elle remonte et tue la tentative. Ne rattrapez que les erreurs « métier » (HorsPerimetre, argument manquant) pour les renvoyer au modèle ; laissez remonter le reste.

La règle de décision

Fixez le budget d'appels à partir de la trajectoire normale mesurée, plus une marge : ici 2 appels suffisent, 6 laissent de la place aux reformulations. Gardez l'atterrissage forcé : un constat partiel avec lacunes vaut mieux qu'un échec. Imposez l'isolation dans l'outil, jamais seulement dans l'invite. Surveillez par sous-agent : nombre d'appels, tokens, et le drapeau force du journal ; un taux d'atterrissages forcés qui monte signale des périmètres trop larges.

Pratique

Exercice. Modifiez la boucle pour qu'un rendre_constat invalide ne fasse pas échouer la tentative : renvoyez l'erreur au modèle, une seule fois.

from pydantic import ValidationError

deja_corrige = False
# ... dans la boucle, à la place du return direct :
if fc.name == "rendre_constat":
    try:
        return Constat.model_validate(args)
    except ValidationError as e:
        if deja_corrige:
            raise
        deja_corrige = True
        reponses.append(types.Part.from_function_response(
            name=fc.name, response={"erreur": f"constat invalide : {e.errors()[:3]}"}))
        continue

• Je sais imposer un périmètre par l'outil et le vérifier par un test
• Je sais compter appels, tokens et temps par sous-agent
• Je sais forcer la conclusion au dernier appel avec allowed_function_names
• Je sais pourquoi on réinjecte le contenu du candidat tel quel