Le contrat d'API : validation et sortie structurée

Pourquoi un « contrat » ?

Un contrat d'API, c'est la promesse faite aux appelants : ce que l'on accepte en entrée, ce que l'on renvoie en sortie, et les erreurs possibles. Avec un LLM au milieu, le contrat protège dans les deux sens :

• en entrée, il refuse ce qui coûterait cher ou n'a pas de sens (un ticket de 2 Mo, une langue non gérée) avant de payer un appel au modèle ;
• en sortie, il garantit à l'appelant un JSON stable (category vaut toujours l'une des 5 valeurs prévues), même si le modèle est créatif.

flowchart LR
  A["Requête JSON"] --> B{"Pydantic : Ticket valide ?"}
  B -->|"non"| E["422 : détail des champs en erreur"]
  B -->|"oui"| C["Gemini avec response_schema"]
  C --> D{"Pydantic : Analysis valide ?"}
  D -->|"oui"| F["200 : JSON garanti"]
  D -->|"non"| G["500 : bug à corriger, jamais renvoyé tel quel"]

FastAPI est un framework web Python : il lit les annotations de type de vos fonctions pour valider les requêtes et générer une documentation OpenAPI (visible sur /docs). Pydantic est la bibliothèque de validation qu'il utilise.

L'entrée : bornes et valeurs permises

class Ticket(BaseModel):
    subject: str = Field(min_length=3, max_length=200)
    body: str = Field(min_length=10, max_length=8000)  # borne le coût d'un appel
    language: Literal["fr", "en"] = "fr"

Chaque contrainte a une raison :

Une requête invalide reçoit un 422 (Unprocessable Entity) avec le détail du champ fautif, sans aucun appel au modèle.

La sortie : demander un schéma au modèle

On décrit la réponse attendue avec un deuxième modèle Pydantic, puis on le passe à Gemini comme response_schema. Gemini produit alors un JSON qui respecte ce schéma (c'est la sortie structurée).

class Analysis(BaseModel):
    summary: str = Field(description="Résumé du ticket en une ou deux phrases")
    category: Category                     # Enum : facturation, livraison, technique, compte, autre
    priority: Literal["basse", "normale", "haute", "urgente"]
    sentiment: Literal["positif", "neutre", "negatif"]

r = await self.client.aio.models.generate_content(
    model=MODEL,
    contents=ticket_prompt(t),
    config=types.GenerateContentConfig(
        system_instruction=SYSTEM,
        response_mime_type="application/json",
        response_schema=Analysis,
        temperature=0,
    ),
)
return Analysis.model_validate_json(r.text), usage

On revalide quand même avec model_validate_json : le schéma réduit fortement les écarts, la validation côté serveur les rend impossibles à propager. La route déclare response_model=Analysis, ce qui documente la sortie dans /docs.

Isoler le modèle pour pouvoir le tester

Le client Gemini est fourni à la route par injection de dépendance (Depends(get_llm)) : FastAPI appelle get_llm() à chaque requête, et les tests peuvent le remplacer par un faux.

def client(llm) -> TestClient:
    main.app.dependency_overrides[main.get_llm] = lambda: llm
    return TestClient(main.app)

def test_validation_rejects_bad_input():
    c = client(FakeLLM())
    assert c.post("/v1/tickets/analyze", json={"subject": "x", "body": "court"}).status_code == 422

get_llm() crée le client une seule fois puis le réutilise : les connexions HTTP vers Gemini restent ouvertes entre deux requêtes, ce qui fait gagner du temps à chaque appel.

Exercice : ajouter un champ

Ajoutez à la sortie un champ needs_human (booléen : le ticket doit-il être traité par un humain en priorité ?) et un test qui vérifie qu'une réponse sans ce champ provoque une erreur de validation.

class Analysis(BaseModel):
    summary: str = Field(description="Résumé du ticket en une ou deux phrases")
    category: Category
    priority: Literal["basse", "normale", "haute", "urgente"]
    sentiment: Literal["positif", "neutre", "negatif"]
    needs_human: bool = Field(description="Vrai si un humain doit traiter le ticket en priorité")

import pytest
from pydantic import ValidationError

def test_needs_human_obligatoire():
    with pytest.raises(ValidationError):
        main.Analysis.model_validate_json(
            '{"summary": "x", "category": "autre", "priority": "basse", "sentiment": "neutre"}')

Résultat obtenu

Test local du 2026-10-15 (uvicorn main:app --port 8765, modèle gemini-3.5-flash via clé d'API) :

$ curl -s -X POST localhost:8765/v1/tickets/analyze -H 'Content-Type: application/json' \
  -d '{"subject":"Impossible de me connecter","body":"Depuis ce matin, la page de connexion affiche une erreur 500. Toute notre équipe est bloquée et nous avons une démo client à 14h !"}'
{"summary":"Erreur 500 sur la page de connexion empêchant toute l'équipe du client d'accéder au service avant une démo.","category":"technique","priority":"urgente","sentiment":"negatif"}

$ curl -s -o /dev/null -w "%{http_code}\n" -X POST localhost:8765/v1/tickets/analyze \
  -H 'Content-Type: application/json' -d '{"subject":"x","body":"court"}'
422

Je sais faire

• Borner les entrées d'une route IA (taille, valeurs permises) avec Pydantic
• Obtenir de Gemini un JSON conforme à un schéma avec response_schema
• Revalider la sortie et la déclarer avec response_model
• Remplacer le modèle par un faux dans les tests avec dependency_overrides