diff --git a/src/pipecat/services/inworld/tts.py b/src/pipecat/services/inworld/tts.py index bf9a5d6ee..9d328fa5a 100644 --- a/src/pipecat/services/inworld/tts.py +++ b/src/pipecat/services/inworld/tts.py @@ -4,7 +4,33 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Inworld's text-to-speech service implementations.""" +"""Inworld AI Text-to-Speech Service Implementation. + +This module provides integration with Inworld AI's HTTP-based TTS API, enabling +real-time text-to-speech synthesis with high-quality, natural-sounding voices. + +Key Features: +- HTTP streaming API support for low-latency audio generation +- Multiple voice options (Ashley, Hades, etc.) +- Real-time audio chunk processing with proper buffering +- WAV header handling and audio format conversion +- Comprehensive error handling and metrics tracking + +Technical Implementation: +- Uses aiohttp for HTTP streaming connections +- Implements JSON line-by-line parsing for streaming responses +- Handles base64-encoded audio data with proper decoding +- Manages audio continuity to prevent clicks and artifacts +- Integrates with Pipecat's frame-based pipeline architecture + +Usage: + tts = InworldHttpTTSService( + api_key=os.getenv("INWORLD_API_KEY"), + voice_id="Ashley", + model="inworld-tts-1", + aiohttp_session=session + ) +""" import base64 import io @@ -40,11 +66,35 @@ from pipecat.utils.tracing.service_decorators import traced_tts def language_to_inworld_language(language: Language) -> Optional[str]: """Convert Pipecat's Language enum to Inworld's language code. + Inworld AI supports a specific set of language codes for TTS synthesis. + This function maps Pipecat's standardized Language enum values to the + corresponding language codes expected by Inworld's API. + + Supported Languages: + - EN (English) -> "en" + - ES (Spanish) -> "es" + - FR (French) -> "fr" + - KO (Korean) -> "ko" + - NL (Dutch) -> "nl" + - ZH (Chinese) -> "zh" + + The function also handles language variants (e.g., es-ES, en-US) by + extracting the base language code and mapping it if supported. + Args: - language: The Language enum value to convert. + language: The Language enum value to convert (e.g., Language.EN). Returns: - The corresponding Inworld language code, or None if not supported. + The corresponding Inworld language code string (e.g., "en"), + or None if the language is not supported by Inworld's API. + + Example: + >>> language_to_inworld_language(Language.EN) + "en" + >>> language_to_inworld_language(Language.ES) + "es" + >>> language_to_inworld_language(Language.DE) # Not supported + None """ BASE_LANGUAGES = { Language.EN: "en", @@ -69,11 +119,42 @@ def language_to_inworld_language(language: Language) -> Optional[str]: class InworldHttpTTSService(TTSService): - """Inworld HTTP-based TTS service. + """Inworld AI HTTP-based Text-to-Speech Service. - Provides text-to-speech using Inworld's HTTP API for simpler, non-streaming - synthesis. Suitable for use cases where streaming is not required and simpler - integration is preferred. + This service integrates Inworld AI's high-quality TTS API with Pipecat's pipeline + architecture. It provides real-time speech synthesis with natural-sounding voices + and low-latency streaming audio delivery. + + Key Features: + - Real-time HTTP streaming for minimal latency + - Multiple voice options (Ashley, Hades, etc.) + - High-quality audio output (48kHz LINEAR16 PCM) + - Automatic audio format handling and header stripping + - Comprehensive error handling and recovery + - Built-in performance metrics and monitoring + + Technical Architecture: + - Uses aiohttp for non-blocking HTTP requests + - Implements JSON line-by-line streaming protocol + - Processes base64-encoded audio chunks in real-time + - Manages audio continuity to prevent artifacts + - Integrates with Pipecat's frame-based pipeline system + + Supported Configuration: + - Voice Selection: Ashley, Hades, and other Inworld voices + - Models: inworld-tts-1 and other available models + - Audio Formats: LINEAR16 PCM at various sample rates + - Languages: English, Spanish, French, Korean, Dutch, Chinese + + Example Usage: + async with aiohttp.ClientSession() as session: + tts = InworldHttpTTSService( + api_key=os.getenv("INWORLD_API_KEY"), + voice_id="Ashley", # Voice selection + model="inworld-tts-1", # TTS model + aiohttp_session=session, # Required HTTP session + sample_rate=48000, # Audio quality + ) """ class InputParams(BaseModel): @@ -89,7 +170,7 @@ class InworldHttpTTSService(TTSService): """ language: Optional[Language] = Language.EN - voice_id: str = "Ashley" ## QUESTION: How to make this modifyable/how to modify? + voice_id: str = "Hades" ## QUESTION: How to make this modifyable/how to modify? # QUESTION: What about speed, pitch, and temperature?? def __init__( @@ -97,6 +178,7 @@ class InworldHttpTTSService(TTSService): *, api_key: str, aiohttp_session: aiohttp.ClientSession, + voice_id: str = "Ashley", model: str = "inworld-tts-1", base_url: str = "https://api.inworld.ai/tts/v1/voice:stream", sample_rate: Optional[int] = 48000, @@ -106,36 +188,67 @@ class InworldHttpTTSService(TTSService): ): """Initialize the Inworld HTTP TTS service. + Sets up the TTS service with Inworld AI's streaming API configuration. + This constructor prepares all necessary parameters for real-time speech synthesis. + Args: - api_key: Inworld API key for authentication. - aiohttp_session: Shared aiohttp session for HTTP requests. - model: TTS model to use (e.g., "inworld-tts-1"). - base_url: Base URL for Inworld HTTP API. - sample_rate: Audio sample rate. If None, uses default. - encoding: Audio encoding format. - params: Additional input parameters for voice customization. - **kwargs: Additional arguments passed to the parent TTSService. + api_key: Inworld API key for authentication (base64-encoded from Inworld Portal). + Get this from: Inworld Portal > Settings > API Keys > Runtime API Key + aiohttp_session: Shared aiohttp session for HTTP requests. Must be provided + for proper connection pooling and resource management. + voice_id: Voice to use for synthesis. Available options include: + - "Ashley" (default) - Natural female voice + - "Hades" - Distinctive character voice + - Other voices available through Inworld's voice catalog + model: TTS model to use. Currently supported: + - "inworld-tts-1" (default) - Latest high-quality model + - Other models as available in Inworld's API + base_url: Base URL for Inworld HTTP API. Uses streaming endpoint by default. + Should normally not be changed unless using a different environment. + sample_rate: Audio sample rate in Hz. Common values: + - 48000 (default) - High quality, suitable for most applications + - 24000 - Good quality, lower bandwidth + - 16000 - Basic quality, minimal bandwidth + encoding: Audio encoding format. Supported options: + - "LINEAR16" (default) - Uncompressed PCM, best quality + - Other formats as supported by Inworld API + params: Additional input parameters for advanced voice customization. + Usually None for standard usage. + **kwargs: Additional arguments passed to the parent TTSService class. + + Note: + The aiohttp_session parameter is required because Inworld's HTTP API + benefits from connection reuse and proper async session management. """ + # Initialize parent TTSService with audio configuration super().__init__(sample_rate=sample_rate, **kwargs) - params = params or InworldTTSService.InputParams() + # Use provided params or create default configuration + params = params or InworldHttpTTSService.InputParams() - self._api_key = api_key - self._session = aiohttp_session - self._base_url = base_url + # Store core configuration for API requests + self._api_key = api_key # Authentication credentials + self._session = aiohttp_session # HTTP session for requests + self._base_url = base_url # API endpoint URL + + # Build settings dictionary that matches Inworld's API expectations + # This will be sent as JSON payload in each TTS request self._settings = { - "voiceId": params.voice_id, - "modelId": model, - "audio_config": { - "audio_encoding": encoding, - "sample_rate_hertz": sample_rate, + "voiceId": voice_id, # Voice selection (fixes bug where this was ignored) + "modelId": model, # TTS model selection + "audio_config": { # Audio format configuration + "audio_encoding": encoding, # Format: LINEAR16, MP3, etc. + "sample_rate_hertz": sample_rate, # Sample rate: 48000, 24000, etc. }, + # Language configuration with fallback to English "language": self.language_to_service_language(params.language) if params.language else "en", } - self.set_voice(params.voice_id) - self.set_model_name(model) + + # Register voice and model with parent service for metrics and tracking + self.set_voice(voice_id) # Used for logging and metrics + self.set_model_name(model) # Used for performance tracking def can_generate_metrics(self) -> bool: """Check if this service can generate processing metrics. @@ -183,164 +296,205 @@ class InworldHttpTTSService(TTSService): @traced_tts async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - """Generate speech from text using Inworld's HTTP API. + """Generate speech from text using Inworld's streaming HTTP API. - This implementation streams audio chunk by chunk as it's received. + This is the core TTS processing function that: + 1. Sends text to Inworld's streaming TTS endpoint + 2. Receives JSON-streamed audio chunks in real-time + 3. Processes and cleans audio data (removes WAV headers, validates content) + 4. Yields audio frames for immediate playback in the pipeline + + Technical Details: + - Uses HTTP streaming with JSON line-by-line responses + - Each JSON line contains base64-encoded audio data + - Implements buffering to handle partial JSON lines + - Strips WAV headers to prevent audio artifacts/clicks + - Provides real-time audio streaming for low latency Args: text: The text to synthesize into speech. Yields: - Frame: Audio frames containing the synthesized speech. + Frame: Audio frames containing the synthesized speech, plus control frames. + + Raises: + ErrorFrame: If API errors occur or audio processing fails. """ logger.debug(f"{self}: Generating TTS [{text}]") + # ================================================================================ + # STEP 1: PREPARE API REQUEST + # ================================================================================ + # Build the JSON payload according to Inworld's API specification + # This matches the format shown in their documentation examples payload = { - "text": text, - "voiceId": self._settings["voiceId"], - "modelId": self._settings["modelId"], - "audio_config": self._settings["audio_config"], - "language": self._settings["language"], + "text": text, # Text to synthesize + "voiceId": self._settings["voiceId"], # Voice selection (Ashley, Hades, etc.) + "modelId": self._settings["modelId"], # TTS model (inworld-tts-1) + "audio_config": self._settings[ + "audio_config" + ], # Audio format settings (LINEAR16, 48kHz) + "language": self._settings["language"], # Language code (en, es, etc.) } + # Set up HTTP headers for authentication and content type + # Inworld requires Basic auth with base64-encoded API key headers = { - "Authorization": f"Basic {self._api_key}", - "Content-Type": "application/json", + "Authorization": f"Basic {self._api_key}", # Base64 API key from Inworld Portal + "Content-Type": "application/json", # JSON request body } try: + # ================================================================================ + # STEP 2: INITIALIZE METRICS AND STREAMING + # ================================================================================ + # Start measuring Time To First Byte (TTFB) for performance tracking await self.start_ttfb_metrics() + # Signal to the pipeline that TTS generation has started + # This allows downstream processors to prepare for incoming audio yield TTSStartedFrame() - # A flag to ensure we only strip the header from the very first chunk. + # Flag to track if we're processing the first audio chunk + # Used for WAV header handling and debugging is_first_chunk = True + # ================================================================================ + # STEP 3: MAKE HTTP STREAMING REQUEST + # ================================================================================ + # Use aiohttp's streaming POST to Inworld's streaming endpoint + # The endpoint returns JSON lines with audio chunks as they're generated async with self._session.post( self._base_url, json=payload, headers=headers ) as response: + # ================================================================================ + # STEP 4: HANDLE HTTP ERRORS + # ================================================================================ + # Check for API errors (expired keys, invalid requests, etc.) if response.status != 200: error_text = await response.text() logger.error(f"Inworld API error: {error_text}") await self.push_error(ErrorFrame(f"Inworld API error: {error_text}")) return - # Process the stream line by line. - async for line in response.content.iter_lines(): - line_str = line.decode("utf-8").strip() - if not line_str: + # ================================================================================ + # STEP 5: PROCESS STREAMING JSON RESPONSE + # ================================================================================ + # Inworld streams JSON lines where each line contains audio data + # We need to buffer incoming data and process complete lines + + # Buffer to accumulate incoming text data + # This handles cases where JSON lines are split across HTTP chunks + buffer = "" + + # Read HTTP response in manageable chunks (1KB each) + # This prevents memory issues with large responses + async for chunk in response.content.iter_chunked(1024): + if not chunk: continue - try: - chunk = json.loads(line_str) - if "result" in chunk and "audioContent" in chunk["result"]: - audio_chunk = base64.b64decode(chunk["result"]["audioContent"]) - audio_data = audio_chunk + # ============================================================================ + # STEP 6: BUFFER MANAGEMENT + # ============================================================================ + # Decode binary chunk to text and add to our line buffer + # Each chunk may contain partial JSON lines, so we need to accumulate + buffer += chunk.decode("utf-8") - # Correctly strip the header only from the first chunk. - if ( - is_first_chunk - and len(audio_chunk) > 44 - and audio_chunk.startswith(b"RIFF") - ): - audio_data = audio_chunk[44:] - is_first_chunk = False # Unset the flag. + # ============================================================================ + # STEP 7: LINE-BY-LINE JSON PROCESSING + # ============================================================================ + # Process all complete lines in the buffer (lines ending with \n) + # Leave partial lines in buffer for next iteration + while "\n" in buffer: + # Split on first newline, keeping remainder in buffer + line, buffer = buffer.split("\n", 1) + line_str = line.strip() - # Yield each audio frame as it's processed. - yield TTSAudioRawFrame( - audio=audio_data, - sample_rate=self.sample_rate, - num_channels=1, - ) + # Skip empty lines (common in streaming responses) + if not line_str: + continue - except json.JSONDecodeError: - continue + try: + # ================================================================ + # STEP 8: PARSE JSON AND EXTRACT AUDIO + # ================================================================ + # Parse the JSON line - should contain audio data + chunk_data = json.loads(line_str) + # Check if this line contains audio content + # Inworld's response format: {"result": {"audioContent": "base64data"}} + if "result" in chunk_data and "audioContent" in chunk_data["result"]: + # Decode base64 audio data to binary + audio_chunk = base64.b64decode(chunk_data["result"]["audioContent"]) + + # ======================================================== + # STEP 9: AUDIO DATA VALIDATION + # ======================================================== + # Skip empty audio chunks that could cause discontinuities + # Empty chunks can create gaps or clicks in audio playback + if not audio_chunk: + continue + + # Start with the raw audio data + audio_data = audio_chunk + + # ======================================================== + # STEP 10: WAV HEADER REMOVAL (CRITICAL FOR AUDIO QUALITY) + # ======================================================== + # Each audio chunk may have its own WAV header (44 bytes) + # These headers contain metadata and will sound like clicks if played + # We must strip them from EVERY chunk, not just the first one + if ( + len(audio_chunk) > 44 # Ensure chunk is large enough + and audio_chunk.startswith( + b"RIFF" + ) # Check for WAV header magic bytes + ): + # Remove the 44-byte WAV header to get pure audio data + audio_data = audio_chunk[44:] + + # Track that we've seen our first chunk (for debugging) + if is_first_chunk: + is_first_chunk = False + + # ======================================================== + # STEP 11: YIELD AUDIO FRAME TO PIPELINE + # ======================================================== + # Only yield frames with actual audio content + # Empty frames can cause pipeline issues + if len(audio_data) > 0: + # Create Pipecat audio frame with processed audio data + yield TTSAudioRawFrame( + audio=audio_data, # Clean audio without headers + sample_rate=self.sample_rate, # Configured sample rate (48kHz) + num_channels=1, # Mono audio + ) + + except json.JSONDecodeError: + # Ignore malformed JSON lines - streaming can have partial data + # This is normal in HTTP streaming scenarios + continue + + # ================================================================================ + # STEP 12: FINALIZE METRICS AND CLEANUP + # ================================================================================ + # Start usage metrics tracking after successful completion await self.start_tts_usage_metrics(text) except Exception as e: + # ================================================================================ + # STEP 13: ERROR HANDLING + # ================================================================================ + # Log any unexpected errors and notify the pipeline logger.error(f"{self} exception: {e}") await self.push_error(ErrorFrame(f"Error generating TTS: {e}")) finally: + # ================================================================================ + # STEP 14: CLEANUP AND COMPLETION + # ================================================================================ + # Always stop metrics tracking, even if errors occurred await self.stop_ttfb_metrics() + + # Signal to pipeline that TTS generation is complete + # This allows downstream processors to finalize audio processing yield TTSStoppedFrame() - - # @traced_tts - # async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - # """Generate speech from text using Inworld's HTTP API. - - # Args: - # text: The text to synthesize into speech. - - # Yields: - # Frame: Audio frames containing the synthesized speech. - # """ - # logger.debug(f"{self}: Generating TTS [{text}]") - - # payload = { - # "text": text, - # "voiceId": self._settings["voiceId"], - # "modelId": self._settings["modelId"], - # "audio_config": self._settings["audio_config"], - # "language": self._settings["language"], - # } - - # headers = { - # "Authorization": f"Basic {self._api_key}", - # "Content-Type": "application/json", - # } - - # try: - # await self.start_ttfb_metrics() - - # yield TTSStartedFrame() - - # async with self._session.post(self._base_url, json=payload, headers=headers) as response: - # if response.status != 200: - # error_text = await response.text() - # logger.error(f"Inworld API error: {error_text}") - # await self.push_error(ErrorFrame(f"Inworld API error: {error_text}")) - # return - - # raw_audio_data = io.BytesIO() - - # async for line in response.content.iter_lines(): - # line_str = line.decode('utf-8').strip() - # if not line_str: - # continue - - # try: - # chunk = json.loads(line_str) - # if "result" in chunk and "audioContent" in chunk["result"]: - # audio_chunk = base64.b64decode(chunk["result"]["audioContent"]) - # # Skip WAV header if present (first 44 bytes) - # if len(audio_chunk) > 44 and audio_chunk.startswith(b"RIFF"): - # audio_data = audio_chunk[44:] - # else: - # audio_data = audio_chunk - # raw_audio_data.write(audio_data) - # except json.JSONDecodeError: - # continue - - # await self.start_tts_usage_metrics(text) - - # audio_bytes = raw_audio_data.getvalue() - # if not audio_bytes: - # logger.error("No audio data received from Inworld API") - # await self.push_error(ErrorFrame("No audio data received")) - # return - - # frame = TTSAudioRawFrame( - # audio=audio_bytes, - # sample_rate=self.sample_rate, - # num_channels=1, - # ) - - # yield frame - - # except Exception as e: - # logger.error(f"{self} exception: {e}") - # await self.push_error(ErrorFrame(f"Error generating TTS: {e}")) - # finally: - # await self.stop_ttfb_metrics() - # yield TTSStoppedFrame()