Connecter l'API

Leçon 8
40 min
Module 3 : Créer le service de prédiction

Leçon 8 sur 15 · Déployer des modèles d'IA pour de vrais utilisateursConnecter l'API

Aperçu du cours

1. Définir le contrat d'inférence

2. Choisir un mode de service

3. Créer le service de prédiction

4. Choisir l'infrastructure et mesurer

5. Mettre en production, surveiller et rétablir

6. Études de cas

Ce que vous saurez faire

Placez le modèle derrière quatre routes HTTP. Elles vérifient chaque billet de soutien selon le contrat et renvoient les erreurs dans un format stable. Chaque réponse indique le modèle et le seuil.

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

Ce que vous devez savoir

Le contrat : TicketIn, clé absente ou valeur nulle, catégories inconnues avec avertissement, dans Entrées et sorties des prédictions : écrire le contrat.
score_tickets() et jobs.py (JobRunner, QueueFull), dans Inférence par lots et Inférence asynchrone.
main.py et le cycle de vie, dans Charger et initialiser : écrire main.py.
Le format d’erreur unique, dans Validation et réponses d’erreur : trois structures, c’est deux de trop, du cours Créer et déployer votre première API. La clé d’API, dans Protection de base : la clé d’API, du cours sur les API.

Ce qu’il vous faut sur votre ordinateur

Le projet ~/projects/escalation-service avec escalation/main.py et pyproject.toml, de Charger et initialiser : écrire main.py.
MODEL_SHA256 défini dans votre terminal, comme dans Charger et initialiser : Démarrer le service.
Si votre projet manque ou ne fonctionne plus, copiez l’instantané end-of-m02 et refaites Charger et initialiser. 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 n’ont pas été testées.

Un reçu avec chaque réponse

À la pharmacie, vous remettez une ordonnance. Si un champ manque, le pharmacien vous la rend et indique le champ. Si le formulaire est complet, vous recevez le médicament avec une étiquette imprimée : le nom, la dose, la date et la pharmacie. Des mois plus tard, n’importe qui peut lire l’étiquette et savoir ce que vous avez reçu, et de qui.

Une API de prédiction a besoin des deux mêmes choses. Elle rejette un billet qui ne respecte pas le contrat et indique le champ. Chaque réponse porte aussi son étiquette : quel modèle, quel seuil, quelle requête.

Le problème : un score sans étiquette

Supposons que l’API réponde seulement {"score": 0.35}. Le service de soutien doit décider : signaler le billet ou non? Avec la version 1.0.0, le seuil est de 0,19, donc un score de 0,35 déclenche un signalement. Au module 5, la version 1.1.0 utilise un seuil de 0,20. La version candidate 1.1.0-rc1 de Tomás augmente tous les scores. Avec l’ancien seuil de 0,19, elle signale 86 % des billets de juillet au lieu de 20 %. Le même nombre signifie différentes choses selon le modèle.

Un score seul pose un deuxième problème. Quand Grace demande en août « pourquoi le billet T-120001 a-t-il été signalé? », personne ne peut identifier le modèle utilisé. Un client qui reçoit trois formats d’erreur a aussi besoin de trois morceaux de code pour les lire. Vous l’avez vu dans le cours sur les API.

L’idée : la réponse contient son contexte

L’API suit donc trois règles :

Vérifier avant de calculer le score. Un billet qui ne respecte pas le contrat reçoit 422 et les noms des champs incorrects. Le modèle ne le voit jamais.
Un seul format d’erreur. Chaque erreur, de 401 à 500, a le même corps : code, message, request_id et fields. Un client lit error.code, jamais le texte.
Chaque score indique sa source. La réponse contient score, flag, threshold et model_version, plus warnings pour les valeurs que le modèle n’a jamais vues. Le service écrit aussi une ligne de journal (log) par score, avec l’identifiant de la requête.

Le diagramme suit une requête à travers ces règles :

