Inférence par lots

Ce que vous saurez faire

Attribuez un score à chaque nouveau billet de soutien du jour dans une seule exécution planifiée. L’équipe de Grace aura ainsi sa liste à examiner chaque matin. Quelques lignes incorrectes ne doivent pas arrêter l’exécution. Un fichier défectueux ne doit pas produire une liste qui semble correcte.

Avant de commencerCe qu’il faut savoir et avoir sous la mainEnviron 40 minutes

Ce que vous devez savoir

• Le modèle d’escalade, son seuil de 0,19 et la capacité d’examen de 20 %, présentés dans Seuils et calibration du cours Apprentissage automatique : du problème à un modèle fiable.
• Comment predict.py a attribué des scores au fichier de juillet et s’est arrêté sur une entrée incorrecte, dans Planifier les vérifications en production du cours d’apprentissage automatique.
• Le contrat (TicketIn, clé absente ou valeur nulle), présenté dans Entrées et sorties des prédictions.
• Le paquet et son empreinte, MODEL_SHA256, présentés dans Empaqueter les dépendances du modèle et Confiance et intégrité des artefacts.

Ce qu’il vous faut sur votre ordinateur

• Le projet ~/projects/escalation-service (Windows : $HOME\projects\escalation-service) avec son environnement .venv, activé dans un terminal.
• escalation/contract.py et les tests, provenant de Entrées et sorties des prédictions.
• bundles/escalation-1.0.0/ avec SHA256SUMS, et escalation/bundle.py, provenant de Confiance et intégrité des artefacts. python -m pytest donne 18 passed (18 tests réussis).
• Le fichier ~/projects/ticket-model/data/july_tickets.csv du cours d’apprentissage automatique. Si vous ne l’avez pas, téléchargez july_tickets.csv depuis le dossier de données de ce cours dans les laboratoires.
• Si votre projet manque ou ne fonctionne plus, copiez l’instantané end-of-m01. Utiliser un instantané présente les étapes.

Testé avec Python 3.14.6 et Docker Desktop (Docker Engine 29.6.1) sur macOS, avec une puce Apple. Les étapes pour Windows et Linux ne sont pas testées.

Le courrier arrive une fois par jour

Le courrier arrive une fois par jour. Personne n’attend chaque lettre à la porte : vous les lisez toutes au déjeuner. Une grande partie du travail avec un modèle peut se faire de la même façon. Rassemblez les cas, attribuez-leur des scores ensemble et livrez les résultats avant que les gens en aient besoin.

Le service de soutien de Larkfield fonctionne ainsi aujourd’hui. Chaque nuit à 2 h, l’exportation de Mei écrit un fichier contenant les billets de la veille. Le modèle attribue des scores au fichier avant que l’équipe senior commence à 8 h. L’équipe de Grace examine d’abord les billets signalés. Les données de ce cours sont les données synthétiques de Larkfield du cours d’apprentissage automatique.

Traitement par lots : les scores sont prêts avant que quelqu'un en ait besoin. Le coût est l'attente : un billet peut avoir presque une journée lorsqu'il est signalé.

Description textuelle du diagramme

Trois acteurs : le système de soutien, le traitement par lots d’escalation-service et l’équipe senior. Toute la journée, de nouveaux billets arrivent sans score. À 2 h, le fichier nocturne de Mei contenant les billets d’une journée passe au traitement par lots. Le traitement vérifie chaque ligne, attribue des scores par segments de 500 et écrit trois fichiers : scores.csv, rejected.csv et summary.json. Le traitement de 1 400 billets prend environ 1 seconde. À 8 h, l’équipe senior ouvre scores.csv et voit les billets signalés de la veille. Dernière note : un billet arrivé hier à 9 h 05 est signalé environ 23 heures plus tard.

Regardez d’abord la dernière note. Ce délai est le prix du traitement par lots, et la prochaine leçon l’élimine. Cette leçon rend l’exécution nocturne sûre.

Le problème : une ligne incorrecte, aucune liste

Dans le cours d’apprentissage automatique, predict.py vérifiait d’abord le fichier entier. Toute vérification échouée arrêtait l’exécution avec le code de sortie 1. Aucun score n’était attribué. C’est sûr, mais trop strict pour un fichier qui arrive chaque nuit.

Calculez le coût d’une seule faute de frappe. Le fichier de juillet contient 1 400 billets. Supposons qu’un agent ait saisi urgent dans le champ priority. Cela représente 1 ligne incorrecte sur 1 400, soit 0,07 %. Avec la règle de ce cours, les 1 399 autres billets n’obtiennent aucun score. L’équipe de Grace commence alors la journée sans aucune liste.

