How to use text-to-speech with websocket streaming in Python or Node.js

How to convert text to speech via websocket and save to mp3

Websocket streaming is a method of sending and receiving data over a single, long-lived connection. This method is useful for real-time applications where you need to stream audio data as it becomes available.

If you want to quickly test out the latency (time to first byte) of a websocket connection to the ElevenLabs text-to-speech API, you can install elevenlabs-latency via npm and follow the instructions here.

Requirements

  • An ElevenLabs account with an API key (here’s how to find your API key).
  • Python or Node.js/Typescript installed on your machine

Setup

Install dotenv package to manage your environmental variables:

$pip install python-dotenv
>pip install websockets

Next, create a .env file in your project directory and fill it with your credentials like so:

.env
$ELEVENLABS_API_KEY=your_elevenlabs_api_key_here

Last, create a new file to write the code in. You can name it text-to-speech-websocket.py for Python or text-to-speech-websocket.ts for Typescript.

Initiate the websocket connection

Pick a voice from the voice library and a text-to-speech model; Then initiate a websocket connection to the text-to-speech API.

1import os
2from dotenv import load_dotenv
3import websockets
4
5# Load the API key from the .env file
6load_dotenv()
7ELEVENLABS_API_KEY = os.getenv("ELEVENLABS_API_KEY")
8
9voice_id = 'kmSVBPu7loj4ayNinwWM'
10model_id = 'eleven_flash_v2_5'
11
12async def text_to_speech_ws_streaming(voice_id, model_id):
13    uri = f"wss://api.elevenlabs.io/v1/text-to-speech/{voice_id}/stream-input?model_id={model_id}"
14
15    async with websockets.connect(uri) as websocket:
16       ...

For TypeScript, create a write stream ahead for saving the audio into mp3 which can be passed to the websocket listener.

text-to-speech-websocket.ts (Typescript)
1import * as fs from "node:fs";
2
3const outputDir = "./output";
4try {
5  fs.accessSync(outputDir, fs.constants.R_OK | fs.constants.W_OK);
6} catch (err) {
7  fs.mkdirSync(outputDir);
8}
9const writeStream = fs.createWriteStream(outputDir + "/test.mp3", {
10  flags: "a",
11});

Send the input text

Once the websocket connection is open, set up voice settings first. Next, send the text message to the API.

1async def text_to_speech_ws_streaming(voice_id, model_id):
2    async with websockets.connect(uri) as websocket:
3        await websocket.send(json.dumps({
4            "text": " ",
5            "voice_settings": {"stability": 0.5, "similarity_boost": 0.8, "use_speaker_boost": False},
6            "generation_config": {
7                "chunk_length_schedule": [120, 160, 250, 290]
8            },
9            "xi_api_key": ELEVENLABS_API_KEY,
10        }))
11
12        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."
13        await websocket.send(json.dumps({"text": text}))
14
15        // Send empty string to indicate the end of the text sequence which will close the websocket connection
16        await websocket.send(json.dumps({"text": ""}))

Save the audio to file

Read the incoming message from the websocket connection and write the audio chunks to a local file.

1import asyncio
2
3async def write_to_local(audio_stream):
4    """Write the audio encoded in base64 string to a local mp3 file."""
5
6    with open(f'./output/test.mp3', "wb") as f:
7        async for chunk in audio_stream:
8            if chunk:
9                f.write(chunk)
10
11async def listen(websocket):
12    """Listen to the websocket for audio data and stream it."""
13
14    while True:
15        try:
16            message = await websocket.recv()
17            data = json.loads(message)
18            if data.get("audio"):
19                yield base64.b64decode(data["audio"])
20            elif data.get('isFinal'):
21                break
22
23        except websockets.exceptions.ConnectionClosed:
24            print("Connection closed")
25            break
26
27async def text_to_speech_ws_streaming(voice_id, model_id):
28    async with websockets.connect(uri) as websocket:
29          ...
30          # Add listen task to submit the audio chunks to the write_to_local function
31          listen_task = asyncio.create_task(write_to_local(listen(websocket)))
32
33          await listen_task
34
35asyncio.run(text_to_speech_ws_streaming(voice_id, model_id))

Run the script

You can run the script by executing the following command in your terminal. An mp3 audio file will be saved in the output directory.

python Python python text-to-speech-websocket.py

Understanding buffering

A key concept to understand when using websockets is buffering. The API only runs model generations when a certain amount of text above a threshold has been sent. This is to optimize the quality of the generated audio by maximising the amount of context available to the model while balancing latency.

For example, if the threshold is set to 120 characters and you send ‘Hello, how are you?’, the audio won’t be generated immediately. This is because the sent message has only 19 characters which is below the threshold. However, if you keep sending text, the API will generate audio once the total text sent since the last generation has at least 120 characters.

In the case that you want force the immediate return of the audio, you can use flush=true to clear out the buffer and force generate any buffered text. This can be useful, for example, when you have reached the end of a document and want to generate audio for the final section.

In addition, closing the websocket will automatically force generate any buffered text.

Best practice

  • We suggest using the default setting for chunk_length_schedule in generation_config. Avoid using try_trigger_generation as it is deprecated.
  • When developing a real-time conversational AI application, we advise using flush=true along with the text at the end of conversation turn to ensure timely audio generation.
  • If the default setting doesn’t provide optimal latency for your use case, you can modify the chunk_length_schedule. However, be mindful that reducing latency through this adjustment may come at the expense of quality.

Tips

  • The API maintains a internal buffer so that it only runs model generations when a certain amount of text above a threshold has been sent. For short texts with a character length smaller than the value set in chunk_length_schedule, you can use flush=true to clear out the buffer and force generate any buffered text.
  • The websocket connection will automatically close after 20 seconds of inactivity. To keep the connection open, you can send a single space character " ". Please note that this string must include a space, as sending a fully empty string, "", will close the websocket.
  • Send an empty string to close the websocket connection after sending the last text message.
  • You can use alignment to get the word-level timestamps for each word in the text. This can be useful for aligning the audio with the text in a video or for other applications that require precise timing.