Transmite voz de baja latencia desde agentes de IA usando Kokoro TTS en Amazon SageMaker AI

Un cliente le hace una pregunta a un agente de voz de IA y espera. El agente ya tiene una respuesta, está a mitad de generar una respuesta larga de varios párrafos, pero el cliente no escucha nada durante cinco segundos. Para cuando llega la respuesta, el cliente está frustrado, confundido y ya no presta atención.

Las integraciones de texto a voz crean exactamente ese vacío. Una respuesta de una sola frase puede generarse en 600 milisegundos, pero la respuesta completa tarda cinco segundos, y la integración espera a tenerla entera antes de que comience la síntesis.

Esta publicación muestra cómo cerrar ese vacío. Transmitimos cada frase a un modelo de texto a voz (TTS) Kokoro-82M alojado en Amazon SageMaker AI tan pronto como el agente la produce, de modo que la síntesis comienza con la primera frase en lugar de con la respuesta terminada. Llevar ese audio de vuelta a un navegador también necesita su propia solución, por lo que también cubrimos ambos transportes para hacerlo y lo que cuesta cada uno en la práctica.

Descripción general de la solución

Esta solución cierra el vacío al transmitir la síntesis en paralelo con la generación en lugar de esperar a que termine. A medida que el agente produce cada frase, esa frase se transmite inmediatamente a un modelo de texto a voz Kokoro-82M alojado, de modo que la reproducción de audio puede comenzar antes de que el agente termine de pronunciar su respuesta completa. Kokoro-82M tiene solo 82 millones de parámetros, lo suficientemente pequeño para ejecutarse de manera económica en una sola instancia GPU de gama media. Esto mantiene bajos tanto la latencia como el costo de inferencia para un agente de voz que gestiona tráfico real.

Kokoro-82M se ejecuta en un contenedor personalizado en un endpoint en tiempo real de SageMaker AI, en una instancia de ml.g5.xlarge GPU. El contenedor expone dos rutas: POST /invocations, que transmite una respuesta para una entrada completa, y WS /invocations-bidirectional-stream, que mantiene una conexión abierta a través de muchos mensajes en ambas direcciones.

El siguiente diagrama muestra la ruta de la solicitud desde el cliente hasta el modelo:

Figura 1: Ruta de solicitud desde el cliente hasta el endpoint en tiempo real de SageMaker AI a través de un flujo bidireccional

Requisitos previos

Antes de comenzar, necesita:

• Una cuenta de AWS con permisos para crear endpoints de SageMaker AI; AgentCore runtime, una capacidad de Amazon Bedrock AgentCore; roles de AWS Identity and Access Management (IAM); y un repositorio de Amazon Elastic Container Registry (Amazon ECR), que incluya como mínimo las acciones sagemaker:CreateEndpoint, sagemaker:CreateEndpointConfig, sagemaker:CreateModel, sagemaker:InvokeEndpointWithBidirectionalStream, y bedrock:InvokeModelWithResponseStream .
• El contenedor de ejemplo fija una versión anterior de PyTorch con un aviso de seguridad conocido (PYSEC-2026-2286, corregido en torch 2.10.0). Revise y actualice la versión fijada antes de cualquier uso que no sea de prueba de valor.
• AWS Cloud Development Kit (AWS CDK) v2, instalado y preparado (bootstrapped) en su cuenta y región de AWS de destino.
• Un endpoint en tiempo real de SageMaker AI implementado desde un contenedor que declara la etiqueta com.amazonaws.sagemaker.capabilities.bidirectional-streaming=true Docker.
• Python 3.12 o posterior, uvy Docker para compilar la imagen del contenedor (el cliente HTTP/2 subyacente solo proporciona wheels precompilados para Python 3.12 y versiones más recientes).
• El AWS SDK para Python (Boto3) con soporte para invoke_endpoint_with_bidirectional_stream.
• Familiaridad con WebSockets y Starlette o un framework ASGI comparable.
• El código de despliegue del repositorio de GitHub referenciado en esta publicación.

Recorrido

El recorrido primero despliega el endpoint y el agente, y luego trabaja con los cambios del lado del cliente que hacen que el streaming incremental funcione de extremo a extremo.

Desplegar el endpoint y el agente

El endpoint Kokoro-82M y el runtime del agente se despliegan mediante dos herramientas de CLI, cada una de las cuales encapsula sus propias pilas de AWS CDK:

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

