Refactor for aiohttp, correct use of settings

examples/foundational/07d-interruptible-elevenlabs-http.py

@@ -48,6 +48,7 @@ async def main():
         tts = ElevenLabsHttpTTSService(
             api_key=os.getenv("ELEVENLABS_API_KEY", ""),
             voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""),
+            # params=ElevenLabsHttpTTSService.InputParams(language="en"),
         )
 
         llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")

pyproject.toml

@@ -51,7 +51,7 @@ cerebras = [ "openai~=1.59.6" ]
 deepseek = [ "openai~=1.59.6" ]
 daily = [ "daily-python~=0.14.2" ]
 deepgram = [ "deepgram-sdk~=3.8.0" ]
-elevenlabs = [ "elevenlabs~=1.50.3","websockets~=13.1" ]
+elevenlabs = [ "websockets~=13.1" ]
 fal = [ "fal-client~=0.5.6" ]
 fish = [ "ormsgpack~=1.7.0", "websockets~=13.1" ]
 gladia = [ "websockets~=13.1" ]

src/pipecat/services/elevenlabs.py

@@ -7,8 +7,9 @@
 import asyncio
 import base64
 import json
-from typing import Any, AsyncGenerator, Dict, List, Literal, Mapping, Optional, Tuple
+from typing import Any, AsyncGenerator, Dict, List, Literal, Mapping, Optional, Tuple, Union
 
+import aiohttp
 from loguru import logger
 from pydantic import BaseModel, model_validator
 
@@ -424,9 +425,20 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService):
 
 
 class ElevenLabsHttpTTSService(TTSService):
+    """ElevenLabs Text-to-Speech service using HTTP streaming.
+
+    Args:
+        api_key: ElevenLabs API key
+        voice_id: ID of the voice to use
+        model: Model ID (default: "eleven_flash_v2_5" for low latency)
+        base_url: API base URL
+        output_format: Audio output format (PCM)
+        params: Additional parameters for voice configuration
+    """
+
     class InputParams(BaseModel):
         language: Optional[Language] = Language.EN
-        optimize_streaming_latency: Optional[str] = None
+        optimize_streaming_latency: Optional[int] = None
         stability: Optional[float] = None
         similarity_boost: Optional[float] = None
         style: Optional[float] = None
@@ -438,109 +450,133 @@ class ElevenLabsHttpTTSService(TTSService):
         api_key: str,
         voice_id: str,
         model: str = "eleven_flash_v2_5",
+        base_url: str = "https://api.elevenlabs.io",
         output_format: ElevenLabsOutputFormat = "pcm_24000",
         params: InputParams = InputParams(),
         **kwargs,
     ):
-        sample_rate = self._sample_rate_from_output_format(output_format)
+        sample_rate = sample_rate_from_output_format(output_format)
-        super().__init__(
+        super().__init__(sample_rate=sample_rate, **kwargs)
-            aggregate_sentences=True,
-            push_text_frames=False,
-            push_stop_frames=True,
-            stop_frame_timeout_s=2.0,
-            sample_rate=sample_rate,
-            **kwargs,
-        )
 
-        self._client = ElevenLabs(api_key=api_key)
+        self._api_key = api_key
-        self._voice_id = voice_id
+        self._base_url = base_url
-        self._model = model
         self._output_format = output_format
+        self._params = params
+        self._session: Optional[aiohttp.ClientSession] = None
 
-        # Create voice settings if provided
+        self._settings = {
-        self._voice_settings = None
+            "sample_rate": sample_rate_from_output_format(output_format),
-        if params.stability is not None and params.similarity_boost is not None:
+            "language": self.language_to_service_language(params.language)
-            self._voice_settings = VoiceSettings(
+            if params.language
-                stability=params.stability,
+            else "en",
-                similarity_boost=params.similarity_boost,
+            "output_format": output_format,
-                style=params.style or 0.0,
+            "optimize_streaming_latency": params.optimize_streaming_latency,
-                use_speaker_boost=params.use_speaker_boost or False,
+            "stability": params.stability,
-            )
+            "similarity_boost": params.similarity_boost,
-
+            "style": params.style,
-        logger.debug(f"Initialized with sample rate: {sample_rate}")
+            "use_speaker_boost": params.use_speaker_boost,
-
+        }
-    @staticmethod
+        self.set_model_name(model)
-    def _sample_rate_from_output_format(output_format: str) -> int:
+        self.set_voice(voice_id)
-        return {
+        self._voice_settings = self._set_voice_settings()
-            "pcm_16000": 16000,
-            "pcm_22050": 22050,
-            "pcm_24000": 24000,
-            "pcm_44100": 44100,
-        }[output_format]
 
     def can_generate_metrics(self) -> bool:
         return True
 
