Generuj audio w czasie rzeczywistym

Strumieniowanie przez WebSocket to metoda wysyłania i odbierania danych przez jedno, długotrwałe połączenie. Jest przydatna w aplikacjach czasu rzeczywistego, w których dane audio trzeba przesyłać strumieniowo, gdy tylko są dostępne.

Jeśli chcesz szybko sprawdzić opóźnienie (czas do pierwszego bajtu) połączenia WebSocket z API zamiany tekstu na mowę ElevenLabs, możesz zainstalować elevenlabs-latency przez npm i postępować zgodnie z instrukcjami tutaj.

Wymagania

• Konto ElevenLabs z kluczem API (zobacz, jak znaleźć klucz API).
• Python lub Node.js (albo inne środowisko JavaScript) zainstalowane na komputerze

Konfiguracja

Zainstaluj wymagane zależności:

pip install python-dotenv
pip install websockets

Następnie utwórz plik .env w katalogu projektu i dodaj klucz API:

ELEVENLABS_API_KEY=your_elevenlabs_api_key_here

Nawiąż połączenie WebSocket

Po wybraniu głosu z Voice Library i modelu zamiany tekstu na mowę, którego chcesz użyć, nawiąż połączenie WebSocket z API zamiany tekstu na mowę.

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:
       ...

Wyślij tekst wejściowy

Gdy połączenie WebSocket jest otwarte, najpierw ustaw parametry głosu. Następnie wyślij wiadomość tekstową do 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": ""}))

Zapisz audio do pliku

Odczytaj przychodzącą wiadomość z połączenia WebSocket i zapisz fragmenty audio w pliku lokalnym.

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

Uruchom skrypt

Uruchom skrypt, wykonując poniższe polecenie w terminalu. Plik audio mp3 zostanie zapisany w katalogu output.

python text-to-speech-websocket.py

Zaawansowana konfiguracja

WebSockety oferują zaawansowane ustawienia, które pozwalają dostroić generowanie audio w czasie rzeczywistym.

Buforowanie

Przy generowaniu audio w czasie rzeczywistym warto uwzględnić dwa ważne pojęcia: czas do pierwszego bajtu (TTFB) i buforowanie. Aby tworzyć audio wysokiej jakości i określać kontekst, model potrzebuje określonej ilości tekstu wejściowego. Im więcej tekstu wysyłasz przez połączenie WebSocket, tym lepsza jakość audio. Jeśli próg nie zostanie osiągnięty, model doda tekst do bufora i wygeneruje audio, gdy bufor się zapełni.

Pod względem opóźnienia TTFB to czas potrzebny na wysłanie pierwszego bajtu audio do klienta. Jest ważny, ponieważ wpływa na odczuwalne opóźnienie audio. Możesz więc chcieć kontrolować rozmiar bufora, aby zachować równowagę między jakością a opóźnieniem.

Aby tym zarządzać, użyj parametru chunk_length_schedule podczas inicjalizacji połączenia WebSocket lub wysyłania tekstu. To tablica liczb całkowitych określających liczbę znaków wysłanych do modelu przed wygenerowaniem audio. Na przykład, jeśli ustawisz chunk_length_schedule na [120, 160, 250, 290], model wygeneruje audio po wysłaniu odpowiednio 120, 160, 250 i 290 znaków.

Oto przykład działania z domyślnymi ustawieniami chunk_length_schedule:

Na powyższym diagramie audio jest generowane dopiero po wysłaniu drugiej wiadomości na serwer. Dzieje się tak, ponieważ pierwsza wiadomość nie osiąga progu 120 znaków, a druga zwiększa łączną liczbę znaków powyżej tego progu. Trzecia wiadomość przekracza próg 160 znaków, więc audio jest od razu generowane i zwracane do klienta.

Możesz podać własną wartość chunk_length_schedule podczas inicjalizacji połączenia WebSocket lub wysyłania tekstu.

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

Jeśli chcesz wymusić natychmiastowe zwrócenie audio, użyj flush: true, aby wyczyścić bufor i wymusić wygenerowanie całego zbuforowanego tekstu. Przydaje się to na przykład po dotarciu do końca dokumentu, gdy chcesz wygenerować audio dla ostatniej sekcji.

Możesz ustawić to dla każdej wiadomości osobno, dodając flush: true w wiadomości.

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

Dodatkowo zamknięcie WebSocketu automatycznie wymusi wygenerowanie całego zbuforowanego tekstu.

Ustawienia głosu

Podczas inicjalizacji połączeń WebSocket możesz określić ustawienia głosu dla kolejnych generacji. Pozwala to kontrolować szybkość, stabilność i inne cechy głosu w wygenerowanym audio.

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

Możesz je zastąpić dla każdej wiadomości osobno, określając inne voice_settings w wiadomości.

Słowniki wymowy

Możesz używać słowników wymowy, aby kontrolować wymowę konkretnych słów lub fraz. Przydaje się to, gdy chcesz zadbać o poprawną wymowę określonych słów albo zaakcentować wybrane słowa lub frazy.

W przeciwieństwie do voice_settings i generation_config, słowniki wymowy musisz określić w wiadomości „Initialize Connection”. Więcej informacji znajdziesz w dokumentacji API.

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

Dobre praktyki

• Zalecamy używanie domyślnego ustawienia chunk_length_schedule w generation_config.
• Przy tworzeniu aplikacji agenta konwersacyjnego w czasie rzeczywistym zalecamy użycie flush: true wraz z tekstem na końcu tury rozmowy, aby audio było generowane na czas.
• Jeśli domyślne ustawienie nie zapewnia optymalnego opóźnienia w twoim przypadku, możesz zmienić chunk_length_schedule. Pamiętaj jednak, że zmniejszenie opóźnienia w ten sposób może odbyć się kosztem jakości.

Wskazówki

• Połączenie WebSocket zostanie automatycznie zamknięte po 20 sekundach bezczynności. Aby je utrzymać, możesz wysłać pojedynczy znak spacji " ". Pamiętaj, że ten ciąg musi zawierać spację, ponieważ wysłanie całkowicie pustego ciągu "" zamknie WebSocket.
• Wyślij pusty ciąg, aby zamknąć połączenie WebSocket po wysłaniu ostatniej wiadomości tekstowej.
• Możesz użyć alignment, aby uzyskać sygnatury czasowe na poziomie słów dla każdego słowa w tekście. Przydaje się to do synchronizowania audio z tekstem w filmie lub w innych zastosowaniach wymagających precyzyjnego czasu. Więcej informacji znajdziesz w dokumentacji API.

Kolejne kroki