Le script complet : schéma, reprise et tests

Le schéma de réponse

files/traiter_lot.py (extrait)

MODEL = os.environ.get("MODEL", "gemini-3.5-flash-lite")  # modèle « volume » : moins cher par ligne

class Analyse(BaseModel):
    resume: str
    categorie: Literal["commande", "facturation", "reclamation", "rendez-vous", "information",
                       "fournisseur", "spam", "autre"]
    priorite: Literal["haute", "normale", "basse"]
    brouillon_reponse: str

Literal[...] est l'équivalent Python d'un enum : le modèle reçoit la liste des valeurs permises, et Pydantic refuse toute autre valeur à la validation.

Un appel par ligne, avec réessais

files/traiter_lot.py (extrait)

def analyser(client, objet, texte, essais=4) -> Analyse:
    from google.genai import errors, types
    config = types.GenerateContentConfig(temperature=0.2, response_mime_type="application/json",
                                         response_schema=Analyse)
    for i in range(essais):
        try:
            r = client.models.generate_content(model=MODEL, contents=PROMPT.format(objet=objet, texte=texte),
                                               config=config)
            return Analyse.model_validate_json(r.text)
        except errors.APIError as e:
            if e.code not in (429, 500, 503) or i == essais - 1:
                raise
            time.sleep(5 * 2 ** i)

L'humain dans la boucle, en code

files/traiter_lot.py (extrait)

def statut_pour(a: Analyse) -> str:
    """Règle déterministe : les cas sensibles passent en tête de la file de relecture."""
    if a.priorite == "haute" or a.categorie in ("reclamation", "autre"):
        return "à relire en priorité"
    return "à valider"

Le script n'a aucune fonction d'envoi d'e-mail. Le brouillon est une suggestion dans une colonne.

La boucle : sauvegarde et reprise

files/traiter_lot.py (extrait)

def traiter(df, client, limite=None, sauvegarder=None) -> int:
    a_faire = df.index[df["statut"] == ""][:limite]
    for n, i in enumerate(a_faire, 1):
        try:
            a = analyser(client, df.at[i, "objet"], df.at[i, "texte"])
            df.loc[i, COLONNES_IA] = [a.resume, a.categorie, a.priorite, a.brouillon_reponse, statut_pour(a)]
        except Exception as e:  # une ligne en échec ne bloque pas le lot
            df.at[i, "statut"] = f"erreur : {type(e).__name__}"
        print(f"[{n}/{len(a_faire)}] ligne {df.at[i, 'id']} -> {df.at[i, 'categorie']} / {df.at[i, 'statut']}")
        if sauvegarder:
            sauvegarder(df)
    return len(a_faire)

flowchart TD
  A["Lignes où statut est vide"] --> B["analyser() : Gemini + schéma"]
  B -- "OK" --> C["remplir les 5 colonnes, statut_pour()"]
  B -- "erreur" --> E["statut = erreur : TypeErreur"]
  C --> S["sauvegarder le fichier"]
  E --> S
  S --> A
  S --> R["Relance : python traiter_lot.py emails_traite.csv"]
  R -- "seules les lignes au statut vide" --> A

Commande	Effet
`python traiter_lot.py emails.csv --limite 3`	3 lignes, pour tester la consigne
`python traiter_lot.py emails.csv`	tout le lot vers `emails_traite.csv`
`python traiter_lot.py emails_traite.csv`	reprise : seules les lignes au statut vide
`python traiter_lot.py emails.xlsx`	même chose avec Excel → `emails_traite.xlsx`

Le script s'arrête (coupure réseau) à la ligne 12 sur 20. Que perdez-vous ?

Que vérifie test_lot_csv_avec_limite_et_reprise ?

Exercice : un contrôle de cohérence

Le modèle peut produire un brouillon qui promet un remboursement malgré la consigne. Ajoutez une règle à statut_pour : si le brouillon contient « rembours » (sans tenir compte des majuscules), le statut devient « à relire en priorité ». Ajoutez le test.

Voir la solution

def statut_pour(a: Analyse) -> str:
    if (a.priorite == "haute" or a.categorie in ("reclamation", "autre")
            or "rembours" in a.brouillon_reponse.lower()):
        return "à relire en priorité"
    return "à valider"

def test_brouillon_qui_parle_de_remboursement():
    a = tl.Analyse(**dict(analyse(), brouillon_reponse="Nous allons vous Rembourser."))
    assert tl.statut_pour(a) == "à relire en priorité"

Un mot-clé est un contrôle grossier, mais il est prévisible et attrape le cas le plus coûteux.

Résultat obtenu

Tests sans clé puis exécution réelle sur les 20 e-mails fictifs, le 2026-10-15 :

pytest -q

.....                                                                    [100%]
5 passed in 0.62s

python traiter_lot.py emails.csv (extrait)

[1/20] ligne 1 -> reclamation / à relire en priorité
[2/20] ligne 2 -> facturation / à valider
[3/20] ligne 3 -> information / à valider
[4/20] ligne 4 -> rendez-vous / à valider
[5/20] ligne 5 -> reclamation / à relire en priorité
...
[19/20] ligne 19 -> autre / à relire en priorité
[20/20] ligne 20 -> reclamation / à relire en priorité
20 ligne(s) traitée(s) -> emails_traite.csv
statut
à valider               15
à relire en priorité     5

20 appels, environ 23 secondes au total, aucune erreur.

Je sais imposer des valeurs fermées avec Literal et response_schema
Je sais écrire la règle de statut en Python, pas dans la consigne
Mon script sauvegarde après chaque ligne et reprend là où il s'est arrêté
Mes tests passent sans clé