services(elevenlabs): implement word-by-word support through websockets

examples/foundational/07d-interruptible-cartesia.py

@@ -1,97 +0,0 @@
-#
-# Copyright (c) 2024, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import aiohttp
-import asyncio
-import os
-import sys
-
-from pipecat.frames.frames import LLMMessagesFrame
-from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.aggregators.llm_response import (
-    LLMAssistantResponseAggregator, LLMUserResponseAggregator)
-from pipecat.services.cartesia import CartesiaTTSService
-from pipecat.services.openai import OpenAILLMService
-from pipecat.transports.services.daily import DailyParams, DailyTransport
-from pipecat.vad.silero import SileroVADAnalyzer
-
-
-from runner import configure
-
-from loguru import logger
-
-from dotenv import load_dotenv
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        (room_url, token) = await configure(session)
-
-        transport = DailyTransport(
-            room_url,
-            token,
-            "Respond bot",
-            DailyParams(
-                audio_out_sample_rate=44100,
-                audio_out_enabled=True,
-                transcription_enabled=True,
-                vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer()
-            )
-        )
-
-        tts = CartesiaTTSService(
-            api_key=os.getenv("CARTESIA_API_KEY"),
-            voice_id="a0e99841-438c-4a64-b679-ae501e7d6091",  # Barbershop Man
-            sample_rate=44100,
-        )
-
-        llm = OpenAILLMService(
-            api_key=os.getenv("OPENAI_API_KEY"),
-            model="gpt-4o")
-
-        messages = [
-            {
-                "role": "system",
-                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
-            },
-        ]
-
-        tma_in = LLMUserResponseAggregator(messages)
-        tma_out = LLMAssistantResponseAggregator(messages)
-
-        pipeline = Pipeline([
-            transport.input(),   # Transport user input
-            tma_in,              # User responses
-            llm,                 # LLM
-            tts,                 # TTS
-            transport.output(),  # Transport bot output
-            tma_out              # Assistant spoken responses
-        ])
-
-        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True))
-
-        @transport.event_handler("on_first_participant_joined")
-        async def on_first_participant_joined(transport, participant):
-            transport.capture_participant_transcription(participant["id"])
-            # Kick off the conversation.
-            messages.append(
-                {"role": "system", "content": "Please introduce yourself to the user."})
-            await task.queue_frames([LLMMessagesFrame(messages)])
-
-        runner = PipelineRunner()
-
-        await runner.run(task)
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

pyproject.toml

@@ -39,7 +39,7 @@ azure = [ "azure-cognitiveservices-speech~=1.40.0" ]
 cartesia = [ "websockets~=12.0" ]
 daily = [ "daily-python~=0.10.1" ]
 deepgram = [ "deepgram-sdk~=3.5.0" ]
-elevenlabs = [ "elevenlabs~=1.7.0" ]
+elevenlabs = [ "websockets~=12.0" ]
 examples = [ "python-dotenv~=1.0.1", "flask~=3.0.3", "flask_cors~=4.0.1" ]
 fal = [ "fal-client~=0.4.1" ]
 gladia = [ "websockets~=12.0" ]

src/pipecat/services/ai_services.py

@@ -6,7 +6,6 @@
 
 import asyncio
 import io
-import time
 import wave
 
 from abc import abstractmethod
@@ -171,7 +170,7 @@ class TTSService(AIService):
             # if True, TTSService will push TTSStoppedFrames, otherwise subclass must do it
             push_stop_frames: bool = False,
             # if push_stop_frames is True, wait for this idle period before pushing TTSStoppedFrame
-            stop_frame_timeout_s: float = 0.8,
+            stop_frame_timeout_s: float = 1.0,
             **kwargs):
         super().__init__(**kwargs)
         self._aggregate_sentences: bool = aggregate_sentences
@@ -319,16 +318,16 @@ class AsyncTTSService(TTSService):
 class AsyncWordTTSService(AsyncTTSService):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
-        self._start_word_timestamp = None
+        self._initial_word_timestamp = -1
         self._words_queue = asyncio.Queue()
         self._words_task = self.get_event_loop().create_task(self._words_task_handler())
 
-    def init_word_timestamps(self):
+    def start_word_timestamps(self):
-        if not self._start_word_timestamp:
+        if self._initial_word_timestamp == -1:
-            self._start_word_timestamp = self.get_clock().get_time()
+            self._initial_word_timestamp = self.get_clock().get_time()
 
     def reset_word_timestamps(self):
-        self._start_word_timestamp = None
+        self._initial_word_timestamp = -1
         self._word_timestamps = []
 
     async def add_word_timestamps(self, word_times: List[Tuple[str, float]]):
@@ -366,7 +365,7 @@ class AsyncWordTTSService(AsyncTTSService):
                     await self.push_frame(LLMFullResponseEndFrame())
                 else:
                     frame = TextFrame(word)
-                    frame.pts = self._start_word_timestamp + timestamp
+                    frame.pts = self._initial_word_timestamp + timestamp
                     await self.push_frame(frame)
                 self._words_queue.task_done()
             except asyncio.CancelledError:

src/pipecat/services/cartesia.py

@@ -181,7 +181,7 @@ class CartesiaTTSService(AsyncWordTTSService):
                     )
                 elif msg["type"] == "chunk":
                     await self.stop_ttfb_metrics()
-                    self.init_word_timestamps()
+                    self.start_word_timestamps()
                     frame = AudioRawFrame(
                         audio=base64.b64decode(msg["data"]),
                         sample_rate=self._output_format["sample_rate"],

src/pipecat/services/elevenlabs.py

@@ -4,17 +4,30 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-from typing import AsyncGenerator, Literal
+import asyncio
+import base64
+import json
+
+from typing import Any, AsyncGenerator, List, Literal, Mapping, Tuple
 from pydantic import BaseModel
 
-from pipecat.frames.frames import AudioRawFrame, Frame, TTSStartedFrame, TTSStoppedFrame
+from pipecat.frames.frames import (
-from pipecat.services.ai_services import TTSService
+    AudioRawFrame,
+    CancelFrame,
+    EndFrame,
+    Frame,
+    StartFrame,
+    StartInterruptionFrame,
+    TTSStartedFrame,
+    TTSStoppedFrame)
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.ai_services import AsyncWordTTSService
 
 from loguru import logger
 
 # See .env.example for ElevenLabs configuration needed
 try:
-    from elevenlabs.client import AsyncElevenLabs
+    import websockets
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error(
@@ -35,7 +48,35 @@ def sample_rate_from_output_format(output_format: str) -> int:
     return 16000
 
 
-class ElevenLabsTTSService(TTSService):
+def calculate_word_times(
+        alignment_info: Mapping[str, Any], cumulative_time: float
+) -> List[Tuple[str, float]]:
+    end_times = [
+        s + d for s,
+        d in zip(
+            alignment_info["charStartTimesMs"],
+            alignment_info["charDurationsMs"])]
+    zipped_end_times = list(zip(alignment_info["chars"], end_times))
+
+    # Get the start time of every character that appears after a space and
+    # match this to the word
+    words = "".join(alignment_info["chars"]).split(" ")
+
+    # Calculate end time for each word. We do this by finding a space character
+    # and using the previous word time, also taking into account there might not
+    # be a space at the end.
+    times = []
+    for (i, (a, b)) in enumerate(zipped_end_times):
+        if a == " " or i == len(zipped_end_times) - 1:
+            t = cumulative_time + (zipped_end_times[i - 1][1] / 1000.0)
+            times.append(t)
+
+    word_times = list(zip(words, times))
+
+    return word_times
+
+
+class ElevenLabsTTSService(AsyncWordTTSService):
     class InputParams(BaseModel):
         output_format: Literal["pcm_16000", "pcm_22050", "pcm_24000", "pcm_44100"] = "pcm_16000"
 
@@ -45,49 +86,186 @@ class ElevenLabsTTSService(TTSService):
             api_key: str,
             voice_id: str,
             model: str = "eleven_turbo_v2_5",
+            url: str = "wss://api.elevenlabs.io",
             params: InputParams = InputParams(),
             **kwargs):
-        super().__init__(**kwargs)
+        # Aggregating sentences still gives cleaner-sounding results and fewer
+        # artifacts than streaming one word at a time. On average, waiting for a
+        # full sentence should only "cost" us 15ms or so with GPT-4o or a Llama
+        # 3 model, and it's worth it for the better audio quality.
+        #
+        # We also don't want to automatically push LLM response text frames,
+        # because the context aggregators will add them to the LLM context even
+        # if we're interrupted. ElevenLabs gives us word-by-word timestamps. We
+        # can use those to generate text frames ourselves aligned with the
+        # playout timing of the audio!
+        #
+        # Finally, ElevenLabs doesn't provide information on when the bot stops
+        # speaking for a while, so we want the parent class to send TTSStopFrame
+        # after a short period not receiving any audio.
+        super().__init__(
+            aggregate_sentences=True,
+            push_text_frames=False,
+            push_stop_frames=True,
+            stop_frame_timeout_s=2.0,
+            **kwargs
+        )
 
+        self._api_key = api_key
         self._voice_id = voice_id
         self._model = model
+        self._url = url
         self._params = params
-        self._client = AsyncElevenLabs(api_key=api_key)
         self._sample_rate = sample_rate_from_output_format(params.output_format)
 
+        # Websocket connection to ElevenLabs.
+        self._websocket = None
+        # Indicates if we have sent TTSStartedFrame. It will reset to False when
+        # there's an interruption or TTSStoppedFrame.
+        self._started = False
+        self._cumulative_time = 0
+
     def can_generate_metrics(self) -> bool:
         return True
 
     async def set_model(self, model: str):
         logger.debug(f"Switching TTS model to: [{model}]")
         self._model = model
+        await self._disconnect()
+        await self._connect()
 
     async def set_voice(self, voice: str):
         logger.debug(f"Switching TTS voice to: [{voice}]")
         self._voice_id = voice
+        await self._disconnect()
+        await self._connect()
+
+    async def start(self, frame: StartFrame):
+        await super().start(frame)
+        await self._connect()
+
+    async def stop(self, frame: EndFrame):
+        await super().stop(frame)
+        await self._disconnect()
+
+    async def cancel(self, frame: CancelFrame):
+        await super().cancel(frame)
+        await self._disconnect()
+
+    async def flush_audio(self):
+        if self._websocket:
+            msg = {"text": " ", "flush": True}
+            await self._websocket.send(json.dumps(msg))
+
+    async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
+        await super().push_frame(frame, direction)
+        if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)):
+            self._started = False
+            if isinstance(frame, TTSStoppedFrame):
+                await self.add_word_timestamps([("LLMFullResponseEndFrame", 0)])
+
+    async def _connect(self):
+        try:
+            voice_id = self._voice_id
+            model = self._model
+            output_format = self._params.output_format
+            url = f"{
+                self._url}/v1/text-to-speech/{voice_id}/stream-input?model_id={model}&output_format={output_format}"
+            self._websocket = await websockets.connect(url)
+            self._receive_task = self.get_event_loop().create_task(self._receive_task_handler())
+            self._keepalive_task = self.get_event_loop().create_task(self._keepalive_task_handler())
+
+            # According to ElevenLabs, we should always start with a single space.
+            msg = {
+                "text": " ",
+                "xi_api_key": self._api_key,
+            }
+            await self._websocket.send(json.dumps(msg))
+        except Exception as e:
+            logger.exception(f"{self} initialization error: {e}")
+            self._websocket = None
+
+    async def _disconnect(self):
+        try:
+            await self.stop_all_metrics()
+
+            if self._receive_task:
+                self._receive_task.cancel()
+                await self._receive_task
+                self._receive_task = None
+
+            if self._keepalive_task:
+                self._keepalive_task.cancel()
+                await self._keepalive_task
+                self._keepalive_task = None
+
+            if self._websocket:
+                await self._websocket.close()
+                self._websocket = None
+
+            self._started = False
+        except Exception as e:
+            logger.exception(f"{self} error closing websocket: {e}")
+
+    async def _receive_task_handler(self):
+        try:
+            async for message in self._websocket:
+                msg = json.loads(message)
+                if msg.get("audio"):
+                    await self.stop_ttfb_metrics()
+                    self.start_word_timestamps()
+
+                    audio = base64.b64decode(msg["audio"])
+                    frame = AudioRawFrame(audio, self._sample_rate, 1)
+                    await self.push_frame(frame)
+
+                if msg.get("alignment"):
+                    word_times = calculate_word_times(msg["alignment"], self._cumulative_time)
+                    await self.add_word_timestamps(word_times)
+                    self._cumulative_time = word_times[-1][1]
+
+        except asyncio.CancelledError:
+            pass
+        except Exception as e:
+            logger.exception(f"{self} exception: {e}")
+
+    async def _keepalive_task_handler(self):
+        while True:
+            try:
+                await asyncio.sleep(10)
+                await self._send_text("")
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                logger.exception(f"{self} exception: {e}")
+
+    async def _send_text(self, text: str):
+        if self._websocket:
+            msg = {"text": text + " "}
+            await self._websocket.send(json.dumps(msg))
 
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         logger.debug(f"Generating TTS: [{text}]")
 
