Speech to text em Python: arquivo, stream e microfone

Para converter fala em texto em Python, você envia áudio a um modelo de reconhecimento de fala e recebe texto de volta. A decisão interessante é o tipo de modelo, porque um arquivo gravado e uma conversa ao vivo são trabalhos diferentes. Um arquivo pode esperar o clipe inteiro. Quem está na linha, não.

Este guia mostra três jeitos de fazer isso, cada um com código que funciona como está: transcrever um arquivo gravado, fazer streaming de áudio com detecção automática de turnos e transcrever seu microfone ao vivo. Os exemplos de streaming usam o Cartesia Ink 2, o modelo que construímos para agentes de voz. Quando um modelo local gratuito é a melhor escolha, dizemos isso e mostramos ele também.

Escolha uma abordagem

A maioria dos tutoriais para no SpeechRecognition com recognize_google(). Esse pacote é um wrapper para uma dúzia de motores, mas o backend padrão do Google usa uma chave compartilhada que o Google pode revogar a qualquer momento, e espera uma frase completa antes de enviar qualquer coisa. Serve para uma demo. Não para atender telefone.

Pré-requisitos

• Python 3.9 ou mais recente
• Para os exemplos da Cartesia, uma chave de API do playground da Cartesia e o SDK:

python -m pip install "cartesia[websockets]"
export CARTESIA_API_KEY="your-api-key"

O extra websockets é necessário para os exemplos de streaming. Guarde a chave numa variável de ambiente ou num gerenciador de segredos, nunca em código que vai para o repositório.

1. Transcreva um arquivo gravado

Se o áudio já está gravado, envie o arquivo inteiro em uma única requisição. O endpoint em lote da Cartesia aceita contêineres como WAV e MP3 e pode devolver um timestamp para cada palavra. Salve isto como transcribe_file.py:

import os
import sys

from cartesia import Cartesia

client = Cartesia(api_key=os.environ["CARTESIA_API_KEY"])

with open(sys.argv[1], "rb") as audio_file:
    result = client.stt.transcribe(
        file=audio_file,
        model="ink-whisper",
        language="en",
        timestamp_granularities=["word"],
    )

print(result.text)
for word in result.words or []:
    print(f"{word.start:6.2f}s  {word.word}")

Rode com python transcribe_file.py meeting.mp3. A transcrição em lote usa o ink-whisper, que cobre cerca de 100 idiomas; passe o código do idioma do seu áudio. O exemplo de STT em lote na documentação tem a versão assíncrona.

A alternativa local

Se o áudio não pode sair da sua máquina, ou se você tem muito áudio e uma GPU, rode o Whisper você mesmo. O faster-whisper reimplementa o Whisper da OpenAI sobre o CTranslate2 e roda na CPU de um notebook:

python -m pip install faster-whisper

from faster_whisper import WhisperModel

model = WhisperModel("base.en", device="cpu", compute_type="int8")
segments, info = model.transcribe("jfk.flac", beam_size=5)

print(f"Detected language: {info.language} ({info.language_probability:.2f})")
for segment in segments:
    print(f"[{segment.start:.2f}s -> {segment.end:.2f}s] {segment.text}")

Na amostra do JFK de 11 segundos do repositório do Whisper, o modelo base.en transcreveu a citação palavra por palavra em cerca de 2,5 segundos numa CPU de servidor de 4 núcleos, depois de um único download do modelo de 140 MB. Modelos maiores são mais precisos e mais lentos. O que você não ganha é streaming: o Whisper trabalha com segmentos prontos, então não é a ferramenta para uma conversa.

2. Faça streaming de áudio e receba turnos

Um agente de voz precisa de duas coisas que um modelo em lote não entrega: texto enquanto a pessoa ainda está falando e uma decisão sobre quando ela terminou. O Ink 2 faz as duas coisas num único WebSocket. A detecção de turnos dele presta atenção ao que está sendo dito, além do silêncio, então quem para no meio da frase para pensar não é cortado só porque a linha ficou quieta.

Este script envia um arquivo WAV como se fosse áudio ao vivo. Salve como stream_file.py:

import os
import sys
import time
import wave

from cartesia import Cartesia

with wave.open(sys.argv[1], "rb") as wav:
    if wav.getnchannels() != 1 or wav.getsampwidth() != 2:
        raise SystemExit("Convert the file to 16-bit mono WAV first.")
    sample_rate = wav.getframerate()
    chunks = []
    while data := wav.readframes(sample_rate // 10):  # 100 ms per chunk
        chunks.append(data)

client = Cartesia(api_key=os.environ["CARTESIA_API_KEY"])
transcript = ""

with client.stt.auto_finalize.websocket(
    model="ink-2",
    encoding="pcm_s16le",
    sample_rate=sample_rate,
) as ws:
    for chunk in chunks:
        ws.send_raw(chunk)
        time.sleep(0.1)  # send at the speed of speech
    ws.send({"type": "close"})  # flush buffered audio; the server then closes

    turn_text = ""
    for event in ws:
        if event.type == "turn.update":
            turn_text = event.transcript or turn_text  # the whole turn so far
            print("...", turn_text)
        elif event.type == "turn.end":
            text = event.transcript or turn_text  # fall back if turn.end is empty
            print("turn:", text)
            transcript += text  # join verbatim, no strip()
            turn_text = ""
        elif event.type == "error":
            raise SystemExit(event.message)

print(transcript)

Se o seu arquivo não for WAV mono de 16 bits, converta antes:

ffmpeg -i input.mp3 -ac 1 -ar 16000 -c:a pcm_s16le input.wav

Três detalhes desse script são fáceis de errar:

• Amostras brutas, não um arquivo. Endpoints em tempo real recebem bytes de áudio sem cabeçalho, e por isso o script lê os frames do WAV em vez de enviar o arquivo. Você declara encoding e sample_rate ao conectar, e eles precisam bater com os bytes. O guia de entrada de áudio avisa que o servidor pode não devolver erro se não baterem, então uma taxa errada aparece como uma transcrição ruim, não como uma exceção.
• Ritmo de tempo real. O sleep(0.1) depois de cada bloco de 100 ms não é enfeite. Endpoints em tempo real esperam mais ou menos um segundo de áudio por segundo, e despejar um arquivo inteiro no socket pode sobrecarregar o buffer do servidor. Para um arquivo que você não precisa ao vivo, use a chamada em lote do passo 1.
• Transcrições cumulativas. Cada turn.update traz o turno inteiro até ali, não um fragmento novo. Concatenar atualizações repete palavras. Pegue o texto do turn.end, uma vez por turno, e junte exatamente como chega. O script também guarda o último turn.update e usa esse texto se algum turn.end chegar com a transcrição vazia, um caso que os nossos próprios clientes tratam.

Para ficar curto, este script lê os eventos depois de enviar o arquivo inteiro, então tudo aparece no final. O exemplo do microfone, mais abaixo, envia e lê ao mesmo tempo.

O Ink 2 transcreve inglês, francês, hindi, japonês e espanhol, e detecta qual está ouvindo. Para outros idiomas em streaming, veja nosso comparativo de modelos de speech to text.

3. Transcreva seu microfone ao vivo

Mesmo endpoint, entrada real. Esta versão usa o sounddevice para capturar áudio mono a 16 kHz e o cliente assíncrono para enviar e receber ao mesmo tempo:

python -m pip install sounddevice

No Linux, talvez você também precise do PortAudio (sudo apt install libportaudio2). Salve isto como microphone.py:

import asyncio
import os

import sounddevice as sd
from cartesia import AsyncCartesia

SAMPLE_RATE = 16000
BLOCK = SAMPLE_RATE // 10  # 100 ms of audio per message


async def main():
    loop = asyncio.get_running_loop()
    audio = asyncio.Queue(maxsize=50)  # about 5 s of audio

    def enqueue(block):
        if audio.full():  # the network fell behind: drop the oldest block
            audio.get_nowait()
        audio.put_nowait(block)

    def on_audio(indata, frames, time_info, status):
        loop.call_soon_threadsafe(enqueue, bytes(indata))

    client = AsyncCartesia(api_key=os.environ["CARTESIA_API_KEY"])
    async with client.stt.auto_finalize.websocket(
        model="ink-2",
        encoding="pcm_s16le",
        sample_rate=SAMPLE_RATE,
    ) as ws:

        async def send():
            with sd.RawInputStream(
                samplerate=SAMPLE_RATE,
                blocksize=BLOCK,
                channels=1,
                dtype="int16",
                callback=on_audio,
            ):
                while True:
                    await ws.send_raw(await audio.get())

        async def receive():
            turn_text = ""
            async for event in ws:
                if event.type == "turn.start":
                    print("(listening)")
                elif event.type == "turn.update":
                    turn_text = event.transcript or turn_text
                elif event.type == "turn.end":
                    print("You said:", event.transcript or turn_text)
                    turn_text = ""
                elif event.type == "error":
                    raise RuntimeError(event.message)

        await asyncio.gather(send(), receive())


asyncio.run(main())

Rode python microphone.py, fale algo, faça uma pausa e a transcrição daquele turno aparece. Aperte Ctrl+C para parar.

O callback de áudio roda em outra thread, então ele entrega cada bloco ao loop de eventos com call_soon_threadsafe em vez de mexer no socket diretamente. O microfone continua enviando enquanto você está em silêncio, e isso importa: o servidor trata a falta de áudio como atraso de rede, não como silêncio, então uma entrada mutada ou pausada deve enviar zeros em vez de nada. A fila tem limite de uns cinco segundos de áudio: se a conexão ficar para trás, os blocos mais antigos são descartados em vez de se acumularem na memória, porque a transcrição de algo dito há um minuto não serve para uma conversa ao vivo.

Num agente de voz, turn.start é onde você faz o bot parar de falar, e turn.end é onde você manda o texto para o seu LLM. Para começar a resposta mais cedo, trate também o turn.eager_end: comece a gerar quando ele chegar e descarte o rascunho se vier um turn.resume. A documentação de detecção de turnos lista os limiares, caso quem liga faça muitas pausas (ou quase nenhuma).

Problemas comuns

Quanto custa

Modelos locais custam processamento e mais nada. Na Cartesia, o Ink 2 usa 3 créditos por segundo de áudio, e uma hora de áudio sai por US$ 0,39 no plano Scale; os créditos do plano gratuito cobrem cerca de uma hora e 50 minutos de Ink 2. Os números atuais estão na página de preços.

Se você prefere não montar o pipeline, o Cartesia Managed Agents liga o Ink, um LLM e o text to speech do Sonic a um número de telefone para você. E se você só precisa transcrever algumas gravações, o faster-whisper no seu notebook provavelmente basta.

Próximos passos

1. Teste o Ink 2 com a sua voz no navegador, sem código, na página do Ink.
2. Mande as transcrições para um LLM e fale a resposta com text to speech em Python.
3. Junte as peças num framework: crie um agente de voz com Pipecat e Cartesia.

Documentação relacionada

• Guia rápido de speech to text em tempo real
• Compare os endpoints de STT: automático, manual e em lote
• Keyterm prompting para nomes e jargões que o modelo precisa escrever certo
• Detecção de atividade de voz, e por que a detecção de turnos a substitui