Diffusez en streaming de la voix à faible latence depuis des agents d'IA avec Kokoro TTS sur Amazon SageMaker AI

Un client pose une question à un agent vocal d'IA et attend. L'agent a déjà une réponse : il est en train de générer une longue réponse de plusieurs paragraphes, mais le client n'entend rien pendant cinq secondes. Au moment où la réponse arrive, le client est frustré, désorienté et ne prête plus attention.

Les intégrations de synthèse vocale créent précisément ce décalage. Une réponse d'une seule phrase peut être générée en 600 millisecondes, mais la réponse complète prend cinq secondes, et l'intégration attend la totalité avant que la synthèse ne commence.

Cet article montre comment combler ce décalage. Nous diffusons en streaming chaque phrase vers un modèle de synthèse vocale (TTS) Kokoro-82M hébergé sur Amazon SageMaker AI dès qu'un agent la produit, de sorte que la synthèse commence dès la première phrase au lieu d'attendre la réponse achevée. Acheminer cet audio vers un navigateur nécessite aussi sa propre solution ; nous couvrons donc les deux transports possibles et ce que chacun coûte en pratique.

Vue d'ensemble de la solution

Cette solution comble le décalage en diffusant la synthèse en parallèle de la génération au lieu d'attendre qu'elle se termine. Au fur et à mesure que l'agent produit chaque phrase, celle-ci est immédiatement diffusée en streaming vers un modèle de synthèse vocale Kokoro-82M hébergé, de sorte que la lecture audio puisse commencer avant que l'agent n'ait fini d'énoncer sa réponse complète. Kokoro-82M ne compte que 82 millions de paramètres, soit une taille suffisamment réduite pour s'exécuter à faible coût sur une seule instance GPU de milieu de gamme. Cela maintient la latence et le coût d'inférence à un niveau bas pour un agent vocal traitant un trafic réel.

Kokoro-82M s'exécute dans un conteneur personnalisé sur un point de terminaison temps réel SageMaker AI, sur une ml.g5.xlarge instance GPU. Le conteneur expose deux routes : POST /invocations, qui diffuse en streaming une réponse pour une entrée complète, et WS /invocations-bidirectional-stream, qui maintient une connexion ouverte sur de nombreux messages dans les deux sens.

Le diagramme suivant montre le chemin de requête du client vers le modèle :

Figure 1 : Chemin de requête du client vers le point de terminaison temps réel SageMaker AI via un flux bidirectionnel

Prérequis

Avant de commencer, vous devez disposer de :

• Un compte AWS avec les autorisations nécessaires pour créer des points de terminaison SageMaker AI ; AgentCore runtime, une capacité d'Amazon Bedrock AgentCore ; des rôles AWS Identity and Access Management (IAM) ; et un dépôt Amazon Elastic Container Registry (Amazon ECR), incluant au minimum les actions sagemaker:CreateEndpoint, sagemaker:CreateEndpointConfig, sagemaker:CreateModel, sagemaker:InvokeEndpointWithBidirectionalStream, et bedrock:InvokeModelWithResponseStream .
• L'exemple de conteneur épingle une ancienne version de PyTorch faisant l'objet d'un avis de sécurité connu (PYSEC-2026-2286, corrigé dans torch 2.10.0). Examinez et mettez à niveau cette épinglage avant toute utilisation autre qu'une preuve de concept.
• AWS Cloud Development Kit (AWS CDK) v2, installé et amorcé dans votre compte cible et votre Région AWS.
• Un point de terminaison temps réel SageMaker AI déployé à partir d'un conteneur déclarant l'étiquette Docker com.amazonaws.sagemaker.capabilities.bidirectional-streaming=true .
• Python 3.12 ou version ultérieure, uv, et Docker pour construire l'image du conteneur (le client HTTP/2 sous-jacent n'est fourni en wheels précompilés que pour Python 3.12 et versions ultérieures).
• Le AWS SDK for Python (Boto3) avec prise en charge de invoke_endpoint_with_bidirectional_stream.
• Une bonne connaissance des WebSockets et de Starlette ou d'un framework ASGI comparable.
• Le code de déploiement issu du dépôt GitHub référencé dans cet article.

Procédure pas à pas

La procédure pas à pas déploie d'abord le point de terminaison et l'agent, puis aborde les modifications côté client qui rendent le streaming incrémental fonctionnel de bout en bout.

Déployer le point de terminaison et l'agent

Le point de terminaison Kokoro-82M et l'exécution de l'agent se déploient via deux outils CLI, chacun encapsulant ses propres piles AWS CDK :

uv run speech-infra deploy kokoro-82m
uv run agent-infra deploy --bedrock-model-id <your-bedrock-model-id>

speech-infra deploy provisionne le point de terminaison Kokoro-82M. Cela comprend les piles de fondation et de cache de modèle qui créent les rôles IAM sous-jacents et synchronisent les poids du modèle dans Amazon Simple Storage Service (Amazon S3). La pile du point de terminaison construit ensuite l'image de conteneur et crée le modèle SageMaker AI, la configuration du point de terminaison et le point de terminaison. agent-infra deploy provisionne le runtime Amazon Bedrock AgentCore hébergeant l'agent Strands, qui appelle le point de terminaison que vous venez de déployer ; déployez donc d'abord le point de terminaison.

Étape 1 : Gérer le type de trame WebSocket que SageMaker AI envoie réellement

Le transport bidirectionnel utilisé par SageMaker AI livre chaque élément du message d'un client à votre conteneur sous forme de trame WebSocket binaire. Une façon courante d'écrire un gestionnaire WebSocket Starlette consiste à appeler receive_text(), qui lit message["text"] et lève KeyError si cette clé n'est pas présente. Comme chaque message envoyé par ce client arrive en binaire, ce gestionnaire échoue à chaque requête.

Cette défaillance est couramment mal diagnostiquée. Un gestionnaire d'exceptions générique qui transmet str(e) au client transforme KeyError('text') en une trame d'erreur indiquant 'text', ce qui ressemble à un champ manquant, et non à un problème de cadrage. Le point de terminaison lui-même signale InService en permanence, puisqu'il accepte les connexions et répond à chaque message, avec une erreur à chaque fois.

Acceptez explicitement l'un ou l'autre type de trame au lieu de supposer qu'un seul apparaîtra :

from starlette.websockets import WebSocket, WebSocketDisconnect

async def _receive_message(websocket: WebSocket) -> str:
    message = await websocket.receive()
    if message["type"] == "websocket.disconnect":
        raise WebSocketDisconnect(message.get("code", 1000))
    payload = message.get("text")
    if payload is None:
        payload = message.get("bytes", b"").decode("utf-8")
    return payload

Ceci décode les octets en UTF-8 lorsqu'aucun champ text n'est présent, plutôt que de supposer un seul type de trame. Testez-le en envoyant une trame binaire, car ce mode de défaillance produit une trame d'erreur bien formée et un état de point de terminaison sain, et non un plantage évident.

Étape 2 : Maintenir la session ouverte entre les messages

Une fois que votre conteneur a fini de synthétiser l'audio d'un message, l'instinct pousse à fermer la session, comme se termine un cycle HTTP. C'est une erreur ici. Après avoir envoyé l'audio d'un message et une trame de contrôle synthesis_complete , la boucle du conteneur devrait retourner en attente sur receive() pour le message suivant sur la même session, toujours ouverte. Ne fermez la connexion que lorsque le client envoie un message explicite {"type": "close"} .

Le diagramme suivant montre une seule session transportant deux phrases en séquence, fermée uniquement à la fin :

Figure 2 : Une session bidirectionnelle transportant deux phrases en séquence, fermée uniquement à la fin

De cette façon, vous pouvez envoyer plusieurs phrases, l'une après l'autre, sur une seule session persistante, sans coût de reconnexion entre elles. Fermer prématurément après une réponse imposerait une nouvelle connexion par phrase, annulant l'essentiel du bénéfice de latence.

Étape 3 : Diffuser les phrases au fur et à mesure que l'agent les produit

Une fois le comportement de session en place, la correction côté client du problème de latence est une classe qui normalise le texte arrivant sous deux formes : une chaîne complète connue à l'avance, ou un flux de jetons en direct. Elle produit la même sortie dans les deux cas : des phrases complètes, prêtes à être envoyées dès que chacune est terminée.

import re

_SENTENCE_RE = re.compile(r"(?<=[.!?])\s+")


class IncrementalSentenceChunker:
    def __init__(self) -> None:
        self._buffer = ""

    def feed(self, fragment: str) -> list[str]:
        self._buffer += fragment
        parts = _SENTENCE_RE.split(self._buffer)
        if len(parts) == 1:
            return []
        self._buffer = parts[-1]
        return [s.strip() for s in parts[:-1] if s.strip()]

    def flush(self) -> str | None:
        remainder = self._buffer.strip()
        self._buffer = ""
        return remainder or None

Appelez feed() une fois par fragment de texte à son arrivée. Elle renvoie les phrases nouvellement complétées, en retenant le texte final. Appelez flush() une seule fois, à la fin de l'entrée, pour libérer tout texte partiel restant, même sans ponctuation finale. Cela évite qu'un flux se terminant avant l'arrivée d'un point ne perde une dernière phrase.

Il ne s'agit pas de synthèse concurrente. Chaque gestionnaire bidirectionnel est une boucle strictement séquentielle : recevoir un message, synthétiser son audio, envoyer cet audio, envoyer synthesis_complete, puis recevoir le message suivant. L'audio de la phrase deux n'est pas généré pendant que celui de la phrase une est encore en cours de transmission. Ce que l'envoi incrémental apporte est plus restreint : la synthèse de la phrase une commence dès qu'elle existe, plutôt que d'attendre la réponse complète. Pour une réponse de plusieurs phrases avec un temps de génération réel entre sa première et sa dernière phrase, c'est là tout le gain de latence.

Nous avons vérifié cela sur un point de terminaison en direct en envoyant une entrée de trois phrases sur une seule session. Les trois morceaux sont arrivés avec un audio correspondant au texte envoyé pour chacun :

Étape 4 : Choisir un transport pour renvoyer l'audio au client

Diffuser le texte vers le modèle de manière incrémentale résout la moitié du problème. L'autre moitié consiste à acheminer l'audio résultant vers ce qui le restitue, généralement un navigateur. La plupart des frontends d'agents et de chat diffusent déjà le texte via Server-Sent Events (SSE) : unidirectionnels et pris en charge nativement par la plupart des navigateurs. SSE est un protocole texte, non conçu pour transporter une charge binaire brute. Une équipe ajoutant la voix à un pipeline SSE existant se tourne naturellement vers le motif qu'elle possède déjà : encoder l'audio en base64 et l'envelopper dans un événement JSON.

Nous avons mesuré le coût de cette approche sur 24 clips, confirmé avec Amazon Transcribe que la qualité ne différait pas selon les encodages, et comparé le MP3 à l'AAC :

Le Base64 gonfle chaque charge utile de près des 33 % attendus. Le coût plus subtil est l'alignement des trames. Les événements SSE sont délimités par une ligne vide, mais l'audio qu'ils contiennent est délimité en trames MP3. Les deux schémas ne s'alignent pas nécessairement avec la façon dont le transport découpe les données en blocs. Dans une réponse mesurée, quatre trames MP3 logiques sont arrivées en six blocs de livraison, dont deux se terminaient au milieu d'une trame. Un consommateur qui traite chaque bloc de livraison comme une unité complète échoue sur du trafic ordinaire et sain. La solution consiste à mettre en mémoire tampon les octets entrants et à n'agir qu'une fois un délimiteur trouvé :

buffer = b""
for event in resp["Body"]:
    if "PayloadPart" not in event:
        continue
    buffer += event["PayloadPart"]["Bytes"]
    while b"\n\n" in buffer:
        frame, buffer = buffer.split(b"\n\n", 1)
        handle(frame)

L'alternative utilisée aujourd'hui pour la liaison orientée modèle abandonne entièrement le base64 et l'encapsulation SSE. Elle multiplexe l'audio binaire brut et les trames de contrôle JSON sur la même session bidirectionnelle qui transporte déjà le texte saisi de façon incrémentale, et les distingue avec une seule règle.

import json


def _decode_control_frame(chunk: bytes) -> dict | None:
    try:
        decoded = json.loads(chunk.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError):
        return None
    return decoded if isinstance(decoded, dict) else None

Un bloc qui échoue à se décoder comme un objet JSON est de l'audio, transmis tel quel, sans étape d'encodage nulle part dans le chemin. Un cas limite est important : l'audio PCM brut peut légitimement commencer par l'octet 0x7b, le caractère ASCII {, le même caractère qui ouvre un objet JSON. Cette règle évite de mal classifier cet audio parce qu'elle exige que le bloc entier s'analyse comme du JSON valide, et non son premier octet seul, chose qu'une trame PCM ne fera pratiquement jamais.

Remarque : Si votre frontend dispose déjà d'un pipeline SSE fonctionnel pour le texte, l'audio encodé en base64 dans un événement SSE est le choix pragmatique. Il coûte environ 33 % de charge utile supplémentaire et une étape de mise en mémoire tampon et de réassemblage, mais réutilise une infrastructure que vous exploitez déjà. Si vous construisez le chemin vocal à neuf, ou si la latence est la priorité, l'approche binaire et JSON unifiée évite ces deux coûts et réutilise la connexion qui transporte déjà le texte.

Étape 5 : Tout assembler dans un vrai agent

Le cas plus complet, de bout en bout, est un agent génératif en direct composant sa réponse un token à la fois, sans aucun moyen de connaître la phrase quatre tant que la phrase une n'existe pas. Cet exemple s'exécute comme un agent Strands, adossé à Amazon Bedrock, déployé sur le runtime Amazon Bedrock AgentCore, appelant le point de terminaison Kokoro avec le modèle de l'étape 4. Trois liaisons de communication se connectent dans cette conception : la liaison orientée navigateur transportant des Server-Sent Events vers l'appelant, l'appel en streaming de l'agent vers Amazon Bedrock pour la génération de texte, et la session bidirectionnelle de l'agent avec le point de terminaison Kokoro pour la synthèse.

Le diagramme suivant montre ces trois liaisons de communication, et où se produit le chevauchement entre génération et synthèse :

Figure 3 : Les trois liaisons de communication reliant l'appelant, Amazon Bedrock et le point de terminaison Kokoro

La charge utile de la requête comporte un champ mode , soit "bidi" soit "batch", acheminé vers deux chemins de code au sein de la même fonction de point d'entrée. Un test de performance entre eux ne change qu'une seule variable : le moment où la synthèse démarre. En run_bidi, deux tâches s'exécutent concurremment. L'une draine le flux de texte de Bedrock et alimente le découpeur avec les phrases complétées. L'autre draine la session Kokoro et émet les blocs audio à leur arrivée. Les deux poussent dans une file partagée unique, si bien qu'un consommateur produit les événements dans leur véritable ordre d'arrivée : l'audio d'une phrase précoce peut atteindre l'appelant pendant que Bedrock génère encore une phrase ultérieure. run_batch est délibérément l'inverse : il attend la réponse complète avant un seul appel de synthèse, conservé comme référence équitable.

Dans l'autre sens, la liaison du conteneur de l'agent vers l'appelant utilise bien le base64, pour une raison différente de celle décrite à l'étape 4. BedrockAgentCoreAppa pour point d'entrée l'obligation de produire des dictionnaires JSON sérialisables simples, que son SDK transforme automatiquement en trames data: {json}\n\n . Il n'y a aucun moyen de lui remettre une trame binaire brute. C'est une conséquence du runtime géré nativement JSON, et non du choix que l'étape 4 a mesuré puis abandonné. Les deux run_bidi et run_batch paient ce coût identique sur cette liaison, si bien que la différence mesurée entre eux est purement temporelle : le moment où le premier bloc audio est émis.

Tests et validation

Nous avons isolé la contribution propre du transport TTS au temps jusqu'au premier audio en envoyant le même bloc de texte de 267 caractères et quatre phrases de deux manières : en un seul message (batch), et en seulement sa première phrase (incrémental). Nous avons effectué cinq essais chacun contre le point de terminaison en direct.

Comme mesuré précédemment, l'envoi de la seule première phrase a réduit le temps moyen jusqu'au premier octet audio de 329.3 ms à 156.7 ms, une amélioration relative de 52.4 %, isolée dans le transport TTS lui-même.

Nous avons ensuite mesuré le cas plus complet, de bout en bout, avec un vrai modèle de langage produisant le texte, en exécutant l'identique prompt à travers run_bidi et run_batch contre le même modèle et point de terminaison Bedrock en direct, si bien que la seule variable est le moment où la synthèse démarre.

Avec cette même méthodologie, le temps jusqu'au premier audio du mode bidi s'est révélé 2.4 fois plus rapide que le mode batch contre un conteneur local, et 3.5 fois plus rapide contre le runtime AgentCore déployé.

Le graphique suivant montre cet écart dans les deux environnements :

Figure 4 : Temps relatif jusqu'au premier audio pour les modes batch et bidi, en local et sur le runtime déployé

L'écart se creuse, plutôt que de se réduire, pour le cas déployé dès qu'un vrai saut réseau est ajouté. run_bidi absorbe ce saut à l'intérieur du chevauchement qu'elle exploite déjà entre génération et synthèse. run_batch ne dispose d'aucun chevauchement équivalent dans lequel masquer ce coût, puisqu'il ne fait qu'une chose à la fois, de sorte que chaque milliseconde supplémentaire de surcharge réseau vient directement s'ajouter à son attente sérielle.

Pour reproduire cette comparaison sur votre propre déploiement, exécutez le même test bidi-vs-batch exposé par le conteneur de l'agent :

uv run agent-infra demo --endpoint-url <your-endpoint-url> --prompt "<your prompt>"

Ceci envoie le prompt dans les deux modes et rapporte l'horodatage du premier audio_chunk événement contre agent_start pour chacun. Si vous voyez des erreurs à la place, vérifiez le KeyError: 'text' symptôme de l'Étape 1. Un point de terminaison rapportant InService alors que chaque requête échoue signale ce bug, et non un problème de déploiement.

Nettoyer

Pour éviter des frais continus, supprimez les ressources créées par ce tutoriel dans l'ordre inverse du déploiement, d'abord l'agent, puis le point de terminaison :

uv run agent-infra destroy --bedrock-model-id <your-bedrock-model-id>
uv run speech-infra destroy kokoro-82m

agent-infra destroy supprime la ressource d'exécution Amazon Bedrock AgentCore. speech-infra destroy supprime le modèle SageMaker AI, la configuration du point de terminaison et le point de terminaison. Aucune des deux commandes ne vide le compartiment Amazon S3 qui met en cache les poids du modèle ni ne supprime le référentiel Amazon ECR contenant l'image du conteneur, car tous deux peuvent être partagés entre plusieurs modèles. Si vous n'en avez plus besoin, videz et supprimez le compartiment Amazon S3 et supprimez le référentiel Amazon ECR manuellement, car un point de contrôle de modèle stocké et une image de conteneur continuent d'entraîner de petits frais de stockage jusqu'à leur suppression.

Conclusion

Nous avons montré comment éliminer une source de latence évitable d'un agent vocal — attendre la réponse complète de l'agent avant de démarrer la synthèse vocale — et comment quantifier un second coût plus discret : l'audio encodé en base64 dans SSE gonfle la charge utile d'environ un tiers et peut désaligner les frontières de trames MP3 si vous ne mettez pas en mémoire tampon en conséquence. Diffuser chaque phrase vers un point de terminaison bidirectionnel SageMaker AI dès qu'un agent la produit a réduit le délai jusqu'au premier audio de plus de la moitié en isolation, et de 2,4 à 3,5 fois dans un agent déployé. Le multiplexage de l'audio binaire et des trames de contrôle JSON sur cette session, au lieu d'encoder l'audio en texte, élimine le gonflement de taille et le travail de réalignement des trames qu'un relais SSE naïf nécessite.

Si vous ajoutez la voix à un agent qui diffuse déjà du texte sur SSE, commencez par l'approche base64 de l'Étape 4 et revenez au transport unifié une fois que la latence devient prioritaire. Pour en savoir plus, consultez la documentation d'inférence en temps réel Amazon SageMaker AI et la documentation Amazon Bedrock AgentCore. Si vous souhaitez de l'aide pour appliquer ce modèle à votre propre architecture, contactez votre équipe de compte AWS.

À propos de l'auteur