Phase 06 - Lesson 12

Construa um pipeline de assistente de voz — o capstone da Fase 6

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

Tudo das lições 01-11, costurado em uma só coisa. Construa um assistente de voz que escuta, raciocina e responde falando. Em 2026 isso é um problema de engenharia resolvido, não um problema de pesquisa — mas os detalhes de integração decidem se ele vai para produção.

Tipo: Build Linguagens: Python Pré-requisitos: Fase 6 · 04, 05, 06, 07, 11; Fase 11 · 09 (Function Calling); Fase 14 · 01 (Agent Loop) Tempo: ~120 minutos

O problema

Construa um assistente ponta a ponta:

  1. Captura entrada do microfone (16 kHz mono).
  2. Detecta o início/fim da fala do usuário.
  3. Transcreve em streaming.
  4. Passa a transcrição para um LLM capaz de chamar ferramentas (timer, clima, agenda).
  5. Faz streaming do texto do LLM para um TTS.
  6. Reproduz o áudio de volta para o usuário.
  7. Para se o usuário interromper no meio da resposta.

Meta de latência: o primeiro byte de áudio do TTS dentro de 800 ms após o usuário terminar de falar, em uma CPU de laptop. Meta de qualidade: nenhuma palavra perdida, nenhuma legenda alucinada no silêncio, nenhum vazamento de clonagem de voz, nenhuma injeção de prompt bem-sucedida.

O conceito

Pipeline de assistente de voz: microfone → VAD → STT → LLM+ferramentas → TTS → alto-falante

Os sete componentes

  1. Captura de áudio. Microfone → 16 kHz mono → blocos de 20 ms. Geralmente sounddevice em Python ou AudioUnit/ALSA/WASAPI nativo em produção.
  2. VAD (Lição 11). Silero VAD @ limiar 0.5, fala mínima de 250 ms, hang-over de silêncio de 500 ms. Sinaliza "início" e "fim".
  3. STT em streaming (Lições 4-5). Whisper-streaming, Parakeet-TDT ou Deepgram Nova-3 (API). Transcrições parciais + finais.
  4. LLM com tool calling. GPT-4o / Claude 3.5 / Gemini 2.5 Flash. Schema JSON para as ferramentas. Streaming de tokens.
  5. TTS em streaming (Lição 7). Kokoro-82M (mais rápido aberto) ou Cartesia Sonic (comercial). Inicie o TTS após 20 tokens do LLM.
  6. Reprodução. Saída para o alto-falante; codificação opus para redes de baixa largura de banda.
  7. Tratador de interrupção. Se o VAD disparar durante a reprodução do TTS, pare a reprodução, cancele o LLM, reinicie o STT.

Os três modos de falha que você vai encontrar

  1. Corte da primeira palavra. O VAD começa um instante tarde demais. O "ei" do usuário fica faltando. Defina o limiar de início em 0.3, não 0.5.
  2. Confusão de interrupção no meio da resposta. O LLM continua gerando depois que o usuário interrompe; o assistente fala por cima do usuário. Conecte VAD → cancelar-LLM.
  3. Alucinação no silêncio. O Whisper produz "Obrigado por assistir" nos frames silenciosos de aquecimento. Sempre faça gating por VAD.

Stacks de referência de produção em 2026

Stack Latência Termos Notas
LiveKit + Deepgram + GPT-4o + Cartesia 350-500 ms API comercial Padrão da indústria em 2026
Pipecat + Whisper-streaming + GPT-4o + Kokoro 500-800 ms majoritariamente aberto Amigável para DIY
Moshi (full-duplex) 200-300 ms CC-BY 4.0 Modelo único; arquitetura diferente, lição 15
Vapi / Retell (gerenciado) 300-500 ms comercial Mais rápido para lançar; customização limitada
Whisper.cpp + llama.cpp + Kokoro-ONNX offline aberto Privacidade / edge

Construa

Passo 1: captura de microfone com 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()

Passo 2: captura de turno com 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)

Passo 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)

Passo 4: tool calling dentro do loop do 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)

Passo 5: tratamento de interrupção

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

Use

Veja code/main.py para uma simulação executável que conecta todos os sete componentes com modelos stub, para que você possa ver o formato do pipeline mesmo sem hardware. Para uma implementação real, troque os stubs por:

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

Armadilhas

  • Registrar PII para sempre. O áudio de um turno completo é PII na maioria das jurisdições. Retenção de 30 dias, criptografado em repouso.
  • Sem barge-in. Os usuários vão interromper. Seu assistente precisa parar de falar.
  • TTS que bloqueia. TTS síncrono bloqueia o event loop. Use async ou uma thread separada.
  • Sem tratamento de erro de tool call. Ferramentas falham. O LLM precisa receber de volta o erro + tentar de novo uma vez, depois degradar de forma graciosa.
  • Filtros de alucinação exagerados. Filtre demais e o assistente repete "Não posso ajudar com isso." Filtre de menos e ele diz qualquer coisa. Calibre em um conjunto de validação separado.
  • Sem opção de wake-word. Sempre escutando é uma responsabilidade de privacidade. Adicione um gate de wake-word (Porcupine ou openWakeWord).

Entregue

Salve como outputs/skill-voice-assistant-architect.md. Dadas as restrições de orçamento + escala + idioma + compliance, produza uma especificação de stack completa.

Exercícios

  1. Fácil. Execute code/main.py. Ele simula um turno completo ponta a ponta com módulos stub e imprime a latência por estágio.
  2. Médio. Substitua o stub de STT por um modelo Whisper real em um .wav pré-gravado. Meça o WER e a latência ponta a ponta.
  3. Difícil. Adicione tool calling: implemente get_weather (qualquer API) e set_timer. Roteie o LLM pelas ferramentas e verifique que, quando o usuário diz "coloca um timer de 5 minutos", a função certa dispara e a resposta falada confirma.

Termos-chave

Termo O que as pessoas dizem O que realmente significa
Turno Um ciclo completo usuário + assistente Uma fala do usuário delimitada por VAD + uma resposta LLM-TTS.
Barge-in Interrupção O usuário fala enquanto o assistente fala; o assistente para.
Wake word "Ei, assistente" Detector de palavra-chave curta; Porcupine, Snowboy, openWakeWord.
End-pointing Fim do turno Decisão de VAD + silêncio mínimo de que o usuário terminou.
Pre-roll Buffer pré-fala Manter 200-400 ms de áudio antes do VAD disparar para evitar o corte da primeira palavra.
Tool call Invocação de função O LLM emite JSON; o runtime despacha; o resultado retorna no loop.

Leituras adicionais

0 lifetime access. Curriculum based on AI Engineering from Scratch by Rohit Ghumare (MIT, used under attribution).