Genera audio in tempo reale

Lo streaming WebSocket è un metodo per inviare e ricevere dati tramite un’unica connessione persistente. È utile per le applicazioni in tempo reale in cui devi trasmettere dati audio non appena diventano disponibili.

Se vuoi testare rapidamente la latenza (time to first byte) di una connessione WebSocket all’API Text to Speech di ElevenLabs, puoi installare elevenlabs-latency tramite npm e seguire le istruzioni qui.

Requisiti

• Un account ElevenLabs con una chiave API (ecco come trovare la tua chiave API).
• Python o Node.js (o un altro runtime JavaScript) installato sul tuo computer

Configurazione

Installa le dipendenze necessarie:

pip install python-dotenv
pip install websockets

Poi, crea un file .env nella directory del progetto e aggiungi la tua chiave API:

ELEVENLABS_API_KEY=your_elevenlabs_api_key_here

Avvia la connessione WebSocket

Dopo aver scelto una voce dalla Voice Library e il modello Text to Speech che vuoi usare, avvia una connessione WebSocket all’API Text to Speech.

import os
from dotenv import load_dotenv
import websockets
# Load the API key from the .env file
load_dotenv()
ELEVENLABS_API_KEY = os.getenv("ELEVENLABS_API_KEY")
voice_id = 'Xb7hH8MSUJpSbSDYk0k2'
# For use cases where latency is important, we recommend using the 'eleven_flash_v2_5' model.
model_id = 'eleven_flash_v2_5'
async def text_to_speech_ws_streaming(voice_id, model_id):
    uri = f"wss://api.elevenlabs.io/v1/text-to-speech/{voice_id}/stream-input?model_id={model_id}"
    async with websockets.connect(uri) as websocket:
       ...

Invia il testo di input

Una volta aperta la connessione WebSocket, configura prima le impostazioni della voce. Poi invia il messaggio di testo all’API.

async def text_to_speech_ws_streaming(voice_id, model_id):
    async with websockets.connect(uri) as websocket:
        await websocket.send(json.dumps({
            "text": " ",
            "voice_settings": {"stability": 0.5, "similarity_boost": 0.8, "use_speaker_boost": False},
            "generation_config": {
                "chunk_length_schedule": [120, 160, 250, 290]
            },
            "xi_api_key": ELEVENLABS_API_KEY,
        }))
        text = "The twilight sun cast its warm golden hues upon the vast rolling fields, saturating the landscape with an ethereal glow. Silently, the meandering brook continued its ceaseless journey, whispering secrets only the trees seemed privy to."
        await websocket.send(json.dumps({"text": text}))
        # Send empty string to indicate the end of the text sequence which will close the WebSocket connection
        await websocket.send(json.dumps({"text": ""}))

Salva l’audio in un file

Leggi il messaggio in arrivo dalla connessione WebSocket e scrivi i chunk audio in un file locale.

import asyncio
async def write_to_local(audio_stream):
    """Write the audio encoded in base64 string to a local mp3 file."""
    with open(f'./output/test.mp3', "wb") as f:
        async for chunk in audio_stream:
            if chunk:
                f.write(chunk)
async def listen(websocket):
    """Listen to the websocket for audio data and stream it."""
    while True:
        try:
            message = await websocket.recv()
            data = json.loads(message)
            if data.get("audio"):
                yield base64.b64decode(data["audio"])
            elif data.get('isFinal'):
                break
        except websockets.exceptions.ConnectionClosed:
            print("Connection closed")
            break
async def text_to_speech_ws_streaming(voice_id, model_id):
    async with websockets.connect(uri) as websocket:
          ...
          # Add listen task to submit the audio chunks to the write_to_local function
          listen_task = asyncio.create_task(write_to_local(listen(websocket)))
          await listen_task
asyncio.run(text_to_speech_ws_streaming(voice_id, model_id))

Esegui lo script

Puoi eseguire lo script con il seguente comando nel terminale. Un file audio mp3 verrà salvato nella directory output.

python text-to-speech-websocket.py

Configurazione avanzata

L’uso dei WebSocket include alcune impostazioni avanzate che puoi usare per ottimizzare la generazione audio in tempo reale.

Buffering

Quando generi audio in tempo reale, devi considerare due concetti importanti: Time To First Byte (TTFB) e buffering. Per produrre audio di alta qualità e dedurre il contesto, il modello richiede una certa soglia di testo di input. Più testo viene inviato tramite una connessione WebSocket, migliore sarà la qualità dell’audio. Se la soglia non viene raggiunta, il modello aggiunge il testo a un buffer e genera l’audio quando il buffer è pieno.

In termini di latenza, il TTFB è il tempo necessario per inviare il primo byte di audio al client. È importante perché influisce sulla latenza percepita dell’audio. Potresti quindi voler controllare la dimensione del buffer per bilanciare qualità e latenza.

Per gestirlo, puoi usare il parametro chunk_length_schedule quando inizializzi la connessione WebSocket o quando invii il testo. Questo parametro è un array di interi che rappresentano il numero di caratteri inviati al modello prima di generare l’audio. Ad esempio, se imposti chunk_length_schedule su [120, 160, 250, 290], il modello genererà audio dopo l’invio rispettivamente di 120, 160, 250 e 290 caratteri.

Ecco un esempio di come funziona con le impostazioni predefinite di chunk_length_schedule:

Nel diagramma qui sopra, l’audio viene generato solo dopo l’invio del secondo messaggio al server. Questo perché il primo messaggio è sotto la soglia di 120 caratteri, mentre il secondo porta il numero totale di caratteri oltre la soglia. Il terzo messaggio supera la soglia di 160 caratteri, quindi l’audio viene generato immediatamente e restituito al client.

Puoi specificare un valore personalizzato per chunk_length_schedule quando inizializzi la connessione WebSocket o quando invii il testo.

await websocket.send(json.dumps({
    "text": text,
    "generation_config": {
        # Generate audio after 50, 120, 160, and 290 characters have been sent
        "chunk_length_schedule": [50, 120, 160, 290]
    },
    "xi_api_key": ELEVENLABS_API_KEY,
}))

Se vuoi forzare la restituzione immediata dell’audio, puoi usare flush: true per svuotare il buffer e forzare la generazione del testo memorizzato nel buffer. Questo può essere utile, ad esempio, quando hai raggiunto la fine di un documento e vuoi generare l’audio per la sezione finale.

Puoi specificarlo per ogni messaggio impostando flush: true nel messaggio.

await websocket.send(json.dumps({"text": "Generate this audio immediately.", "flush": True}))

Inoltre, chiudendo il WebSocket verrà automaticamente forzata la generazione del testo nel buffer.

Impostazioni della voce

Quando inizializzi le connessioni WebSocket, puoi specificare le impostazioni della voce per le generazioni successive. Ciò ti consente di controllare la velocità, la stabilità e altre caratteristiche vocali dell’audio generato.

await websocket.send(json.dumps({
    "text": text,
    "voice_settings": {"stability": 0.5, "similarity_boost": 0.8, "use_speaker_boost": False},
}))

Puoi sostituirle per ogni messaggio specificando voice_settings diversi nel messaggio.

Dizionari di pronuncia

Puoi usare i dizionari di pronuncia per controllare la pronuncia di parole o frasi specifiche. Questo può essere utile per assicurarti che determinate parole siano pronunciate correttamente o per aggiungere enfasi a parole o frasi specifiche.

A differenza di voice_settings e generation_config, i dizionari di pronuncia devono essere specificati nel messaggio “Initialize Connection”. Per maggiori informazioni, consulta il Riferimento API.

wss://api.elevenlabs.io/v1/text-to-speech/{voice_id}/stream-input?model_id={model_id}&enable_ssml_parsing=true

Buone pratiche

• Ti consigliamo di usare l’impostazione predefinita per chunk_length_schedule in generation_config.
• Quando sviluppi un’applicazione di agente conversazionale in tempo reale, ti consigliamo di usare flush: true insieme al testo alla fine del turno di conversazione per garantire una generazione audio tempestiva.
• Se l’impostazione predefinita non offre una latenza ottimale per il tuo caso d’uso, puoi modificare chunk_length_schedule. Tieni però presente che ridurre la latenza con questa modifica può compromettere la qualità.

Suggerimenti

• La connessione WebSocket si chiude automaticamente dopo 20 secondi di inattività. Per tenerla aperta, puoi inviare un singolo carattere spazio " ". Tieni presente che questa stringa deve includere uno spazio, poiché l’invio di una stringa completamente vuota, "", chiuderà il WebSocket.
• Invia una stringa vuota per chiudere la connessione WebSocket dopo aver inviato l’ultimo messaggio di testo.
• Puoi usare alignment per ottenere i timestamp a livello di parola per ogni parola nel testo. Può essere utile per allineare l’audio al testo in un video o per altre applicazioni che richiedono una temporizzazione precisa. Per maggiori informazioni, consulta il Riferimento API.

Passaggi successivi