L’erreur inverse est pire. Supposons que le traitement ne vérifie rien, ou qu’il plante après avoir écrit la moitié de scores.csv. L’équipe de Grace ouvre un fichier qui semble normal, avec 700 billets plutôt que 1 400. Personne ne voit que la moitié des signalements manque.

Tomás demande : « Pourquoi ne pas parcourir le fichier dans une boucle et appeler le modèle? Ça fait dix lignes. » Les dix lignes donnent des scores. Elles ne répondent pas aux trois questions dont dépend l’examen du matin :

1. Quelles lignes n’ont pas reçu de score, et pourquoi?
2. Le fichier de sortie est-il complet, ou l’exécution s’est-elle arrêtée à mi-chemin?
3. Quel modèle a attribué les scores à ce fichier, et une deuxième exécution donne-t-elle les mêmes scores?

L’idée : l’attribution de scores par lots

L’attribution de scores par lots traite plusieurs cas ensemble dans une seule exécution, généralement planifiée, puis stocke les résultats. Ces résultats serviront plus tard. Personne n’attend une réponse individuelle. Le nombre important est donc le débit, soit le nombre de cas traités par seconde, plutôt que la durée d’un cas. Sur l’ordinateur de test, le traitement a attribué des scores aux 1 400 billets de juillet en 0,75 seconde. Cette durée comprend le chargement du modèle. Cela représente environ 1 900 billets par seconde.

Un traitement par lots sûr répond aux trois questions avec trois règles.

Isolez les lignes incorrectes, mais arrêtez le traitement d’un fichier défectueux. Chaque ligne passe par le contrat du module 1. Une ligne correcte reçoit un score. Une ligne incorrecte va dans rejected.csv, avec son numéro de ligne et le problème, puis l’exécution continue. Mais si beaucoup de lignes sont incorrectes, le fichier lui-même est probablement défectueux, par exemple après une modification de l’exportation de Mei. Le traitement impose donc une limite. Si plus de 5 % des lignes sont incorrectes, il échoue avec le code de sortie 1 et n’écrit aucun scores.csv. C’est la gestion des échecs partiels : un petit échec reste limité, et un échec important arrête l’exécution.

Ne montrez jamais la moitié d’un fichier. Le traitement écrit chaque sortie sous un nom temporaire, comme scores.csv.partial. À la fin, il renomme le fichier en scores.csv. Le système de fichiers effectue le renommage en une seule étape. Le lecteur voit donc soit aucun fichier, soit le fichier complet. C’est une écriture atomique : elle se fait entièrement, ou pas du tout.

Consignez ce qui a produit la sortie. Chaque exécution écrit summary.json. Ce fichier consigne les lignes traitées et rejetées, le taux de signalement, la version du modèle et l’empreinte du paquet. Il consigne aussi l’empreinte SHA-256 du fichier d’entrée. Chaque exécution écrit aussi dans son propre dossier, comme out/2026-07-31/. Une nouvelle exécution n’écrase donc jamais un ancien résultat.

Échec partiel dans le traitement par lots. Une colonne manquante arrête immédiatement l'exécution. Les lignes incorrectes sont isolées. Trop de lignes incorrectes font échouer l'exécution. Ainsi, un fichier défectueux ne produit jamais de liste à examiner.

Description textuelle du diagramme

Le fichier de Mei mène à une première question : toutes les colonnes sont-elles présentes? Sinon, l’exécution s’arrête avec le code de sortie 2, et rien n’est écrit. Si toutes les colonnes sont présentes, le contrat vérifie chaque ligne. Les lignes correctes reçoivent un score, 500 à la fois. Les lignes incorrectes vont dans rejected.csv, avec le numéro de ligne, le ticket_id et le problème. Puis vient une deuxième question : les lignes incorrectes représentent-elles au plus 5 % du fichier? Si oui, l’exécution se termine avec le code de sortie 0, et scores.csv.partial est renommé en scores.csv. Sinon, l’exécution échoue avec le code de sortie 1, et aucun scores.csv n’existe. summary.json est écrit dans les deux cas.

Créer le traitement par lots

Le traitement par lots nécessite trois nouveaux fichiers dans escalation/. Chacun a un seul rôle, et le service du module 3 réutilise les deux premiers.

settings.py : les paramètres de l’environnement

Le traitement doit savoir quel paquet charger et quelle empreinte attendre. Comme dans les cours sur les API et Docker, les paramètres viennent de variables d’environnement. Un paramètre incorrect arrête le programme avant tout travail. Créez escalation/settings.py :

escalation/settings.py

"""Settings from environment variables. A wrong setting stops the service at start."""

import os
from dataclasses import dataclass, field
from pathlib import Path