speech-infra deploy provisiona el endpoint Kokoro-82M. Esto incluye las pilas de foundation y model-cache que crean los roles de IAM subyacentes y sincronizan los pesos del modelo en Amazon Simple Storage Service (Amazon S3). Luego, la pila del endpoint construye la imagen de contenedor y crea el modelo de SageMaker AI, la configuración del endpoint y el endpoint. agent-infra deploy provisiona el runtime de Amazon Bedrock AgentCore que aloja el agente de Strands, el cual llama al endpoint que acaba de desplegar, así que despliegue primero el endpoint.

Paso 1: Manejar el tipo de trama WebSocket que SageMaker AI realmente envía

El transporte bidireccional que utiliza SageMaker AI entrega cada parte del mensaje de un cliente a su contenedor como una trama WebSocket binaria. Una forma común de escribir un manejador WebSocket de Starlette es llamando receive_text(), que lee message["text"] y lanza KeyError si esa clave no está presente. Debido a que cada mensaje que este cliente envía llega como binario, ese manejador falla en cada solicitud.

La falla suele diagnosticarse erróneamente. Un controlador de excepciones genérico que reenvía str(e) al cliente convierte KeyError('text') en un frame de error que dice 'text', lo cual parece un campo faltante, no un problema de framing. El endpoint en sí reporta InService durante todo el tiempo, ya que acepta conexiones y responde a cada mensaje, con un error cada vez.

Acepta explícitamente cualquiera de los dos tipos de frame en lugar de asumir que aparecerá uno:

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

Esto decodifica los bytes como UTF-8 cuando no hay un campo de texto presente, en lugar de asumir un tipo de frame. Pruébalo enviando un frame binario, porque este modo de falla produce un frame de error bien formado y un estado de endpoint saludable, no un fallo evidente.

Paso 2: Mantener la sesión abierta entre mensajes

Después de que tu contenedor termina de sintetizar el audio de un mensaje, el instinto es cerrar la sesión, de la manera en que termina un ciclo HTTP. Eso está mal aquí. Después de enviar el audio de un mensaje y un frame de control synthesis_complete , el bucle del contenedor debe volver a esperar en receive() el siguiente mensaje en la misma sesión, que sigue abierta. Cierra la conexión solo cuando el cliente envíe un mensaje {"type": "close"} explícito.

El siguiente diagrama muestra una sesión que transporta dos frases en secuencia, cerrándose solo al final:

Figura 2: Una sesión bidireccional que transporta dos frases en secuencia, cerrándose solo al final

De esta manera, puedes enviar varias frases, una tras otra, en una sola sesión persistente, sin costo de reconexión entre ellas. Cerrar de inmediato después de una respuesta forzaría una nueva conexión por frase, deshaciendo la mayor parte del beneficio de latencia.

Paso 3: Transmitir las frases a medida que el agente las produce

Con el comportamiento de la sesión en su lugar, la corrección en el lado del cliente para el problema de latencia es una clase que normaliza el texto que llega en dos formas: una cadena completa conocida de antemano, o un flujo de tokens en vivo. Produce la misma salida en cualquier caso: frases completas, listas para enviarse tan pronto como cada una esté completa.

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

Llama a feed() una vez por fragmento de texto a medida que llega. Devuelve las frases recién completadas, reteniendo el texto final. Llama a flush() una vez, al final de la entrada, para liberar el texto parcial que quede, incluso sin puntuación final. Esto evita que un flujo que termina antes de que llegue un punto pierda una última frase.

Esto no es síntesis concurrente. Cada controlador bidireccional es un bucle estrictamente secuencial: recibir un mensaje, sintetizar su audio, enviar ese audio, enviar synthesis_complete, y luego recibir el siguiente mensaje. El audio de la frase dos no se genera mientras la frase uno aún se transmite. Lo que compra el envío incremental es algo más limitado: la síntesis de la frase uno comienza en el instante en que existe, en lugar de esperar la respuesta completa. Para una respuesta de varias frases con tiempo de generación real entre su primera y última frase, esa es toda la ganancia de latencia.

Lo verificamos contra un endpoint en vivo enviando una entrada de tres frases por una sola sesión. Los tres fragmentos llegaron con audio que coincidía con el texto enviado para cada uno:

Paso 4: Elegir un transporte para devolver el audio al cliente

Transmitir el texto al modelo incrementalmente resuelve la mitad del problema. La otra mitad es sacar el audio resultante hacia lo que lo reproduce, comúnmente un navegador. La mayoría de los frontends de agentes y chat ya transmiten texto mediante Server-Sent Events (SSE): unidireccionales, y soportados nativamente por la mayoría de los navegadores. SSE es un protocolo de texto, no diseñado para transportar un payload binario crudo. Un equipo que añade voz a un pipeline SSE existente recurre naturalmente al patrón que ya tiene: codificar el audio en base64 y envolverlo en un evento JSON.

Medimos el costo de ese enfoque en 24 clips, confirmamos con Amazon Transcribe que la calidad no difería entre codificaciones y comparamos MP3 con AAC:

Base64 infla cada payload en cerca del 33% esperado. El costo más sutil es la alineación de marcos. Los eventos SSE se delimitan con una línea en blanco, pero el audio interior está delimitado en tramas MP3. Los dos esquemas no necesariamente se alinean con la forma en que el transporte divide los datos en fragmentos. En una respuesta medida, cuatro tramas lógicas de MP3 llegaron como seis fragmentos de entrega, dos de los cuales terminaban a mitad de trama. Un consumidor que trata cada fragmento de entrega como una unidad completa falla con tráfico ordinario y saludable. La solución almacena en búfer los bytes entrantes y actúa solo cuando encuentra un delimitador:

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)

La alternativa que se usa hoy para la pata orientada al modelo omite por completo base64 y el encuadre SSE. Multiplexa audio binario sin procesar y tramas de control JSON sobre la misma sesión bidireccional que ya transporta entrada de texto incremental, y las distingue con una sola regla.

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 fragmento que no se puede decodificar como objeto JSON es audio, y se reenvía tal como se recibe, sin ningún paso de codificación en la ruta. Hay un caso límite importante: el audio PCM sin procesar puede legítimamente comenzar con el byte 0x7b, el carácter ASCII {, el mismo carácter que abre un objeto JSON. Esta regla evita clasificar erróneamente ese audio porque exige que todo el fragmento se analice como JSON válido, no solo su primer byte, algo que un fotograma PCM prácticamente nunca hará.

Nota: Si tu frontend ya tiene un pipeline SSE funcional para texto, el audio codificado en base64 dentro de un evento SSE es la opción pragmática. Cuesta aproximadamente un 33 por ciento adicional de carga y un paso de buffering y reensamblaje, pero reutiliza infraestructura que ya operas. Si estás construyendo la ruta de voz desde cero, o la latencia es la prioridad, el enfoque unificado binario y JSON evita ambos costes y reutiliza la conexión que ya transporta el texto.

Paso 5: Ponerlo todo junto en un agente real

El caso más completo, de extremo a extremo, es un agente generativo en vivo que compone su respuesta token a token, sin forma de conocer la cuarta frase hasta que existe la primera. Este ejemplo se ejecuta como un agente Strands, respaldado por Amazon Bedrock, desplegado en el runtime de Amazon Bedrock AgentCore, llamando al endpoint de Kokoro con el patrón del Paso 4. Tres tramos de comunicación se conectan en este diseño: el tramo orientado al navegador que lleva Server-Sent Events de vuelta al llamante, la llamada en streaming del agente a Amazon Bedrock para la generación de texto, y la sesión bidireccional del agente con el endpoint de Kokoro para la síntesis.

El siguiente diagrama muestra esos tres tramos de comunicación, y dónde ocurre la superposición entre generación y síntesis:

Figura 3: Los tres tramos de comunicación que conectan al llamante, Amazon Bedrock y el endpoint de Kokoro

La carga de la petición lleva un campo mode , ya sea "bidi" o "batch", despachado a dos rutas de código dentro de la misma función de entrada. Un benchmark entre ellas cambia exactamente una variable: cuándo comienza la síntesis. En run_bidi, dos tareas se ejecutan concurrentemente. Una drena el stream de texto de Bedrock y alimenta el fragmentador con las frases completadas. La otra drena la sesión de Kokoro y emite fragmentos de audio a medida que llegan. Ambas empujan a una cola compartida, de modo que un consumidor entrega los eventos en verdadero orden de llegada: el audio de una frase temprana puede llegar al llamante mientras Bedrock todavía genera una posterior. run_batch es deliberadamente lo opuesto: espera la respuesta completa antes de una única llamada de síntesis, y se mantiene como línea base justa.

En la dirección contraria, el tramo desde el contenedor del agente de vuelta al llamante sí usa base64, por una razón distinta a la descrita en el Paso 4. BedrockAgentCoreAppexige que su función de entrada produzca diccionarios planos serializables a JSON, que su SDK convierte en data: {json}\n\n fotogramas automáticamente. No hay forma de entregarle un fotograma binario en bruto. Esa es una consecuencia del runtime gestionado nativo de JSON, no de la elección que el Paso 4 midió y abandonó. Tanto run_bidi como run_batch pagan ese coste idéntico en este tramo, por lo que la diferencia medida entre ellos es puramente temporal: cuándo se emite el primer fragmento de audio.

Pruebas y validación

Aislamos la contribución propia del transporte TTS al tiempo hasta el primer audio enviando el mismo bloque de texto de 267 caracteres y cuatro frases de dos maneras: como un solo mensaje (batch), y como solo su primera frase (incremental). Ejecutamos cinco pruebas de cada tipo contra el endpoint en vivo.

Como se midió anteriormente, enviar solo la primera frase redujo el tiempo medio hasta el primer byte de audio de 329.3 ms a 156.7 ms, una mejora relativa del 52.4 por ciento, aislada en el propio transporte TTS.

Luego medimos el caso más completo, de extremo a extremo, con un modelo de lenguaje real produciendo el texto, ejecutando la misma prompt a través de run_bidi y run_batch contra el mismo modelo y endpoint de Bedrock en vivo, de modo que la única variable es cuándo comienza la síntesis.

Usando esa misma metodología, el tiempo hasta el primer audio del modo bidi resultó 2.4 veces más rápido que el modo batch contra un contenedor local, y 3.5 veces más rápido contra el runtime de AgentCore desplegado.

El siguiente gráfico muestra esa brecha en ambos entornos:

Figura 4: Tiempo relativo hasta el primer audio para los modos batch y bidi, localmente y en el runtime desplegado

La brecha se amplía, en lugar de reducirse, para el caso desplegado una vez que se añade un salto de red real. run_bidi absorbe ese salto dentro del solapamiento que ya está explotando entre generación y síntesis. run_batch no tiene una superposición equivalente en la que ocultar ese costo, ya que hace una cosa a la vez, por lo que cada milisegundo adicional de sobrecarga de red recae directamente en su espera en serie.

Para reproducir esta comparación en su propia implementación, ejecute la misma prueba de bidi frente a lotes que expone el contenedor del agente:

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

Esto envía el prompt a través de ambos modos e informa la marca de tiempo de la primera. audio_chunk evento contra agent_start por cada uno. Si en cambio ves errores, revisa el KeyError: 'text' síntoma del Paso 1. Un punto de interés (endpoint) de informe InService mientras que cada solicitud falla señala ese error, no un problema de despliegue.

Limpieza

Para evitar cargos continuos, elimine los recursos que crea esta guía en orden inverso al de despliegue: primero el agente, luego el endpoint:

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

agent-infra destroy elimina el recurso de runtime de Amazon Bedrock AgentCore. speech-infra destroy elimina el modelo de SageMaker AI, la configuración del endpoint y el endpoint. Ninguno de los dos vacía el bucket de Amazon S3 que almacena en caché los pesos del modelo ni elimina el repositorio de Amazon ECR que contiene la imagen del contenedor, ya que ambos pueden compartirse entre varios modelos. Si ya no los necesita, vacíe y elimine el bucket de Amazon S3 y quite el repositorio de Amazon ECR manualmente, ya que un checkpoint de modelo almacenado y una imagen de contenedor siguen generando pequeños cargos de almacenamiento hasta que se eliminen.

Conclusión

Mostramos cómo eliminar una fuente de latencia evitable de un agente de voz, la espera de la respuesta completa del agente antes de iniciar la síntesis de voz, y cómo cuantificar un segundo costo más silencioso: el audio codificado en base64 dentro de SSE infla la carga útil en aproximadamente un tercio y puede desalinearse con los límites de los fotogramas MP3 si no se usa un búfer adecuado. Transmitir cada frase a un endpoint bidireccional de SageMaker AI tan pronto como el agente la produce redujo el tiempo hasta el primer audio a menos de la mitad de forma aislada, y entre 2.4 y 3.5 veces en un agente desplegado. La multiplexación de audio binario y tramas de control JSON sobre esa sesión, en lugar de codificar el audio como texto, elimina la inflación de tamaño y el trabajo de realineación de tramas que requiere un relé SSE ingenuo.

Si está añadiendo voz a un agente que ya transmite texto a través de SSE, comience con el enfoque base64 en el Paso 4 y reconsidere el transporte unificado cuando la latencia se convierta en la prioridad. Para obtener más información, consulte la documentación de inferencia en tiempo real de Amazon SageMaker AI y la documentación de Amazon Bedrock AgentCore. Si desea ayuda para aplicar este patrón a su propia arquitectura, contacte a su equipo de cuenta de AWS.

Sobre el autor