La Live API : une conversation sur WebSocket

Pourquoi pas une simple requête ?

Avec generate_content, on envoie tout, on attend, on reçoit tout. Pour une conversation, c'est trop lent : il faudrait attendre la fin de la phrase, téléverser, attendre la réponse complète, la télécharger. La Live API ouvre une connexion WebSocket : un canal réseau qui reste ouvert et dans lequel les deux côtés peuvent envoyer des messages à tout moment.

• Le client envoie l'audio du micro par petits morceaux (ici 100 ms), au fil de l'eau.
• Le serveur détecte tout seul que la personne a fini de parler : c'est la VAD (Voice Activity Detection, détection d'activité vocale).
• La réponse arrive en flux : les premiers morceaux d'audio se jouent pendant que la suite est encore générée.
• Si la personne reprend la parole pendant la réponse, le serveur s'interrompt (barge-in) : c'est ce qui rend l'échange naturel.

sequenceDiagram
  participant U as "Client (micro)"
  participant WS as "WebSocket Live API"
  participant M as "gemini-3.8-live"
  U->>WS: "connect(config : AUDIO, consigne système)"
  loop "toutes les 100 ms"
    U->>WS: "send_realtime_input(PCM 16 kHz)"
  end
  Note over WS: "VAD : silence détecté, fin du tour"
  WS->>M: "tour utilisateur"
  M-->>U: "morceaux audio 24 kHz + transcription"
  M-->>U: "turn_complete"

Les formats à respecter

Notre question vient d'un TTS (24 kHz) : to_pcm16k() la rééchantillonne par interpolation linéaire avec numpy, ce qui suffit pour la voix (la doc précise d'ailleurs que l'API rééchantillonne si besoin).

Le code

cfg = types.LiveConnectConfig(
    response_modalities=["AUDIO"],
    system_instruction=SYSTEM,
    input_audio_transcription=types.AudioTranscriptionConfig(),   # ce que l'API a entendu
    output_audio_transcription=types.AudioTranscriptionConfig(),  # ce qu'elle dit
)
async with c.aio.live.connect(model=LIVE_MODEL, config=cfg) as session:
    async def micro():  # simule un micro : envoi en temps réel
        for i, ch in enumerate(speech + silence):
            if i == len(speech):
                speech_end["t"] = time.time()  # fin de la parole, début du silence
            await session.send_realtime_input(audio=types.Blob(data=ch, mime_type=f"audio/pcm;rate={IN_RATE}"))
            await asyncio.sleep(CHUNK_MS / 1000)

    sender = asyncio.create_task(micro())     # on envoie ET on écoute en même temps
    async for msg in session.receive():
        if msg.data:                            # un morceau d'audio de la réponse
            first = first or time.time(); audio += msg.data
        ...
        if msg.server_content and msg.server_content.turn_complete:
            break

Deux détails que nous avons appris en testant :

• Envoyer tout l'audio d'un coup, sans pause, puis attendre : la session est restée muette jusqu'au délai d'expiration. Il faut cadencer l'envoi comme un vrai micro (asyncio.sleep) et finir par du silence pour que la VAD conclue le tour.
• Envoi et réception doivent tourner en parallèle (une tâche asyncio pour le micro), sinon on ne voit pas la réponse commencer pendant qu'on parle encore.

Limites d'une session

Résultat obtenu

Sorties réelles du 2026-10-15. D'abord une question en texte (send_client_content), réponse en audio :

gemini-3.8-live connected 0.61
first audio 0.66 total 6.54 282240 Bonjour, je vais vérifier immédiatement le statut de votre commande 4521 et vous tenir informé.

Puis python live_demo.py avec une question en voix (fichier exemple_question.wav) :

WebSocket ouvert en 0.32 s (gemini-3.8-live)
Entendu : Bonjour, ma commande 4521 n'est toujours pas arrivée. Vous pouvez me dire où elle en est ?
Répondu : Bonjour, je vais vérifier l'état de votre commande 4521 et revenir vers vous rapidement. Pourriez-vous me confirmer votre numéro de téléphone ?
Latence fin de parole -> 1er audio : 1.00 s ; réponse de 7.9 s -> sortie/live_reponse.wav

Exercice : afficher les sous-titres en direct

Modifiez la boucle de réception pour afficher la transcription de sortie au fur et à mesure qu'elle arrive (sous-titres en direct), au lieu de l'afficher à la fin.

async for msg in session.receive():
    sc = msg.server_content
    if sc and sc.output_transcription and sc.output_transcription.text:
        print(sc.output_transcription.text, end="", flush=True)  # morceau par morceau
    if msg.data:
        audio += msg.data
    if sc and sc.turn_complete:
        print()
        break

Je sais faire

• Expliquer ce qu'apporte un WebSocket par rapport à une requête classique
• Ouvrir une session Live, envoyer de l'audio 16 kHz cadencé et lire la réponse en flux
• Expliquer le rôle de la VAD et de l'interruption
• Citer les limites de session et protéger la clé avec un jeton éphémère