-        await self.start_tts_usage_metrics(text)
+        try:
-        await self.start_ttfb_metrics()
+            if not self._websocket:
+                await self._connect()
 
-        results = await self._client.generate(
+            try:
-            text=text,
+                if not self._started:
-            voice=self._voice_id,
+                    await self.push_frame(TTSStartedFrame())
-            model=self._model,
+                    await self.start_ttfb_metrics()
-            output_format=self._params.output_format
+                    self._started = True
-        )
+                    self._cumulative_time = 0
 
-        tts_started = False
+                await self._send_text(text)
-        async for audio in results:
+                await self.start_tts_usage_metrics(text)
-            # This is so we send TTSStartedFrame when we have the first audio
+            except Exception as e:
-            # bytes.
+                logger.error(f"{self} error sending message: {e}")
-            if not tts_started:
+                await self.push_frame(TTSStoppedFrame())
-                await self.push_frame(TTSStartedFrame())
+                await self._disconnect()
-                tts_started = True
+                await self._connect()
-            await self.stop_ttfb_metrics()
+                return
-            frame = AudioRawFrame(audio, self._sample_rate, 1)
+            yield None
-            yield frame
+        except Exception as e:
-
+            logger.exception(f"{self} exception: {e}")
-        await self.push_frame(TTSStoppedFrame())

src/pipecat/services/lmnt.py

@@ -60,6 +60,8 @@ class LmntTTSService(AsyncTTSService):
         self._speech = None
         self._connection = None
         self._receive_task = None
+        # Indicates if we have sent TTSStartedFrame. It will reset to False when
+        # there's an interruption or TTSStoppedFrame.
         self._started = False
 
     def can_generate_metrics(self) -> bool: