diff --git a/CHANGELOG.md b/CHANGELOG.md index 10aa1c662..268022300 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,9 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added `ElevenLabsRealtimeSTTService` which implements the Realtime STT service from ElevenLabs. -- Added ai-coustics integrated VAD (`AICVADAnalyzer`) with `AICFilter` factory and - example wiring; leverages the enhancement model for robust detection with no - ONNX dependency or added processing complexity. +- Added word-level timestamps support to Hume TTS service ### Changed diff --git a/examples/foundational/07ae-interruptible-hume.py b/examples/foundational/07ae-interruptible-hume.py index 046f2d4c8..c5de34c85 100644 --- a/examples/foundational/07ae-interruptible-hume.py +++ b/examples/foundational/07ae-interruptible-hume.py @@ -13,24 +13,29 @@ from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams -from pipecat.frames.frames import LLMRunFrame +from pipecat.frames.frames import LLMRunFrame, TTSTextFrame +from pipecat.observers.loggers.debug_log_observer import DebugLogObserver, FrameEndpoint from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.llm_context import LLMContext -from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair +from pipecat.processors.aggregators.llm_response_universal import ( + LLMContextAggregatorPair, +) from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.hume.tts import HUME_SAMPLE_RATE, HumeTTSService from pipecat.services.openai.llm import OpenAILLMService +from pipecat.transports.base_output import BaseOutputTransport from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams load_dotenv(override=True) + # We store functions so objects (e.g. SileroVADAnalyzer) don't get # instantiated. The function will be called when the desired transport gets # selected. @@ -88,7 +93,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): stt, context_aggregator.user(), # User responses llm, # LLM - tts, # TTS + tts, # TTS (HumeTTSService with word timestamps) transport.output(), # Transport bot output context_aggregator.assistant(), # Assistant spoken responses ] @@ -102,7 +107,14 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): audio_out_sample_rate=HUME_SAMPLE_RATE, ), idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, - observers=[RTVIObserver(rtvi)], + observers=[ + RTVIObserver(rtvi), + DebugLogObserver( + frame_types={ + TTSTextFrame: (BaseOutputTransport, FrameEndpoint.SOURCE), + } + ), + ], ) @rtvi.event_handler("on_client_ready") @@ -112,6 +124,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): logger.info(f"Client connected") + logger.info( + "💡 Word timestamps are enabled! Watch the console for TTSTextFrame logs showing each word with its PTS." + ) # Kick off the conversation. messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMRunFrame()]) diff --git a/src/pipecat/services/hume/tts.py b/src/pipecat/services/hume/tts.py index a3a7e9a4c..278626748 100644 --- a/src/pipecat/services/hume/tts.py +++ b/src/pipecat/services/hume/tts.py @@ -14,12 +14,14 @@ from pydantic import BaseModel from pipecat.frames.frames import ( ErrorFrame, Frame, + InterruptionFrame, StartFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, ) -from pipecat.services.tts_service import TTSService +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.tts_service import WordTTSService from pipecat.utils.tracing.service_decorators import traced_tts try: @@ -29,6 +31,7 @@ try: PostedUtterance, PostedUtteranceVoiceWithId, ) + from hume.tts.types import TimestampMessage except ModuleNotFoundError as e: # pragma: no cover - import-time guidance logger.error(f"Exception: {e}") logger.error("In order to use Hume, you need to `pip install pipecat-ai[hume]`.") @@ -38,7 +41,7 @@ except ModuleNotFoundError as e: # pragma: no cover - import-time guidance HUME_SAMPLE_RATE = 48_000 # Hume TTS streams at 48 kHz -class HumeTTSService(TTSService): +class HumeTTSService(WordTTSService): """Hume Octave Text-to-Speech service. Streams PCM audio via Hume's HTTP output streaming (JSON chunks) endpoint @@ -48,6 +51,7 @@ class HumeTTSService(TTSService): - Generates speech from text using Hume TTS. - Streams PCM audio. + - Supports word-level timestamps for precise audio-text synchronization. - Supports dynamic updates of voice and synthesis parameters at runtime. - Provides metrics for Time To First Byte (TTFB) and TTS usage. """ @@ -92,7 +96,13 @@ class HumeTTSService(TTSService): f"Hume TTS streams at {HUME_SAMPLE_RATE} Hz; configured sample_rate={sample_rate}" ) - super().__init__(sample_rate=sample_rate, **kwargs) + # WordTTSService sets push_text_frames=False by default, which we want + super().__init__( + sample_rate=sample_rate, + push_text_frames=False, + push_stop_frames=True, + **kwargs, + ) self._client = AsyncHumeClient(api_key=api_key) self._params = params or HumeTTSService.InputParams() @@ -102,6 +112,10 @@ class HumeTTSService(TTSService): self._audio_bytes = b"" + # Track cumulative time for word timestamps across utterances + self._cumulative_time = 0.0 + self._started = False + def can_generate_metrics(self) -> bool: """Can generate metrics. @@ -117,6 +131,27 @@ class HumeTTSService(TTSService): frame: The start frame. """ await super().start(frame) + self._reset_state() + + def _reset_state(self): + """Reset internal state variables.""" + self._cumulative_time = 0.0 + self._started = False + + async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): + """Push a frame and handle state changes. + + Args: + frame: The frame to push. + direction: The direction to push the frame. + """ + await super().push_frame(frame, direction) + if isinstance(frame, (InterruptionFrame, TTSStoppedFrame)): + # Reset timing on interruption or stop + self._reset_state() + + if isinstance(frame, TTSStoppedFrame): + await self.add_word_timestamps([("Reset", 0)]) async def update_setting(self, key: str, value: Any) -> None: """Runtime updates via `TTSUpdateSettingsFrame`. @@ -133,7 +168,7 @@ class HumeTTSService(TTSService): if key_l == "voice_id": self.set_voice(str(value)) - logger.info(f"HumeTTSService voice_id set to: {self.voice}") + logger.debug(f"HumeTTSService voice_id set to: {self.voice}") elif key_l == "description": self._params.description = None if value is None else str(value) elif key_l == "speed": @@ -146,7 +181,7 @@ class HumeTTSService(TTSService): @traced_tts async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - """Generate speech from text using Hume TTS. + """Generate speech from text using Hume TTS with word timestamps. Args: text: The text to be synthesized. @@ -177,7 +212,12 @@ class HumeTTSService(TTSService): await self.start_ttfb_metrics() await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + + # Start TTS sequence if not already started + if not self._started: + self.start_word_timestamps() + yield TTSStartedFrame() + self._started = True try: # Instant mode is always enabled here (not user-configurable) @@ -188,23 +228,50 @@ class HumeTTSService(TTSService): # Use version "2" by default if no description is provided # Version "1" is needed when description is used version = "1" if self._params.description is not None else "2" + + # Track the duration of this utterance based on the last timestamp + utterance_duration = 0.0 + async for chunk in self._client.tts.synthesize_json_streaming( utterances=[utterance], format=pcm_fmt, instant_mode=True, version=version, + include_timestamp_types=["word"], # Request word-level timestamps ): + # Process audio chunks audio_b64 = getattr(chunk, "audio", None) - if not audio_b64: - continue + if audio_b64: + await self.stop_ttfb_metrics() + pcm_bytes = base64.b64decode(audio_b64) + self._audio_bytes += pcm_bytes - pcm_bytes = base64.b64decode(audio_b64) - self._audio_bytes += pcm_bytes + # Buffer audio until we have enough to avoid glitches + if len(self._audio_bytes) >= self.chunk_size: + frame = TTSAudioRawFrame( + audio=self._audio_bytes, + sample_rate=self.sample_rate, + num_channels=1, + ) + yield frame + self._audio_bytes = b"" - # Buffer audio until we have enough to avoid glitches - if len(self._audio_bytes) < self.chunk_size: - continue + # Process timestamp messages + if isinstance(chunk, TimestampMessage): + timestamp = chunk.timestamp + if timestamp.type == "word": + # Convert milliseconds to seconds and add cumulative offset + word_start_time = self._cumulative_time + (timestamp.time.begin / 1000.0) + word_end_time = self._cumulative_time + (timestamp.time.end / 1000.0) + # Track the maximum end time for this utterance + utterance_duration = max(utterance_duration, word_end_time) + + # Add word timestamp + await self.add_word_timestamps([(timestamp.text, word_start_time)]) + + # Flush any remaining audio bytes + if self._audio_bytes: frame = TTSAudioRawFrame( audio=self._audio_bytes, sample_rate=self.sample_rate, @@ -215,10 +282,14 @@ class HumeTTSService(TTSService): self._audio_bytes = b"" + # Update cumulative time for next utterance + if utterance_duration > 0: + self._cumulative_time = utterance_duration + except Exception as e: logger.error(f"{self} exception: {e}") await self.push_error(ErrorFrame(error=f"{self} error: {e}")) finally: # Ensure TTFB timer is stopped even on early failures await self.stop_ttfb_metrics() - yield TTSStoppedFrame() + # Let the parent class handle TTSStoppedFrame via push_stop_frames