Le pont WebSocket et l'interface

La secrétaire du cabinet veut voir ce que fait l'assistant pendant qu'il parle : quel créneau il propose, s'il a réservé, s'il a transféré. Le directeur veut pouvoir l'essayer depuis son navigateur, sans téléphone. Et personne ne veut que la clé d'API se retrouve dans le JavaScript d'une page web.

D'où l'architecture : le navigateur ne parle jamais à la Live API. Il envoie le son de son micro à notre serveur Python par WebSocket ; le serveur tient la session Live, exécute les outils, et renvoie au navigateur la voix de l'assistant et des événements lisibles.

Le mécanisme : un protocole minimal sur une seule socket

sequenceDiagram
  participant N as "Navigateur"
  participant S as "Serveur /ws"
  participant A as "Appel (run_live)"
  N->>S: ouverture WebSocket
  S-->>N: {"type": "debut", "appel": "...", "modele": "..."}
  loop toutes les 100 ms
    N->>S: binaire PCM 16 kHz (3 200 octets)
    S->>A: envoyer_audio()
  end
  A-->>S: audio + transcription + outils
  S-->>N: 0x01 + PCM 24 kHz (voix de l'assistant)
  S-->>N: {"type": "transcription" | "outil" | "agenda" | "latence" | "interrompu"}
  N->>S: {"type": "raccrocher"}
  S-->>N: {"type": "fin", "duree_s": ..., "latences": [...]}

Le préfixe d'un octet évite un second canal : une seule socket, un seul ordre d'arrivée.

Mise en œuvre : le serveur

@app.websocket("/ws")
async def ws_appel(ws: WebSocket, scenario: str | None = Query(None)):
    await ws.accept()
    verrou = asyncio.Lock()  # un seul envoi à la fois sur la socket

    async def emettre(m):
        async with verrou:
            await (ws.send_bytes(m) if isinstance(m, bytes) else ws.send_json(m))

    appel = fabrique(cfg, agenda, emettre)
    taches = [asyncio.create_task(appel.executer()), asyncio.create_task(_ecouter(ws, appel, scenario))]
    if scenario in SCENARIOS:  # mode démonstration : un appel enregistré tient lieu de micro
        taches.append(asyncio.create_task(jouer(appel, scenario, echo=lambda pcm: emettre(VOIX_APPELANT + pcm))))
    await asyncio.wait(taches, return_when=asyncio.FIRST_COMPLETED)

Trois choix :

• Un verrou d'émission. La voix de l'assistant, les événements d'outils (émis depuis un thread du pool ADK via call_soon_threadsafe) et les latences partent de tâches différentes. Sans verrou, deux send concurrents corrompent la trame WebSocket.
• La première tâche qui finit clôt l'appel. Raccrocher, transfert ou coupure de la session Live : dans tous les cas, appel.fermer() écrit la transcription et le bilan.
• Les trames aberrantes sont rejetées : plus de 64 000 octets (2 s d'audio) ou une taille impaire ne viennent pas d'un micro.

Le serveur écoute sur 127.0.0.1 par défaut. La page n'a pas d'authentification : c'est un poste de démonstration, pas un service exposé.

Mise en œuvre : le micro et le haut-parleur

Le navigateur capte le micro à 44,1 ou 48 kHz selon la carte son. Un AudioWorklet (code audio exécuté hors du fil principal de la page) ramène le signal à 16 kHz et poste des trames de 1 600 échantillons :

for (; this.pos < ch.length; this.pos += this.ratio) {
  const a = Math.floor(this.pos), b = Math.min(ch.length, Math.floor(this.pos + this.ratio))
  let s = 0
  for (let i = a; i < Math.max(b, a + 1); i++) s += ch[i]
  this.buf[this.n++] = Math.max(-1, Math.min(1, s / Math.max(b - a, 1))) * 0x7fff

En lecture, chaque fragment de voix est planifié à la suite du précédent (src.start(etat.prochain)), ce qui évite les micro-coupures entre fragments. Sur l'événement interrompu, l'interface arrête toutes les sources planifiées : l'appelant qui coupe la parole n'entend pas la fin d'une phrase devenue caduque.

getUserMedia est appelé avec echoCancellation: true. Sans annulation d'écho, sur haut-parleurs, le micro renvoie la voix de l'assistant à la Live API, qui la prend pour une interruption et se coupe elle-même : le troisième défaut décrit dans la leçon précédente.

make run          # http://localhost:8080 : cliquez sur le micro, ou choisissez un appel test et « Rejouer »

Ce qui casse en production

La règle de décision

Gardez un serveur entre le navigateur et la Live API dès que l'assistant a des outils métier ou une clé à protéger. La connexion directe du navigateur à la Live API, avec des jetons éphémères, se justifie pour un assistant sans outil côté serveur. Pour un vrai numéro de téléphone, remplacez le navigateur par une passerelle de téléphonie qui transporte l'audio en WebSocket (8 kHz, μ-law le plus souvent) : Appel ne change pas, seule la conversion audio s'ajoute.

Résultat obtenu

Capture Playwright du 2026-10-16 : Chromium avec un faux périphérique audio qui lit la première réplique du scénario « réservation » comme un micro (--use-file-for-fake-audio-capture), contre gemini-3.8-live. La latence affichée est celle du serveur pour ce tour.

tests/test_serveur.py::test_routes PASSED
tests/test_serveur.py::test_websocket PASSED

Pratique

Exercice. Ajoutez au protocole un message {"type": "niveau", "rms": 0.12} envoyé au navigateur toutes les secondes, pour afficher un vumètre de l'appelant. Où le calculez-vous, et quel test ajoutez-vous ?

• Je sais décrire le protocole navigateur ↔ serveur et justifier le préfixe d'un octet
• Je sais pourquoi le micro exige HTTPS hors de localhost
• J'ai lancé make run et passé un appel au micro ou rejoué un appel test