La session Live avec ADK

Premier essai d'un assistant vocal, en test interne : la voix de l'assistant sort accélérée et aiguë, comme un disque en 45 tours. Deuxième essai : la voix est correcte, mais l'assistant attend quatre secondes de silence avant de répondre. Troisième essai : il se coupe lui-même la parole. Les trois défauts viennent de la même couche, celle qui transporte l'audio et découpe les tours de parole, et aucun ne se voit dans un test en texte.

Le mécanisme : une file en entrée, un flux d'événements en sortie

ADK expose la Live API par Runner.run_live(). On ne lui envoie pas des messages : on remplit une LiveRequestQueue (audio en continu, ou contenu texte), et on lit un flux d'événements jusqu'à la fin de l'appel.

sequenceDiagram
  participant M as "Micro (16 kHz)"
  participant Q as "LiveRequestQueue"
  participant R as "run_live (ADK)"
  participant L as "Live API"
  participant O as "Outils"
  M->>Q: send_realtime(Blob PCM 100 ms)
  Q->>L: audio en continu
  L-->>R: input_transcription (partielle puis finale)
  L-->>R: function_call chercher_creneaux
  R->>O: exécute l'outil
  O-->>R: dict résultat
  R->>L: function_response
  L-->>R: audio 24 kHz + output_transcription
  L-->>R: turn_complete

Les formats sont fixés par l'API : entrée PCM 16 bits mono 16 kHz, sortie PCM 16 bits mono 24 kHz. Lire la sortie à 16 kHz donne une voix ralentie et grave ; lire l'entrée comme du 24 kHz donne le « 45 tours » de l'introduction. Les deux constantes sont dans audio.py et nulle part ailleurs.

Mise en œuvre : ouvrir la session

rc = RunConfig(
    streaming_mode=StreamingMode.BIDI,
    response_modalities=["AUDIO"],
    speech_config=types.SpeechConfig(voice_config=types.VoiceConfig(
        prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name=self.cfg.voix))),
    input_audio_transcription=types.AudioTranscriptionConfig(),
    output_audio_transcription=types.AudioTranscriptionConfig(),
)
# L'appelant décroche : le modèle parle en premier (accueil + mention de l'assistant automatique).
self._fin_parole, self._accueil = time.time(), True
self.queue.send_content(types.Content(role="user", parts=[types.Part(text="[décroché]")]))
async for ev in self.runner.run_live(user_id=self.id, session_id=session.id,
                                     live_request_queue=self.queue, run_config=rc):
    await self.traiter(ev)

Les deux AudioTranscriptionConfig activent la transcription des deux voix. Elles servent trois fois : à l'affichage, au journal, et au garde-fou de confirmation (la dernière phrase transcrite de l'appelant est ce que reserver_creneau vérifie).

Le message [décroché] règle un détail d'ergonomie téléphonique : c'est l'appelé qui parle en premier. La consigne dit exactement quoi répondre à ce signal, y compris la phrase « Vous êtes en ligne avec l'assistant vocal automatique du cabinet ».

Mise en œuvre : traduire les événements

traiter(ev) est le seul endroit qui connaît la forme des événements ADK. Il les traduit en état, en journal et en messages pour l'interface :

Un piège : un turn_complete arrive aussi après un tour qui n'a produit qu'un appel d'outil, sans un mot. Le code ne considère un tour terminé que s'il a contenu de l'audio (_audio_ce_tour), sinon le testeur automatique reprendrait la parole pendant que l'assistant s'apprête à répondre.

La latence de première réponse

La mesure qui compte pour l'appelant : entre la fin de sa phrase et le premier son de l'assistant. Le code la prend au premier fragment audio reçu après fin_de_parole(t) :

async def _audio_assistant(self, pcm: bytes, t: float) -> None:
    if self._fin_parole is not None:
        lat = round(t - self._fin_parole, 2)
        self._fin_parole = None

Dans les appels de test, fin_de_parole est appelée à l'instant exact où le fichier audio se termine. Au micro, on ne connaît pas cet instant : le code prend le dernier fragment de transcription reçu, ce qui sous-estime la latence (commentaire ponytail: dans le code ; l'extrait de journal plus bas montre pourquoi). La latence d'accueil (décroché → première syllabe) est mesurée à part : elle inclut l'ouverture de la session.

Ce qui casse en production

La règle de décision

Passez par ADK (run_live) quand l'assistant a des outils : ADK exécute les appels de fonction, renvoie les réponses au modèle et garde la session. Utilisez le SDK google-genai directement (client.aio.live.connect) pour un flux audio sans outil, ou si vous avez besoin d'un champ de configuration qu'ADK ne transmet pas encore. Dans les deux cas, gardez un seul module qui connaît la forme des événements.

Résultat obtenu

Extrait réel du journal d'un appel de test, exécution du 2026-10-15 (sorties/appels/e2e-reservation/journal.jsonl, numéro masqué par le code) :

{"t": 1.145, "appel": "e2e-reservation", "type": "latence_accueil", "secondes": 1.13}
{"t": 25.968, "appel": "e2e-reservation", "type": "transcription", "role": "appelant", "texte": "Bonjour, je m'appelle Claire Martin. Je voudrais un rendez-vous de kiné pour une rééducation du genou. […]"}
{"t": 25.97, "appel": "e2e-reservation", "type": "latence", "secondes": 0.68}
{"t": 37.226, "appel": "e2e-reservation", "type": "transcription", "role": "appelant", "texte": "plutôt le matin à partir de lundi prochain"}
{"t": 37.228, "appel": "e2e-reservation", "type": "outil", "nom": "chercher_creneaux", "args": {"a_partir_du": "2026-10-19", "type_seance": "kinesitherapie", "moment": "matin"}}
{"t": 37.23, "appel": "e2e-reservation", "type": "resultat", "nom": "chercher_creneaux", "resultat": {"jour_demande_disponible": true, "creneaux": [{"creneau_id": 27, "quand": "lundi 19 octobre à 8h30", "praticien": "Mme Durand"}, […]]}}
{"t": 37.231, "appel": "e2e-reservation", "type": "usage", "entree": 2687, "sortie": 48, "total": 2735, "entree_detail": {"TEXT": 1447, "AUDIO": 979}, "sortie_detail": {}}
{"t": 37.931, "appel": "e2e-reservation", "type": "latence", "secondes": 1.47}

Deux lectures utiles. L'outil répond en 2 ms : la latence de 1,47 s de ce tour est celle du modèle, pas de SQLite. Et la transcription finale de l'appelant arrive au moment où l'assistant commence à répondre (25,968 s puis 25,97 s) : c'est pourquoi le serveur ne peut pas s'en servir pour dater la fin de parole.

.venv/bin/python -m pytest -q tests/test_appel.py tests/test_audio.py    # 10 tests, faux événements ADK

Pratique

Exercice. Écrivez un test sans réseau qui vérifie qu'un événement interrupted envoie bien {"type": "interrompu"} à l'interface. Ajoutez-le à test_appel.py, qui fournit déjà la fabrique ev() et les fixtures.

def test_interruption_relayee(cfg, agenda):
    recus = []

    async def emettre(m):
        recus.append(m)

    async def scenario():
        a = Appel(cfg, agenda, emettre, appel_id="t4")
        await a.traiter(ev(interrupted=True))
        a.fermer()

    asyncio.run(scenario())
    assert {"type": "interrompu"} in recus

• Je sais quels formats audio la Live API attend et renvoie
• Je sais lire un flux d'événements run_live et dire lequel clôt un tour
• Je sais mesurer une latence de première réponse et dire ce que ma mesure sous-estime