Speech to text in Python: file, streaming e microfono

Per convertire la voce in testo in Python, invii audio a un modello di riconoscimento vocale e ricevi del testo. La decisione interessante è il tipo di modello, perché un file registrato e una conversazione dal vivo sono lavori diversi. Un file può aspettare la fine della clip. Chi chiama, no.

Questa guida copre tre modi per farlo, ognuno con codice che funziona così com’è: trascrivere un file registrato, inviare audio in streaming con rilevamento automatico dei turni e trascrivere il microfono dal vivo. Gli esempi in streaming usano Cartesia Ink 2, il modello che costruiamo per gli agenti vocali. Quando un modello locale gratuito è la scelta migliore, lo diciamo e mostriamo anche quello.

Scegli un approccio

La maggior parte dei tutorial si ferma a SpeechRecognition con recognize_google(). Quel pacchetto fa da wrapper a una dozzina di motori, ma il suo backend Google predefinito usa una chiave condivisa che Google può revocare in qualsiasi momento, e aspetta una frase completa prima di inviare qualsiasi cosa. Va bene per una demo. Non per rispondere al telefono.

Prerequisiti

• Python 3.9 o successivo
• Per gli esempi Cartesia, una chiave API dal playground di Cartesia e l’SDK:

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

L’extra websockets serve per gli esempi in streaming. Tieni la chiave in una variabile d’ambiente o in un secret manager, mai nel codice che metti sotto versione.

1. Trascrivi un file registrato

Se l’audio è già registrato, invia l’intero file in una sola richiesta. L’endpoint batch di Cartesia accetta contenitori come WAV e MP3 e può restituire un timestamp per ogni parola. Salva questo come 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}")

Eseguilo con python transcribe_file.py meeting.mp3. La trascrizione batch usa ink-whisper, che copre circa 100 lingue; passa il codice della lingua del tuo audio. L’esempio STT batch nella documentazione ha la versione asincrona.

L’alternativa locale

Se l’audio non può lasciare la tua macchina, o ne hai molto e hai una GPU, esegui Whisper da solo. faster-whisper reimplementa Whisper di OpenAI su CTranslate2 e gira sulla CPU di un portatile:

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}")

Sul campione di JFK di 11 secondi del repository di Whisper, il modello base.en ha trascritto la citazione parola per parola in circa 2,5 secondi su una CPU server a 4 core, dopo un download una tantum del modello da 140 MB. I modelli più grandi sono più precisi e più lenti. Quello che non ottieni è lo streaming: Whisper lavora su segmenti finiti, quindi non è lo strumento per una conversazione.

2. Invia audio in streaming e ricevi i turni

Un agente vocale ha bisogno di due cose che un modello batch non può dargli: testo mentre la persona sta ancora parlando e una decisione su quando ha finito. Ink 2 fa entrambe le cose su un solo WebSocket. Il suo rilevamento dei turni ascolta cosa viene detto oltre al silenzio, quindi chi si ferma a metà frase per pensare non viene interrotto solo perché la linea è rimasta muta.

Questo script invia un file WAV come se fosse audio dal vivo. Salvalo come 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 il tuo file non è un WAV mono a 16 bit, convertilo prima:

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

Tre dettagli di quello script sono facili da sbagliare:

• Campioni grezzi, non un file. Gli endpoint in tempo reale ricevono byte audio senza intestazione, ed è per questo che lo script legge i frame dal WAV invece di inviare il file. Dichiari encoding e sample_rate quando ti connetti, e devono corrispondere ai byte. La guida all’input audio avverte che il server potrebbe non restituire un errore se non corrispondono, quindi una frequenza sbagliata si presenta come una trascrizione sbagliata, non come un’eccezione.
• Ritmo da tempo reale. Lo sleep(0.1) dopo ogni blocco da 100 ms non è decorativo. Gli endpoint in tempo reale si aspettano circa un secondo di audio al secondo, e scaricare un file intero nel socket può sovraccaricare il buffer del server. Per un file che non ti serve dal vivo, usa la chiamata batch del passo 1.
• Trascrizioni cumulative. Ogni turn.update contiene l’intero turno fino a quel momento, non un nuovo frammento. Concatenare gli aggiornamenti ripete le parole. Prendi il testo da turn.end, una volta per turno, e uniscilo esattamente come arriva. Lo script conserva anche l’ultimo turn.update e lo usa se un turn.end dovesse arrivare con la trascrizione vuota, un caso che i nostri stessi client gestiscono.

Per brevità, questo script legge gli eventi dopo aver inviato tutto il file, quindi tutto viene stampato alla fine. L’esempio del microfono, più sotto, invia e legge allo stesso tempo.

Ink 2 trascrive inglese, francese, hindi, giapponese e spagnolo, e riconosce quale sta ascoltando. Per altre lingue in streaming, vedi il nostro confronto tra modelli di speech to text.

3. Trascrivi il microfono dal vivo

Stesso endpoint, input reale. Questa versione usa sounddevice per catturare audio mono a 16 kHz e il client asincrono per inviare e ricevere contemporaneamente:

python -m pip install sounddevice

Su Linux potrebbe servire anche PortAudio (sudo apt install libportaudio2). Salva questo come 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())

Esegui python microphone.py, di’ qualcosa, fai una pausa e viene stampata la trascrizione di quel turno. Premi Ctrl+C per fermarlo.

La callback audio gira su un thread separato, quindi passa ogni blocco al loop degli eventi con call_soon_threadsafe invece di toccare direttamente il socket. Il microfono continua a inviare mentre stai zitto, e questo conta: il server interpreta l’assenza di audio come un ritardo di rete, non come silenzio, quindi un input disattivato o in pausa deve inviare zeri invece di niente. La coda è limitata a circa cinque secondi di audio: se la connessione resta indietro, i blocchi più vecchi vengono scartati invece di accumularsi in memoria, perché la trascrizione di qualcosa detto un minuto fa non serve a una conversazione dal vivo.

In un agente vocale, turn.start è dove fai smettere di parlare il bot, e turn.end è dove invii il testo al tuo LLM. Per iniziare prima la risposta, gestisci anche turn.eager_end: inizia a generare quando arriva e scarta la bozza se segue turn.resume. La documentazione sul rilevamento dei turni elenca le soglie se chi chiama fa molte pause (o quasi nessuna).

Problemi comuni

Quanto costa

I modelli locali costano calcolo e nient’altro. Su Cartesia, Ink 2 usa 3 crediti per secondo di audio, e un’ora di audio costa 0,39 $ sul piano Scale; i crediti del piano gratuito coprono circa un’ora e 50 minuti di Ink 2. I numeri aggiornati sono nella pagina dei prezzi.

Se preferisci non costruire la pipeline da solo, Cartesia Managed Agents collega per te Ink, un LLM e la sintesi vocale Sonic a un numero di telefono. E se devi trascrivere solo qualche registrazione, probabilmente ti basta faster-whisper sul portatile.

Prossimi passi

1. Prova Ink 2 con la tua voce nel browser, senza codice, nella pagina di Ink.
2. Passa le trascrizioni a un LLM e pronuncia la risposta con il text to speech in Python.
3. Metti insieme i pezzi in un framework: crea un agente vocale Pipecat con Cartesia.

Documentazione correlata

• Guida rapida allo speech to text in tempo reale
• Confronta gli endpoint STT: automatico, manuale e batch
• Keyterm prompting per nomi e gergo che il modello deve scrivere bene
• Rilevamento dell’attività vocale, e perché il rilevamento dei turni lo sostituisce