Atelier

Leçon 3 — Construire pas à pas

Assembleur structuré et contrôle du budget

6 min de lecture

Assembleur structuré et contrôle du budget

1. Un schéma pour la sortie

Une sortie structurée oblige le modèle à répondre en JSON conforme à un schéma. Avec ADK, on décrit ce schéma par une classe Pydantic (bibliothèque Python de validation de données) et on la passe dans output_schema. ADK valide la réponse et range le dictionnaire obtenu dans state[output_key].

files/agent.py (schéma)
class Jour(BaseModel):
    jour: int
    programme: str = Field(description="Programme du jour en une ou deux phrases")
    activite_ids: list[str] = Field(default_factory=list, description="Identifiants A... prévus ce jour")


class Itineraire(BaseModel):
    titre: str
    transport_id: str = Field(description="Identifiant T... retenu")
    hebergement_id: str = Field(description="Identifiant H... retenu")
    jours: list[Jour]
    conseils: str

Le schéma ne contient aucun prix : volontairement. Le modèle choisit des identifiants ; les montants viendront du catalogue.

2. L'assembleur lit l'état

files/agent.py (assembleur)
assembleur = Agent(
    name="assembleur", model=m("assembleur"), output_schema=Itineraire, output_key="itineraire",
    instruction="Tu assembles un itinéraire de {nuits} nuits pour {voyageurs} voyageur(s), budget total "
                "{budget_max} EUR. Propositions reçues :\nTRANSPORT : {transport}\nHÉBERGEMENT : {hebergement}\n"
                "ACTIVITÉS : {activites}\nChoisis UNIQUEMENT des identifiants présents ci-dessus. Calcule "
                "mentalement le total (transport x voyageurs + nuit x nuits + activités x voyageurs) et "
                "choisis des options qui tiennent dans le budget. N'invente aucun prix.")
flowchart LR
  ST["state : transport, hebergement, activites, nuits, voyageurs, budget_max"] -->|"injection {cle}"| I["instruction de l'assembleur"]
  I --> G["Gemini + schéma Itineraire"]
  G -->|"JSON validé par Pydantic"| O["state['itineraire'] (dict)"]

Pourquoi le schéma Itineraire ne contient-il aucun champ de prix ?

3. Le contrôle du budget : un BaseAgent sans LLM

Tout agent ADK hérite de BaseAgent. Pour écrire le sien, on implémente la méthode asynchrone _run_async_impl(ctx) qui produit des événements (Event). Un événement peut porter du texte (content) et des actions, dont state_delta : les clés à écrire dans l'état.

D'abord, la logique, sous forme de fonction pure (facile à tester) :

files/agent.py (calcul)
def calculer_budget(itineraire: dict, voyageurs: int, nuits: int, budget_max: float) -> dict:
    """Recalcule le coût à partir du CATALOGUE (jamais à partir des prix écrits par le LLM)."""
    t = {x["id"]: x for x in TRANSPORTS}.get(itineraire.get("transport_id"))
    h = {x["id"]: x for x in HEBERGEMENTS}.get(itineraire.get("hebergement_id"))
    cat_a = {x["id"]: x for x in ACTIVITES}
    ids_a = [i for j in itineraire.get("jours", []) for i in j.get("activite_ids", [])]
    erreurs = [f"identifiant inconnu : {i}" for i in ids_a if i not in cat_a]
    ...
    postes = {
        "transport": (t["prix_eur"] * voyageurs) if t else 0,
        "hebergement": (h["prix_nuit"] * nuits) if h else 0,
        "activites": sum(cat_a[i]["prix_eur"] for i in set(ids_a) if i in cat_a) * voyageurs,
    }
    total = sum(postes.values())
    return {"postes": postes, "total": total, "budget_max": budget_max,
            "respecte": total <= budget_max and not erreurs, "erreurs": erreurs}