Une requête traverse le service. Chaque chemin aboutit à la réponse 200 ou au format d'erreur unique.

Description textuelle du diagramme

Une requête à POST /v1/score reçoit d’abord un identifiant de requête. La vérification de la clé d’API vient ensuite : une clé absente ou incorrecte mène à 401 unauthorized (non autorisé). Puis vient la vérification du contrat : un billet non conforme mène à 422 invalid_ticket (billet invalide). Un billet conforme passe à score_tickets(), qui utilise l’ensemble empaqueté en mémoire. La fonction produit la réponse 200 avec score, flag, threshold, model_version et warnings, et écrit une ligne de journal de prédiction. Une erreur pendant le calcul mène à 500 internal_error (erreur interne). Les trois réponses d’erreur ont un seul format : code, message, request_id et fields.

Ce sont les pratiques du cours sur les API, tirées de ticket-api et appliquées à un modèle. Les nouveautés sont le seuil et la version du modèle dans chaque réponse, les avertissements et le journal de prédiction.

errors.py : un format pour chaque erreur

Créez escalation/errors.py. Il contient le format et trois gestionnaires, comme dans le cours sur les API :

"""One error shape for every failure, as in the ticket API:
{"error": {"code": ..., "message": ..., "request_id": ..., "fields": [...]}}"""

import logging

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from starlette.exceptions import HTTPException

logger = logging.getLogger("escalation")

CODES = {401: "unauthorized", 404: "not_found", 405: "method_not_allowed", 429: "too_many_jobs"}


class ErrorDetail(BaseModel):
    code: str = Field(description="A stable code that a program can check.")
    message: str = Field(description="A short explanation for a person.")
    request_id: str | None = None
    fields: list[str] = []


class ErrorResponse(BaseModel):
    error: ErrorDetail


def field_name(error: dict) -> str:
    if error["type"] == "json_invalid":
        return "body"
    return ".".join(str(part) for part in error["loc"][1:]) or "body"


def error_response(request: Request, status: int, code: str, message: str, fields=()) -> JSONResponse:
    detail = ErrorDetail(
        code=code,
        message=message,
        request_id=getattr(request.state, "request_id", None),
        fields=list(fields),
    )
    return JSONResponse(status_code=status, content=ErrorResponse(error=detail).model_dump())


def add_error_handlers(app: FastAPI) -> None:
    @app.exception_handler(RequestValidationError)
    async def invalid_request(request: Request, exc: RequestValidationError):
        fields = sorted({field_name(e) for e in exc.errors()})
        return error_response(
            request, 422, "invalid_ticket", "The ticket does not match the model's contract.", fields
        )

    @app.exception_handler(HTTPException)
    async def http_error(request: Request, exc: HTTPException):
        code = CODES.get(exc.status_code, "http_error")
        return error_response(request, exc.status_code, code, str(exc.detail))

    @app.exception_handler(Exception)
    async def unexpected(request: Request, exc: Exception):
        logger.exception("unexpected error id=%s", getattr(request.state, "request_id", None))
        return error_response(request, 500, "internal_error", "Something went wrong on the server.")

invalid_request transforme chaque erreur de contrat en 422 invalid_ticket, avec les noms triés des champs incorrects. Un corps qui n’est pas du tout en JSON donne le champ body (corps). unexpected consigne l’erreur complète pour vous, mais envoie au client seulement un court message et l’identifiant de requête. Les détails internes du serveur restent internes.

security.py : la clé d’API

Créez escalation/security.py. C’est la vérification de clé du cours sur les API :

import secrets

from fastapi import Request
from fastapi.security import APIKeyHeader
from starlette.exceptions import HTTPException

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)


def require_api_key(request: Request) -> None:
    """Allow the request only with the right X-API-Key, if a key is set."""
    expected = request.app.state.settings.api_key
    if expected is None:
        return
    given = request.headers.get("X-API-Key") or ""
    if not secrets.compare_digest(given.encode(), expected.encode()):
        raise HTTPException(401, "A valid X-API-Key header is required.")

Sans API_KEY, chaque requête passe, ce qui est pratique sur votre ordinateur. Si API_KEY est défini, une requête doit inclure l’en-tête X-API-Key avec cette valeur. secrets.compare_digest compare en temps constant. Le temps de réponse ne révèle donc pas à un attaquant combien de caractères étaient corrects. Définissez REQUIRE_API_KEY=true en production : settings.py refuse alors de démarrer sans clé.

routes.py : quatre routes

Créez escalation/routes.py. La première partie contient le routeur et les modèles de réponse :

import json
import logging
from typing import Annotated

from fastapi import APIRouter, Body, Depends, Request, Response
from pydantic import BaseModel, Field

from escalation.contract import TicketIn
from escalation.errors import ErrorResponse, error_response
from escalation.jobs import QueueFull
from escalation.scoring import score_tickets
from escalation.security import require_api_key

predictions = logging.getLogger("escalation.predictions")

router = APIRouter(prefix="/v1", dependencies=[Depends(require_api_key)])

ERRORS = {
    401: {"model": ErrorResponse, "description": "The API key is missing or wrong."},
    422: {"model": ErrorResponse, "description": "The ticket does not match the contract."},
}


class ScoreOut(BaseModel):
    ticket_id: str
    score: float = Field(ge=0, le=1, description="0 to 1. Higher means a higher risk of escalation.")
    flag: bool = Field(description="True: send the ticket to the senior team.")
    threshold: float = Field(description="The score at and above which a ticket is flagged.")
    model_version: str
    warnings: list[str] = Field(description="Values the model never saw. The score is less certain.")


class ModelInfo(BaseModel):
    model_version: str
    bundle_sha256: str
    threshold: float
    trained_at: str
    features: list[str]


class JobIn(BaseModel):
    tickets: list[TicketIn] = Field(min_length=1)


class JobStatus(BaseModel):
    job_id: str
    status: str = Field(description="queued, running, done or failed")
    tickets: int
    created_at: str
    finished_at: str | None = None
    seconds: float | None = None
    model_version: str | None = None
    results: list[dict] | None = None
    error: str | None = None

dependencies=[Depends(require_api_key)] ajoute la vérification de clé à chaque route sous /v1. /health et /ready sont dans main.py, hors du routeur, pour qu’une plateforme puisse les appeler sans clé. ScoreOut est l’étiquette de la pharmacie : chaque champ a une description, qui devient la documentation de l’API.

La deuxième partie contient les deux routes du modèle :

@router.post("/score", tags=["scoring"], summary="Score one new ticket", responses=ERRORS)
def score(ticket: TicketIn, request: Request, response: Response) -> ScoreOut:
    state = request.app.state
    [result], timings = score_tickets(state.bundle, [ticket])
    response.headers["Server-Timing"] = timings.header()
    record = {
        "request_id": request.state.request_id,
        "ticket_id": ticket.ticket_id,
        "model_version": state.bundle.version,
        "score": result.score,
        "flag": result.flag,
    }
    predictions.info(json.dumps(record))
    return ScoreOut(
        ticket_id=ticket.ticket_id,
        score=result.score,
        flag=result.flag,
        threshold=state.bundle.threshold,
        model_version=state.bundle.version,
        warnings=result.warnings,
    )


@router.get("/model", tags=["model"], summary="Which model answers, and how to check it")
def model(request: Request) -> ModelInfo:
    bundle = request.app.state.bundle
    return ModelInfo(
        model_version=bundle.version,
        bundle_sha256=bundle.digest,
        threshold=bundle.threshold,
        trained_at=bundle.metadata["trained_at"],
        features=bundle.contract["features"],
    )

Le paramètre ticket: TicketIn vérifie le contrat. FastAPI valide le corps avec la classe du module 1 avant l’exécution de la fonction. La route appelle ensuite la même fonction score_tickets() que le traitement par lots. L’API et le traitement par lots calculent donc toujours le score d’un billet de la même façon. L’en-tête Server-Timing contient les trois étapes chronométrées, pour les mesures du module 4.

La route écrit une ligne de journal par score, en JSON, dans le consignateur escalation.predictions. Pourquoi? Un score est une prédiction pour les 72 prochaines heures. Larkfield sait seulement plus tard si le billet a été transféré à un niveau supérieur de soutien. Au module 5, Surveiller la santé du modèle joint ces lignes aux résultats réels avec ticket_id. Cela permet de mesurer la précision et le rappel sur les billets en service. Sans le journal, les preuves sont perdues. La ligne ne contient aucune donnée client : seulement les identifiants, la version et le résultat.

La troisième partie relie le gestionnaire de tâches du module 2 à deux routes :

def status_of(job) -> JobStatus:
    return JobStatus(
        job_id=job.id,
        status=job.status,
        tickets=len(job.results) if job.results is not None else len(job.tickets),
        created_at=job.created_at,
        finished_at=job.finished_at,
        seconds=job.seconds,
        model_version=job.model_version,
        results=job.results if job.status == "done" else None,
        error=job.error,
    )


@router.post(
    "/jobs",
    tags=["jobs"],
    status_code=202,
    summary="Score many tickets later",
    responses={**ERRORS, 429: {"model": ErrorResponse, "description": "Too many jobs are waiting."}},
)
def submit_job(job_in: Annotated[JobIn, Body()], request: Request, response: Response) -> JobStatus:
    limit = request.app.state.settings.max_job_tickets
    if len(job_in.tickets) > limit:
        return error_response(request, 413, "job_too_large", f"A job can have at most {limit} tickets.")
    try:
        job = request.app.state.jobs.submit(job_in.tickets)
    except QueueFull:
        return error_response(request, 429, "too_many_jobs", "Too many jobs are waiting. Try again later.")
    response.headers["Location"] = f"/v1/jobs/{job.id}"
    response.headers["Retry-After"] = "2"
    return status_of(job)


@router.get("/jobs/{job_id}", tags=["jobs"], summary="The status of a job, and its scores when done")
def get_job(job_id: str, request: Request) -> JobStatus:
    job = request.app.state.jobs.jobs.get(job_id)
    if job is None:
        return error_response(request, 404, "job_not_found", "There is no job with this ID here.")
    return status_of(job)

Chaque billet d’une tâche passe la même vérification du contrat, car JobIn contient une liste de TicketIn. Le reste reprend la conception du module 2 : 202 Accepted (accepté), le Location (emplacement) du statut, Retry-After (réessayer après), et 429 quand quatre tâches attendent déjà.

Exemple détaillé : appeler chaque route

Démarrez le service comme dans Charger et initialiser, sans API_KEY. Dans un deuxième terminal, calculez le score du billet d’exemple de contract.py. L’option -i affiche aussi les en-têtes :

macOS et Linux
Windows (PowerShell)

curl -i http://127.0.0.1:8000/v1/score -H "Content-Type: application/json" -d '{"ticket_id":"T-120001","channel":"email","team":"payment","segment":"home","region":"west","priority":1,"order_value":89.5,"word_count":112,"customer_tenure_days":400,"prior_tickets_90d":2,"created_hour":10,"prior_escalations_90d":1}'

$ticket = '{"ticket_id":"T-120001","channel":"email","team":"payment","segment":"home","region":"west","priority":1,"order_value":89.5,"word_count":112,"customer_tenure_days":400,"prior_tickets_90d":2,"created_hour":10,"prior_escalations_90d":1}'
Invoke-WebRequest -Method Post -Uri http://127.0.0.1:8000/v1/score -ContentType "application/json" -Body $ticket

La sortie :

HTTP/1.1 200 OK
date: Thu, 08 Oct 2026 14:35:05 GMT
server: uvicorn
content-length: 119
content-type: application/json
server-timing: frame;dur=1.07, prepare;dur=1.81, model;dur=0.07
x-request-id: 53ffd3d9d351

{"ticket_id":"T-120001","score":0.35065699401024064,"flag":true,"threshold":0.19,"model_version":"1.0.0","warnings":[]}

Le score de 0,351 dépasse le seuil de 0,19, donc flag vaut vrai. Dans le terminal du service, la ligne de journal de prédiction porte le même identifiant de requête :

INFO escalation.predictions: {"request_id": "53ffd3d9d351", "ticket_id": "T-120001", "model_version": "1.0.0", "score": 0.35065699401024064, "flag": true}
INFO escalation: POST /v1/score 200 4.9ms id=53ffd3d9d351

Maintenant, enfreignez le contrat deux fois dans un billet : priority "high" (élevée) et created_hour 24. Modifiez ces deux valeurs dans la commande et envoyez-la de nouveau. La sortie :

HTTP/1.1 422 Unprocessable Content
...
x-request-id: e53bd384c649

{"error":{"code":"invalid_ticket","message":"The ticket does not match the model's contract.","request_id":"e53bd384c649","fields":["created_hour","priority"]}}

Les deux champs incorrects sont nommés, et le modèle n’a jamais été exécuté. Ensuite, deux cas du module 1. Retirez complètement la clé "order_value":89.5,. Puis remettez-la sous la forme "order_value":null et définissez "channel":"social". Les deux sorties :

{"error":{"code":"invalid_ticket","message":"The ticket does not match the model's contract.","request_id":"179de6f59408","fields":["order_value"]}}
{"ticket_id":"T-120001","score":0.4310802648610386,"flag":true,"threshold":0.19,"model_version":"1.0.0","warnings":["channel: 'social' is a value the model never saw"]}

Une clé absente est une erreur; une valeur nulle est une valeur manquante connue, que l’outil d’imputation du pipeline remplit. Le nouveau canal reçoit un score, avec un avertissement dans la réponse.

Demandez au service quel modèle répond :

curl http://127.0.0.1:8000/v1/model

La sortie :

{"model_version":"1.0.0","bundle_sha256":"8f233029aa4c896dce1a5b3dfa970cd085088386bbe7784519f15a5b1a433b74","threshold":0.19,"trained_at":"2026-10-08 13:53:54 UTC","features":["channel","team","segment","region","priority","order_value","word_count","customer_tenure_days","prior_tickets_90d","created_hour","prior_escalations_90d"]}

bundle_sha256 est l’empreinte complète, que n’importe qui peut comparer au registre de mise en production. Votre empreinte et votre trained_at vous sont propres.

Enfin, une tâche avec le même billet trois fois. Envoyez {"tickets":[…]} avec le billet trois fois à /v1/jobs, puis appelez l’URL dans Location. Les sorties (en-têtes abrégés) :

HTTP/1.1 202 Accepted
location: /v1/jobs/b28f86e15848
retry-after: 2
...
{"job_id":"b28f86e15848","status":"queued","tickets":3,"created_at":"2026-10-08T14:35:05+00:00","finished_at":null,"seconds":null,"model_version":null,"results":null,"error":null}
{"job_id":"b28f86e15848","status":"done","tickets":3,"created_at":"2026-10-08T14:35:05+00:00","finished_at":"2026-10-08T14:35:05+00:00","seconds":0.003,"model_version":"1.0.0","results":[{"ticket_id":"T-120001","score":0.35065699401024064,"flag":true},...],"error":null}

Les scores de la tâche égalent le score en ligne, 0,35065699401024064. Les deux chemins appellent score_tickets() avec le même ensemble empaqueté.

La clé d’API

Arrêtez le service. Redémarrez-le avec une clé, puis envoyez le billet d’exemple sans l’en-tête et avec l’en-tête :

API_KEY=change-me-1234 fastapi run

curl -i http://127.0.0.1:8000/v1/score -H "Content-Type: application/json" -d '…the example ticket…'
curl http://127.0.0.1:8000/v1/score -H "Content-Type: application/json" -H "X-API-Key: change-me-1234" -d '…the example ticket…'

Sur Windows, définissez $env:API_KEY = "change-me-1234" avant fastapi run, et ajoutez -Headers @{"X-API-Key"="change-me-1234"} à Invoke-WebRequest. La sortie (en-têtes abrégés) :

HTTP/1.1 401 Unauthorized
...
{"error":{"code":"unauthorized","message":"A valid X-API-Key header is required.","request_id":"68782a77b2f0","fields":[]}}
{"ticket_id":"T-120001","score":0.35065699401024064,"flag":true,"threshold":0.19,"model_version":"1.0.0","warnings":[]}

/health répond toujours {"status":"ok"} sans clé. Dans L’essentiel de Docker et du nuage pour les développeurs en IA, la clé provenait d’un fichier secret avec API_KEY_FILE. settings.py lit aussi ce fichier. Arrêtez le service et redémarrez-le sans API_KEY pour la prochaine partie.

Swagger UI

FastAPI documente les routes pour vous à /docs. Ouvrez http://127.0.0.1:8000/docs :

Sélectionnez POST /v1/score pour voir le billet d’exemple de contract.py et les champs de ScoreOut avec leurs descriptions. Vous verrez aussi les réponses 401 et 422 avec ErrorResponse. Sélectionnez Try it out (essayer) et Execute (exécuter) pour calculer le score de l’exemple. Cette page n’a pas de bouton Authorize (autoriser) : require_api_key lit lui-même l’en-tête. Le document OpenAPI ne décrit donc pas la clé. Utilisez curl quand une clé est définie. En production, SHOW_DOCS=false désactive la page.

Tests de l’API

Les tests de l’API ont besoin d’un client. Ce client de test envoie des requêtes à l’application en mémoire, sans serveur ni port. Ajoutez-le à tests/conftest.py. D’abord, l’importation, après import pytest :

from fastapi.testclient import TestClient

Puis la fixture (préparation de test), après les autres fixtures à la fin du fichier :

@pytest.fixture
def client():
    from escalation.main import create_app

    with TestClient(create_app()) as test_client:
        yield test_client

create_app est importé à l’intérieur de la fixture. Il s’exécute donc après que la fixture settings_env (de Inférence par lots) a défini MODEL_SHA256 pour l’ensemble empaqueté de test. Le bloc with exécute le cycle de vie. Chaque test reçoit donc le véritable ensemble empaqueté, chargé et vérifié comme dans le serveur.

Créez maintenant tests/test_api.py. Chaque test prouve une règle de cette leçon, avec les fixtures client et ticket :

import time


def test_score_one_ticket(client, ticket):
    response = client.post("/v1/score", json=ticket)
    assert response.status_code == 200
    body = response.json()
    assert body["model_version"] == "1.0.0"
    assert 0 <= body["score"] <= 1
    assert body["flag"] == (body["score"] >= body["threshold"])
    assert "model;dur=" in response.headers["Server-Timing"]


def test_a_contract_error_has_the_error_shape(client, ticket):
    response = client.post("/v1/score", json={**ticket, "priority": "high", "created_hour": 24})
    assert response.status_code == 422
    error = response.json()["error"]
    assert error["code"] == "invalid_ticket"
    assert error["fields"] == ["created_hour", "priority"]


def test_an_unknown_category_is_scored_with_a_warning(client, ticket):
    body = client.post("/v1/score", json={**ticket, "channel": "social"}).json()
    assert body["warnings"] == ["channel: 'social' is a value the model never saw"]


def test_the_model_endpoint_names_the_bundle(client):
    body = client.get("/v1/model").json()
    assert body["model_version"] == "1.0.0"
    assert len(body["bundle_sha256"]) == 64


def test_ready_names_the_model(client):
    assert client.get("/ready").json() == {"status": "ready", "model_version": "1.0.0"}


def test_a_job_is_accepted_then_done(client, ticket):
    response = client.post("/v1/jobs", json={"tickets": [ticket] * 3})
    assert response.status_code == 202
    location = response.headers["Location"]
    for _ in range(50):
        body = client.get(location).json()
        if body["status"] == "done":
            break
        time.sleep(0.02)
    assert body["status"] == "done"
    assert len(body["results"]) == 3


def test_an_unknown_job_is_404(client):
    assert client.get("/v1/jobs/nothere").json()["error"]["code"] == "job_not_found"


def test_a_key_is_required_when_set(monkeypatch, ticket):
    from fastapi.testclient import TestClient

    from escalation.main import create_app

    monkeypatch.setenv("API_KEY", "test-key")
    with TestClient(create_app()) as client:
        assert client.post("/v1/score", json=ticket).status_code == 401
        assert client.post("/v1/score", json=ticket, headers={"X-API-Key": "test-key"}).status_code == 200

Exécutez-les :

python -m pytest tests/test_api.py -v

La sortie (abrégée) :

tests/test_api.py::test_score_one_ticket PASSED                          [ 12%]
tests/test_api.py::test_a_contract_error_has_the_error_shape PASSED      [ 25%]
...
tests/test_api.py::test_a_key_is_required_when_set PASSED                [100%]

============================== 8 passed in 0.95s ===============================

python -m pytest exécute toute la suite : 28 passed (28 tests réussis).

Exercice guidé : intégrer le classificateur à l’API

Voici l’exercice prévu au plan du module. Placez le classificateur derrière l’API avec les mêmes pratiques que ticket-api du cours sur les API.

Écrivez escalation/errors.py, escalation/security.py et escalation/routes.py comme ci-dessus. Si vous les avez copiés de l’instantané dans la dernière leçon, comparez-les avec cette page.
Ajoutez la fixture client à tests/conftest.py, et écrivez tests/test_api.py.
Exécutez python -m pytest. La dernière ligne indique 28 passed.
Démarrez le service avec fastapi run.
Calculez le score du billet d’exemple avec curl -i. Vérifiez les quatre champs score, flag, threshold et model_version, et l’en-tête x-request-id.
Trouvez la ligne de journal de prédiction avec le même identifiant de requête dans le terminal du service.
Appelez /v1/model. Vérifiez que bundle_sha256 commence par les 12 caractères de la ligne de journal de démarrage.

Afficher le résultat attendu des étapes 5 et 6

Une réponse 200 avec "score":0.35065699401024064,"flag":true,"threshold":0.19,"model_version":"1.0.0","warnings":[], et une ligne de journal INFO escalation.predictions: {"request_id": "…", "ticket_id": "T-120001", …} avec le même identifiant que l’en-tête x-request-id. L’identifiant lui-même est aléatoire.

À vous de jouer : un champ de plus

Le contrat contient extra="forbid". Envoyez le billet d’exemple avec un champ de plus, "colour":"red". Prédisez d’abord : quel sera le code d’état, ainsi que code et fields?

Afficher la réponse

{"error":{"code":"invalid_ticket","message":"The ticket does not match the model's contract.","request_id":"5c935dcc4129","fields":["colour"]}}

Le code d’état est 422. Le champ supplémentaire est rejeté et nommé. Un client qui envoie colour peut croire que le modèle l’utilise. Le modèle ne l’utilise pas, alors l’API le signale tout de suite.

Erreur courante

Chaque requête reçoit 422 avec le champ body

Symptôme : un script sous Windows envoie des billets, et chaque réponse est :

{"error":{"code":"invalid_ticket","message":"The ticket does not match the model's contract.","request_id":"c2c4f45d10bf","fields":["body"]}}

Le journal du service affiche POST /v1/score 422 pour chaque requête, sans aucune ligne de prédiction.

Cause : le corps n’est pas du JSON valide. La vérification du contrat ne lit donc aucun champ. Ici, l’interpréteur de commandes a modifié les guillemets. Le corps est arrivé sous la forme {'ticket_id':'T-120001'}, avec des guillemets simples, que JSON n’accepte pas. field_name() signale une erreur JSON en indiquant le champ body (corps).

Solution : envoyez le corps exactement au format JSON. Dans PowerShell, gardez le JSON dans une variable avec des guillemets simples à l’extérieur et doubles à l’intérieur. Suivez l’exemple détaillé, ou laissez ConvertTo-Json le créer. Vérifiez une requête avec -i avant d’en envoyer plusieurs.

Pour un vrai service d’IA : une réponse 422 coûte peu et ne présente pas de risque. Aucun modèle n’a été exécuté, et le client apprend quoi corriger. Comptez les réponses 422 par fields (champs). Une hausse soudaine des erreurs body indique un client défectueux. Une hausse pour un champ indique qu’un client a changé ce qu’il envoie. Le module 5 compte ces signaux.

Vérifiez votre compréhension

Pourquoi chaque réponse de POST /v1/score contient-elle le seuil et la version du modèle, et pas seulement le score?

Afficher la réponse

Un score ne signifie rien sans le modèle et le seuil qui l’accompagnent. La version 1.0.0 signale les billets à partir de 0,19, et la version 1.1.0, à partir de 0,20. La version candidate défectueuse rc1 augmente tous les scores. Un seuil de 0,19 signale donc 86 % des billets. Avec la version et le seuil dans la réponse et le journal, un client prend la bonne décision. Plus tard, n’importe qui peut déterminer quel modèle a attribué un score à un billet.

Vérifiez votre compréhension

Le service écrit une ligne de journal JSON par score avec l'identifiant de requête, l'identifiant de billet, la version du modèle, le score et l'indicateur. Pourquoi, et que doit-elle exclure?

Afficher la réponse

Le résultat réel (transmis au niveau supérieur ou non) arrive plusieurs jours plus tard. Les lignes du journal permettent à l’équipe de relier chaque prédiction à son résultat grâce à l’identifiant du billet. L’équipe peut alors mesurer la précision et le rappel sur les billets réels, par version du modèle (module 5). L’identifiant de requête relie la ligne à la réponse reçue par le client. La ligne ne doit contenir aucune donnée client, comme le texte du billet ou un nom : seulement les identifiants et le résultat.

Vérifiez votre compréhension

Un client envoie un billet sans la clé order_value, et un autre envoie order_value : null. Que reçoit chacun, et pourquoi cette différence?

Afficher la réponse

Sans la clé : 422 invalid_ticket avec fields: ["order_value"], car le contrat exige cette clé. Avec null : 200 et un score. Null est la façon convenue de dire « il n’y a pas de commande ». L’outil d’imputation du pipeline remplace cette valeur comme lors de l’entraînement. Une clé absente indique plus probablement un client défectueux qu’une véritable valeur manquante.

Récapitulatif

POST /v1/score vérifie le billet selon le contrat avant d’exécuter le modèle. La réponse contient score, flag, threshold, model_version et warnings.
GET /v1/model indique la version, l’empreinte complète, le seuil et les caractéristiques. Les routes des tâches sont reliées au JobRunner du module 2.
Chaque erreur a la même structure : code, message, request_id, fields. Chaque réponse contient X-Request-ID.
Une ligne de journal JSON par score conserve les preuves nécessaires à la surveillance quand les résultats arrivent.
La clé d’API protège /v1; /health et /ready restent accessibles à la plateforme.
tests/test_api.py prouve chaque règle : 28 passed (28 tests réussis).

Ensuite, vous prouvez que les scores fournis par le service sont ceux obtenus à l’entraînement : c’est la parité.

Leçon suivante : Vérifier la parité