class SettingsError(Exception):
    """A required setting is missing or not valid. The service must not start."""


@dataclass(frozen=True)
class Settings:
    model_bundle: Path
    model_sha256: str
    api_key: str | None = field(repr=False)
    log_level: str = "INFO"
    show_docs: bool = True
    max_job_tickets: int = 5000
    max_queued_jobs: int = 4
    extra_ms_per_ticket: float = 0.0


def read_secret(name: str) -> str | None:
    """Return the value of NAME, or the contents of the file in NAME_FILE."""
    path = os.environ.get(f"{name}_FILE")
    if path:
        try:
            return Path(path).read_text(encoding="utf-8").strip() or None
        except OSError as error:
            raise SettingsError(f"{name}_FILE: cannot read {path}: {error.strerror}")
    return os.environ.get(name) or None


def load_settings() -> Settings:
    env = os.environ
    sha = env.get("MODEL_SHA256", "").strip().lower()
    if len(sha) != 64:
        raise SettingsError("MODEL_SHA256 must be the 64-character digest of the bundle")
    settings = Settings(
        model_bundle=Path(env.get("MODEL_BUNDLE", "bundles/escalation-1.0.0")),
        model_sha256=sha,
        api_key=read_secret("API_KEY"),
        log_level=env.get("LOG_LEVEL", "INFO").upper(),
        show_docs=env.get("SHOW_DOCS", "true").lower() == "true",
        max_job_tickets=int(env.get("MAX_JOB_TICKETS", "5000")),
        max_queued_jobs=int(env.get("MAX_QUEUED_JOBS", "4")),
        extra_ms_per_ticket=float(env.get("EXTRA_MS_PER_TICKET", "0")),
    )
    if env.get("REQUIRE_API_KEY", "false").lower() == "true" and not settings.api_key:
        raise SettingsError("REQUIRE_API_KEY is true, but API_KEY is not set")
    return settings

Le traitement par lots utilise seulement les deux premiers paramètres. MODEL_BUNDLE a une valeur par défaut. MODEL_SHA256 n’en a pas, car l’empreinte doit venir de vous, et non du paquet. La clé d’API et les paramètres de documentation servent au service HTTP du module 3. Les trois paramètres de traitement servent à l’Inférence asynchrone.

scoring.py : une fonction pour chaque mode

Créez escalation/scoring.py. Le traitement par lots, l’API et les traitements appellent tous cette fonction. Chaque mode attribue donc un score à un billet exactement de la même façon :

escalation/scoring.py

"""Score tickets with a loaded bundle. The API and the batch job both use this.

The work for each request is split into three timed steps, so that a
benchmark can show where the time goes:

    frame     the checked tickets → a table with the training columns
    prepare   the pipeline's preprocessing (one-hot, impute, scale)
    model     the logistic regression's score
"""

import time
from dataclasses import dataclass, field

from escalation.bundle import Bundle
from escalation.contract import TicketFeatures, to_frame, unknown_categories


@dataclass
class Result:
    score: float
    flag: bool
    warnings: list[str]


@dataclass
class Timings:
    """Milliseconds spent in each step of one call."""

    steps: dict[str, float] = field(default_factory=dict)

    def header(self) -> str:
        """The value of a Server-Timing header, for example "frame;dur=0.41"."""
        return ", ".join(f"{name};dur={ms:.2f}" for name, ms in self.steps.items())


def score_tickets(bundle: Bundle, tickets: list[TicketFeatures]) -> tuple[list[Result], Timings]:
    timings = Timings()
    started = time.perf_counter()

    frame = to_frame(tickets)
    now = time.perf_counter()
    timings.steps["frame"] = (now - started) * 1000

    preprocess, model = bundle.pipeline[:-1], bundle.pipeline[-1]
    prepared = preprocess.transform(frame)
    started, now = now, time.perf_counter()
    timings.steps["prepare"] = (now - started) * 1000

    scores = model.predict_proba(prepared)[:, 1]
    started, now = now, time.perf_counter()
    timings.steps["model"] = (now - started) * 1000

    results = [
        Result(
            score=float(score),
            flag=bool(score >= bundle.threshold),
            warnings=unknown_categories(ticket, bundle.known),
        )
        for ticket, score in zip(tickets, scores)
    ]
    return results, timings

score_tickets prend une liste de billets vérifiés et chronomètre ses trois étapes. Le traitement par lots ignore les durées; la prochaine leçon les utilise. Le pipeline est divisé uniquement pour chronométrer les étapes. pipeline[:-1] correspond au prétraitement et pipeline[-1] au modèle. Le score est donc identique à celui de pipeline.predict_proba.

batch.py : le traitement

Créez escalation/batch.py. La première partie lit et vérifie les lignes :

