diff --git a/examples/foundational/07aa-interruptible-inworld-http.py b/examples/foundational/07aa-interruptible-inworld-http.py index 65d7babca..dbfbcc878 100644 --- a/examples/foundational/07aa-interruptible-inworld-http.py +++ b/examples/foundational/07aa-interruptible-inworld-http.py @@ -16,7 +16,7 @@ from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext -from pipecat.services.inworld.tts import InworldHttpNonStreamingService, InworldHttpStreamingService +from pipecat.services.inworld.tts import InworldTTSService from pipecat.services.openai.llm import OpenAILLMService from pipecat.services.openai.stt import OpenAISTTService from pipecat.transports.base_transport import BaseTransport, TransportParams @@ -58,30 +58,20 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si prompt="Expect words related to dogs, such as breed names.", ) - streaming = True + # Inworld TTS Service - Unified streaming and non-streaming + # Set streaming=True for real-time audio, streaming=False for complete audio generation + streaming = False # Toggle this to switch between modes - if streaming: - # Streaming TTS - Real-time audio generation as text is processed - tts = InworldHttpStreamingService( - api_key=os.getenv("INWORLD_API_KEY", ""), - aiohttp_session=session, - params=InworldHttpStreamingService.InputParams( - voice_id="Ashley", - model="inworld-tts-1", - temperature=0.8, - ), - ) - else: - # Non-streaming TTS - Complete audio generation then playback - tts = InworldHttpNonStreamingService( - api_key=os.getenv("INWORLD_API_KEY", ""), - aiohttp_session=session, - params=InworldHttpNonStreamingService.InputParams( - voice_id="Ashley", - model="inworld-tts-1", - temperature=0.8, - ), - ) + tts = InworldTTSService( + api_key=os.getenv("INWORLD_API_KEY", ""), + aiohttp_session=session, + streaming=streaming, # True: real-time chunks, False: complete audio then playback + params=InworldTTSService.InputParams( + voice_id="Ashley", + model="inworld-tts-1", + temperature=0.8, + ), + ) llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) diff --git a/src/pipecat/services/inworld/tts.py b/src/pipecat/services/inworld/tts.py index 3a70b7499..94ef5aa32 100644 --- a/src/pipecat/services/inworld/tts.py +++ b/src/pipecat/services/inworld/tts.py @@ -7,11 +7,12 @@ """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. +both streaming and non-streaming text-to-speech synthesis with high-quality, +natural-sounding voices. Key Features: -- HTTP streaming API support for low-latency audio generation +- HTTP streaming and non-streaming API support for flexible audio generation - Multiple voice options (Ashley, Hades, etc.) - Automatic language detection from input text (no manual language setting required) - Real-time audio chunk processing with proper buffering @@ -20,8 +21,8 @@ Key Features: Technical Implementation: -- Uses aiohttp for HTTP streaming connections -- Implements JSON line-by-line parsing for streaming responses +- Uses aiohttp for HTTP connections +- Implements both JSON line-by-line parsing (streaming) and complete response (non-streaming) - Handles base64-encoded audio data with proper decoding - Manages audio continuity to prevent clicks and artifacts - Integrates with Pipecat's frame-based pipeline architecture @@ -29,15 +30,29 @@ Technical Implementation: Usage:: async with aiohttp.ClientSession() as session: - tts = InworldHttpStreamingService( + # Streaming mode (default) - real-time audio generation + tts = InworldTTSService( api_key=os.getenv("INWORLD_API_KEY"), aiohttp_session=session, - params=InworldHttpStreamingService.InputParams( + streaming=True, # Default + params=InworldTTSService.InputParams( voice_id="Ashley", model="inworld-tts-1", temperature=0.8, # Optional: control synthesis variability (range: [0, 2]) ), ) + + # Non-streaming mode - complete audio generation then playback + tts = InworldTTSService( + api_key=os.getenv("INWORLD_API_KEY"), + aiohttp_session=session, + streaming=False, + params=InworldTTSService.InputParams( + voice_id="Ashley", + model="inworld-tts-1", + temperature=0.8, + ), + ) """ import base64 @@ -70,27 +85,30 @@ from pipecat.utils.text.skip_tags_aggregator import SkipTagsAggregator from pipecat.utils.tracing.service_decorators import traced_tts -class InworldHttpStreamingService(TTSService): +class InworldTTSService(TTSService): """Inworld AI HTTP-based Text-to-Speech Service. - 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. + This unified service integrates Inworld AI's high-quality TTS API with Pipecat's pipeline + architecture. It supports both streaming and non-streaming modes, providing flexible + speech synthesis with natural-sounding voices. Key Features: - - Real-time HTTP streaming for minimal latency + - **Streaming Mode**: Real-time HTTP streaming for minimal latency + - **Non-Streaming Mode**: Complete audio synthesis then chunked playback - 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 + - Unified interface for both modes Technical Architecture: - Uses aiohttp for non-blocking HTTP requests - - Implements JSON line-by-line streaming protocol - - Processes base64-encoded audio chunks in real-time + - **Streaming**: Implements JSON line-by-line streaming protocol + - **Non-Streaming**: Single HTTP POST with complete response + - Processes base64-encoded audio chunks in real-time or batch - Manages audio continuity to prevent artifacts - Integrates with Pipecat's frame-based pipeline system @@ -100,31 +118,38 @@ class InworldHttpStreamingService(TTSService): - Models: inworld-tts-1 and other available models - Audio Formats: LINEAR16 PCM at various sample rates - Language Detection: Automatically inferred from input text (no explicit language setting required) + - Mode Selection: streaming=True for real-time, streaming=False for complete synthesis Example Usage:: async with aiohttp.ClientSession() as session: - # Using default settings (Ashley voice, inworld-tts-1 model) - tts = InworldHttpStreamingService( + # Streaming mode (default) - Real-time audio generation + tts_streaming = InworldTTSService( api_key=os.getenv("INWORLD_API_KEY"), aiohttp_session=session, + streaming=True, # Default behavior + params=InworldTTSService.InputParams( + voice_id="Ashley", + model="inworld-tts-1", + temperature=0.8, # Add variability to speech synthesis (range: [0, 2]) + ), ) - # Or with custom voice, model, and temperature via params - params = InworldHttpStreamingService.InputParams( - voice_id="Hades", - model="inworld-tts-1-max", - temperature=0.8, # Add variability to speech synthesis (range: [0, 2]) - ) - tts = InworldHttpStreamingService( + # Non-streaming mode - Complete audio then playback + tts_complete = InworldTTSService( api_key=os.getenv("INWORLD_API_KEY"), aiohttp_session=session, - params=params, + streaming=False, + params=InworldTTSService.InputParams( + voice_id="Hades", + model="inworld-tts-1-max", + temperature=0.8, + ), ) """ class InputParams(BaseModel): - """Input parameters for Inworld HTTP TTS configuration. + """Input parameters for Inworld TTS configuration. Parameters: voice_id: Voice selection for speech synthesis (e.g., "Ashley", "Hades"). @@ -146,23 +171,29 @@ class InworldHttpStreamingService(TTSService): *, api_key: str, aiohttp_session: aiohttp.ClientSession, - base_url: str = "https://api.inworld.ai/tts/v1/voice:stream", + streaming: bool = True, + base_url: Optional[str] = None, sample_rate: Optional[int] = None, encoding: str = "LINEAR16", params: Optional[InputParams] = None, **kwargs, ): - """Initialize the Inworld HTTP TTS service. + """Initialize the Inworld 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. + Sets up the TTS service with Inworld AI's API configuration. + This constructor prepares all necessary parameters for speech synthesis. Args: 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. - base_url: Base URL for Inworld HTTP API. Uses streaming endpoint by default. + streaming: Whether to use streaming mode (True) or non-streaming mode (False). + - True: Real-time audio chunks as they're generated (lower latency) + - False: Complete audio file generated first, then chunked for playback (simpler) + base_url: Base URL for Inworld HTTP API. If None, automatically selected based on streaming mode: + - Streaming: "https://api.inworld.ai/tts/v1/voice:stream" + - Non-streaming: "https://api.inworld.ai/tts/v1/voice" Should normally not be changed unless using a different environment. sample_rate: Audio sample rate in Hz. If None, uses default from StartFrame. Common values: 48000 (high quality), 24000 (good quality), 16000 (basic) @@ -185,11 +216,19 @@ class InworldHttpStreamingService(TTSService): super().__init__(sample_rate=sample_rate, **kwargs) # Use provided params or create default configuration - params = params or InworldHttpStreamingService.InputParams() + params = params or InworldTTSService.InputParams() # Store core configuration for API requests self._api_key = api_key # Authentication credentials self._session = aiohttp_session # HTTP session for requests + self._streaming = streaming # Streaming mode selection + + # Set base URL based on streaming mode if not provided + if base_url is None: + if streaming: + base_url = "https://api.inworld.ai/tts/v1/voice:stream" # Streaming endpoint + else: + base_url = "https://api.inworld.ai/tts/v1/voice" # Non-streaming endpoint self._base_url = base_url # API endpoint URL # Build settings dictionary that matches Inworld's API expectations @@ -216,12 +255,12 @@ class InworldHttpStreamingService(TTSService): """Check if this service can generate processing metrics. Returns: - True, as Inworld HTTP service supports metrics generation. + True, as Inworld TTS service supports metrics generation. """ return True async def start(self, frame: StartFrame): - """Start the Inworld HTTP TTS service. + """Start the Inworld TTS service. Args: frame: The start frame containing initialization parameters. @@ -230,7 +269,7 @@ class InworldHttpStreamingService(TTSService): self._settings["audio_config"]["sample_rate_hertz"] = self.sample_rate async def stop(self, frame: EndFrame): - """Stop the Inworld HTTP TTS service. + """Stop the Inworld TTS service. Args: frame: The end frame. @@ -238,7 +277,7 @@ class InworldHttpStreamingService(TTSService): await super().stop(frame) async def cancel(self, frame: CancelFrame): - """Cancel the Inworld HTTP TTS service. + """Cancel the Inworld TTS service. Args: frame: The cancel frame. @@ -247,21 +286,30 @@ class InworldHttpStreamingService(TTSService): @traced_tts async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - """Generate speech from text using Inworld's streaming HTTP API. + """Generate speech from text using Inworld's HTTP API. - This is the core TTS processing function that: + This is the core TTS processing function that adapts its behavior based on the streaming mode: + + **Streaming Mode (streaming=True)**: 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 + **Non-Streaming Mode (streaming=False)**: + 1. Sends text to Inworld's non-streaming TTS endpoint + 2. Receives complete audio file as base64-encoded response + 3. Processes entire audio and chunks for playback + 4. Yields audio frames in manageable pieces + 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 + - **Streaming**: Uses HTTP streaming with JSON line-by-line responses + - **Non-Streaming**: Single HTTP POST with complete JSON response + - Each audio chunk contains base64-encoded audio data + - Implements buffering to handle partial data (streaming mode) - Strips WAV headers to prevent audio artifacts/clicks - - Provides real-time audio streaming for low latency + - Provides optimized audio delivery for each mode Args: text: The text to synthesize into speech. @@ -272,7 +320,7 @@ class InworldHttpStreamingService(TTSService): Raises: ErrorFrame: If API errors occur or audio processing fails. """ - logger.debug(f"{self}: Generating TTS [{text}]") + logger.debug(f"{self}: Generating TTS [{text}] (streaming={self._streaming})") # ================================================================================ # STEP 1: PREPARE API REQUEST @@ -302,7 +350,7 @@ class InworldHttpStreamingService(TTSService): try: # ================================================================================ - # STEP 2: INITIALIZE METRICS AND STREAMING + # STEP 2: INITIALIZE METRICS AND PROCESSING # ================================================================================ # Start measuring Time To First Byte (TTFB) for performance tracking await self.start_ttfb_metrics() @@ -312,10 +360,10 @@ class InworldHttpStreamingService(TTSService): yield TTSStartedFrame() # ================================================================================ - # STEP 3: MAKE HTTP STREAMING REQUEST + # STEP 3: MAKE HTTP REQUEST (MODE-SPECIFIC) # ================================================================================ - # Use aiohttp's streaming POST to Inworld's streaming endpoint - # The endpoint returns JSON lines with audio chunks as they're generated + # Use aiohttp to make request to Inworld's endpoint + # Behavior differs based on streaming mode async with self._session.post( self._base_url, json=payload, headers=headers ) as response: @@ -330,115 +378,34 @@ class InworldHttpStreamingService(TTSService): return # ================================================================================ - # STEP 5: PROCESS STREAMING JSON RESPONSE + # STEP 5: PROCESS RESPONSE (MODE-SPECIFIC) # ================================================================================ - # 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 - - # ============================================================================ - # 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") - - # ============================================================================ - # 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() - - # Skip empty lines (common in streaming responses) - if not line_str: - 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:] - - # ======================================================== - # 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 + # Choose processing method based on streaming mode + if self._streaming: + # Stream processing: JSON line-by-line with real-time audio + async for frame in self._process_streaming_response(response): + yield frame + else: + # Non-stream processing: Complete JSON response with batch audio + async for frame in self._process_non_streaming_response(response): + yield frame # ================================================================================ - # STEP 12: FINALIZE METRICS AND CLEANUP + # STEP 6: 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 + # STEP 7: 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 + # STEP 8: CLEANUP AND COMPLETION # ================================================================================ # Always stop metrics tracking, even if errors occurred await self.stop_ttfb_metrics() @@ -447,363 +414,187 @@ class InworldHttpStreamingService(TTSService): # This allows downstream processors to finalize audio processing yield TTSStoppedFrame() + async def _process_streaming_response( + self, response: aiohttp.ClientResponse + ) -> AsyncGenerator[Frame, None]: + """Process streaming JSON response with real-time audio chunks. -class InworldHttpNonStreamingService(TTSService): - """Inworld AI HTTP-based Text-to-Speech Service (Non-Streaming). - - This service integrates with Inworld AI's non-streaming TTS API for simpler, - complete audio synthesis. Suitable for use cases where streaming is not required - and you prefer to receive the complete audio file at once. - - Key Features: - - - Simple HTTP request/response for complete audio synthesis - - Same voice options as streaming version (Ashley, Hades, etc.) - - High-quality audio output (48kHz LINEAR16 PCM) - - Automatic language detection from input text - - Support for temperature parameter for synthesis variability - - Lower complexity compared to streaming implementation - - Technical Architecture: - - - Uses aiohttp for single HTTP POST request - - Downloads complete audio as base64-encoded data - - Processes entire audio file and chunks for playback - - Integrates with Pipecat's frame-based pipeline system - - Usage:: - - async with aiohttp.ClientSession() as session: - # Using default settings (Ashley voice, inworld-tts-1 model) - tts = InworldHttpNonStreamingService( - api_key=os.getenv("INWORLD_API_KEY"), - aiohttp_session=session, - ) - - # Or with custom voice, model, and temperature - params = InworldHttpNonStreamingService.InputParams( - voice_id="Hades", - model="inworld-tts-1-max", - temperature=0.8, # Control synthesis variability (range: [0, 2]) - ) - tts = InworldHttpNonStreamingService( - api_key=os.getenv("INWORLD_API_KEY"), - aiohttp_session=session, - params=params, - ) - """ - - class InputParams(BaseModel): - """Input parameters for Inworld non-streaming TTS configuration. - - Parameters: - voice_id: Voice selection for speech synthesis (e.g., "Ashley", "Hades"). - model: TTS model to use (e.g., "inworld-tts-1", "inworld-tts-1-max"). - temperature: Voice temperature control for synthesis variability (e.g., 0.8). - Valid range: [0, 2]. Higher values increase variability. - - Note: - Language is automatically inferred from the input text by Inworld's TTS models, - so no explicit language parameter is required. - """ - - voice_id: Optional[str] = "Ashley" # defaults to the Ashley voice - model: Optional[str] = "inworld-tts-1" # defaults to the inworld-tts-1 model - temperature: Optional[float] = None # optional temperature control (range: [0, 2]) - - def __init__( - self, - *, - api_key: str, - aiohttp_session: Optional[aiohttp.ClientSession] = None, - base_url: str = "https://api.inworld.ai/tts/v1/voice", # Non-streaming endpoint - sample_rate: Optional[int] = None, - encoding: str = "LINEAR16", - params: Optional[InputParams] = None, - **kwargs, - ): - """Initialize the Inworld non-streaming TTS service. - - Sets up the TTS service with Inworld AI's non-streaming API configuration. - This constructor prepares all necessary parameters for complete audio synthesis. + This method handles Inworld's streaming endpoint response format: + - JSON lines containing base64-encoded audio chunks + - Real-time processing as data arrives + - Line buffering to handle partial JSON data Args: - 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. - base_url: Base URL for Inworld non-streaming HTTP API. Uses non-streaming endpoint by default. - Should normally not be changed unless using a different environment. - sample_rate: Audio sample rate in Hz. If None, uses default from StartFrame. - Common values: 48000 (high quality), 24000 (good quality), 16000 (basic) - encoding: Audio encoding format. Supported options: - - "LINEAR16" (default) - Uncompressed PCM, best quality - - Other formats as supported by Inworld API - params: Input parameters for voice and model configuration. Use this to specify: - - voice_id: Voice selection ("Ashley", "Hades", etc.) - - model: TTS model ("inworld-tts-1", "inworld-tts-1-max", etc.) - - temperature: Voice temperature control for variability (range: [0, 2], e.g., 0.8, optional) - If None, uses default values (Ashley voice, inworld-tts-1 model). - Note: Language is automatically inferred from input text. - **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) - - # Use provided params or create default configuration - params = params or InworldHttpNonStreamingService.InputParams() - - # Store core configuration for API requests - self._api_key = api_key # Authentication credentials - self._session = aiohttp_session # HTTP session for requests (optional) - 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 the TTS request - # Note: Language is automatically inferred from text by Inworld's models - self._settings = { - "voiceId": params.voice_id or "Ashley", # Voice selection from params - "modelId": params.model or "inworld-tts-1", # TTS model selection from params - "audio_config": { # Audio format configuration - "audio_encoding": encoding, # Format: LINEAR16, MP3, etc. - "sample_rate_hertz": 0, # Will be set in start() from parent service - }, - } - - # Add optional temperature parameter if provided (valid range: [0, 2]) - if params.temperature is not None: - self._settings["temperature"] = params.temperature - - # Register voice and model with parent service for metrics and tracking - self.set_voice(params.voice_id or "Ashley") # Used for logging and metrics - self.set_model_name(params.model or "inworld-tts-1") # Used for performance tracking - - def can_generate_metrics(self) -> bool: - """Check if this service can generate processing metrics. - - Returns: - True, as Inworld non-streaming service supports metrics generation. - """ - return True - - async def start(self, frame: StartFrame): - """Start the Inworld non-streaming TTS service. - - Args: - frame: The start frame containing initialization parameters. - """ - await super().start(frame) - self._settings["audio_config"]["sample_rate_hertz"] = self.sample_rate - - async def stop(self, frame: EndFrame): - """Stop the Inworld non-streaming TTS service. - - Args: - frame: The end frame. - """ - await super().stop(frame) - - async def cancel(self, frame: CancelFrame): - """Cancel the Inworld non-streaming TTS service. - - Args: - frame: The cancel frame. - """ - await super().cancel(frame) - - @traced_tts - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - """Generate speech from text using Inworld's non-streaming HTTP API. - - This method sends text to Inworld's non-streaming TTS endpoint and receives - the complete audio file as a base64-encoded response. The audio is then - chunked and yielded for playback in the pipeline. - - Args: - text: The text to synthesize into speech. + response: The aiohttp response object from streaming endpoint. Yields: - Frame: Audio frames containing the synthesized speech, plus control frames. - - Raises: - ErrorFrame: If API errors occur or audio processing fails. + Frame: Audio frames as they're processed from the stream. """ - logger.debug(f"{self}: Generating TTS [{text}]") + # ================================================================================ + # STREAMING: PROCESS JSON LINE-BY-LINE 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 + + # ============================================================================ + # 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") + + # ============================================================================ + # 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() + + # Skip empty lines (common in streaming responses) + if not line_str: + continue + + try: + # ================================================================ + # 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"]: + # Process the audio chunk + async for frame in self._process_audio_chunk( + base64.b64decode(chunk_data["result"]["audioContent"]) + ): + yield frame + + except json.JSONDecodeError: + # Ignore malformed JSON lines - streaming can have partial data + # This is normal in HTTP streaming scenarios + continue + + async def _process_non_streaming_response( + self, response: aiohttp.ClientResponse + ) -> AsyncGenerator[Frame, None]: + """Process complete JSON response with full audio content. + + This method handles Inworld's non-streaming endpoint response format: + - Single JSON response with complete base64-encoded audio + - Full audio download then chunked playback + - Simpler processing without line buffering + + Args: + response: The aiohttp response object from non-streaming endpoint. + + Yields: + Frame: Audio frames chunked from the complete audio. + """ + # ================================================================================ + # NON-STREAMING: PARSE COMPLETE JSON RESPONSE + # ================================================================================ + # Parse the complete JSON response containing base64 audio data + response_data = await response.json() # ================================================================================ - # STEP 1: PREPARE API REQUEST + # EXTRACT AND VALIDATE AUDIO CONTENT # ================================================================================ - # Build the JSON payload according to Inworld's non-streaming API specification - # This matches the format shown in their documentation examples - # Note: Language is automatically inferred from the input text by Inworld's models - payload = { - "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 - } + # Extract the base64-encoded audio content from response + if "audioContent" not in response_data: + logger.error("No audioContent in Inworld API response") + await self.push_error(ErrorFrame("No audioContent in response")) + return - # Add optional temperature parameter if configured (valid range: [0, 2]) - if "temperature" in self._settings: - payload["temperature"] = self._settings["temperature"] + # ================================================================================ + # DECODE AND PROCESS COMPLETE AUDIO DATA + # ================================================================================ + # Decode the base64 audio data to binary + audio_data = base64.b64decode(response_data["audioContent"]) - # 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}", # Base64 API key from Inworld Portal - "Content-Type": "application/json", # JSON request body - } + # Strip WAV header if present (Inworld may include WAV header) + # This prevents audio clicks and ensures clean audio playback + if len(audio_data) > 44 and audio_data.startswith(b"RIFF"): + audio_data = audio_data[44:] - try: - # ================================================================================ - # STEP 2: INITIALIZE METRICS AND STREAMING - # ================================================================================ - # Start measuring Time To First Byte (TTFB) for performance tracking - await self.start_ttfb_metrics() + # ================================================================================ + # CHUNK AND YIELD COMPLETE AUDIO FOR PLAYBACK + # ================================================================================ + # Chunk the complete audio for streaming playback + # This allows the pipeline to process audio in manageable pieces + CHUNK_SIZE = self.chunk_size - # Signal to the pipeline that TTS generation has started - # This allows downstream processors to prepare for incoming audio - yield TTSStartedFrame() + for i in range(0, len(audio_data), CHUNK_SIZE): + chunk = audio_data[i : i + CHUNK_SIZE] + if len(chunk) > 0: + await self.stop_ttfb_metrics() + yield TTSAudioRawFrame( + audio=chunk, + sample_rate=self.sample_rate, + num_channels=1, + ) - # ================================================================================ - # STEP 3: MAKE HTTP NON-STREAMING REQUEST - # ================================================================================ - # Make single HTTP POST request to Inworld's non-streaming endpoint - # This endpoint returns complete audio as base64-encoded data - # Create session if none was provided - if self._session: - session = self._session - else: - session = aiohttp.ClientSession() + async def _process_audio_chunk(self, audio_chunk: bytes) -> AsyncGenerator[Frame, None]: + """Process a single audio chunk (common logic for both modes). - async with ( - session - if not self._session - else session.post( - self._base_url, json=payload, headers=headers - ) as context_or_response - ): - if self._session: - response = context_or_response - else: - async with context_or_response.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 + This method handles audio chunk processing that's common to both streaming + and non-streaming modes: + - WAV header removal + - Audio validation + - Frame creation and yielding - # ================================================================ - # STEP 5: PARSE COMPLETE JSON RESPONSE - # ================================================================ - # Parse the complete JSON response containing base64 audio data - response_data = await response.json() + Args: + audio_chunk: Raw audio data bytes to process. - # ================================================================ - # STEP 6: EXTRACT AND VALIDATE AUDIO CONTENT - # ================================================================ - # Extract the base64-encoded audio content from response - if "audioContent" not in response_data: - logger.error("No audioContent in Inworld API response") - await self.push_error(ErrorFrame("No audioContent in response")) - return + Yields: + Frame: Audio frame if chunk contains valid audio data. + """ + # ======================================================== + # 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: + return - # ================================================================ - # STEP 7: DECODE AND PROCESS AUDIO DATA - # ================================================================ - # Decode the base64 audio data to binary - audio_data = base64.b64decode(response_data["audioContent"]) + # Start with the raw audio data + audio_data = audio_chunk - # Strip WAV header if present (Inworld may include WAV header) - # This prevents audio clicks and ensures clean audio playback - if len(audio_data) > 44 and audio_data.startswith(b"RIFF"): - audio_data = audio_data[44:] + # ======================================================== + # 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:] - # ================================================================ - # STEP 8: START USAGE METRICS TRACKING - # ================================================================ - await self.start_tts_usage_metrics(text) - - # ================================================================ - # STEP 9: CHUNK AND YIELD AUDIO FOR PLAYBACK - # ================================================================ - # Chunk the complete audio for streaming playback - # This allows the pipeline to process audio in manageable pieces - CHUNK_SIZE = self.chunk_size - - for i in range(0, len(audio_data), CHUNK_SIZE): - chunk = audio_data[i : i + CHUNK_SIZE] - if len(chunk) > 0: - await self.stop_ttfb_metrics() - yield TTSAudioRawFrame( - audio=chunk, - sample_rate=self.sample_rate, - num_channels=1, - ) - - if self._session: - # Handle HTTP errors - 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 - - # Parse the complete JSON response - response_data = await response.json() - - # Extract the base64-encoded audio content - if "audioContent" not in response_data: - logger.error("No audioContent in Inworld API response") - await self.push_error(ErrorFrame("No audioContent in response")) - return - - # Decode the base64 audio data - audio_data = base64.b64decode(response_data["audioContent"]) - - # Strip WAV header if present (Inworld may include WAV header) - if len(audio_data) > 44 and audio_data.startswith(b"RIFF"): - audio_data = audio_data[44:] - - await self.start_tts_usage_metrics(text) - - # Chunk the complete audio for streaming playback - CHUNK_SIZE = self.chunk_size - - for i in range(0, len(audio_data), CHUNK_SIZE): - chunk = audio_data[i : i + CHUNK_SIZE] - if len(chunk) > 0: - await self.stop_ttfb_metrics() - yield TTSAudioRawFrame( - audio=chunk, - sample_rate=self.sample_rate, - num_channels=1, - ) - - except Exception as e: - # ================================================================================ - # STEP 10: 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 11: 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() + # ======================================================== + # 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 + )