Puis l'agent, qui ne fait que lire l'état, appeler la fonction et publier le verdict :

files/agent.py (agent personnalisé)
class ControleBudget(BaseAgent):
    async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
        s = ctx.session.state
        verdict = calculer_budget(s.get("itineraire") or {}, s["voyageurs"], s["nuits"], s["budget_max"])
        etat = "OK" if verdict["respecte"] else "À REVOIR"
        texte = f"Budget {etat} : {verdict['total']} EUR pour {verdict['budget_max']} EUR max. ..."
        yield Event(author=self.name, invocation_id=ctx.invocation_id,
                    content=types.Content(role="model", parts=[types.Part(text=texte)]),
                    actions=EventActions(state_delta={"controle_budget": verdict}))
Agent LLMBaseAgent personnalisé
Résultatprobable, varie d'un appel à l'autredéterministe
Coûtun ou plusieurs appels au modèlenul
Bon pourchoisir, rédiger, résumercalculer, vérifier, router selon une règle
Se testeavec un faux modèleavec un simple assert

L'assembleur choisit l'activité A99, qui n'existe pas dans le catalogue. Que renvoie calculer_budget ?

Exercice : replanifier en cas de dépassement

Quand respecte vaut False, on voudrait relancer l'assembleur avec une consigne « réduis le coût de X EUR », au plus 2 fois. Quelle structure ADK utiliseriez-vous, et comment le contrôle budgétaire arrêterait-il la boucle quand tout va bien ?

Voir la solution

Placez l'assembleur et le contrôle dans un LoopAgent(max_iterations=3), après le ParallelAgent. Le contrôle écrit une consigne dans l'état et, si le budget est respecté, escalade pour sortir de la boucle :

files/agent.py (variante)
depassement = verdict["total"] - verdict["budget_max"]
yield Event(author=self.name, invocation_id=ctx.invocation_id,
            actions=EventActions(
                escalate=verdict["respecte"],           # True = sortir du LoopAgent
                state_delta={"controle_budget": verdict,
                             "consigne_budget": f"Réduis le coût d'au moins {depassement} EUR."
                             if depassement > 0 else ""}))

Et ajoutez {consigne_budget?} dans l'instruction de l'assembleur. Le lab « Rédacteur et critique en boucle » détaille ce mécanisme.

Résultat obtenu

Itinéraire réel produit le 2026-10-15 (gemini-3.5-flash), puis vérifié par le code :

python agent.py (extrait)
{
  "itineraire": {
    "titre": "Séjour culturel et gourmand à Lisbonne",
    "transport_id": "T2",
    "hebergement_id": "H2",
    "jours": [
      {"jour": 1, "programme": "Arrivée à Lisbonne, installation dans l'Alfama et découverte du quartier historique avec une visite guidée.", "activite_ids": ["A1"]},
      {"jour": 2, "programme": "Matinée culturelle au Musée des Azulejos suivie d'un atelier gourmand ...", "activite_ids": ["A6", "A2"]},
      {"jour": 3, "programme": "Journée d'excursion féerique à Sintra ...", "activite_ids": ["A3"]},
      {"jour": 4, "programme": "Dernière promenade le long du Tage ...", "activite_ids": []}
    ]
  },
  "controle_budget": {
    "postes": {"transport": 192, "hebergement": 294, "activites": 314},
    "total": 800, "budget_max": 900, "respecte": true, "erreurs": []
  }
}

Vérification à la main : 2 × 96 = 192 ; 3 × 98 = 294 ; (25 + 8 + 45 + 79) × 2 = 314 ; total 800 EUR.

Je sais faire

  • Je décris une sortie avec une classe Pydantic et je la passe dans output_schema
  • J'injecte les résultats d'agents précédents dans une instruction avec {cle} et {cle?}
  • J'écris un BaseAgent qui lit l'état et publie un state_delta
  • Je sépare la logique (fonction pure) de l'agent qui l'appelle
Suivant