Speech-to-Speech (WebSocket)

Le Speech-to-Speech (S2S) de Gradium envoie l’audio d’entrée en flux au serveur sur une seule connexion WebSocket bidirectionnelle, et renvoie en flux l’audio de sortie traduit ainsi que sa transcription. L’audio est transcrit, traduit dans la langue target_language, puis resynthétisé avec la voix que vous avez choisie. Ce guide couvre le streaming avec le SDK Python.

Vous hésitez sur l'API à utiliser ?

La vue d’ensemble compare s2s_realtime, s2s_stream et le helper bufferisé s2s.

Démarrage rapide

Le chemin le plus court : ouvrez un gestionnaire de contexte s2s_realtime, envoyez des fragments audio et itérez pour recevoir l’audio et le texte en sortie.

import asyncio
import gradium

async def main():
    client = gradium.client.GradiumClient(api_key="your-api-key")

    async with client.s2s_realtime(
        model_name="s2s-translate",
        stt_model_name="stt-translate",
        tts_model_name="default",
        input_format="pcm",
        output_format="pcm",
        voice_id="YTpq7expH9539ERJ",  # required; must be a voice in the target language
        json_config={"target_language": "en"},
        wait_for_ready_on_start=True,
    ) as s2s:
        # Producer: push 80 ms PCM frames as they're captured.
        async def producer():
            async for chunk in microphone_chunks():  # 1920-sample frames
                await s2s.send_audio(chunk)
            await s2s.send_eos()

        # Consumer: collect output audio, print the transcript as it arrives.
        out = []
        async def consumer():
            async for msg in s2s:
                if msg["type"] == "audio":
                    out.append(msg["audio"])  # already-decoded bytes
                elif msg["type"] == "text":
                    print(msg["text"], end=" ", flush=True)
                elif msg["type"] == "end_of_stream":
                    return

        await asyncio.gather(producer(), consumer())

asyncio.run(main())

Entrée attendue : PCM mono 16 bits signé à 24 kHz, en fragments de 1920 échantillons (80 ms). La sortie pcm est en PCM mono 16 bits signé à 48 kHz. D’autres formats sont pris en charge, voir Paramètres de configuration.

Choisir l’API du SDK

Le SDK Python propose trois façons de piloter le S2S via l’endpoint WebSocket. Toutes communiquent avec wss://api.gradium.ai/api/speech/s2s : choisissez selon la forme de votre source audio.

Cas d’usage	Méthode
Vous produisez l’audio en direct (micro, téléphonie, boucle d’agent) et voulez la sortie au fil de l’eau	`s2s_realtime`
Vous pouvez itérer sur l’audio de bout en bout (fichier, buffer en mémoire, générateur fini) et voulez recevoir la sortie en flux	`s2s_stream`
Vous avez un fichier complet et voulez simplement l’audio de sortie final et la transcription	`s2s` (bufferisé)

La configuration se passe en arguments nommés pour s2s_realtime et sous forme de dict pour s2s_stream / s2s. Le message setup envoyé sur le réseau est le même dans tous les cas.

Envoi en temps réel : `s2s_realtime`

Utilisez s2s_realtime lorsque l’audio est produit en direct (micro, téléphonie, agent conversationnel) et que vous avez besoin de l’audio et du texte de sortie dès qu’ils arrivent. C’est un gestionnaire de contexte asynchrone : envoyez l’audio avec send_audio() et itérez sur le même objet pour recevoir tous les types de messages.

import asyncio
import gradium

async def main():
    client = gradium.client.GradiumClient(api_key="your-api-key")

    async with client.s2s_realtime(
        model_name="s2s-translate",
        stt_model_name="stt-translate",
        tts_model_name="default",
        input_format="pcm",
        output_format="pcm",
        voice_id="YTpq7expH9539ERJ",  # required; must be a voice in the target language
        json_config={"target_language": "en"},
        wait_for_ready_on_start=True,
    ) as s2s:
        print("Ready:", s2s.ready)

        async def producer():
            async for chunk in microphone_chunks():  # 80 ms PCM frames
                await s2s.send_audio(chunk)
            await s2s.send_eos()

        out = []
        async def consumer():
            async for msg in s2s:
                if msg["type"] == "audio":
                    out.append(msg["audio"])
                elif msg["type"] == "text":
                    print(msg["text"], end=" ", flush=True)
                elif msg["type"] == "end_of_stream":
                    return

        await asyncio.gather(producer(), consumer())

if __name__ == "__main__":
    asyncio.run(main())

send_audio() accepte des bytes ou un tableau numpy 1-D (int16 ou float32) lorsque input_format est une variante de pcm. Dans l’itérateur, les messages audio contiennent déjà des bytes décodés dans msg["audio"].

Mode pull : `s2s_stream`

Utilisez s2s_stream lorsque vous disposez déjà de l’audio (un fichier ou un buffer complet en mémoire) mais souhaitez tout de même recevoir la sortie en flux au fur et à mesure de sa production. La fonction consomme votre itérable audio et renvoie un S2SStream.

import asyncio
import gradium

async def main():
    client = gradium.client.GradiumClient(api_key="your-api-key")

    async def audio_generator(audio_data, chunk_size=1920):
        for i in range(0, len(audio_data), chunk_size):
            yield audio_data[i : i + chunk_size]

    setup = {
        "model_name": "s2s-translate",
        "stt_model_name": "stt-translate",
        "tts_model_name": "default",
        "input_format": "pcm",
        "output_format": "wav",
        "voice_id": "YTpq7expH9539ERJ",
        "json_config": {"target_language": "en"},
    }

    stream = await client.s2s_stream(setup, audio_generator(audio_data))

    with open("output.wav", "wb") as f:
        async for chunk in stream.iter_audio():
            f.write(chunk)

    # Text segments are collected as audio is consumed.
    for twt in stream._text_with_timestamps:
        print(f"[{twt.start_s:.2f}s] {twt.text}")

if __name__ == "__main__":
    asyncio.run(main())

iter_audio() produit les octets audio de sortie et collecte les segments de texte au passage. Pour traiter vous-même chaque message du serveur (audio comme text, sous forme de dicts, avec les champs audio décodés), utilisez plutôt iter_events().

Bufferisé : `s2s`

Utilisez s2s lorsque vous n’avez pas besoin des résultats au fil de l’eau : passez-lui un buffer audio complet et récupérez un S2SResult contenant l’audio de sortie complet et la transcription.

import asyncio
import gradium

async def main():
    client = gradium.client.GradiumClient(api_key="your-api-key")

    setup = {
        "model_name": "s2s-translate",
        "stt_model_name": "stt-translate",
        "tts_model_name": "default",
        "input_format": "wav",
        "output_format": "wav",
        "voice_id": "YTpq7expH9539ERJ",
        "json_config": {"target_language": "en"},
    }

    with open("speech.wav", "rb") as f:
        result = await client.s2s(setup, f.read())

    print(result.text)
    with open("translated.wav", "wb") as f:
        f.write(result.raw_data)

if __name__ == "__main__":
    asyncio.run(main())

audio peut être des bytes bruts, un tableau numpy 1-D (int16/float32, entrée pcm uniquement, sample_rate=24000 obligatoire) ou un générateur asynchrone de fragments d’octets. Le S2SResult renvoyé expose raw_data, sample_rate, output_format, request_id, text et text_with_timestamps ; pcm16() / pcm() convertissent la sortie PCM en tableaux numpy.

Paramètres de configuration

model_name : le modèle S2S à utiliser. Utilisez "s2s-translate" pour la traduction en direct, le seul modèle actuellement pris en charge.
stt_model_name : modèle de transcription utilisé pour transcrire l’entrée. Indiquez "stt-translate" pour la traduction en direct.
tts_model_name : modèle de synthèse vocale utilisé pour synthétiser la sortie. Indiquez "default" pour la traduction en direct.
voice_id (obligatoire) : UID de la voix utilisée pour la sortie synthétisée. Il doit s’agir d’une voix dans la même langue que target_language. Voir Voix.
input_format : format audio de l’entrée. Valeurs prises en charge :
- "pcm" : PCM brut, 24 kHz, 16 bits signé little-endian, mono. Envoyez des fragments de 1920 échantillons (80 ms) pour de meilleurs résultats.
- "pcm_8000", "pcm_16000", "pcm_22050", "pcm_24000", "pcm_44100", "pcm_48000" : PCM brut à la fréquence d’échantillonnage indiquée (16 bits signé little-endian, mono).
- "wav" : un fichier WAV valide contenant des données PCM.
- "opus" : flux Opus encapsulé dans Ogg.
- "ulaw_8000" (alias "mulaw_8000") : PCM encodé en mu-law à 8 kHz.
- "alaw_8000" : PCM encodé en A-law à 8 kHz.
output_format : format audio de la sortie. Mêmes valeurs possibles que input_format. Pour "pcm", la sortie est en 48 kHz, 16 bits signé little-endian, mono.
json_config : définissez target_language (par ex. "en") sur la langue dans laquelle traduire la parole. Le voice_id doit correspondre à une voix dans cette langue. Voir Paramètres Speech-to-Speech.

Une valeur de format non reconnue provoque une erreur serveur.

Types de messages

Lorsque vous itérez sur une session s2s_realtime, vous recevez tous les messages envoyés par le serveur :

