Une API web en streaming avec FastAPI

Le rôle de l'API

Une API (interface de programmation) est ici un petit serveur HTTP : la page web lui envoie une question en JSON, il renvoie la réponse. FastAPI est un framework Python qui transforme des fonctions en routes HTTP et valide les données grâce à Pydantic (des classes qui décrivent la forme attendue). Uvicorn est le serveur qui fait tourner l'application.

sequenceDiagram
  participant N as Navigateur
  participant A as FastAPI (app.py)
  participant C as chatbot.repondre
  participant G as Gemini
  N->>A: POST /chat {session, message}
  A->>A: validation Pydantic (sinon 422)
  A->>C: générateur
  C->>G: generate_content_stream
  loop chaque morceau
    G-->>C: texte
    C-->>A: yield texte
    A-->>N: morceau HTTP (Transfer-Encoding chunked)
  end
  C->>C: mémoriser + journaliser

Le code de l'API

app.py ne contient aucune logique métier : il traduit HTTP en appels à chatbot.py.

files/app.py (extrait)

class Question(BaseModel):
    session: str = Field(min_length=1, max_length=64)
    message: str = Field(min_length=1, max_length=2000)


@app.post("/chat")
def chat(q: Question):
    echange_id = uuid.uuid4().hex[:12]
    return StreamingResponse(
        chatbot.repondre(q.session, q.message, echange_id),
        media_type="text/plain; charset=utf-8",
        headers={"X-Echange-Id": echange_id})

StreamingResponse accepte un générateur Python (une fonction qui fait yield) : chaque yield part immédiatement vers le navigateur.
X-Echange-Id est un en-tête HTTP : l'identifiant de l'échange, envoyé avant le texte. La page s'en servira pour rattacher l'avis 👍/👎 à la bonne réponse.
max_length=2000 protège votre facture : personne ne pourra envoyer un roman à chaque requête.

Les erreurs du modèle

Si Gemini renvoie une erreur passagère (429 : trop de requêtes ; 503 : service surchargé), repondre réessaie deux fois avec une attente croissante, seulement si rien n'a encore été envoyé. Sinon l'utilisateur reçoit un message d'excuse poli, jamais une trace d'erreur Python.

files/chatbot.py (extrait de repondre)

except errors.APIError as e:
    erreur = f"{e.code} {e.status}"
    if e.code in (429, 500, 503) and not morceaux and essai < 2:
        time.sleep(2 ** essai * 2)
        continue
    break

Code	Signification	Réaction
200	réponse en streaming	afficher au fil de l'eau
422	entrée invalide (message vide, trop long)	renvoyé par FastAPI, sans appel au modèle
429 / 503 côté Gemini	quota ou surcharge	réessai, puis message d'excuse (le HTTP reste 200)

Que se passe-t-il si la page envoie un message vide ?

Pourquoi ne réessaie-t-on pas si des morceaux ont déjà été envoyés ?

Tester sans clé

test_chatbot.py remplace Gemini par un faux modèle qui renvoie des morceaux prévus à l'avance, et appelle l'API avec le TestClient de FastAPI :

files/test_chatbot.py (extrait)

class FauxModele:
    def __init__(self, morceaux=("Bonjour, ", "la salle ouvre à 6h.")):
        self.morceaux, self.appels = morceaux, []

    def generate_content_stream(self, model, contents, config):
        self.appels.append(contents)
        for m in self.morceaux:
            yield SimpleNamespace(text=m)

Exercice : limiter la taille de la session

Un client malveillant pourrait créer des milliers de sessions et remplir la RAM. Proposez une protection simple dans chatbot.py, sans base de données.

Voir la solution

Limiter le nombre de sessions gardées, en oubliant les plus anciennes (un dict Python garde l'ordre d'insertion) :

chatbot.py (modification)

MAX_SESSIONS = 1000


def historique(session: str) -> list[types.Content]:
    if session not in MEMOIRE and len(MEMOIRE) >= MAX_SESSIONS:
        MEMOIRE.pop(next(iter(MEMOIRE)))  # la plus ancienne
    return MEMOIRE.setdefault(session, [])

Côté infrastructure, Cloud Run permet aussi de limiter le nombre d'instances (--max-instances), ce qui plafonne la dépense.

Résultat obtenu

Serveur lancé avec uvicorn app:app --port 18001, requêtes réelles avec curl, le 2026-10-15 :

sortie réelle, 2026-10-15

$ curl -s localhost:18001/sante
{"ok":true,"modele":"gemini-3.5-flash"}

$ curl -sN -D - -X POST localhost:18001/chat -H 'Content-Type: application/json' \
       -d '{"session":"demo","message":"Puis-je venir avec un invité ?"}'
HTTP/1.1 200 OK
x-echange-id: b181c0955625
content-type: text/plain; charset=utf-8
Transfer-Encoding: chunked

Si vous disposez de la formule Premium, vous pouvez venir avec un invité une fois par mois.
Pour les autres formules, votre invité peut bénéficier d'une séance découverte gratuite, une
fois par personne et sur réservation. [...]

$ curl -s -X POST localhost:18001/chat -H 'Content-Type: application/json' -d '{"session":"demo","message":""}'
{"detail":[{"type":"string_too_short","loc":["body","message"],"msg":"String should have at least 1 character",...}]}

Transfer-Encoding: chunked confirme le streaming. Et pytest -q test_chatbot.py : 8 passed in 0.55s, sans clé.

Je sais faire

Écrire une route FastAPI validée par un modèle Pydantic
Renvoyer un générateur avec StreamingResponse
Passer une information dans un en-tête (X-Echange-Id)
Tester une API sans clé grâce à un faux modèle