escalation/batch.py (part 1 of 2)

"""Score a file of tickets with the same bundle and contract as the API.

    python -m escalation.batch data/july_tickets.csv out/2026-07

Writes three files into the output folder:

    scores.csv      ticket_id, score, flag, model_version (one row per good ticket)
    rejected.csv    row, ticket_id, problem (one row per ticket that fails the contract)
    summary.json    what was read, what was written, with which model, and how long it took

A bad row does not stop the job: it is set aside in rejected.csv, and the
other rows are scored. But if more than --max-rejected of the rows are bad,
the input itself is probably broken, so the job fails (exit code 1) and
writes no scores.csv. Files are written under a temporary name and renamed at
the end, so a reader never sees half a file.
"""

import argparse
import hashlib
import json
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

import pandas as pd
from pydantic import ValidationError

from escalation.bundle import BundleError, load_bundle
from escalation.contract import FEATURES, TicketIn
from escalation.scoring import score_tickets
from escalation.settings import SettingsError, load_settings

COLUMNS = ["ticket_id"] + FEATURES


def problem(error: ValidationError) -> str:
    first = error.errors()[0]
    where = ".".join(str(p) for p in first["loc"]) or "row"
    return f"{where}: {first['msg']}"


def check_rows(chunk: pd.DataFrame, first_row: int):
    good, bad = [], []
    for offset, row in enumerate(chunk.to_dict("records")):
        values = {k: (None if pd.isna(v) else v) for k, v in row.items()}
        try:
            good.append(TicketIn(**values))
        except ValidationError as error:
            bad.append({"row": first_row + offset, "ticket_id": values.get("ticket_id"), "problem": problem(error)})
    return good, bad


def write_atomic(frame: pd.DataFrame, path: Path) -> None:
    temporary = path.with_suffix(path.suffix + ".partial")
    frame.to_csv(temporary, index=False)
    temporary.replace(path)

Trois détails comptent. check_rows fait passer chaque ligne par TicketIn, le même contrat que l’API. Un billet refusé par l’API est donc aussi refusé ici. Une cellule vide devient None (valeur nulle), et non 0. Le composant d’imputation du pipeline doit reconnaître une valeur manquante. write_atomic écrit le fichier .partial, puis le renomme avec replace.

La deuxième partie exécute le traitement :

escalation/batch.py (part 2 of 2)

def main() -> int:
    parser = argparse.ArgumentParser(description="Score a CSV file of tickets.")
    parser.add_argument("input", type=Path)
    parser.add_argument("output", type=Path, help="a folder for scores.csv, rejected.csv and summary.json")
    parser.add_argument("--chunk", type=int, default=500, help="tickets per chunk")
    parser.add_argument("--max-rejected", type=float, default=0.05, help="the largest share of bad rows")
    args = parser.parse_args()

    started_at = datetime.now(timezone.utc)
    started = time.perf_counter()
    try:
        settings = load_settings()
        bundle = load_bundle(settings.model_bundle, settings.model_sha256)
    except (SettingsError, BundleError) as error:
        print(f"Start-up error: {error}", file=sys.stderr)
        return 2
    try:
        reader = pd.read_csv(args.input, usecols=COLUMNS, dtype=str, keep_default_na=False,
                             na_values=[""], chunksize=args.chunk)
    except ValueError as error:  # a missing column
        print(f"Stopped: {error}", file=sys.stderr)
        return 2

    scored, rejected, rows = [], [], 0
    for chunk in reader:
        good, bad = check_rows(chunk, first_row=rows + 2)  # row 1 is the header
        rows += len(chunk)
        rejected += bad
        if good:
            results, _ = score_tickets(bundle, good)
            scored += [{"ticket_id": t.ticket_id, "score": r.score, "flag": int(r.flag),
                        "model_version": bundle.version} for t, r in zip(good, results)]

    args.output.mkdir(parents=True, exist_ok=True)
    scores = pd.DataFrame(scored, columns=["ticket_id", "score", "flag", "model_version"])
    write_atomic(pd.DataFrame(rejected, columns=["row", "ticket_id", "problem"]), args.output / "rejected.csv")
    failed = rows == 0 or len(rejected) / rows > args.max_rejected
    if not failed:
        write_atomic(scores, args.output / "scores.csv")
    summary = {
        "input": str(args.input),
        "input_sha256": hashlib.sha256(args.input.read_bytes()).hexdigest(),
        "rows": rows,
        "scored": len(scores),
        "rejected": len(rejected),
        "flagged": int(scores["flag"].sum()),
        "flag_rate": round(float(scores["flag"].mean()), 3) if len(scores) else None,
        "model_version": bundle.version,
        "bundle_sha256": bundle.digest,
        "started_at": started_at.isoformat(timespec="seconds"),
        "seconds": round(time.perf_counter() - started, 2),
        "status": "failed: too many rejected rows" if failed else "done",
    }
    (args.output / "summary.json").write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8")
    print(f"Read {rows} rows: scored {len(scores)}, rejected {len(rejected)}, "
          f"flagged {summary['flagged']} ({summary['flag_rate']}). Model {bundle.version}. {summary['seconds']} s.")
    if failed:
        print(f"Failed: {len(rejected)} of {rows} rows were rejected (the limit is {args.max_rejected:.0%}). "
              f"No scores.csv was written. See {args.output / 'rejected.csv'}.", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())

Le traitement charge le paquet une seule fois, avec les six vérifications de load_bundle, avant de lire la moindre ligne. Il lit le fichier par segments de 500. Ainsi, un fichier d’un million de lignes n’exige jamais un million de lignes en mémoire. Le code de sortie indique au planificateur ce qui s’est passé : 0 signifie terminé, 1 signifie trop de lignes incorrectes. 2 signifie que le traitement n’a pas pu démarrer, ou que le fichier a les mauvaises colonnes.

Exemple détaillé : 20 billets, puis juillet

Amira commence avec peu de données, pour que chaque nombre soit visible. Elle copie l’en-tête et les 20 premiers billets de juillet dans data/twenty.csv. À la ligne 4 (billet T-119235), elle remplace la priorité 3 par urgent. Une ligne incorrecte sur 20 représente exactement 5 %, la limite.

Prédisez : l’exécution se termine-t-elle? Combien de billets reçoivent un score?

Terminal, in ~/projects/escalation-service

python -m escalation.batch data/twenty.csv out/twenty

La sortie :

Read 20 rows: scored 19, rejected 1, flagged 3 (0.158). Model 1.0.0. 1.16 s.

out/twenty/rejected.csv indique la ligne, le billet et le problème :

row,ticket_id,problem
4,T-119235,"priority: Input should be a valid integer, unable to parse string as an integer"

scores.csv contient 19 lignes. T-119235 n’y figure pas, et les autres billets conservent leur ordre :

ticket_id,score,flag,model_version
T-119233,0.037624397312637836,0,1.0.0
T-119234,0.255169256886106,1,1.0.0
T-119236,0.06880610647988575,0,1.0.0
...

La limite dit « plus de 5 % », et 1 sur 20 ne la dépasse pas. Amira remplace maintenant aussi created_hour par 24 à la ligne 9. Deux lignes incorrectes sur 20 représentent 10 %. Elle exécute le traitement avec la sortie dans out/twenty-two.

La sortie :

Read 20 rows: scored 18, rejected 2, flagged 2 (0.111). Model 1.0.0. 0.79 s.
Failed: 2 of 20 rows were rejected (the limit is 5%). No scores.csv was written. See out/twenty-two/rejected.csv.

Le code de sortie est 1. out/twenty-two/ contient seulement rejected.csv, avec les lignes 4 et 9, et summary.json, avec "status": "failed: too many rejected rows" (état : échec, trop de lignes rejetées). Le traitement a attribué des scores à 18 billets, mais il ne les écrit pas. L’équipe de Grace ne devrait pas se fier à une liste où 10 % des billets manquent.

Passons au vrai fichier. Le fichier nocturne de Mei contient environ 45 billets. Pour tester avec davantage de données, Amira attribue des scores aux 1 400 billets de juillet dans une seule exécution :

Terminal

python -m escalation.batch data/july_tickets.csv out/2026-07

La sortie :

Read 1400 rows: scored 1400, rejected 0, flagged 282 (0.201). Model 1.0.0. 0.75 s.

282 billets signalés, soit 20,1 % : le même résultat que predict.py du cours d’apprentissage automatique pour juillet. Le même modèle avec le même prétraitement donne la même réponse. summary.json consigne l’exécution :

out/2026-07/summary.json

{
  "input": "data/july_tickets.csv",
  "input_sha256": "10dffc093993341a361172d0aa6ea0e7c06521f140cc2a80ee90c713361f7c99",
  "rows": 1400,
  "scored": 1400,
  "rejected": 0,
  "flagged": 282,
  "flag_rate": 0.201,
  "model_version": "1.0.0",
  "bundle_sha256": "8f233029aa4c896dce1a5b3dfa970cd085088386bbe7784519f15a5b1a433b74",
  "started_at": "2026-10-08T14:52:07+00:00",
  "seconds": 0.75,
  "status": "done"
}

