wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add TTFB metrics for STT services (#3495)

Browse Source
This commit is contained in:
Mark Backman
2026-01-23 18:47:34 -05:00
committed by GitHub
parent 4ea546785f
commit bcb019e8ab
22 changed files with 309 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/pipecat/frames/frames.py
View File

@@ -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})"

3
src/pipecat/services/assemblyai/stt.py
View File

@@ -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
):

2
src/pipecat/services/aws/stt.py
View File

@@ -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(

2
src/pipecat/services/azure/stt.py
View File

@@ -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):

9
src/pipecat/services/cartesia/stt.py
View File

@@ -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(

3
src/pipecat/services/deepgram/flux/stt.py
View File

@@ -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:

17
src/pipecat/services/deepgram/stt.py
View File

@@ -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=}")

10
src/pipecat/services/deepgram/stt_sagemaker.py
View File

@@ -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

15
src/pipecat/services/elevenlabs/stt.py
View File

@@ -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

2
src/pipecat/services/fal/stt.py
View File

@@ -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")

2
src/pipecat/services/gladia/stt.py
View File

@@ -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):

2
src/pipecat/services/google/stt.py
View File

@@ -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,

1
src/pipecat/services/gradium/stt.py
View File

@@ -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:

2
src/pipecat/services/hathora/stt.py
View File

@@ -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:

4
src/pipecat/services/nvidia/stt.py
View File

@@ -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

44
src/pipecat/services/sarvam/stt.py
View File

@@ -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()

9
src/pipecat/services/soniox/stt.py
View File

@@ -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)

234
src/pipecat/services/stt_service.py
View File

@@ -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)

2
src/pipecat/services/whisper/base_stt.py
View File

@@ -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()

4
src/pipecat/services/whisper/stt.py
View File

@@ -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: