From bcb019e8abd2abc5bd8c8b26459e3bba38ed2435 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 23 Jan 2026 18:47:34 -0500 Subject: [PATCH] Add TTFB metrics for STT services (#3495) --- changelog/3495.changed.2.md | 1 + changelog/3495.changed.md | 1 + src/pipecat/frames/frames.py | 3 + src/pipecat/services/assemblyai/stt.py | 3 +- src/pipecat/services/aws/stt.py | 2 - src/pipecat/services/azure/stt.py | 2 - src/pipecat/services/cartesia/stt.py | 9 +- src/pipecat/services/deepgram/flux/stt.py | 3 + src/pipecat/services/deepgram/stt.py | 17 +- .../services/deepgram/stt_sagemaker.py | 10 +- src/pipecat/services/elevenlabs/stt.py | 15 +- src/pipecat/services/fal/stt.py | 2 - src/pipecat/services/gladia/stt.py | 2 - src/pipecat/services/google/stt.py | 2 - src/pipecat/services/gradium/stt.py | 1 - src/pipecat/services/hathora/stt.py | 2 - src/pipecat/services/nvidia/stt.py | 4 - src/pipecat/services/sarvam/stt.py | 44 +++- src/pipecat/services/soniox/stt.py | 9 +- src/pipecat/services/stt_service.py | 234 ++++++++++++++++++ src/pipecat/services/whisper/base_stt.py | 2 - src/pipecat/services/whisper/stt.py | 4 - 22 files changed, 309 insertions(+), 63 deletions(-) create mode 100644 changelog/3495.changed.2.md create mode 100644 changelog/3495.changed.md diff --git a/changelog/3495.changed.2.md b/changelog/3495.changed.2.md new file mode 100644 index 000000000..cf1f526b8 --- /dev/null +++ b/changelog/3495.changed.2.md @@ -0,0 +1 @@ +- `SarvamSTTService` now defaults `vad_signals` and `high_vad_sensitivity` to `None` (omitted from connection parameters), improving latency by ~300ms compared to the previous defaults. diff --git a/changelog/3495.changed.md b/changelog/3495.changed.md new file mode 100644 index 000000000..c690ebc09 --- /dev/null +++ b/changelog/3495.changed.md @@ -0,0 +1 @@ +- Improved the STT TTFB (Time To First Byte) measurement, reporting the delay between when the user stops speaking and when the final transcription is received. Note: Unlike traditional TTFB which measures from a discrete request, STT services receive continuous audio input—so we measure from speech end to final transcript, which captures the latency that matters for voice AI applications. In support of this change, added `finalized` field to `TranscriptionFrame` to indicate when a transcript is the final result for an utterance. diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index a27bb780e..26f9af9d2 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -426,12 +426,15 @@ class TranscriptionFrame(TextFrame): timestamp: When the transcription occurred. language: Detected or specified language of the speech. result: Raw result from the STT service. + finalized: Whether this is the final transcription for an utterance. + Set by STT services that support commit/finalize signals. """ user_id: str timestamp: str language: Optional[Language] = None result: Optional[Any] = None + finalized: bool = False def __str__(self): return f"{self.name}(user: {self.user_id}, text: [{self.text}], language: {self.language}, timestamp: {self.timestamp})" diff --git a/src/pipecat/services/assemblyai/stt.py b/src/pipecat/services/assemblyai/stt.py index 3fde9491c..99b4f423d 100644 --- a/src/pipecat/services/assemblyai/stt.py +++ b/src/pipecat/services/assemblyai/stt.py @@ -161,7 +161,7 @@ class AssemblyAISTTService(WebsocketSTTService): """ await super().process_frame(frame, direction) if isinstance(frame, VADUserStartedSpeakingFrame): - await self.start_ttfb_metrics() + pass elif isinstance(frame, VADUserStoppedSpeakingFrame): if ( self._vad_force_turn_endpoint @@ -354,7 +354,6 @@ class AssemblyAISTTService(WebsocketSTTService): """Handle transcription results.""" if not message.transcript: return - await self.stop_ttfb_metrics() if message.end_of_turn and ( not self._connection_params.formatted_finals or message.turn_is_formatted ): diff --git a/src/pipecat/services/aws/stt.py b/src/pipecat/services/aws/stt.py index 5ae99b02f..8beefe051 100644 --- a/src/pipecat/services/aws/stt.py +++ b/src/pipecat/services/aws/stt.py @@ -158,7 +158,6 @@ class AWSTranscribeSTTService(WebsocketSTTService): await self._websocket.send(event_message) # Start metrics after first chunk sent await self.start_processing_metrics() - await self.start_ttfb_metrics() except Exception as e: yield ErrorFrame(error=f"Error sending audio: {e}") @@ -470,7 +469,6 @@ class AWSTranscribeSTTService(WebsocketSTTService): is_final = not result.get("IsPartial", True) if transcript: - await self.stop_ttfb_metrics() if is_final: await self.push_frame( TranscriptionFrame( diff --git a/src/pipecat/services/azure/stt.py b/src/pipecat/services/azure/stt.py index aaa1e2da4..d45eb71df 100644 --- a/src/pipecat/services/azure/stt.py +++ b/src/pipecat/services/azure/stt.py @@ -116,7 +116,6 @@ class AzureSTTService(STTService): """ try: await self.start_processing_metrics() - await self.start_ttfb_metrics() if self._audio_stream: self._audio_stream.write(audio) yield None @@ -191,7 +190,6 @@ class AzureSTTService(STTService): self, transcript: str, is_final: bool, language: Optional[Language] = None ): """Handle a transcription result with tracing.""" - await self.stop_ttfb_metrics() await self.stop_processing_metrics() def _on_handle_recognized(self, event): diff --git a/src/pipecat/services/cartesia/stt.py b/src/pipecat/services/cartesia/stt.py index 625df6366..2fb08b981 100644 --- a/src/pipecat/services/cartesia/stt.py +++ b/src/pipecat/services/cartesia/stt.py @@ -207,9 +207,8 @@ class CartesiaSTTService(WebsocketSTTService): await super().cancel(frame) await self._disconnect() - async def start_metrics(self): + async def _start_metrics(self): """Start performance metrics collection for transcription processing.""" - await self.start_ttfb_metrics() await self.start_processing_metrics() async def process_frame(self, frame: Frame, direction: FrameDirection): @@ -222,10 +221,13 @@ class CartesiaSTTService(WebsocketSTTService): await super().process_frame(frame, direction) if isinstance(frame, VADUserStartedSpeakingFrame): - await self.start_metrics() + # Reset finalize state for new utterance + self.set_finalize_pending(False) + await self._start_metrics() elif isinstance(frame, VADUserStoppedSpeakingFrame): # Send finalize command to flush the transcription session if self._websocket and self._websocket.state is State.OPEN: + self.set_finalize_pending(True) await self._websocket.send("finalize") async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: @@ -342,7 +344,6 @@ class CartesiaSTTService(WebsocketSTTService): pass if len(transcript) > 0: - await self.stop_ttfb_metrics() if is_final: await self.push_frame( TranscriptionFrame( diff --git a/src/pipecat/services/deepgram/flux/stt.py b/src/pipecat/services/deepgram/flux/stt.py index 045af9811..756827171 100644 --- a/src/pipecat/services/deepgram/flux/stt.py +++ b/src/pipecat/services/deepgram/flux/stt.py @@ -659,6 +659,8 @@ class DeepgramFluxSTTService(WebsocketSTTService): average_confidence = self._calculate_average_confidence(data) if not self._params.min_confidence or average_confidence > self._params.min_confidence: + # EndOfTurn means Flux has determined the turn is complete, + # so this TranscriptionFrame is always finalized await self.push_frame( TranscriptionFrame( transcript, @@ -666,6 +668,7 @@ class DeepgramFluxSTTService(WebsocketSTTService): time_now_iso8601(), self._language, result=data, + finalized=True, ) ) else: diff --git a/src/pipecat/services/deepgram/stt.py b/src/pipecat/services/deepgram/stt.py index 2469b25a3..20dfba724 100644 --- a/src/pipecat/services/deepgram/stt.py +++ b/src/pipecat/services/deepgram/stt.py @@ -276,9 +276,8 @@ class DeepgramSTTService(STTService): # GH issue: https://github.com/deepgram/deepgram-python-sdk/issues/570 await self._connection.finish() - async def start_metrics(self): - """Start TTFB and processing metrics collection.""" - await self.start_ttfb_metrics() + async def _start_metrics(self): + """Start processing metrics collection for this utterance.""" await self.start_processing_metrics() async def _on_error(self, *args, **kwargs): @@ -292,7 +291,7 @@ class DeepgramSTTService(STTService): await self._connect() async def _on_speech_started(self, *args, **kwargs): - await self.start_metrics() + await self._start_metrics() await self._call_event_handler("on_speech_started", *args, **kwargs) await self.broadcast_frame(UserStartedSpeakingFrame) if self._should_interrupt: @@ -320,8 +319,12 @@ class DeepgramSTTService(STTService): language = result.channel.alternatives[0].languages[0] language = Language(language) if len(transcript) > 0: - await self.stop_ttfb_metrics() if is_final: + # Check if this response is from a finalize() call. + # Only mark as finalized when both we requested it AND Deepgram confirms it. + from_finalize = getattr(result, "from_finalize", False) + if from_finalize: + self.confirm_finalize() await self.push_frame( TranscriptionFrame( transcript, @@ -356,8 +359,10 @@ class DeepgramSTTService(STTService): if isinstance(frame, VADUserStartedSpeakingFrame) and not self.vad_enabled: # Start metrics if Deepgram VAD is disabled & pipeline VAD has detected speech - await self.start_metrics() + await self._start_metrics() elif isinstance(frame, VADUserStoppedSpeakingFrame): # https://developers.deepgram.com/docs/finalize + # Mark that we're awaiting a from_finalize response + self.request_finalize() await self._connection.finalize() logger.trace(f"Triggered finalize event on: {frame.name=}, {direction=}") diff --git a/src/pipecat/services/deepgram/stt_sagemaker.py b/src/pipecat/services/deepgram/stt_sagemaker.py index 7eb6a072b..0f7e6f988 100644 --- a/src/pipecat/services/deepgram/stt_sagemaker.py +++ b/src/pipecat/services/deepgram/stt_sagemaker.py @@ -363,9 +363,6 @@ class DeepgramSageMakerSTTService(STTService): if not transcript.strip(): return - # Stop TTFB metrics on first transcript - await self.stop_ttfb_metrics() - is_final = parsed.get("is_final", False) speech_final = parsed.get("speech_final", False) @@ -417,9 +414,8 @@ class DeepgramSageMakerSTTService(STTService): """ pass - async def start_metrics(self): - """Start TTFB and processing metrics collection.""" - await self.start_ttfb_metrics() + async def _start_metrics(self): + """Start processing metrics collection.""" await self.start_processing_metrics() async def process_frame(self, frame: Frame, direction: FrameDirection): @@ -433,7 +429,7 @@ class DeepgramSageMakerSTTService(STTService): # Start metrics when user starts speaking (if VAD is not provided by Deepgram) if isinstance(frame, VADUserStartedSpeakingFrame): - await self.start_metrics() + await self._start_metrics() elif isinstance(frame, VADUserStoppedSpeakingFrame): # Send finalize message to Deepgram when user stops speaking # This tells Deepgram to flush any remaining audio and return final results diff --git a/src/pipecat/services/elevenlabs/stt.py b/src/pipecat/services/elevenlabs/stt.py index 8f9020aa7..a1be56a0d 100644 --- a/src/pipecat/services/elevenlabs/stt.py +++ b/src/pipecat/services/elevenlabs/stt.py @@ -310,7 +310,6 @@ class ElevenLabsSTTService(SegmentedSTTService): self, transcript: str, is_final: bool, language: Optional[str] = None ): """Handle a transcription result with tracing.""" - await self.stop_ttfb_metrics() await self.stop_processing_metrics() async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: @@ -328,7 +327,6 @@ class ElevenLabsSTTService(SegmentedSTTService): """ try: await self.start_processing_metrics() - await self.start_ttfb_metrics() # Upload audio and get transcription result directly result = await self._transcribe_audio(audio) @@ -539,9 +537,8 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService): await super().cancel(frame) await self._disconnect() - async def start_metrics(self): + async def _start_metrics(self): """Start performance metrics collection for transcription processing.""" - await self.start_ttfb_metrics() await self.start_processing_metrics() async def process_frame(self, frame: Frame, direction: FrameDirection): @@ -554,13 +551,17 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService): await super().process_frame(frame, direction) if isinstance(frame, VADUserStartedSpeakingFrame): + # Reset finalize state for new utterance + self.set_finalize_pending(False) # Start metrics when user starts speaking - await self.start_metrics() + await self._start_metrics() elif isinstance(frame, VADUserStoppedSpeakingFrame): # Send commit when user stops speaking (manual commit mode) if self._params.commit_strategy == CommitStrategy.MANUAL: if self._websocket and self._websocket.state is State.OPEN: try: + # Mark that the next committed transcript should be finalized + self.set_finalize_pending(True) commit_message = { "message_type": "input_audio_chunk", "audio_base_64": "", @@ -764,8 +765,6 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService): if not text: return - await self.stop_ttfb_metrics() - # Get language if provided language = data.get("language_code") @@ -803,7 +802,6 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService): if not text: return - await self.stop_ttfb_metrics() await self.stop_processing_metrics() # Get language if provided @@ -845,7 +843,6 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService): if not text: return - await self.stop_ttfb_metrics() await self.stop_processing_metrics() # Get language if provided diff --git a/src/pipecat/services/fal/stt.py b/src/pipecat/services/fal/stt.py index a71915743..114865778 100644 --- a/src/pipecat/services/fal/stt.py +++ b/src/pipecat/services/fal/stt.py @@ -249,7 +249,6 @@ class FalSTTService(SegmentedSTTService): self, transcript: str, is_final: bool, language: Optional[str] = None ): """Handle a transcription result with tracing.""" - await self.stop_ttfb_metrics() await self.stop_processing_metrics() async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: @@ -267,7 +266,6 @@ class FalSTTService(SegmentedSTTService): """ try: await self.start_processing_metrics() - await self.start_ttfb_metrics() # Send to Fal directly (audio is already in WAV format from base class) data_uri = fal_client.encode(audio, "audio/x-wav") diff --git a/src/pipecat/services/gladia/stt.py b/src/pipecat/services/gladia/stt.py index 8c5ec7aa4..e4229fe97 100644 --- a/src/pipecat/services/gladia/stt.py +++ b/src/pipecat/services/gladia/stt.py @@ -385,7 +385,6 @@ class GladiaSTTService(WebsocketSTTService): Yields: None (processing is handled asynchronously via WebSocket). """ - await self.start_ttfb_metrics() await self.start_processing_metrics() # Add audio to buffer @@ -513,7 +512,6 @@ class GladiaSTTService(WebsocketSTTService): async def _handle_transcription( self, transcript: str, is_final: bool, language: Optional[str] = None ): - await self.stop_ttfb_metrics() await self.stop_processing_metrics() async def _on_speech_started(self): diff --git a/src/pipecat/services/google/stt.py b/src/pipecat/services/google/stt.py index dcf0adf89..edb2ccec0 100644 --- a/src/pipecat/services/google/stt.py +++ b/src/pipecat/services/google/stt.py @@ -823,7 +823,6 @@ class GoogleSTTService(STTService): """ if self._streaming_task: # Queue the audio data - await self.start_ttfb_metrics() await self.start_processing_metrics() await self._request_queue.put(audio) yield None @@ -875,7 +874,6 @@ class GoogleSTTService(STTService): ) else: self._last_transcript_was_final = False - await self.stop_ttfb_metrics() await self.push_frame( InterimTranscriptionFrame( transcript, diff --git a/src/pipecat/services/gradium/stt.py b/src/pipecat/services/gradium/stt.py index b66b18070..1c378ff9f 100644 --- a/src/pipecat/services/gradium/stt.py +++ b/src/pipecat/services/gradium/stt.py @@ -122,7 +122,6 @@ class GradiumSTTService(WebsocketSTTService): None (processing handled via WebSocket messages). """ self._audio_buffer.extend(audio) - await self.start_ttfb_metrics() await self.start_processing_metrics() while len(self._audio_buffer) >= self._chunk_size_bytes: diff --git a/src/pipecat/services/hathora/stt.py b/src/pipecat/services/hathora/stt.py index 8aadb6b65..81e8f0ea6 100644 --- a/src/pipecat/services/hathora/stt.py +++ b/src/pipecat/services/hathora/stt.py @@ -111,7 +111,6 @@ class HathoraSTTService(SegmentedSTTService): """ try: await self.start_processing_metrics() - await self.start_ttfb_metrics() url = f"{self._base_url}" @@ -153,7 +152,6 @@ class HathoraSTTService(SegmentedSTTService): result=response, ) - await self.stop_ttfb_metrics() await self.stop_processing_metrics() except Exception as e: diff --git a/src/pipecat/services/nvidia/stt.py b/src/pipecat/services/nvidia/stt.py index 7d52f5130..3af3fe78c 100644 --- a/src/pipecat/services/nvidia/stt.py +++ b/src/pipecat/services/nvidia/stt.py @@ -307,7 +307,6 @@ class NvidiaSTTService(STTService): transcript = result.alternatives[0].transcript if transcript and len(transcript) > 0: - await self.stop_ttfb_metrics() if result.is_final: await self.stop_processing_metrics() await self.push_frame( @@ -344,7 +343,6 @@ class NvidiaSTTService(STTService): Yields: None - transcription results are pushed to the pipeline via frames. """ - await self.start_ttfb_metrics() await self.start_processing_metrics() await self._queue.put(audio) yield None @@ -598,12 +596,10 @@ class NvidiaSegmentedSTTService(SegmentedSTTService): assert self._config is not None, "Recognition config not created" await self.start_processing_metrics() - await self.start_ttfb_metrics() # Process audio with NVIDIA Riva ASR - explicitly request non-future response raw_response = self._asr_service.offline_recognize(audio, self._config, future=False) - await self.stop_ttfb_metrics() await self.stop_processing_metrics() # Process the response - handle different possible return types diff --git a/src/pipecat/services/sarvam/stt.py b/src/pipecat/services/sarvam/stt.py index f4e5c676b..96447dce9 100644 --- a/src/pipecat/services/sarvam/stt.py +++ b/src/pipecat/services/sarvam/stt.py @@ -15,9 +15,15 @@ from pipecat.frames.frames import ( CancelFrame, EndFrame, ErrorFrame, + Frame, StartFrame, TranscriptionFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, ) +from pipecat.processors.frame_processor import FrameDirection from pipecat.services.sarvam._sdk import sdk_headers from pipecat.services.stt_service import STTService from pipecat.transcriptions.language import Language, resolve_language @@ -75,14 +81,14 @@ class SarvamSTTService(STTService): language: Target language for transcription. Defaults to None (required for saarika models). prompt: Optional prompt to guide translation style/context for STT-Translate models. Only applicable to saaras (STT-Translate) models. Defaults to None. - vad_signals: Enable VAD signals in response. Defaults to True. - high_vad_sensitivity: Enable high VAD (Voice Activity Detection) sensitivity. Defaults to False. + vad_signals: Enable VAD signals in response. Defaults to None. + high_vad_sensitivity: Enable high VAD (Voice Activity Detection) sensitivity. Defaults to None. """ language: Optional[Language] = None prompt: Optional[str] = None - vad_signals: bool = True - high_vad_sensitivity: bool = False + vad_signals: bool = None + high_vad_sensitivity: bool = None def __init__( self, @@ -155,6 +161,7 @@ class SarvamSTTService(STTService): self._websocket_context = None self._socket_client = None self._receive_task = None + logger.info(f"Sarvam STT initialized with SDK headers: {self._sdk_headers}") def language_to_service_language(self, language: Language) -> str: """Convert pipecat Language enum to Sarvam's language code. @@ -175,6 +182,24 @@ class SarvamSTTService(STTService): """ return True + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process incoming frames. + + Handles VAD frames for TTFB tracking when using Pipecat's VAD + instead of Sarvam's built-in VAD. + """ + await super().process_frame(frame, direction) + + # Only handle VAD frames when not using Sarvam's VAD signals + if not self._vad_signals: + if isinstance(frame, VADUserStartedSpeakingFrame): + self.set_finalize_pending(False) + await self._start_metrics() + elif isinstance(frame, VADUserStoppedSpeakingFrame): + if self._socket_client: + self.set_finalize_pending(True) + await self._socket_client.flush() + async def set_language(self, language: Language): """Set the recognition language and reconnect. @@ -411,16 +436,18 @@ class SarvamSTTService(STTService): logger.debug(f"VAD Signal: {signal}, Occurred at: {timestamp}") if signal == "START_SPEECH": - await self.start_metrics() + await self._start_metrics() logger.debug("User started speaking") await self._call_event_handler("on_speech_started") + await self.broadcast_frame(UserStartedSpeakingFrame) + await self.push_interruption_task_frame_and_wait() elif signal == "END_SPEECH": logger.debug("User stopped speaking") await self._call_event_handler("on_speech_stopped") + await self.broadcast_frame(UserStoppedSpeakingFrame) elif message.type == "data": - await self.stop_ttfb_metrics() transcript = message.data.transcript language_code = message.data.language_code # Prefer language from message (auto-detected for translate models). Fallback to configured. @@ -482,7 +509,6 @@ class SarvamSTTService(STTService): } return mapping.get(language_code, Language.HI_IN) - async def start_metrics(self): - """Start TTFB and processing metrics collection.""" - await self.start_ttfb_metrics() + async def _start_metrics(self): + """Start processing metrics collection.""" await self.start_processing_metrics() diff --git a/src/pipecat/services/soniox/stt.py b/src/pipecat/services/soniox/stt.py index 476ae0762..6f8c92d26 100644 --- a/src/pipecat/services/soniox/stt.py +++ b/src/pipecat/services/soniox/stt.py @@ -21,7 +21,7 @@ from pipecat.frames.frames import ( InterimTranscriptionFrame, StartFrame, TranscriptionFrame, - UserStoppedSpeakingFrame, + VADUserStoppedSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.stt_service import WebsocketSTTService @@ -162,7 +162,7 @@ class SonioxSTTService(WebsocketSTTService): sample_rate: Audio sample rate. params: Additional configuration parameters, such as language hints, context and speaker diarization. - vad_force_turn_endpoint: Listen to `UserStoppedSpeakingFrame` to send finalize message to Soniox. If disabled, Soniox will detect the end of the speech. + vad_force_turn_endpoint: Listen to `VADUserStoppedSpeakingFrame` to send finalize message to Soniox. If disabled, Soniox will detect the end of the speech. **kwargs: Additional arguments passed to the STTService. """ super().__init__(sample_rate=sample_rate, **kwargs) @@ -247,7 +247,7 @@ class SonioxSTTService(WebsocketSTTService): """ await super().process_frame(frame, direction) - if isinstance(frame, UserStoppedSpeakingFrame) and self._vad_force_turn_endpoint: + if isinstance(frame, VADUserStoppedSpeakingFrame) and self._vad_force_turn_endpoint: # Send finalize message to Soniox so we get the final tokens asap. if self._websocket and self._websocket.state is State.OPEN: await self._websocket.send(FINALIZE_MESSAGE) @@ -374,12 +374,15 @@ class SonioxSTTService(WebsocketSTTService): async def send_endpoint_transcript(): if self._final_transcription_buffer: text = "".join(map(lambda token: token["text"], self._final_transcription_buffer)) + # Soniox only pushes TranscriptionFrame when an end token is received, + # so every TranscriptionFrame is inherently finalized await self.push_frame( TranscriptionFrame( text=text, user_id=self._user_id, timestamp=time_now_iso8601(), result=self._final_transcription_buffer, + finalized=True, ) ) await self._handle_transcription(text, is_final=True) diff --git a/src/pipecat/services/stt_service.py b/src/pipecat/services/stt_service.py index 46c40ae70..24b1aafcf 100644 --- a/src/pipecat/services/stt_service.py +++ b/src/pipecat/services/stt_service.py @@ -6,7 +6,9 @@ """Base classes for Speech-to-Text services with continuous and segmented processing.""" +import asyncio import io +import time import wave from abc import abstractmethod from typing import Any, AsyncGenerator, Dict, Mapping, Optional @@ -17,12 +19,17 @@ from pipecat.frames.frames import ( AudioRawFrame, ErrorFrame, Frame, + InterruptionFrame, + MetricsFrame, + SpeechControlParamsFrame, StartFrame, STTMuteFrame, STTUpdateSettingsFrame, + TranscriptionFrame, VADUserStartedSpeakingFrame, VADUserStoppedSpeakingFrame, ) +from pipecat.metrics.metrics import TTFBMetricsData from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_service import AIService from pipecat.services.websocket_service import WebsocketService @@ -61,6 +68,8 @@ class STTService(AIService): audio_passthrough=True, # STT input sample rate sample_rate: Optional[int] = None, + # STT TTFB timeout - time to wait after VAD stop before reporting TTFB + stt_ttfb_timeout: float = 2.0, **kwargs, ): """Initialize the STT service. @@ -70,6 +79,12 @@ class STTService(AIService): Defaults to True. sample_rate: The sample rate for audio input. If None, will be determined from the start frame. + stt_ttfb_timeout: Time in seconds to wait after VAD stop before reporting + TTFB. This delay allows the final transcription to arrive. Defaults to 2.0. + Note: STT "TTFB" differs from traditional TTFB (which measures from a discrete + request to first response byte). Since STT receives continuous audio, we measure + from when the user stops speaking to when the final transcript arrives—capturing + the latency that matters for voice AI applications. **kwargs: Additional arguments passed to the parent AIService. """ super().__init__(**kwargs) @@ -81,6 +96,16 @@ class STTService(AIService): self._muted: bool = False self._user_id: str = "" + # STT TTFB tracking state + self._stt_ttfb_timeout = stt_ttfb_timeout + self._ttfb_timeout_task: Optional[asyncio.Task] = None + self._vad_stop_secs: Optional[float] = None + self._speech_end_time: Optional[float] = None + self._user_speaking: bool = False + self._last_transcription_time: Optional[float] = None + self._finalize_pending: bool = False + self._finalize_requested: bool = False + self._register_event_handler("on_connected") self._register_event_handler("on_disconnected") self._register_event_handler("on_connection_error") @@ -94,6 +119,44 @@ class STTService(AIService): """ return self._muted + def set_finalize_pending(self, value: bool): + """Set whether the next TranscriptionFrame should be marked as finalized. + + When True, the next TranscriptionFrame pushed will have its `finalized` + field set to True, and this flag will automatically reset to False. + This is used to signal that a transcript is the final result for an + utterance, enabling immediate TTFB reporting. + + Args: + value: True to mark the next transcription as finalized. + """ + self._finalize_pending = value + + def request_finalize(self): + """Mark that a finalize request has been sent, awaiting server confirmation. + + For providers that require server confirmation before marking transcripts + as finalized (e.g., Deepgram's from_finalize field), call this when sending + the finalize request. Then call confirm_finalize() when the server confirms. + + This is an alternative to set_finalize_pending() for providers that need + two-step finalization. + """ + self._finalize_requested = True + + def confirm_finalize(self): + """Confirm that the server has acknowledged the finalize request. + + Call this when the server response confirms finalization (e.g., Deepgram's + from_finalize=True). The next TranscriptionFrame pushed will be marked + as finalized. + + Only has effect if request_finalize() was previously called. + """ + if self._finalize_requested: + self._finalize_pending = True + self._finalize_requested = False + @property def sample_rate(self) -> int: """Get the current sample rate for audio processing. @@ -144,6 +207,11 @@ class STTService(AIService): self._sample_rate = self._init_sample_rate or frame.audio_in_sample_rate self._tracing_enabled = frame.enable_tracing + async def cleanup(self): + """Clean up STT service resources.""" + await super().cleanup() + await self._cancel_ttfb_timeout() + async def _update_settings(self, settings: Mapping[str, Any]): logger.info(f"Updating STT settings: {self._settings}") for key, value in settings.items(): @@ -206,14 +274,166 @@ class STTService(AIService): await self.process_audio_frame(frame, direction) if self._audio_passthrough: await self.push_frame(frame, direction) + elif isinstance(frame, SpeechControlParamsFrame): + await self._handle_speech_control_params(frame) + await self.push_frame(frame, direction) + elif isinstance(frame, VADUserStartedSpeakingFrame): + await self._handle_vad_user_started_speaking(frame) + await self.push_frame(frame, direction) + elif isinstance(frame, VADUserStoppedSpeakingFrame): + await self._handle_vad_user_stopped_speaking(frame) + await self.push_frame(frame, direction) elif isinstance(frame, STTUpdateSettingsFrame): await self._update_settings(frame.settings) elif isinstance(frame, STTMuteFrame): self._muted = frame.mute logger.debug(f"STT service {'muted' if frame.mute else 'unmuted'}") + elif isinstance(frame, InterruptionFrame): + await self._reset_stt_ttfb_state() + await self.push_frame(frame, direction) else: await self.push_frame(frame, direction) + async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): + """Push a frame downstream, tracking TranscriptionFrame timestamps for TTFB. + + Stores the timestamp of each TranscriptionFrame for TTFB calculation. + If the frame is marked as finalized (either directly or via set_finalize_pending), + reports TTFB immediately and cancels any pending timeout. Otherwise, TTFB is + reported after a timeout. + + Args: + frame: The frame to push. + direction: The direction to push the frame. + """ + if isinstance(frame, TranscriptionFrame): + # Store the transcription time for TTFB calculation + self._last_transcription_time = time.time() + + # Set finalized from pending state and auto-reset + if self._finalize_pending: + frame.finalized = True + self._finalize_pending = False + + # If this is a finalized transcription, report TTFB immediately + if frame.finalized and self._speech_end_time is not None: + ttfb = self._last_transcription_time - self._speech_end_time + await self._emit_stt_ttfb_metric(ttfb) + # Cancel the timeout since we've already reported + await self._cancel_ttfb_timeout() + # Clear state + self._speech_end_time = None + self._last_transcription_time = None + + await super().push_frame(frame, direction) + + async def _handle_speech_control_params(self, frame: SpeechControlParamsFrame): + """Handle speech control parameters frame to extract VAD stop_secs. + + Args: + frame: The speech control parameters frame. + """ + if frame.vad_params is not None: + self._vad_stop_secs = frame.vad_params.stop_secs + + async def _cancel_ttfb_timeout(self): + """Cancel any pending TTFB timeout task.""" + if self._ttfb_timeout_task: + await self.cancel_task(self._ttfb_timeout_task) + self._ttfb_timeout_task = None + + async def _reset_stt_ttfb_state(self): + """Reset STT TTFB measurement state. + + Called when starting a new utterance or on interruption to ensure + we don't use stale state for TTFB calculations. This specifically guards + against the case where a TranscriptionFrame is received without corresponding + VADUserStartedSpeakingFrame and VADUserStoppedSpeakingFrame frames. + + Note: Does not reset _user_speaking since InterruptionFrame can arrive + while user is still speaking. + """ + await self._cancel_ttfb_timeout() + self._speech_end_time = None + self._last_transcription_time = None + + async def _handle_vad_user_started_speaking(self, frame: VADUserStartedSpeakingFrame): + """Handle VAD user started speaking frame to start tracking transcriptions. + + Cancels any pending TTFB timeout, resets TTFB tracking state, and marks user as speaking. + + Args: + frame: The VAD user started speaking frame. + """ + await self._reset_stt_ttfb_state() + self._user_speaking = True + self._finalize_requested = False + + async def _handle_vad_user_stopped_speaking(self, frame: VADUserStoppedSpeakingFrame): + """Handle VAD user stopped speaking frame. + + Calculates the actual speech end time and starts a timeout task to wait + for the final transcription before reporting TTFB. + + Args: + frame: The VAD user stopped speaking frame. + """ + self._user_speaking = False + + # Skip TTFB measurement if we don't have VAD params + if self._vad_stop_secs is None: + return + + # Calculate the actual speech end time (current time minus VAD stop delay). + # This approximates when the last user audio was sent to the STT service, + # which we use to measure against the eventual transcription response. + self._speech_end_time = time.time() - self._vad_stop_secs + + # Start timeout task (any previous timeout was cancelled by VADUserStartedSpeakingFrame + # or InterruptionFrame) + self._ttfb_timeout_task = self.create_task( + self._ttfb_timeout_handler(), name="stt_ttfb_timeout" + ) + + async def _ttfb_timeout_handler(self): + """Wait for timeout then report TTFB using the last transcription timestamp. + + This timeout allows the final transcription to arrive before we calculate + and report TTFB. If no transcription arrived, no TTFB is reported. + """ + try: + await asyncio.sleep(self._stt_ttfb_timeout) + + # Report TTFB if we have both speech end time and transcription time + if self._speech_end_time is not None and self._last_transcription_time is not None: + ttfb = self._last_transcription_time - self._speech_end_time + await self._emit_stt_ttfb_metric(ttfb) + + # Clear state after reporting + self._speech_end_time = None + self._last_transcription_time = None + except asyncio.CancelledError: + # Task was cancelled (new utterance or interruption), which is expected behavior + pass + finally: + self._ttfb_timeout_task = None + + async def _emit_stt_ttfb_metric(self, ttfb: float): + """Emit STT TTFB metric if value is non-negative. + + Args: + ttfb: The TTFB value in seconds. + """ + if ttfb >= 0: + logger.debug(f"{self} TTFB: {ttfb:.3f}s") + if self.metrics_enabled: + ttfb_data = TTFBMetricsData( + processor=self.name, + model=self.model_name, + value=ttfb, + ) + await super().push_frame(MetricsFrame(data=[ttfb_data])) + class SegmentedSTTService(STTService): """STT service that processes speech in segments using VAD events. @@ -250,6 +470,20 @@ class SegmentedSTTService(STTService): await super().start(frame) self._audio_buffer_size_1s = self.sample_rate * 2 + async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): + """Push a frame, marking TranscriptionFrames as finalized. + + Segmented STT services process complete speech segments and return a single + TranscriptionFrame per segment, so every transcription is inherently finalized. + + Args: + frame: The frame to push. + direction: The direction of frame flow in the pipeline. + """ + if isinstance(frame, TranscriptionFrame): + frame.finalized = True + await super().push_frame(frame, direction) + async def process_frame(self, frame: Frame, direction: FrameDirection): """Process frames, handling VAD events and audio segmentation.""" await super().process_frame(frame, direction) diff --git a/src/pipecat/services/whisper/base_stt.py b/src/pipecat/services/whisper/base_stt.py index 32e08de8e..79d2723f1 100644 --- a/src/pipecat/services/whisper/base_stt.py +++ b/src/pipecat/services/whisper/base_stt.py @@ -204,11 +204,9 @@ class BaseWhisperSTTService(SegmentedSTTService): """ try: await self.start_processing_metrics() - await self.start_ttfb_metrics() response = await self._transcribe(audio) - await self.stop_ttfb_metrics() await self.stop_processing_metrics() text = response.text.strip() diff --git a/src/pipecat/services/whisper/stt.py b/src/pipecat/services/whisper/stt.py index 3865578e1..f11978cc2 100644 --- a/src/pipecat/services/whisper/stt.py +++ b/src/pipecat/services/whisper/stt.py @@ -289,7 +289,6 @@ class WhisperSTTService(SegmentedSTTService): return await self.start_processing_metrics() - await self.start_ttfb_metrics() # Divide by 32768 because we have signed 16-bit data. audio_float = np.frombuffer(audio, dtype=np.int16).astype(np.float32) / 32768.0 @@ -303,7 +302,6 @@ class WhisperSTTService(SegmentedSTTService): if segment.no_speech_prob < self._no_speech_prob: text += f"{segment.text} " - await self.stop_ttfb_metrics() await self.stop_processing_metrics() if text: @@ -388,7 +386,6 @@ class WhisperSTTServiceMLX(WhisperSTTService): import mlx_whisper await self.start_processing_metrics() - await self.start_ttfb_metrics() # Divide by 32768 because we have signed 16-bit data. audio_float = np.frombuffer(audio, dtype=np.int16).astype(np.float32) / 32768.0 @@ -413,7 +410,6 @@ class WhisperSTTServiceMLX(WhisperSTTService): if len(text.strip()) == 0: text = None - await self.stop_ttfb_metrics() await self.stop_processing_metrics() if text: