Tutorial: Como construir agentes de IA controlados por voz — pipeline completo com código

Por que agentes de voz são o próximo salto — e por que 2026 é o ano

Até 2024, construir um agente de voz com IA que soasse natural exigia integrar cinco serviços diferentes, lidar com WebSockets manualmente e rezar para a latência não passar de 3 segundos. Em 2026, o cenário mudou radicalmente: APIs unificadas como OpenAI Realtime, AssemblyAI Voice Agent e ElevenLabs Conversational AI empacotam todo o pipeline em uma única conexão WebSocket. Mas entender o que cada componente faz — e onde ele quebra — continua sendo a diferença entre um agente que funciona e um que realmente parece humano.

✅ O que você ganha com este tutorial

• Domínio do pipeline completo de voz: STT → LM → TTS com streaming
• Detecção de turno (turn-taking) configurável para o seu caso de uso
• Tratamento de interrupção (barge-in) com três sinais combinados
• Chamadas de ferramentas (tool calling) em conversas de voz sem dead air
• Código executável em Python puro, sem dependências externas

⚠️ O que você NÃO ganha

• Uma solução pronta para produção com microfone real — o código é didático
• Cobertura de ASR (reconhecimento de fala) multilíngue — foco em inglês
• Integração com hardware telefônico (SIP, WebRTC)

Pré-requisitos

Passo a passo

1. Entenda por que o padrão sequencial não funciona

O modelo ingênuo de um agente de voz é linear: o usuário fala → STT transcreve → LLM gera resposta → TTS sintetiza áudio. Esse padrão funciona, mas é lento demais para conversas naturais. Cada etapa espera a anterior terminar completamente, acumulando latência.

O padrão correto em produção é streaming: STT emite transcrições parciais enquanto o usuário ainda fala, o LLM gera tokens incrementalmente, e o TTS começa a sintetizar a primeira frase completa antes mesmo do LLM terminar a resposta inteira.

Por que isso importa: O limiar psicológico para uma conversa soar natural é ~500ms de tempo até o primeiro token (TTFT). Sistemas sequenciais raramente entregam abaixo de 2 segundos. O streaming é o que permite ficar abaixo de 500ms.

2. Implemente streaming de Speech-to-Text

O STT em produção para agentes de voz usa WebSocket persistente, não chamadas HTTP. O áudio é enviado em chunks de ~50ms e o servidor retorna eventos de transcrição parcial e final.

Por que isso importa: A transcrição parcial permite mostrar feedback em tempo real (“o agente está te ouvindo”), mas só a transcrição final deve ser usada para ações downstream. Transcrições parciais mudam conforme mais áudio chega — agir sobre elas causaria ações incorretas.

# streaming_stt.py — Python 3.10+, somente stdlib
import asyncio
from dataclasses import dataclass
from enum import Enum

class TranscriptEventType(Enum):
    PARTIAL = "transcript.user.delta"
    FINAL = "transcript.user"

@dataclass
class TranscriptEvent:
    event_type: TranscriptEventType
    text: str
    confidence: float = 1.0

class MockStreamingSTT:
    """Simula uma conexão WebSocket real com STT."""
    def __init__(self, simulated_utterance: str):
        words = simulated_utterance.split()
        self._partial_stages = [" ".join(words[:i]) for i in range(1, len(words) + 1)]

    async def stream_events(self):
        for stage in self._partial_stages[:-1]:
            yield TranscriptEvent(TranscriptEventType.PARTIAL, stage, confidence=0.7)
            await asyncio.sleep(0)
        yield TranscriptEvent(TranscriptEventType.FINAL, self._partial_stages[-1], confidence=0.97)

async def consume_transcript_stream(stt: MockStreamingSTT):
    """Renderiza parciais para feedback, age só no FINAL."""
    final_transcript = None
    async for event in stt.stream_events():
        if event.event_type == TranscriptEventType.PARTIAL:
            print(f"  [partial] '{event.text}'")
        elif event.event_type == TranscriptEventType.FINAL:
            final_transcript = event.text
            print(f"  [FINAL]   '{event.text}'")
    return final_transcript

# Execute: python streaming_stt.py
async def main():
    stt = MockStreamingSTT("Meu número de pedido é A B 3 7 9 2")
    final_text = await consume_transcript_stream(stt)
    print(f"\nTranscrição usada downstream: '{final_text}'")