input_sha256 identifie le fichier d’entrée exact, et bundle_sha256 identifie le modèle exact. Ensemble, ils répondent à la question de Grace : « Quel modèle a attribué les scores à cette liste, et à partir de quel fichier? » Votre bundle_sha256 diffère, car votre paquet contient votre propre date et heure d’entraînement. Seul le paquet du cours dans les laboratoires a l’empreinte 8f233029…. Votre valeur seconds diffère aussi.

Une nouvelle exécution. Omar demande : « Si le traitement s’exécute deux fois par erreur, on obtient deux listes différentes? » Amira l’exécute de nouveau dans out/2026-07-again et compare les deux fichiers scores.csv :

Terminal (Windows: Get-FileHash out\2026-07\scores.csv, out\2026-07-again\scores.csv)

shasum -a 256 out/2026-07/scores.csv out/2026-07-again/scores.csv

La sortie :

2b89ad330e0be98c2fadf3647c83304f6f134fac21511e029234f8d6fec18515  out/2026-07/scores.csv
2b89ad330e0be98c2fadf3647c83304f6f134fac21511e029234f8d6fec18515  out/2026-07-again/scores.csv

Le même fichier d’entrée et le même paquet donnent les mêmes octets. Une nouvelle exécution est donc sûre. Si l’exécution de 2 h échoue, le planificateur peut la relancer.

Quand choisir le traitement par lots

L’attribution de scores par lots convient quand les gens ont besoin des résultats plus tard, à une heure connue. Elle convient aussi quand plusieurs cas arrivent ensemble. Exemples : une liste à examiner le matin, un rapport mensuel, une liste d’envoi. C’est le mode le plus simple à exploiter. Un planificateur lance une commande, et rien ne doit rester actif toute la journée. Si l’exécution échoue, vous la relancez avant que quelqu’un ait besoin des résultats.

Ce mode ne convient pas quand une personne ou un programme attend une réponse immédiate. Chez Larkfield, un billet arrivé à 9 h 05 est signalé le lendemain matin, environ 23 heures plus tard. Pour un billet qui peut nécessiter une escalade dans les 72 heures, c’est un tiers du temps écoulé.

Les coûts : le délai, un planificateur pour lancer le traitement, le stockage des dossiers de sortie et une personne qui lit rejected.csv. Un traitement qui échoue la nuit nécessite aussi une alerte. Sinon, personne ne le remarque avant 8 h.

Tomás demande : « La prochaine leçon attribue un score à chaque billet immédiatement. Alors pourquoi créer un traitement par lots? » Parce que le traitement par lots reste une solution de secours. Si le service en ligne tombe en panne pendant une heure, l’exécution nocturne attribue quand même un score à chaque billet du jour.

Exercice guidé : exécuter le traitement par lots

1. Créez les trois fichiers de Créer le traitement par lots dans escalation/ : settings.py, scoring.py et batch.py.

2. Créez le dossier data/ dans le projet et copiez-y le fichier de juillet :

   macOS / Linux

   mkdir -p data
   cp ../ticket-model/data/july_tickets.csv data/

   Windows (PowerShell)

   New-Item -ItemType Directory -Force data
   Copy-Item ..\ticket-model\data\july_tickets.csv data\

3. Définissez MODEL_SHA256 avec l’empreinte de votre paquet. make_bundle.py l’a affichée dans Confiance et intégrité des artefacts. Sur votre ordinateur, vous pouvez aussi la calculer à partir de SHA256SUMS :

   macOS / Linux

   export MODEL_SHA256=$(shasum -a 256 bundles/escalation-1.0.0/SHA256SUMS | cut -c1-64)

   Windows (PowerShell)

   $env:MODEL_SHA256 = (Get-FileHash bundles\escalation-1.0.0\SHA256SUMS -Algorithm SHA256).Hash.ToLower()

4. Attribuez des scores au fichier de juillet :

   python -m escalation.batch data/july_tickets.csv out/2026-07

   La ligne commence par Read 1400 rows: scored 1400, rejected 0, flagged 282 (0.201). (1 400 lignes lues, 1 400 traitées, 0 rejetée, 282 signalées, soit 0,201).

5. Ouvrez out/2026-07/summary.json. Vérifiez que input_sha256 commence par 10dffc09 et que bundle_sha256 égale votre MODEL_SHA256.