ready : connexion acceptée ; contient le sample_rate et le frame_size de la sortie. Accessible via s2s.ready lorsque vous passez wait_for_ready_on_start=True.
text : un segment de transcription traduit (dans target_language) avec les horodatages start_s / stop_s.
audio : un fragment audio de sortie. Dans l’itérateur du SDK, le champ audio contient déjà des bytes décodés.
end_of_stream : message final après send_eos().

Pour le schéma au niveau du protocole (types de champs complets et JSON tel qu’envoyé sur le réseau), consultez la référence WebSocket S2S.

WebSocket direct

Si vous ne souhaitez pas utiliser le SDK gradium, par exemple parce que vous appelez l’API depuis un runtime autre que Python, ou que vous voulez un contrôle total sur les échanges, vous pouvez dialoguer directement avec le protocole WebSocket. La forme des messages est identique à celle qu’envoie le SDK. Pour explorer de manière interactive depuis un terminal, wscat est le moyen le plus rapide de vérifier la connectivité et d’envoyer les messages un à un à la main :

wscat -c "wss://api.gradium.ai/api/speech/s2s" \
  -H "x-api-key: your_api_key"
# After connection, type:
# {"type":"setup","model_name":"s2s-translate","stt_model_name":"stt-translate","tts_model_name":"default","input_format":"pcm","output_format":"pcm","voice_id":"YTpq7expH9539ERJ","json_config":{"target_language":"en"}}

Pour un client sans le SDK, utilisez n’importe quelle bibliothèque WebSocket (ici websockets en Python) :

import asyncio
import base64
import json

import websockets

CHUNK_BYTES = 1920 * 2  # 80 ms at 24 kHz, 16-bit mono.


async def translate(api_key: str, pcm_audio: bytes):
    setup = {
        "type": "setup",
        "model_name": "s2s-translate",
        "stt_model_name": "stt-translate",
        "tts_model_name": "default",
        "input_format": "pcm",
        "output_format": "pcm",
        "voice_id": "YTpq7expH9539ERJ",
        "json_config": {"target_language": "en"},
    }

    async with websockets.connect(
        "wss://api.gradium.ai/api/speech/s2s",
        additional_headers={"x-api-key": api_key},
    ) as ws:
        await ws.send(json.dumps(setup))
        ready = json.loads(await ws.recv())
        assert ready["type"] == "ready"

        async def producer():
            for off in range(0, len(pcm_audio), CHUNK_BYTES):
                chunk = pcm_audio[off : off + CHUNK_BYTES]
                await ws.send(json.dumps({
                    "type": "audio",
                    "audio": base64.b64encode(chunk).decode(),
                }))
            await ws.send(json.dumps({"type": "end_of_stream"}))

        out = []
        async def consumer():
            while True:
                msg = json.loads(await ws.recv())
                if msg["type"] == "audio":
                    out.append(base64.b64decode(msg["audio"]))
                elif msg["type"] == "text":
                    print(msg["text"])
                elif msg["type"] == "end_of_stream":
                    return
                elif msg["type"] == "error":
                    raise RuntimeError(msg["message"])

        await asyncio.gather(producer(), consumer())

asyncio.run(translate("your_api_key", open("input.pcm", "rb").read()))

Pour le schéma complet des messages (chaque champ, chaque code d’erreur), consultez la référence WebSocket S2S.

Étapes suivantes

Référence WebSocket S2S

Schéma complet au niveau du protocole : chaque type de message, chaque champ, chaque code d’erreur.

Paramètres S2S

voice_id, target_language, choix des modèles internes, formats.

Voix

Choisissez la voix utilisée pour la sortie synthétisée.

Erreurs

Contrats d’erreur pour REST, WebSocket et les réponses en streaming.

Premiers pas

Synthèse vocale (TTS)

Transcription (STT)

Speech-to-Speech

Commun

Recettes temps réel

Migrations

Voix

Auto-hébergé

Ressources

Speech-to-Speech (WebSocket)

Vous hésitez sur l'API à utiliser ?

Démarrage rapide

Choisir l’API du SDK

Envoi en temps réel : `s2s_realtime`

Mode pull : `s2s_stream`

Bufferisé : `s2s`

Paramètres de configuration

Types de messages

WebSocket direct

Étapes suivantes

Référence WebSocket S2S

Paramètres S2S

Voix

Erreurs

Vous hésitez sur l'API à utiliser ?

​Démarrage rapide

​Choisir l’API du SDK

​Envoi en temps réel : s2s_realtime

​Mode pull : s2s_stream

​Bufferisé : s2s

​Paramètres de configuration

​Types de messages

​WebSocket direct

​Étapes suivantes

Référence WebSocket S2S

Paramètres S2S

Voix

Erreurs

Démarrage rapide

Choisir l’API du SDK

Envoi en temps réel : `s2s_realtime`

Mode pull : `s2s_stream`

Bufferisé : `s2s`

Paramètres de configuration

Types de messages

WebSocket direct

Étapes suivantes