En este tutorial, creamos un agente de transmisión de voz de extremo a extremo que refleja cómo funcionan los sistemas conversacionales modernos de baja latencia en tiempo real. Simulamos todo el proceso, desde la entrada de audio fragmentada y el reconocimiento de voz en streaming hasta el razonamiento incremental del modelo de lenguaje y la salida de texto a voz en streaming, mientras realizamos un seguimiento explícito de la latencia en cada etapa. Al trabajar con presupuestos de latencia estrictos y observar métricas como el tiempo hasta el primer token y el tiempo hasta el primer audio, nos centramos en las compensaciones prácticas de ingeniería que dan forma a las experiencias de usuario receptivas basadas en voz. Consulta los CÓDIGOS COMPLETOS aquí.
Definimos las estructuras de datos centrales y las representaciones de estado que nos permiten rastrear la latencia en todo el proceso de voz. Formalizamos señales de sincronización para ASR, LLM y TTS para garantizar una medición consistente en todas las etapas. También establecemos una máquina de estado de agente clara que guía cómo el sistema realiza la transición durante un turno conversacional. Consulta los CÓDIGOS COMPLETOS aquí.
Simulamos la entrada de audio en tiempo real dividiendo la voz en fragmentos de duración fija que llegan de forma asincrónica. Modelamos velocidades de conversación y comportamiento de transmisión realistas para imitar la entrada de micrófono en vivo. Usamos esta transmisión como base para probar componentes posteriores sensibles a la latencia. Consulta los CÓDIGOS COMPLETOS aquí.
Implementamos un módulo ASR de streaming que produce transcripciones parciales antes de emitir un resultado final. Progresivamente revelamos palabras para reflejar cómo funcionan los sistemas ASR modernos en tiempo real. También introducimos la finalización basada en silencio para aproximar la detección del final de la expresión. Consulta los CÓDIGOS COMPLETOS aquí.
para clave en respuestas: si clave en solicitud.inferior(): respuesta = respuestas[key]
break await asyncio.sleep(self.time_to_first_token) para palabra en respuesta.split(): producir palabra + ” ” await asyncio.sleep(1.0 / self.tokens_per_segundo) clase StreamingTTS: def __init__(self, time_to_first_chunk: float = 0.2, chars_per_segundo: float = 15): self.time_to_first_chunk = time_to_first_chunk self.chars_per_segundo = caracteres_por_segundo async def synthesize_stream(self, text_stream: AsyncIterator[str]) -> Asincrónico[np.ndarray]: first_chunk = True buffer = “” async para texto en text_stream: buffer += texto if len(buffer) >= 20 o first_chunk: if first_chunk: await asyncio.sleep(self.time_to_first_chunk) first_chunk = Duración falsa = len(buffer) / self.chars_per_segundo rendimiento np.random.randn(int(16000 * duración)).astype(np.float32) * 0.1 buffer = “” espera asyncio.sleep(duración * 0.5)
En este fragmento, modelamos un modelo de lenguaje de transmisión y un motor de transmisión de texto a voz trabajando juntos. Generamos respuestas token por token para capturar el comportamiento del tiempo hasta el primer token. Luego convertimos texto incremental en fragmentos de audio para simular la síntesis de voz temprana y continua. Consulta los CÓDIGOS COMPLETOS aquí.
async def Process_turn(self, user_input: str) -> LatencyMetrics: metrics = LatencyMetrics() start_time = time.time() metrics.audio_chunk_received = time.time() – start_time audio_gen = self.audio_stream.stream_audio(user_input) metrics.asr_started = time.time() – start_time asíncrono para texto, final en self.asr.transcribe_stream(audio_gen, user_input): si es final: metrics.asr_complete = time.time() – start_time transcripción = texto metrics.llm_started = time.time() – start_time respuesta = “” async para token en self.llm.generate_response(transcripción): si no metrics.llm_first_token: metrics.llm_first_token = time.time() – respuesta start_time += token metrics.llm_complete = time.time() – start_time metrics.tts_started = time.time() – start_time async def text_stream(): para palabra en respuesta.split(): produce palabra + ” ” async para _ en self.tts.synthesize_stream(text_stream()): si no metrics.tts_first_chunk: metrics.tts_first_chunk = time.time() – start_time metrics.tts_complete = time.time() – start_time self.metrics_history.append(metrics) devuelve métricas
Orquestamos el agente de voz completo conectando la entrada de audio, ASR, LLM y TTS en un único flujo asincrónico. Registramos marcas de tiempo precisas en cada transición para calcular métricas de latencia críticas. Tratamos cada turno de usuario como un experimento aislado para permitir un análisis sistemático del rendimiento. Consulta los CÓDIGOS COMPLETOS aquí.
“Hello, how are you today?”,
“What’s the weather like?”,
“Can you tell me the time?”
]
para texto en entradas: await agent.process_turn(text) await asyncio.sleep(1) if __name__ == “__main__”: asyncio.run(run_demo())
Ejecutamos todo el sistema en múltiples turnos de conversación para observar la coherencia y variación de la latencia. Aplicamos presupuestos de latencia agresivos para estresar el proceso bajo restricciones realistas. Usamos estas ejecuciones para validar si el sistema cumple con los objetivos de capacidad de respuesta en todas las interacciones.
En conclusión, demostramos cómo se puede orquestar un agente de voz totalmente en streaming como un único canal asíncrono con límites de escenario claros y garantías de rendimiento mensurables. Demostramos que la combinación de ASR parcial, transmisión LLM a nivel de token y TTS de inicio temprano reduce la latencia percibida, incluso cuando el tiempo total de cálculo no es trivial. Este enfoque nos ayuda a razonar sistemáticamente sobre la toma de turnos, la capacidad de respuesta y las palancas de optimización, y proporciona una base sólida para extender el sistema hacia implementaciones del mundo real utilizando modelos de producción ASR, LLM y TTS.
Consulta los CÓDIGOS COMPLETOS aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 100.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.
Asif Razzaq es el director ejecutivo de Marktechpost Media Inc.. Como emprendedor e ingeniero visionario, Asif está comprometido a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, Marktechpost, que se destaca por su cobertura en profundidad del aprendizaje automático y las noticias sobre aprendizaje profundo que es técnicamente sólida y fácilmente comprensible para una amplia audiencia. La plataforma cuenta con más de 2 millones de visitas mensuales, lo que ilustra su popularidad entre el público.