6. Donnez aux tests leurs propres paramètres. Le traitement lit MODEL_BUNDLE et MODEL_SHA256 dans l’environnement. Les tests doivent donc les définir, plutôt que votre terminal. Dans tests/conftest.py, ajoutez cette importation après import pytest :

   tests/conftest.py

   from escalation.bundle import bundle_digest

   Ajoutez ensuite cette fixture (fonction de préparation des tests) à la fin du fichier :

   tests/conftest.py

   @pytest.fixture(autouse=True)
   def settings_env(monkeypatch):
       monkeypatch.chdir(ROOT)
       monkeypatch.setenv("MODEL_BUNDLE", str(BUNDLE))
       monkeypatch.setenv("MODEL_SHA256", bundle_digest(BUNDLE))
       for name in ["API_KEY", "API_KEY_FILE", "REQUIRE_API_KEY", "SHADOW_BUNDLE", "SHADOW_SHA256"]:
           monkeypatch.delenv(name, raising=False)

   autouse=True exécute la fixture avant chaque test. Elle calcule l’empreinte du paquet du projet et supprime les paramètres des modules suivants. Ainsi, une valeur dans votre terminal ne peut pas modifier un test. Le traitement qu’un test lance comme commande reçoit les mêmes paramètres.

7. Créez tests/test_batch.py. Les deux tests exécutent le traitement comme commande, exactement comme le planificateur :

   tests/test_batch.py

   import json
   import subprocess
   import sys
   
   import pandas as pd
   
   from tests.conftest import TICKET
   
   
   def run_batch(tmp_path, rows):
       source = tmp_path / "tickets.csv"
       pd.DataFrame(rows).to_csv(source, index=False)
       out = tmp_path / "out"
       result = subprocess.run([sys.executable, "-m", "escalation.batch", str(source), str(out)],
                               capture_output=True, text=True)
       return result, out
   
   
   def test_bad_rows_are_set_aside(tmp_path):
       rows = [{**TICKET, "ticket_id": f"T-{120000 + i}"} for i in range(40)]
       rows[5]["priority"] = "high"
       result, out = run_batch(tmp_path, rows)
       assert result.returncode == 0, result.stderr
       assert len(pd.read_csv(out / "scores.csv")) == 39
       rejected = pd.read_csv(out / "rejected.csv")
       assert rejected.to_dict("records")[0]["row"] == 7
       assert json.loads((out / "summary.json").read_text())["status"] == "done"
   
   
   def test_too_many_bad_rows_fail_the_job(tmp_path):
       rows = [{**TICKET, "ticket_id": f"T-{120000 + i}", "created_hour": 25} for i in range(10)]
       result, out = run_batch(tmp_path, rows)
       assert result.returncode == 1
       assert not (out / "scores.csv").exists()
       assert len(pd.read_csv(out / "rejected.csv")) == 10

   Le premier test place une ligne incorrecte parmi 40, à l’indice 5. L’en-tête est la ligne 1; la ligne incorrecte est donc la ligne 7 du fichier. Le deuxième test rend toutes les lignes incorrectes.

8. Exécutez tous les tests avec python -m pytest. La dernière ligne indique 20 passed (20 tests réussis).

Afficher la sortie attendue des tests de l’étape 8

tests/test_batch.py ..                                                   [100%]
...
============================== 20 passed in 2.47s ==============================

Les 18 tests du module 1 et les 2 nouveaux réussissent. Ils réussissent aussi dans un nouveau terminal, sans MODEL_SHA256 : la fixture settings_env la définit pour chaque test.

À vous de jouer : deux cellules vides

Dans l’exemple détaillé, les mauvaises valeurs étaient des fautes de frappe. Essayez maintenant des cellules vides. Créez un nouveau data/twenty.csv avec l’en-tête et les 20 premiers billets de juillet :

macOS / Linux

head -21 data/july_tickets.csv > data/twenty.csv

Windows (PowerShell)

Get-Content data\july_tickets.csv -TotalCount 21 | Set-Content data\twenty.csv

Ouvrez-le dans VS Code. À la ligne 4 (T-119235), supprimez la valeur de order_value, 186.4, et gardez les virgules. À la ligne 9 (T-119240), supprimez la valeur de team, returns (retours). Avant d’exécuter python -m escalation.batch data/twenty.csv out/twenty-empty, notez votre prédiction. Quelles lignes seront rejetées, quel sera le code de sortie et combien de billets contiendra scores.csv?

Afficher la réponse

La sortie :

Read 20 rows: scored 19, rejected 1, flagged 2 (0.105). Model 1.0.0. 0.8 s.

rejected.csv contient une ligne :

row,ticket_id,problem
9,T-119240,team: Input should be a valid string

Une cellule vide devient une valeur nulle. Le contrat autorise une valeur nulle pour order_value (aucune commande), donc la ligne 4 reçoit un score. T-119235 obtient 0,0978 au lieu de 0,1061 avec sa commande. Le contrat n’autorise pas de valeur nulle pour team, donc la ligne 9 est rejetée. Un billet sur 20 représente 5 %, pas plus. Le code de sortie est donc 0, et scores.csv contient 19 billets. Pour voir le code de sortie, exécutez echo $? sur macOS et Linux, ou $LASTEXITCODE dans PowerShell.