asyncio.run(main())

3. Implemente detecção de turno (turn detection)

A detecção de turno é o componente mais negligenciado e o que mais causa frustração em agentes de voz. Ela decide quando o usuário terminou de falar — não é parte do STT, é uma política separada que consome o padrão de silêncio do stream de áudio.

A configuração usa dois números: silêncio mínimo (~600ms) que só encerra o turno se a transcrição também indica que a fala soa completa, e um teto máximo (~1500ms) que força o encerramento mesmo em pausas ambíguas.

# turn_detection.py — Python 3.10+, somente stdlib
from dataclasses import dataclass
from enum import Enum

class TurnState(Enum):
    LISTENING = "listening"
    SILENCE_PENDING = "silence_pending"
    END_OF_TURN = "end_of_turn"

@dataclass
class AudioFrame:
    is_speech: bool
    timestamp_ms: int

class TurnDetector:
    """Máquina de estados para detecção de fim de turno."""
    def __init__(self, min_silence_ms: int = 600, max_silence_ms: int = 1500):
        self.min_silence_ms = min_silence_ms
        self.max_silence_ms = max_silence_ms
        self._silence_start: int | None = None
        self.state = TurnState.LISTENING

    def process_frame(self, frame: AudioFrame, utterance_looks_complete: bool = True) -> TurnState:
        if frame.is_speech:
            self._silence_start = None
            self.state = TurnState.LISTENING
            return self.state

        if self._silence_start is None:
            self._silence_start = frame.timestamp_ms

        silence_duration = frame.timestamp_ms - self._silence_start

        if silence_duration >= self.max_silence_ms:
            self.state = TurnState.END_OF_TURN
        elif silence_duration >= self.min_silence_ms and utterance_looks_complete:
            self.state = TurnState.END_OF_TURN
        else:
            self.state = TurnState.SILENCE_PENDING

        return self.state

Ajuste por domínio: contextos de saúde ou atendimento a idosos se beneficiam de tetos mais altos (~2500ms). Bate-papo casual pode usar mínimo mais baixo (~300ms).

4. Streaming da resposta para Text-to-Speech

A unidade que faz a ponte entre o streaming do LLM e a síntese de voz não é o token, nem a resposta inteira — é a frase completa. O chunker de sentenças detecta o exato momento em que uma fronteira de frase aparece no buffer de tokens e a entrega para o TTS imediatamente.

# sentence_chunker.py
import asyncio, re

SENTENCE_END_PATTERN = re.compile(r'(?<=[.!?])\s+')

async def mock_llm_token_stream(text: str):
    for word in text.split(" "):
        yield word + " "
        await asyncio.sleep(0)

async def stream_sentences(token_stream) -> list[str]:
    """Entrega frases completas para TTS enquanto o LLM ainda gera."""
    buffer = ""
    sentences = []
    async for token in token_stream:
        buffer += token
        match = SENTENCE_END_PATTERN.search(buffer)
        while match:
            sentence = buffer[:match.start() + 1].strip()
            sentences.append(sentence)
            print(f"  [frase pronta para TTS] '{sentence}'")
            buffer = buffer[match.end():]
            match = SENTENCE_END_PATTERN.search(buffer)
    if buffer.strip():
        sentences.append(buffer.strip())
    return sentences

5. Tratamento de interrupção (barge-in)

Barge-in — o usuário interromper o agente no meio da fala — é tratado como o problema mais difícil em engenharia de voz. Requer quatro ações simultâneas: parar playback de TTS, cancelar geração em andamento, cancelar geração do LLM, e resetar o estado do stream.

O maior risco é falso positivo: tosse, ruído de fundo ou conversa paralela disparando interrupção. A prevenção combina três sinais:

• Limiar de energia (~-40 dBFS): ignora sons abaixo do volume de fala
• Classificador de voz (Silero VAD, WebRTC VAD): distingue fala de ruído
• Duração mínima (200-300ms): exige voz sustentada antes de disparar

# bargein_detector.py
@dataclass
class AudioChunk:
    energy_dbfs: float
    voice_confidence: float
    timestamp_ms: int

