Atelier

Lección 4 — Robustesse : reprise et validation

Regrouper les documents par lot et valider chaque réponse

5 min de lectura

Este contenido aún no está traducido. Se muestra la versión original.

Regrouper les documents par lot et valider chaque réponse

Pourquoi regrouper ?

Chaque requête porte un coût fixe : l'instruction système (le rôle et les règles, répétés à chaque appel), la latence réseau, et une place dans le quota de requêtes par minute. En envoyant 10 avis courts par requête :

  • on divise par 10 le nombre de requêtes (donc la pression sur le RPM) ;
  • l'instruction système n'est payée qu'une fois pour 10 avis.
flowchart LR
  subgraph U["1 avis par requête"]
    S1["Instruction + avis 1"] --> R1["Réponse 1"]
    S2["Instruction + avis 2"] --> R2["Réponse 2"]
    S3["... x 10"]
  end
  subgraph L["10 avis par requête"]
    SL["Instruction + avis 1 à 10"] --> RL["Liste de 10 réponses"]
  end
Taille du lotAvantagesInconvénients
1une erreur ne touche qu'un document ; prompt simplebeaucoup de requêtes, instruction répétée
5 à 20 (documents courts)bon compromisune réponse incomplète touche plusieurs documents
100 et plustrès peu de requêtesréponses longues, oublis plus probables, délais plus longs

Le prompt d'un lot

Chaque avis est préfixé par son id, pour que le modèle puisse le renvoyer :

files/classer.py
def prompt_lot(avis: list[dict]) -> str:
    return "\n".join(f"[{a['id']}] {a['texte']}" for a in avis)
[A0001] Génial, mon grille-pain Nimbus est arrivé avec trois semaines de retard. Vraiment bravo.
[A0002] Colis du ventilateur Lumo perdu par le transporteur, toujours pas de nouvelles.
...

Le schéma de réponse

On passe un JSON Schema (une description formelle d'une structure JSON) avec response_json_schema. Les valeurs permises sont des enum : le modèle ne peut pas inventer une catégorie « logistique ».

files/classer.py
SCHEMA = {
    "type": "array",
    "items": {
        "type": "object",
        "properties": {
            "id": {"type": "string"},
            "sentiment": {"type": "string", "enum": SENTIMENTS},
            "categorie": {"type": "string", "enum": CATEGORIES},
            "a_recontacter": {"type": "boolean",
                              "description": "vrai si le client attend une action du support"},
        },
        "required": ["id", "sentiment", "categorie", "a_recontacter"],
    },
}

Ce schéma est un simple dictionnaire : il sert tel quel dans l'appel synchrone et dans le fichier JSONL de la Batch API (module 3).

Valider : ne garder que ce qui a été demandé

Le schéma garantit la forme, pas le contenu : le modèle pourrait oublier un avis, en renvoyer un deux fois, ou renvoyer un id qu'on n'a jamais envoyé.

files/classer.py
def valider(avis: list[dict], brut: str) -> list[dict]:
    """Garde uniquement les réponses valides pour des ids demandés (le reste sera refait)."""
    attendus = {a["id"] for a in avis}
    ok = []
    for r in json.loads(brut):
        if (r.get("id") in attendus and r.get("sentiment") in SENTIMENTS
                and r.get("categorie") in CATEGORIES and isinstance(r.get("a_recontacter"), bool)):
            ok.append(r)
            attendus.discard(r["id"])  # un id en double n'est compté qu'une fois
    return ok

Un avis oublié n'apparaît pas dans les résultats du lot : deja_faits ne le connaît pas, il sera donc refait à la prochaine exécution. Validation et reprise travaillent ensemble.

Le modèle renvoie 9 réponses valides pour un lot de 10 avis. Que fait le pipeline ?

Quels cas la fonction valider écarte-t-elle ? (plusieurs réponses)

Exercice : un test pour l'avis oublié

Écrivez un test qui vérifie qu'un avis oublié par le modèle est bien retraité à l'exécution suivante.

Voir la solution
test_lot.py
def test_avis_oublie_refait(tmp_path):
    sortie = tmp_path / "r.jsonl"
    avis = generer(10, 1)

    async def oublie_le_dernier(client, lot):
        brut = json.dumps([{"id": a["id"], "sentiment": "neutre", "categorie": "autre",
                            "a_recontacter": False} for a in lot[:-1]])
        return {"resultats": classer.valider(lot, brut), "tokens_in": 1, "tokens_out": 1}

    lancer(avis, sortie, traiter_lot=oublie_le_dernier)
    assert len(classer.deja_faits(str(sortie))) == 9
    appels = []
    lancer(avis, sortie, traiter_lot=faux_lot(appels=appels))
    assert appels == [["A0010"]]

Résultat obtenu

Mesure réelle du 2026-10-15 sur les 10 premiers avis, gemini-3.5-flash-lite :

lot de 1  : 10 appels, 861 jetons en entrée, 501 en sortie, 2.8 s (concurrence 4)
lot de 10 :  1 appel,  339 jetons en entrée, 474 en sortie, 1.8 s

Le regroupement divise ici par environ 2,5 les jetons d'entrée (l'instruction n'est envoyée qu'une fois) et par 10 le nombre de requêtes. Sur les 200 avis du projet final, 20 lots sur 20 ont renvoyé 10 réponses valides sur 10.

Je sais faire

  • Choisir une taille de lot selon la longueur des documents
  • Imposer un schéma JSON avec des enum via response_json_schema
  • Valider ids, valeurs et doublons avant d'écrire
  • Laisser la reprise retraiter ce que la validation a écarté
Siguiente