From 384f80983ffab845f25d9d2be9ede5f505354bab Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 15 Apr 2025 21:29:19 -0400 Subject: [PATCH 1/4] Added word/timestamp pairs to ElevenLabsHttpTTSService --- CHANGELOG.md | 4 +- src/pipecat/services/elevenlabs/tts.py | 165 ++++++++++++++++++++++--- 2 files changed, 149 insertions(+), 20 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 80b1369d5..d3333c27d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 you to control aggregator settings. You can now pass these arguments when creating aggregator pairs with `create_context_aggregator()`. +- Added word/timestamp pairs to `ElevenLabsHttpTTSService`. + - It is now possible to disable `SoundfileMixer` when created. You can then use `MixerEnableFrame` to dynamically enable it when necessary. @@ -55,7 +57,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed an issue in `SmallWebRTCTransport` where an error was thrown if the client did not create a video transceiver. -- Fixed an issue where LLM input parameters were not working and applied correctly in `GoogleVertexLLMService`, causing +- Fixed an issue where LLM input parameters were not working and applied correctly in `GoogleVertexLLMService`, causing unexpected behavior during inference. ## [0.0.63] - 2025-04-11 diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index cc9a72889..d77e4199c 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -25,7 +25,7 @@ from pipecat.frames.frames import ( TTSStoppedFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.tts_service import InterruptibleWordTTSService, TTSService +from pipecat.services.tts_service import InterruptibleWordTTSService, WordTTSService from pipecat.transcriptions.language import Language # See .env.example for ElevenLabs configuration needed @@ -441,8 +441,8 @@ class ElevenLabsTTSService(InterruptibleWordTTSService): logger.error(f"{self} exception: {e}") -class ElevenLabsHttpTTSService(TTSService): - """ElevenLabs Text-to-Speech service using HTTP streaming. +class ElevenLabsHttpTTSService(WordTTSService): + """ElevenLabs Text-to-Speech service using HTTP streaming with word timestamps. Args: api_key: ElevenLabs API key @@ -475,7 +475,13 @@ class ElevenLabsHttpTTSService(TTSService): params: InputParams = InputParams(), **kwargs, ): - super().__init__(sample_rate=sample_rate, **kwargs) + super().__init__( + aggregate_sentences=True, + push_text_frames=False, + push_stop_frames=True, + sample_rate=sample_rate, + **kwargs, + ) self._api_key = api_key self._base_url = base_url @@ -498,28 +504,109 @@ class ElevenLabsHttpTTSService(TTSService): self._output_format = "" # initialized in start() self._voice_settings = self._set_voice_settings() + # Track cumulative time to properly sequence word timestamps across utterances + self._cumulative_time = 0 + self._started = False + + def language_to_service_language(self, language: Language) -> Optional[str]: + """Convert pipecat Language to ElevenLabs language code.""" + return language_to_elevenlabs_language(language) + def can_generate_metrics(self) -> bool: + """Indicate that this service can generate usage metrics.""" return True def _set_voice_settings(self): return build_elevenlabs_voice_settings(self._settings) async def start(self, frame: StartFrame): + """Initialize the service upon receiving a StartFrame.""" await super().start(frame) self._output_format = output_format_from_sample_rate(self.sample_rate) + self._cumulative_time = 0 + self._started = False - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - """Generate speech from text using ElevenLabs streaming API. + async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): + await super().push_frame(frame, direction) + if isinstance(frame, (StartInterruptionFrame, TTSStoppedFrame)): + # Reset timing on interruption or stop + self._started = False + self._cumulative_time = 0 + if isinstance(frame, TTSStoppedFrame): + await self.add_word_timestamps([("LLMFullResponseEndFrame", 0), ("Reset", 0)]) + + def calculate_word_times(self, alignment_info: Mapping[str, Any]) -> List[Tuple[str, float]]: + """Calculate word timing from character alignment data. + + Example input data: + { + "characters": [" ", "H", "e", "l", "l", "o", " ", "w", "o", "r", "l", "d"], + "character_start_times_seconds": [0.0, 0.1, 0.15, 0.2, 0.25, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9], + "character_end_times_seconds": [0.1, 0.15, 0.2, 0.25, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0] + } + + Would produce word times (with cumulative_time=0): + [("Hello", 0.1), ("world", 0.5)] Args: - text: The text to convert to speech + alignment_info: Character timing data from ElevenLabs + + Returns: + List of (word, timestamp) pairs + """ + chars = alignment_info.get("characters", []) + char_start_times = alignment_info.get("character_start_times_seconds", []) + + if not chars or not char_start_times or len(chars) != len(char_start_times): + logger.warning( + f"Invalid alignment data: chars={len(chars)}, times={len(char_start_times)}" + ) + return [] + + # Build the words and find their start times + words = [] + word_start_times = [] + current_word = "" + first_char_idx = -1 + + for i, char in enumerate(chars): + if char == " ": + if current_word: # Only add non-empty words + words.append(current_word) + # Use time of the first character of the word, offset by cumulative time + word_start_times.append( + self._cumulative_time + char_start_times[first_char_idx] + ) + current_word = "" + first_char_idx = -1 + else: + if not current_word: # This is the first character of a new word + first_char_idx = i + current_word += char + + # Don't forget the last word if there's no trailing space + if current_word and first_char_idx >= 0: + words.append(current_word) + word_start_times.append(self._cumulative_time + char_start_times[first_char_idx]) + + # Create word-time pairs + word_times = list(zip(words, word_start_times)) + + return word_times + + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + """Generate speech from text using ElevenLabs streaming API with timestamps. + + Args: + text: Text to convert to speech Yields: - Frames containing audio data and status information + Audio and control frames """ logger.debug(f"{self}: Generating TTS [{text}]") - url = f"{self._base_url}/v1/text-to-speech/{self._voice_id}/stream" + # Use the with-timestamps endpoint + url = f"{self._base_url}/v1/text-to-speech/{self._voice_id}/stream/with-timestamps" payload: Dict[str, Union[str, Dict[str, Union[float, bool]]]] = { "text": text, @@ -550,8 +637,6 @@ class ElevenLabsHttpTTSService(TTSService): if self._settings["optimize_streaming_latency"] is not None: params["optimize_streaming_latency"] = self._settings["optimize_streaming_latency"] - logger.debug(f"{self} ElevenLabs request - payload: {payload}, params: {params}") - try: await self.start_ttfb_metrics() @@ -566,17 +651,59 @@ class ElevenLabsHttpTTSService(TTSService): await self.start_tts_usage_metrics(text) - # Process the streaming response - CHUNK_SIZE = 1024 + # Start TTS sequence if not already started + if not self._started: + self.start_word_timestamps() + yield TTSStartedFrame() + self._started = True + + # Track the duration of this utterance based on the last character's end time + utterance_duration = 0 + async for line in response.content: + line_str = line.decode("utf-8").strip() + if not line_str: + continue + + try: + # Parse the JSON object + data = json.loads(line_str) + + # Process audio if present + if data and "audio_base64" in data: + await self.stop_ttfb_metrics() + audio = base64.b64decode(data["audio_base64"]) + yield TTSAudioRawFrame(audio, self.sample_rate, 1) + + # Process alignment if present + if data and "alignment" in data: + alignment = data["alignment"] + if alignment: # Ensure alignment is not None + # Get end time of the last character in this chunk + char_end_times = alignment.get("character_end_times_seconds", []) + if char_end_times: + chunk_end_time = char_end_times[-1] + # Update to the longest end time seen so far + utterance_duration = max(utterance_duration, chunk_end_time) + + # Calculate word timestamps + word_times = self.calculate_word_times(alignment) + if word_times: + await self.add_word_timestamps(word_times) + except json.JSONDecodeError as e: + logger.warning(f"Failed to parse JSON from stream: {e}") + continue + except Exception as e: + logger.error(f"Error processing response: {e}", exc_info=True) + continue + + # After processing all chunks, add the total utterance duration + # to the cumulative time to ensure next utterance starts after this one + if utterance_duration > 0: + self._cumulative_time += utterance_duration - yield TTSStartedFrame() - async for chunk in response.content.iter_chunked(CHUNK_SIZE): - if len(chunk) > 0: - await self.stop_ttfb_metrics() - yield TTSAudioRawFrame(chunk, self.sample_rate, 1) except Exception as e: logger.error(f"Error in run_tts: {e}") yield ErrorFrame(error=str(e)) finally: await self.stop_ttfb_metrics() - yield TTSStoppedFrame() + # Let the parent class handle TTSStoppedFrame From 4a23e138b14ecdfe497135a6045e8e8031953922 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 15 Apr 2025 21:29:19 -0400 Subject: [PATCH 2/4] Added word/timestamp pairs to ElevenLabsHttpTTSService --- src/pipecat/services/elevenlabs/tts.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index d77e4199c..832ed7309 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -12,6 +12,7 @@ from typing import Any, AsyncGenerator, Dict, List, Literal, Mapping, Optional, import aiohttp from loguru import logger from pydantic import BaseModel, model_validator +from sentry_sdk import push_scope from pipecat.frames.frames import ( CancelFrame, From 1e0a9d7b06164c22b50a30f59fd26d5d33cd3d11 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 15 Apr 2025 22:39:36 -0400 Subject: [PATCH 3/4] Add previous_text context to ElevenLabsHttpTTSService --- CHANGELOG.md | 3 +++ src/pipecat/services/elevenlabs/tts.py | 28 +++++++++++++++++++++++++- 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d3333c27d..1324e2dbb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 you to control aggregator settings. You can now pass these arguments when creating aggregator pairs with `create_context_aggregator()`. +- Added `previous_text` context support to ElevenLabsHttpTTSService, improving + speech consistency across sentences within an LLM response. + - Added word/timestamp pairs to `ElevenLabsHttpTTSService`. - It is now possible to disable `SoundfileMixer` when created. You can then use diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index 832ed7309..b3e203b35 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -12,13 +12,13 @@ from typing import Any, AsyncGenerator, Dict, List, Literal, Mapping, Optional, import aiohttp from loguru import logger from pydantic import BaseModel, model_validator -from sentry_sdk import push_scope from pipecat.frames.frames import ( CancelFrame, EndFrame, ErrorFrame, Frame, + LLMFullResponseEndFrame, StartFrame, StartInterruptionFrame, TTSAudioRawFrame, @@ -509,6 +509,9 @@ class ElevenLabsHttpTTSService(WordTTSService): self._cumulative_time = 0 self._started = False + # Store previous text for context within a turn + self._previous_text = "" + def language_to_service_language(self, language: Language) -> Optional[str]: """Convert pipecat Language to ElevenLabs language code.""" return language_to_elevenlabs_language(language) @@ -526,6 +529,7 @@ class ElevenLabsHttpTTSService(WordTTSService): self._output_format = output_format_from_sample_rate(self.sample_rate) self._cumulative_time = 0 self._started = False + self._previous_text = "" async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): await super().push_frame(frame, direction) @@ -533,9 +537,15 @@ class ElevenLabsHttpTTSService(WordTTSService): # Reset timing on interruption or stop self._started = False self._cumulative_time = 0 + self._previous_text = "" + if isinstance(frame, TTSStoppedFrame): await self.add_word_timestamps([("LLMFullResponseEndFrame", 0), ("Reset", 0)]) + elif isinstance(frame, LLMFullResponseEndFrame): + # End of turn - reset previous text + self._previous_text = "" + def calculate_word_times(self, alignment_info: Mapping[str, Any]) -> List[Tuple[str, float]]: """Calculate word timing from character alignment data. @@ -598,6 +608,10 @@ class ElevenLabsHttpTTSService(WordTTSService): async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using ElevenLabs streaming API with timestamps. + Makes a request to the ElevenLabs API to generate audio and timing data. + Tracks the duration of each utterance to ensure correct sequencing. + Includes previous text as context for better prosody continuity. + Args: text: Text to convert to speech @@ -614,6 +628,11 @@ class ElevenLabsHttpTTSService(WordTTSService): "model_id": self._model_name, } + # Include previous text as context if available + if self._previous_text: + payload["previous_text"] = self._previous_text + print(f"Previous text: {self._previous_text}") + if self._voice_settings: payload["voice_settings"] = self._voice_settings @@ -702,6 +721,13 @@ class ElevenLabsHttpTTSService(WordTTSService): if utterance_duration > 0: self._cumulative_time += utterance_duration + # Append the current text to previous_text for context continuity + # Only add a space if there's already text + if self._previous_text: + self._previous_text += " " + text + else: + self._previous_text = text + except Exception as e: logger.error(f"Error in run_tts: {e}") yield ErrorFrame(error=str(e)) From 2dafbee2aa26b9c7062168a8e4f8a597f010c1cf Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 16 Apr 2025 22:29:33 -0400 Subject: [PATCH 4/4] Code review fixes --- src/pipecat/services/elevenlabs/tts.py | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index b3e203b35..acb4a7b12 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -523,21 +523,24 @@ class ElevenLabsHttpTTSService(WordTTSService): def _set_voice_settings(self): return build_elevenlabs_voice_settings(self._settings) + def _reset_state(self): + """Reset internal state variables.""" + self._cumulative_time = 0 + self._started = False + self._previous_text = "" + logger.debug(f"{self}: Reset internal state") + async def start(self, frame: StartFrame): """Initialize the service upon receiving a StartFrame.""" await super().start(frame) self._output_format = output_format_from_sample_rate(self.sample_rate) - self._cumulative_time = 0 - self._started = False - self._previous_text = "" + self._reset_state() async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): await super().push_frame(frame, direction) if isinstance(frame, (StartInterruptionFrame, TTSStoppedFrame)): # Reset timing on interruption or stop - self._started = False - self._cumulative_time = 0 - self._previous_text = "" + self._reset_state() if isinstance(frame, TTSStoppedFrame): await self.add_word_timestamps([("LLMFullResponseEndFrame", 0), ("Reset", 0)]) @@ -631,7 +634,6 @@ class ElevenLabsHttpTTSService(WordTTSService): # Include previous text as context if available if self._previous_text: payload["previous_text"] = self._previous_text - print(f"Previous text: {self._previous_text}") if self._voice_settings: payload["voice_settings"] = self._voice_settings