Le script complet : schéma, reprise et tests
Le schéma de réponse
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: strLiteral[...] 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
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
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
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 ?
sauvegarder(df) est appelé après chaque ligne, et la relance sur emails_traite.csv saute les lignes déjà dotées d'un statut. L'original n'est jamais modifié.
Que vérifie test_lot_csv_avec_limite_et_reprise ?
Le test utilise un faux client qui compte ses appels : 3 puis 17, jamais 20 deux fois. La qualité du classement, elle, se juge sur l'exécution réelle.
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 :
..... [100%]
5 passed in 0.62s[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é 520 appels, environ 23 secondes au total, aucune erreur.
- Je sais imposer des valeurs fermées avec
Literaletresponse_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é