From dfe5fec8f964ec21dd9dd543154c06b18cd6e22c Mon Sep 17 00:00:00 2001 From: Ian Lee Date: Thu, 12 Mar 2026 14:42:56 -0700 Subject: [PATCH 1/2] [inworld] prewarm context on llm response start --- changelog/4013.changed.md | 1 + src/pipecat/services/inworld/tts.py | 37 +++++++++++++++++++++++++++++ 2 files changed, 38 insertions(+) create mode 100644 changelog/4013.changed.md diff --git a/changelog/4013.changed.md b/changelog/4013.changed.md new file mode 100644 index 000000000..4f8a37680 --- /dev/null +++ b/changelog/4013.changed.md @@ -0,0 +1 @@ +- Added context prewarming path for `InworldTTSService` to improve first audio latency \ No newline at end of file diff --git a/src/pipecat/services/inworld/tts.py b/src/pipecat/services/inworld/tts.py index 334c3c617..f6d0bf466 100644 --- a/src/pipecat/services/inworld/tts.py +++ b/src/pipecat/services/inworld/tts.py @@ -56,6 +56,7 @@ from pipecat.frames.frames import ( ErrorFrame, Frame, InterruptionFrame, + LLMFullResponseStartFrame, StartFrame, TTSAudioRawFrame, TTSStartedFrame, @@ -653,6 +654,11 @@ class InworldTTSService(WebsocketTTSService): # Track the end time of the last word in the current generation self._generation_end_time = 0.0 + # Context ID that was pre-opened on the server during process_frame + # (LLMFullResponseStartFrame) to avoid context creation latency when + # enough context for TTS is available. + self._prewarmed_context_id: Optional[str] = None + # Init-only config (not runtime-updatable). self._audio_encoding = encoding self._audio_sample_rate = 0 # Set in start() @@ -726,6 +732,29 @@ class InworldTTSService(WebsocketTTSService): if isinstance(frame, TTSStoppedFrame): await self.add_word_timestamps([("Reset", 0)]) + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process incoming frames and pre-open context on LLM response start. + + Eagerly sends the context configuration to the server when + LLMFullResponseStartFrame arrives, so the context is ready by the time + enough context for TTS is available. The base class assigns ``_turn_context_id`` before + this runs, which is reused for all ``run_tts`` calls within the turn. + """ + await super().process_frame(frame, direction) + + if isinstance(frame, LLMFullResponseStartFrame): + if self._prewarmed_context_id: + try: + await self._send_close_context(self._prewarmed_context_id) + except Exception as e: + logger.warning(f"{self}: Failed to close previous prewarmed context: {e}") + self._prewarmed_context_id = None + try: + await self._send_context(self._turn_context_id) + self._prewarmed_context_id = self._turn_context_id + except Exception as e: + logger.warning(f"{self}: Failed to pre-open context: {e}") + def _calculate_word_times(self, timestamp_info: Dict[str, Any]) -> List[Tuple[str, float]]: """Calculate word timestamps from Inworld WebSocket API response. @@ -887,6 +916,7 @@ class InworldTTSService(WebsocketTTSService): finally: await self.remove_active_audio_context() self._websocket = None + self._prewarmed_context_id = None self._cumulative_time = 0.0 self._generation_end_time = 0.0 await self._call_event_handler("on_disconnected") @@ -1001,9 +1031,16 @@ class InworldTTSService(WebsocketTTSService): async def _send_context(self, context_id: str): """Send a context to the Inworld WebSocket TTS service. + Skips the send if this context was already pre-opened on the server + (prewarmed during process_frame). + Args: context_id: The context ID. """ + if context_id == self._prewarmed_context_id: + self._prewarmed_context_id = None + return + audio_config = { "audioEncoding": self._audio_encoding, "sampleRateHertz": self._audio_sample_rate, From e9f3086ea39c359abdb839721323f7901b296625 Mon Sep 17 00:00:00 2001 From: Ian Lee Date: Fri, 20 Mar 2026 10:33:25 -0700 Subject: [PATCH 2/2] add on_turn_context_created hook instead --- src/pipecat/services/inworld/tts.py | 57 +++++++++++------------------ src/pipecat/services/tts_service.py | 13 +++++++ 2 files changed, 35 insertions(+), 35 deletions(-) diff --git a/src/pipecat/services/inworld/tts.py b/src/pipecat/services/inworld/tts.py index f6d0bf466..857aaa276 100644 --- a/src/pipecat/services/inworld/tts.py +++ b/src/pipecat/services/inworld/tts.py @@ -56,7 +56,6 @@ from pipecat.frames.frames import ( ErrorFrame, Frame, InterruptionFrame, - LLMFullResponseStartFrame, StartFrame, TTSAudioRawFrame, TTSStartedFrame, @@ -654,10 +653,10 @@ class InworldTTSService(WebsocketTTSService): # Track the end time of the last word in the current generation self._generation_end_time = 0.0 - # Context ID that was pre-opened on the server during process_frame - # (LLMFullResponseStartFrame) to avoid context creation latency when - # enough context for TTS is available. - self._prewarmed_context_id: Optional[str] = None + # Context IDs already sent to the server via _send_context, used to + # make _send_context idempotent so on_turn_context_created can eagerly + # open contexts without causing duplicate creates in run_tts. + self._sent_context_ids: set[str] = set() # Init-only config (not runtime-updatable). self._audio_encoding = encoding @@ -732,28 +731,16 @@ class InworldTTSService(WebsocketTTSService): if isinstance(frame, TTSStoppedFrame): await self.add_word_timestamps([("Reset", 0)]) - async def process_frame(self, frame: Frame, direction: FrameDirection): - """Process incoming frames and pre-open context on LLM response start. + async def on_turn_context_created(self, context_id: str): + """Eagerly open the context on the server when a new turn starts. - Eagerly sends the context configuration to the server when - LLMFullResponseStartFrame arrives, so the context is ready by the time - enough context for TTS is available. The base class assigns ``_turn_context_id`` before - this runs, which is reused for all ``run_tts`` calls within the turn. + This overlaps server-side context creation with sentence aggregation + time, so the context is ready by the time text arrives in run_tts. """ - await super().process_frame(frame, direction) - - if isinstance(frame, LLMFullResponseStartFrame): - if self._prewarmed_context_id: - try: - await self._send_close_context(self._prewarmed_context_id) - except Exception as e: - logger.warning(f"{self}: Failed to close previous prewarmed context: {e}") - self._prewarmed_context_id = None - try: - await self._send_context(self._turn_context_id) - self._prewarmed_context_id = self._turn_context_id - except Exception as e: - logger.warning(f"{self}: Failed to pre-open context: {e}") + try: + await self._send_context(context_id) + except Exception as e: + logger.warning(f"{self}: Failed to pre-open context: {e}") def _calculate_word_times(self, timestamp_info: Dict[str, Any]) -> List[Tuple[str, float]]: """Calculate word timestamps from Inworld WebSocket API response. @@ -800,6 +787,7 @@ class InworldTTSService(WebsocketTTSService): await self._send_close_context(context_id) except Exception as e: await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) + self._sent_context_ids.discard(context_id) self._cumulative_time = 0.0 self._generation_end_time = 0.0 @@ -916,7 +904,7 @@ class InworldTTSService(WebsocketTTSService): finally: await self.remove_active_audio_context() self._websocket = None - self._prewarmed_context_id = None + self._sent_context_ids.clear() self._cumulative_time = 0.0 self._generation_end_time = 0.0 await self._call_event_handler("on_disconnected") @@ -961,8 +949,11 @@ class InworldTTSService(WebsocketTTSService): await self.push_error(error_msg=str(msg["error"])) continue + # Handle context created confirmation + if "contextCreated" in result: + logger.trace(f"{self}: Context created on server: {ctx_id}") # If the context isn't available recreate it (handles race conditions during interruption recovery). - if ctx_id and not self.audio_context_available(ctx_id): + elif ctx_id and not self.audio_context_available(ctx_id): logger.trace(f"{self}: Recreating audio context for current context: {ctx_id}") await self.create_audio_context(ctx_id) @@ -987,10 +978,6 @@ class InworldTTSService(WebsocketTTSService): if word_times: await self.add_word_timestamps(word_times, ctx_id) - # Handle context created confirmation - if "contextCreated" in result: - logger.trace(f"{self}: Context created on server: {ctx_id}") - # Handle flush completion, which indicates the end of a generation if "flushCompleted" in result: logger.trace( @@ -1031,15 +1018,15 @@ class InworldTTSService(WebsocketTTSService): async def _send_context(self, context_id: str): """Send a context to the Inworld WebSocket TTS service. - Skips the send if this context was already pre-opened on the server - (prewarmed during process_frame). + Idempotent: skips the send if this context was already opened on the + server (e.g., eagerly via on_turn_context_created). Args: context_id: The context ID. """ - if context_id == self._prewarmed_context_id: - self._prewarmed_context_id = None + if context_id in self._sent_context_ids: return + self._sent_context_ids.add(context_id) audio_config = { "audioEncoding": self._audio_encoding, diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index da1bcaf87..bc499586a 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -633,6 +633,18 @@ class TTSService(AIService): await self.queue_frame(TTSSpeakFrame(text)) + async def on_turn_context_created(self, context_id: str): + """Called when a new turn context ID has been created. + + Override to perform provider-specific setup (e.g., eagerly opening a + server-side context) before text starts flowing. This is called from + ``process_frame`` when an ``LLMFullResponseStartFrame`` arrives. + + Args: + context_id: The newly created turn context ID. + """ + pass + async def on_turn_context_completed(self): """Handle the completion of a turn.""" # For HTTP services they emit the frames synchronously, so close the audio context here @@ -685,6 +697,7 @@ class TTSService(AIService): self._llm_response_started = True # New LLM turn → assign a fresh context ID shared by all sentences self._turn_context_id = self.create_context_id() + await self.on_turn_context_created(self._turn_context_id) await self.push_frame(frame, direction) elif isinstance(frame, (LLMFullResponseEndFrame, EndFrame)): # We pause processing incoming frames if the LLM response included