Speech-to-text en Python : fichier, flux et micro

Pour convertir la parole en texte en Python, on envoie de l’audio à un modèle de reconnaissance vocale et on récupère du texte. La vraie décision, c’est le type de modèle, car un fichier enregistré et une conversation en direct sont deux métiers différents. Un fichier peut attendre la fin du clip. Un appelant, non.

Ce guide présente trois façons de faire, chacune avec du code qui fonctionne tel quel : transcrire un fichier enregistré, streamer de l’audio avec détection automatique des tours de parole, et transcrire votre micro en direct. Les exemples de streaming utilisent Cartesia Ink 2, le modèle que nous construisons pour les agents vocaux. Quand un modèle local gratuit convient mieux, nous le disons et le montrons aussi.

Choisir une approche

La plupart des tutoriels s’arrêtent à SpeechRecognition avec recognize_google(). Ce paquet encapsule une douzaine de moteurs, mais son backend Google par défaut utilise une clé partagée que Google peut révoquer à tout moment, et il attend une phrase complète avant d’envoyer quoi que ce soit. Bien pour une démo. Pas pour répondre au téléphone.

Prérequis

• Python 3.9 ou plus récent
• Pour les exemples Cartesia, une clé d’API du playground Cartesia et le SDK :

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

L’extra websockets est nécessaire pour les exemples de streaming. Gardez la clé dans une variable d’environnement ou un gestionnaire de secrets, jamais dans du code versionné.

1. Transcrire un fichier enregistré

Si l’audio est déjà enregistré, envoyez le fichier entier en une seule requête. L’endpoint batch de Cartesia accepte des conteneurs comme WAV et MP3, et peut renvoyer un horodatage pour chaque mot. Enregistrez ceci sous 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}")

Lancez-le avec python transcribe_file.py meeting.mp3. La transcription batch utilise ink-whisper, qui couvre une centaine de langues ; passez le code de langue de votre audio. L’exemple STT batch de la documentation donne la version asynchrone.

L’alternative locale

Si l’audio ne doit pas quitter votre machine, ou si vous en avez beaucoup et un GPU, faites tourner Whisper vous-même. faster-whisper réimplémente Whisper d’OpenAI sur CTranslate2 et tourne sur le CPU d’un portable :

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

Sur l’extrait de JFK de 11 secondes du dépôt Whisper, le modèle base.en a restitué la citation mot pour mot en 2,5 secondes environ sur un CPU serveur 4 cœurs, après un téléchargement unique du modèle de 140 Mo. Les modèles plus gros sont plus précis et plus lents. Ce qu’on n’obtient pas, c’est le streaming : Whisper travaille sur des segments terminés, ce n’est donc pas l’outil pour une conversation.

2. Streamer l’audio et récupérer des tours de parole

Un agent vocal a besoin de deux choses qu’un modèle batch ne fournit pas : du texte pendant que la personne parle encore, et une décision sur le moment où elle a fini. Ink 2 fait les deux sur un seul WebSocket. Sa détection des tours de parole écoute ce qui est dit en plus du silence, donc un appelant qui s’arrête au milieu d’une phrase pour réfléchir n’est pas coupé juste parce que la ligne s’est tue.

Ce script envoie un fichier WAV comme s’il s’agissait d’audio en direct. Enregistrez-le sous 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)

Si votre fichier n’est pas un WAV mono 16 bits, convertissez-le d’abord :

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

Trois détails de ce script sont faciles à rater :

• Des échantillons bruts, pas un fichier. Les endpoints temps réel prennent des octets audio sans en-tête, d’où la lecture des frames du WAV au lieu d’envoyer le fichier. On déclare encoding et sample_rate à la connexion, et ils doivent correspondre aux octets. Le guide sur l’entrée audio précise que le serveur peut ne pas renvoyer d’erreur s’ils ne correspondent pas : une mauvaise fréquence se traduit par une mauvaise transcription, pas par une exception.
• Le rythme du temps réel. Le sleep(0.1) après chaque morceau de 100 ms n’est pas décoratif. Les endpoints temps réel attendent environ une seconde d’audio par seconde, et vider un fichier entier dans le socket peut saturer le tampon côté serveur. Pour un fichier qui n’a pas besoin du direct, utilisez l’appel batch de l’étape 1.
• Des transcriptions cumulatives. Chaque turn.update contient tout le tour jusque-là, pas un nouveau fragment. Concaténer les mises à jour répète des mots. Prenez le texte de turn.end, une fois par tour, et assemblez-le tel quel. Le script garde aussi le dernier turn.update et l’utilise si un turn.end arrive un jour avec une transcription vide, un cas que nos propres clients prévoient.

Pour rester court, ce script lit les événements après avoir envoyé tout le fichier, donc tout s’affiche à la fin. L’exemple du micro, plus bas, envoie et lit en même temps.

Ink 2 transcrit l’anglais, le français, l’hindi, le japonais et l’espagnol, et détecte la langue qu’il entend. Pour d’autres langues en streaming, voyez notre comparatif des modèles de speech-to-text.

3. Transcrire votre micro en direct

Même endpoint, vraie entrée. Cette version utilise sounddevice pour capturer de l’audio mono à 16 kHz, et le client asynchrone pour envoyer et recevoir en même temps :

python -m pip install sounddevice

Sous Linux, il vous faudra peut-être aussi PortAudio (sudo apt install libportaudio2). Enregistrez ceci sous 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())

Lancez python microphone.py, dites quelque chose, faites une pause, et la transcription de ce tour s’affiche. Ctrl+C pour arrêter.

Le callback audio tourne dans un autre thread : il confie chaque bloc à la boucle d’événements avec call_soon_threadsafe au lieu de toucher directement au socket. Le micro continue d’envoyer pendant que vous vous taisez, et c’est important : le serveur traite l’absence d’audio comme un retard réseau, pas comme du silence, donc une entrée coupée ou en pause doit envoyer des zéros plutôt que rien. La file est limitée à environ cinq secondes d’audio : si la connexion prend du retard, les blocs les plus anciens sont abandonnés au lieu de s’accumuler en mémoire, car la transcription d’une phrase dite il y a une minute ne sert à rien dans une conversation en direct.

Dans un agent vocal, turn.start est le moment où le bot arrête de parler, et turn.end celui où vous envoyez le texte à votre LLM. Pour démarrer la réponse plus tôt, gérez aussi turn.eager_end : commencez à générer quand il arrive, et jetez le brouillon si turn.resume suit. La documentation sur la détection des tours liste les seuils si vos appelants font beaucoup de pauses (ou presque aucune).

Problèmes fréquents

Combien ça coûte

Les modèles locaux coûtent du calcul, rien d’autre. Chez Cartesia, Ink 2 utilise 3 crédits par seconde d’audio, et une heure d’audio revient à 0,39 $ sur le plan Scale ; les crédits du plan gratuit couvrent environ une heure et 50 minutes d’Ink 2. Les chiffres à jour sont sur la page des tarifs.

Si vous préférez ne pas monter le pipeline vous-même, Cartesia Managed Agents branche Ink, un LLM et la synthèse vocale Sonic sur un numéro de téléphone pour vous. Et si vous n’avez que quelques enregistrements à transcrire, faster-whisper sur votre portable suffira probablement.

Étapes suivantes

1. Essayez Ink 2 avec votre propre voix dans le navigateur, sans code, sur la page Ink.
2. Envoyez les transcriptions à un LLM et lisez la réponse avec la synthèse vocale en Python.
3. Assemblez le tout dans un framework : créer un agent vocal Pipecat avec Cartesia.

Documentation associée

• Démarrage rapide du speech-to-text en temps réel
• Comparer les endpoints STT : automatique, manuel et batch
• Keyterm prompting pour les noms et le jargon que le modèle doit bien orthographier
• Détection d’activité vocale, et pourquoi la détection des tours de parole la remplace