Merge pull request #3070 from ivaaan/hume-timestamps

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
+- Added word-level timestamps support to Hume TTS service
-  example wiring; leverages the enhancement model for robust detection with no
-  ONNX dependency or added processing complexity.
 
 ### Changed
 

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()])

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:
+                if audio_b64:
-                    continue
+                    await self.stop_ttfb_metrics()
+                    pcm_bytes = base64.b64decode(audio_b64)
+                    self._audio_bytes += pcm_bytes
 
-                pcm_bytes = base64.b64decode(audio_b64)
+                    # Buffer audio until we have enough to avoid glitches
-                self._audio_bytes += pcm_bytes
+                    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
+                # Process timestamp messages
-                if len(self._audio_bytes) < self.chunk_size:
+                if isinstance(chunk, TimestampMessage):
-                    continue
+                    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