Erreur courante

Le traitement s'arrête avec Stopped : Usecols do not match columns (arrêt : les colonnes demandées ne correspondent pas)

Symptôme : une nuit, le traitement n’écrit rien du tout et se termine avec le code 2 :

Stopped: Usecols do not match columns, columns expected but not found: ['prior_escalations_90d']

Cause : l’exportation de Mei a perdu une colonne. Ici, un changement dans l’exportation a omis prior_escalations_90d. Le traitement lit seulement les colonnes du contrat (usecols=COLUMNS), et pandas refuse un fichier auquel il en manque une. Aucune ligne n’est vérifiée, car sans cette colonne, aucune ligne ne peut recevoir un score correct.

Correction : corrigez l’exportation, puis relancez le traitement dans le même dossier daté. Ne « réparez » pas le fichier avec une colonne de zéros. Le modèle a appris que prior_escalations_90d compte. Mettre 0 à chaque ligne change discrètement tous les scores.

Pour un vrai service d’IA : une colonne manquante est un problème du fichier, pas d’une ligne. Arrêtez toute l’exécution, avec un message clair, avant de calculer les scores. Une valeur par défaut silencieuse amorce un écart entre l’entraînement et le service, que le module 3 vérifie.

Vérifiez votre compréhension

Le fichier de Mei contient 1 000 billets, dont 30 ont created_hour égal à 24. Que fait le traitement, et que voit l'équipe de Grace le matin?

Afficher la réponse

30 sur 1 000 représente 3 %, ce qui ne dépasse pas la limite de 5 %. Le traitement se termine donc avec le code de sortie 0. scores.csv contient les 970 billets valides. rejected.csv liste les 30 billets invalides avec leurs numéros de ligne et le problème created_hour: Input should be less than or equal to 23 (la valeur doit être inférieure ou égale à 23). summary.json indique "rejected": 30 (30 rejets) et "status": "done" (état : terminé). L’équipe de Grace a sa liste, et quelqu’un devrait examiner les 30 billets rejetés manuellement.

Vérifiez votre compréhension

Pourquoi le traitement écrit-il scores.csv.partial, puis le renomme-t-il, au lieu d'écrire directement scores.csv?

Afficher la réponse

Une exécution peut s’arrêter en cours de route, par exemple si l’ordinateur redémarre. Si le traitement écrivait directement scores.csv, un lecteur pourrait trouver un fichier incomplet qui semble complet. Le système de fichiers renomme le fichier en une seule étape. Ainsi, scores.csv est soit absent, soit complet. L’écriture atomique protège le lecteur contre une liste à laquelle il manque des billets.

Vérifiez votre compréhension

Deux exécutions le même jour ont produit deux fichiers scores.csv avec des valeurs SHA-256 différentes. Quels sont les deux champs de summary.json à comparer d'abord, et pourquoi?

Afficher la réponse

Comparez input_sha256 et bundle_sha256. Avec les mêmes données d’entrée et le même paquet, le traitement produit les mêmes octets. La nouvelle exécution dans cette leçon l’a montré. Une sortie différente signifie donc que le fichier d’entrée a changé ou qu’une autre version du modèle a calculé les scores. Par exemple, l’exportation de Mei a été relancée avec plus de billets.

De bout en boutPrédire le défaut de crédit par lots et en ligneÉtude de cas

L’étude de cas exécute un traitement mensuel par lots pour tous les clients avec les mêmes règles. Elle utilise des lignes rejetées, une limite d’échec et un résumé.

Récapitulatif

• Le calcul des scores par lots traite plusieurs cas dans une seule exécution planifiée et conserve les résultats pour plus tard. Le débit, soit le nombre de cas par seconde, compte davantage que le temps pour un seul cas : 1 400 billets en 0,75 s.
• Les lignes invalides vont dans rejected.csv avec leur numéro de ligne et leur problème. Plus de 5 % de lignes invalides font échouer l’exécution, sans scores.csv. Une colonne manquante l’arrête immédiatement.
• Les fichiers sont écrits avec l’extension .partial, puis renommés, pour que personne ne lise un fichier incomplet.
• summary.json consigne l’empreinte des données d’entrée, l’empreinte du paquet et l’état. Les mêmes données d’entrée et le même paquet produisent la même sortie. Une nouvelle exécution est donc sûre.
• Le coût est l’attente : un billet peut dater de presque un jour quand son indicateur arrive.

Ensuite, vous calculez le score d’un billet au moment où un agent l’ouvre. Vous convenez aussi du temps d’attente acceptable pour le service de soutien.

Leçon suivante : Inférence en ligne