# Copyright (c) 2024-2026, Daily # # SPDX-License-Identifier: BSD 2-Clause License """Hume Text-to-Speech service implementation.""" import base64 import os from typing import Any, AsyncGenerator, Optional import httpx from loguru import logger from pydantic import BaseModel from pipecat import version as pipecat_version from pipecat.frames.frames import ( CancelFrame, EndFrame, Frame, InterruptionFrame, StartFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.tts_service import WordTTSService from pipecat.utils.tracing.service_decorators import traced_tts try: from hume import AsyncHumeClient from hume.tts import FormatPcm, 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]`.") raise Exception(f"Missing module: {e}") HUME_SAMPLE_RATE = 48_000 # Hume TTS streams at 48 kHz # Tracking headers for Hume API requests DEFAULT_HEADERS = { "X-Hume-Client-Name": "pipecat", "X-Hume-Client-Version": pipecat_version(), } class HumeTTSService(WordTTSService): """Hume Octave Text-to-Speech service. Streams PCM audio via Hume's HTTP output streaming (JSON chunks) endpoint using the Python SDK and emits ``TTSAudioRawFrame`` frames suitable for Pipecat transports. Supported features: - 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. """ class InputParams(BaseModel): """Optional synthesis parameters for Hume TTS. Parameters: description: Natural-language acting directions (up to 100 characters). speed: Speaking-rate multiplier (0.5-2.0). trailing_silence: Seconds of silence to append at the end (0-5). """ description: Optional[str] = None speed: Optional[float] = None trailing_silence: Optional[float] = None def __init__( self, *, api_key: Optional[str] = None, voice_id: str, params: Optional[InputParams] = None, sample_rate: Optional[int] = HUME_SAMPLE_RATE, **kwargs, ) -> None: """Initialize the HumeTTSService. Args: api_key: Hume API key. If omitted, reads the ``HUME_API_KEY`` environment variable. voice_id: ID of the voice to use. Only voice IDs are supported; voice names are not. params: Optional synthesis controls (acting instructions, speed, trailing silence). sample_rate: Output sample rate for emitted PCM frames. Defaults to 48_000 (Hume). **kwargs: Additional arguments passed to the parent class. """ api_key = api_key or os.getenv("HUME_API_KEY") if not api_key: raise ValueError("HumeTTSService requires an API key (env HUME_API_KEY or api_key=)") if sample_rate != HUME_SAMPLE_RATE: logger.warning( f"Hume TTS streams at {HUME_SAMPLE_RATE} Hz; configured sample_rate={sample_rate}" ) # 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, ) # Create a custom httpx.AsyncClient with tracking headers # Headers are included in all requests made by the Hume SDK self._http_client = httpx.AsyncClient(headers=DEFAULT_HEADERS) self._client = AsyncHumeClient(api_key=api_key, httpx_client=self._http_client) self._params = params or HumeTTSService.InputParams() # Store voice in the base class (mirrors other services) self.set_voice(voice_id) 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. Returns: True if metrics can be generated, False otherwise. """ return True async def start(self, frame: StartFrame) -> None: """Start the service. Args: 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 stop(self, frame: EndFrame) -> None: """Stop the service and cleanup resources. Args: frame: The end frame. """ await super().stop(frame) if hasattr(self, "_http_client") and self._http_client: await self._http_client.aclose() async def cancel(self, frame: CancelFrame) -> None: """Cancel the service and cleanup resources. Args: frame: The cancel frame. """ await super().cancel(frame) if hasattr(self, "_http_client") and self._http_client: await self._http_client.aclose() 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`. Args: key: The name of the setting to update. Recognized keys are: - "voice_id" - "description" - "speed" - "trailing_silence" value: The new value for the setting. """ key_l = (key or "").lower() if key_l == "voice_id": self.set_voice(str(value)) 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": self._params.speed = None if value is None else float(value) elif key_l == "trailing_silence": self._params.trailing_silence = None if value is None else float(value) else: # Defer unknown keys to the base class await super().update_setting(key, value) @traced_tts async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using Hume TTS with word timestamps. Args: text: The text to be synthesized. Returns: An async generator that yields `Frame` objects, including `TTSStartedFrame`, `TTSAudioRawFrame`, `ErrorFrame`, and `TTSStoppedFrame`. """ logger.debug(f"{self}: Generating Hume TTS: [{text}]") # Build the request payload utterance_kwargs: dict[str, Any] = { "text": text, "voice": PostedUtteranceVoiceWithId(id=self._voice_id), } if self._params.description is not None: utterance_kwargs["description"] = self._params.description if self._params.speed is not None: utterance_kwargs["speed"] = self._params.speed if self._params.trailing_silence is not None: utterance_kwargs["trailing_silence"] = self._params.trailing_silence utterance = PostedUtterance(**utterance_kwargs) # Request raw PCM chunks in the streaming JSON pcm_fmt = FormatPcm(type="pcm") await self.start_ttfb_metrics() await self.start_tts_usage_metrics(text) # Start TTS sequence if not already started if not self._started: await self.start_word_timestamps() yield TTSStartedFrame() self._started = True try: # Instant mode is always enabled here (not user-configurable) # Hume emits mono PCM at 48 kHz; downstream can resample if needed. # We buffer audio bytes before sending to prevent glitches. self._audio_bytes = b"" # 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 audio_b64: await self.stop_ttfb_metrics() 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"" # 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, num_channels=1, ) yield frame self._audio_bytes = b"" # Update cumulative time for next utterance if utterance_duration > 0: self._cumulative_time = utterance_duration except Exception as e: await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: # Ensure TTFB timer is stopped even on early failures await self.stop_ttfb_metrics() # Let the parent class handle TTSStoppedFrame via push_stop_frames