class BargeInDetector:
    def __init__(self, energy_threshold_dbfs=-40.0, voice_confidence_threshold=0.6, min_duration_ms=250):
        self.energy_threshold = energy_threshold_dbfs
        self.voice_threshold = voice_confidence_threshold
        self.min_duration_ms = min_duration_ms
        self._candidate_start_ms: int | None = None

    def process_chunk(self, chunk: AudioChunk) -> bool:
        passes_energy = chunk.energy_dbfs > self.energy_threshold
        passes_voice = chunk.voice_confidence > self.voice_threshold

        if not (passes_energy and passes_voice):
            self._candidate_start_ms = None
            return False

        if self._candidate_start_ms is None:
            self._candidate_start_ms = chunk.timestamp_ms

        sustained_duration = chunk.timestamp_ms - self._candidate_start_ms
        if sustained_duration >= self.min_duration_ms:
            self._candidate_start_ms = None
            return True
        return False

6. Tool calling em conversas de voz

Chamadas de ferramenta em voz têm um problema que não existe em chat de texto: o silêncio entre a chamada disparar e o resultado chegar é audível. Em texto, 3 segundos de pausa são invisíveis. Em uma chamada telefônica, o usuário assume que a ligação caiu e começa a falar — interrompendo a tool call em andamento.

A solução tem duas partes:

1. Técnica do preâmbulo: instrua o modelo a narrar o que está fazendo (“Deixa eu verificar isso para você…”) enquanto a função executa.
2. Buffer de resultados: acumule resultados de tool calls e só envie quando o turno terminar limpo. Se houve interrupção, descarte tudo.

# tool_result_buffer.py
class TurnOutcome(Enum):
    CLEAN_COMPLETION = "clean_completion"
    INTERRUPTED = "interrupted"

class ToolResultBuffer:
    def __init__(self):
        self._pending: list = []

    def accumulate(self, call_id: str, result: dict) -> None:
        self._pending.append((call_id, result))

    def resolve_turn(self, outcome: TurnOutcome) -> list:
        pending = list(self._pending)
        self._pending.clear()
        if outcome == TurnOutcome.CLEAN_COMPLETION:
            return pending
        return []  # interrompido: descarta tudo

Casos de uso reais

• Contact center inteligente: agente de voz que consulta CRM, verifica pedidos e agenda retornos — tudo sem transferir para humano
• Assistente de telemedicina: triagem por voz que coleta sintomas, verifica disponibilidade e agenda consultas
• Suporte técnico automatizado: diagnóstico de primeiro nível para ISPs e operadoras
• Tutor de idiomas: agente que corrige pronúncia em tempo real usando barge-in para interromper erros
• Vendas por voz: agente que qualifica leads por telefone enquanto consulta catálogo de produtos via tool calling

Troubleshooting

FAQ

Preciso implementar tudo isso do zero? Não. A maioria dos times em 2026 usa APIs integradas como OpenAI Realtime, AssemblyAI Voice Agent ou ElevenLabs Conversational AI. Mas entender cada componente é o que permite debugar quando o agente “simplesmente não funciona”.

Qual a latência aceitável para um agente de voz? Abaixo de 500ms de TTFT é o ideal. Entre 500ms e 1s é aceitável. Acima de 3s, a maioria dos usuários desengaja.

Funciona em português? Sim, mas a qualidade depende do motor de STT/TTS. Azure Speech Services e Google Cloud TTS têm bom suporte para português brasileiro. Whisper da OpenAI também funciona. O código de orquestração (turn detection, barge-in, buffer) é independente do idioma.

Quanto custa rodar um agente de voz em produção? Com APIs atuais, ~US$ 0.05-0.15 por minuto de conversa. Para 10.000 chamadas de 3 minutos, o custo fica entre US$ 1.500 e US$ 4.500/mês, mais infraestrutura de telefonia.

Posso usar modelos locais em vez de APIs? Sim. Modelos como Whisper (STT), Llama 3 (LLM) e Piper TTS podem rodar localmente com latência competitiva, eliminando custo de API. A troca é complexidade operacional.

O futuro dos agentes de voz

Em 2027, a distinção entre agentes de voz e texto tende a desaparecer — modelos multimodais nativos processarão áudio diretamente, sem pipeline STT→texto→TTS. Enquanto isso não chega, o domínio dos cinco componentes cobertos neste tutorial (streaming STT, turn detection, streaming TTS, barge-in e tool calling com buffer) é o que separa um agente de voz funcional de um que realmente parece humano.