+    def _set_voice_settings(self) -> Optional[Dict[str, Union[float, bool]]]:
+        voice_settings: Dict[str, Union[float, bool]] = {}
+        if (
+            self._settings["stability"] is not None
+            and self._settings["similarity_boost"] is not None
+        ):
+            voice_settings["stability"] = float(self._settings["stability"])
+            voice_settings["similarity_boost"] = float(self._settings["similarity_boost"])
+            if self._settings["style"] is not None:
+                voice_settings["style"] = float(self._settings["style"])
+            if self._settings["use_speaker_boost"] is not None:
+                voice_settings["use_speaker_boost"] = bool(self._settings["use_speaker_boost"])
+        else:
+            if self._settings["style"] is not None:
+                logger.warning(
+                    "'style' is set but will not be applied because 'stability' and 'similarity_boost' are not both set."
+                )
+            if self._settings["use_speaker_boost"] is not None:
+                logger.warning(
+                    "'use_speaker_boost' is set but will not be applied because 'stability' and 'similarity_boost' are not both set."
+                )
+
+        return voice_settings or None
+
     async def start(self, frame: StartFrame):
         await super().start(frame)
+        self._session = aiohttp.ClientSession()
 
     async def stop(self, frame: EndFrame):
         await super().stop(frame)
+        if self._session:
+            await self._session.close()
+            self._session = None
 
     async def cancel(self, frame: CancelFrame):
         await super().cancel(frame)
+        if self._session:
+            await self._session.close()
+            self._session = None
 
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
-        def read_audio_stream(**kwargs):
-            audio_chunks = []
-            stream = self._client.text_to_speech.convert_as_stream(**kwargs)
-            for chunk in stream:
-                if chunk:
-                    audio_chunks.append(chunk)
-            return b"".join(audio_chunks)
-
         logger.debug(f"Generating TTS: [{text}]")
 
+        if not self._session:
+            self._session = aiohttp.ClientSession()
+
+        url = f"{self._base_url}/v1/text-to-speech/{self._voice_id}/stream"
+
+        payload = {
+            "text": text,
+            "model_id": self._model_name,
+        }
+
+        if self._voice_settings:
+            payload["voice_settings"] = json.dumps(self._voice_settings)
+
+        if self._settings["language"]:
+            payload["language_code"] = self._settings["language"]
+
+        headers = {
+            "xi-api-key": self._api_key,
+            "Content-Type": "application/json",
+        }
+
+        # Build query parameters
+        params = {
+            "output_format": self._output_format,
+        }
+        if self._settings["optimize_streaming_latency"] is not None:
+            params["optimize_streaming_latency"] = self._settings["optimize_streaming_latency"]
+
+        logger.debug(f"ElevenLabs request - payload: {payload}, params: {params}")
+
         try:
-            # Start TTFB metrics before any processing
             await self.start_ttfb_metrics()
 
-            # Prepare parameters
+            async with self._session.post(
-            params = {
+                url, json=payload, headers=headers, params=params
-                "text": text,
+            ) as response:
-                "voice_id": self._voice_id,
+                if response.status != 200:
-                "model_id": self._model,
+                    error_text = await response.text()
-                "output_format": self._output_format,
+                    logger.error(f"{self} error: {error_text}")
-                "voice_settings": self._voice_settings,
+                    yield ErrorFrame(error=f"ElevenLabs API error: {error_text}")
-                "optimize_streaming_latency": 4,
+                    return
-            }
 
-            # Get audio data in a separate thread
+                await self.start_tts_usage_metrics(text)
-            audio_data = await asyncio.to_thread(read_audio_stream, **params)
+                yield TTSStartedFrame()
 
-            if not audio_data:
+                async for chunk in response.content:
-                logger.error(f"{self} No audio data returned")
+                    if chunk:
-                yield None
+                        await self.stop_ttfb_metrics()
-                return
+                        yield TTSAudioRawFrame(chunk, self._settings["sample_rate"], 1)
 
-            # Start usage metrics before sending any frames
+                yield TTSStoppedFrame()
-            await self.start_tts_usage_metrics(text)
-
-            yield TTSStartedFrame()
-
-            # Stream the audio data in chunks
-            chunk_size = 4096
-            for i in range(0, len(audio_data), chunk_size):
-                chunk = audio_data[i : i + chunk_size]
-                if len(chunk) > 0:
-                    # Stop TTFB metrics on first chunk
-                    await self.stop_ttfb_metrics()
-
-                    yield TTSAudioRawFrame(
-                        chunk, self._sample_rate_from_output_format(self._output_format), 1
-                    )
-
-            yield TTSStoppedFrame()
 
         except Exception as e:
             logger.error(f"Error in run_tts: {e}")