Phase 06 - Lesson 12

Construye una canalización de asistente de voz — el capstone de la Fase 6

This lesson includes a graded coding exercise that runs in your browser, unlocked with lifetime access.

Todo de las lecciones 01-11, unido en una sola cosa. Construye un asistente de voz que escucha, razona y responde hablando. En 2026 esto es un problema de ingeniería resuelto, no un problema de investigación — pero los detalles de integración deciden si llega a producción.

Tipo: Build Lenguajes: Python Requisitos previos: Fase 6 · 04, 05, 06, 07, 11; Fase 11 · 09 (Function Calling); Fase 14 · 01 (Agent Loop) Tiempo: ~120 minutos

El problema

Construye un asistente de extremo a extremo:

1. Captura la entrada del micrófono (16 kHz mono).
2. Detecta el inicio/fin del habla del usuario.
3. Transcribe en streaming.
4. Pasa la transcripción a un LLM capaz de llamar herramientas (temporizador, clima, calendario).
5. Hace streaming del texto del LLM hacia un TTS.
6. Reproduce el audio de vuelta al usuario.
7. Se detiene si el usuario interrumpe a mitad de la respuesta.

Objetivo de latencia: el primer byte de audio del TTS dentro de 800 ms después de que el usuario termina de hablar, en una CPU de laptop. Objetivo de calidad: ninguna palabra perdida, ningún subtítulo alucinado en el silencio, ninguna fuga de clonación de voz, ninguna inyección de prompt exitosa.

El concepto

Los siete componentes

1. Captura de audio. Micrófono → 16 kHz mono → bloques de 20 ms. Generalmente sounddevice en Python o AudioUnit/ALSA/WASAPI nativo en producción.
2. VAD (Lección 11). Silero VAD @ umbral 0.5, habla mínima de 250 ms, hang-over de silencio de 500 ms. Señala "inicio" y "fin".
3. STT en streaming (Lecciones 4-5). Whisper-streaming, Parakeet-TDT o Deepgram Nova-3 (API). Transcripciones parciales + finales.
4. LLM con tool calling. GPT-4o / Claude 3.5 / Gemini 2.5 Flash. Schema JSON para las herramientas. Streaming de tokens.
5. TTS en streaming (Lección 7). Kokoro-82M (el abierto más rápido) o Cartesia Sonic (comercial). Inicia el TTS después de 20 tokens del LLM.
6. Reproducción. Salida al altavoz; codificación opus para redes de bajo ancho de banda.
7. Manejador de interrupción. Si el VAD se dispara durante la reproducción del TTS, detén la reproducción, cancela el LLM, reinicia el STT.

Los tres modos de falla que vas a encontrar

1. Corte de la primera palabra. El VAD arranca un instante demasiado tarde. El "hey" del usuario falta. Pon el umbral de inicio en 0.3, no 0.5.
2. Confusión de interrupción a mitad de respuesta. El LLM sigue generando después de que el usuario interrumpe; el asistente habla por encima del usuario. Conecta VAD → cancelar-LLM.
3. Alucinación en el silencio. Whisper produce "Gracias por ver" en los frames silenciosos de calentamiento. Siempre aplica gating por VAD.

Stacks de referencia de producción en 2026

Construye

Paso 1: captura de micrófono con chunking (pseudocódigo)

import sounddevice as sd

def mic_stream(chunk_ms=20, sr=16000):
    q = queue.Queue()
    def cb(indata, frames, time, status):
        q.put(indata.copy().flatten())
    with sd.InputStream(channels=1, samplerate=sr, blocksize=int(sr * chunk_ms/1000), callback=cb):
        while True:
            yield q.get()

Paso 2: captura de turno con gating por VAD

def capture_turn(stream, vad, pre_roll_ms=300, silence_ms=500):
    buf, pre, triggered = [], collections.deque(maxlen=pre_roll_ms // 20), False
    silent = 0
    for chunk in stream:
        pre.append(chunk)
        if vad(chunk):
            if not triggered:
                buf = list(pre)
                triggered = True
            buf.append(chunk)
            silent = 0
        elif triggered:
            silent += 20
            buf.append(chunk)
            if silent >= silence_ms:
                return b"".join(buf)

Paso 3: streaming STT → LLM → TTS

async def turn(audio_bytes):
    transcript = await stt.transcribe(audio_bytes)
    async for token in llm.stream(transcript):
        async for audio in tts.stream(token):
            await speaker.play(audio)

Paso 4: tool calling dentro del loop del LLM

tools = [
    {"name": "get_weather", "parameters": {"location": "string"}},
    {"name": "set_timer", "parameters": {"seconds": "int"}},
]

async for chunk in llm.stream(user_text, tools=tools):
    if chunk.type == "tool_call":
        result = dispatch(chunk.name, chunk.args)
        continue_streaming(result)
    if chunk.type == "text":
        await tts.stream(chunk.text)

Paso 5: manejo de interrupción

tts_task = asyncio.create_task(tts_loop())
while True:
    chunk = await mic.get()
    if vad(chunk):
        tts_task.cancel()
        await speaker.stop()
        await new_turn()
        break

Úsalo

Mira code/main.py para una simulación ejecutable que conecta los siete componentes con modelos stub, de modo que puedas ver la forma de la canalización incluso sin hardware. Para una implementación real, reemplaza los stubs por:

• silero-vad (pip install silero-vad)
• deepgram-sdk u openai-whisper
• openai (gpt-4o) o anthropic
• kokoro o cartesia
• sounddevice para I/O

Trampas

• Registrar PII para siempre. El audio de un turno completo es PII en la mayoría de las jurisdicciones. Retención de 30 días, cifrado en reposo.
• Sin barge-in. Los usuarios van a interrumpir. Tu asistente debe dejar de hablar.
• TTS que bloquea. El TTS síncrono bloquea el event loop. Usa async o un hilo separado.
• Sin manejo de errores de tool call. Las herramientas fallan. El LLM debe recibir de vuelta el error + reintentar una vez, luego degradar de forma elegante.
• Filtros de alucinación demasiado celosos. Filtra de más y el asistente repite "No puedo ayudar con eso." Filtra de menos y dice cualquier cosa. Calibra en un conjunto de validación reservado.
• Sin opción de wake-word. Escuchar siempre es una responsabilidad de privacidad. Agrega un gate de wake-word (Porcupine u openWakeWord).

Entrégalo

Guarda como outputs/skill-voice-assistant-architect.md. Dadas las restricciones de presupuesto + escala + idioma + cumplimiento, produce una especificación de stack completa.

Ejercicios

1. Fácil. Ejecuta code/main.py. Simula un turno completo de extremo a extremo con módulos stub e imprime la latencia por etapa.
2. Medio. Reemplaza el stub de STT por un modelo Whisper real en un .wav pregrabado. Mide el WER y la latencia de extremo a extremo.
3. Difícil. Agrega tool calling: implementa get_weather (cualquier API) y set_timer. Enruta el LLM a través de las herramientas y verifica que, cuando el usuario dice "pon un temporizador de 5 minutos", la función correcta se dispara y la respuesta hablada lo confirma.

Términos clave

Lecturas adicionales

• LiveKit — quickstart de agente de voz — referencia de nivel de producción.
• Pipecat — ejemplos de agente de voz — framework amigable para DIY.
• OpenAI Realtime API — el camino gestionado nativo de voz.
• Kyutai Moshi — referencia full-duplex (Lección 15).
• Porcupine wake-word — gating de wake-word.
• Anthropic — guía de uso de herramientas — function calling de LLM.