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:
- Captura entrada do microfone (16 kHz mono).
- Detecta o início/fim da fala do usuário.
- Transcreve em streaming.
- Passa a transcrição para um LLM capaz de chamar ferramentas (timer, clima, agenda).
- Faz streaming do texto do LLM para um TTS.
- Reproduz o áudio de volta para o usuário.
- 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
Os sete componentes
- Captura de áudio. Microfone → 16 kHz mono → blocos de 20 ms. Geralmente
sounddeviceem Python ou AudioUnit/ALSA/WASAPI nativo em produção. - 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".
- STT em streaming (Lições 4-5). Whisper-streaming, Parakeet-TDT ou Deepgram Nova-3 (API). Transcrições parciais + finais.
- LLM com tool calling. GPT-4o / Claude 3.5 / Gemini 2.5 Flash. Schema JSON para as ferramentas. Streaming de tokens.
- 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.
- Reprodução. Saída para o alto-falante; codificação opus para redes de baixa largura de banda.
- 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
- 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.
- 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.
- 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-sdkouopenai-whisperopenai(gpt-4o) ouanthropickokorooucartesiasounddevicepara 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
- 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. - Médio. Substitua o stub de STT por um modelo Whisper real em um
.wavpré-gravado. Meça o WER e a latência ponta a ponta. - Difícil. Adicione tool calling: implemente
get_weather(qualquer API) eset_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
- LiveKit — quickstart de agente de voz — referência de nível de produção.
- Pipecat — exemplos de agente de voz — framework amigável para DIY.
- OpenAI Realtime API — o caminho gerenciado nativo de voz.
- Kyutai Moshi — referência full-duplex (Lição 15).
- Porcupine wake-word — gating de wake-word.
- Anthropic — guia de uso de ferramentas — function calling de LLM.