Merge pull request #2979 from thsunkid/feature/whisper-stt-probability-metrics

Feat: Access prob metrics for Whisper STT services using include_prob_metrics

CHANGELOG.md

@@ -31,6 +31,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   you cancel a task with `PipelineTask.cancel(reason="cancellation your
 reason")`.
 
+- Added `include_prob_metrics` parameter to Whisper STT services to enable access
+  to probability metrics from transcription results.
+
+- Added utility functions `extract_whisper_probability()`,
+  `extract_openai_gpt4o_probability()`, and `extract_deepgram_probability()` to
+  extract probability metrics from `TranscriptionFrame` objects for Whisper-based,
+  OpenAI GPT-4o-transcribe, and Deepgram STT services respectively.
+
 ### Changed
 
 - Updated the default model for `GoogleVertexLLMService` to `gemini-2.5-flash`.

src/pipecat/services/groq/stt.py

@@ -58,7 +58,8 @@ class GroqSTTService(BaseWhisperSTTService):
         kwargs = {
             "file": ("audio.wav", audio, "audio/wav"),
             "model": self.model_name,
-            "response_format": "json",
+            # Use verbose_json to get probability metrics
+            "response_format": "verbose_json" if self._include_prob_metrics else "json",
             "language": self._language,
         }
 

src/pipecat/services/openai/stt.py

@@ -61,6 +61,15 @@ class OpenAISTTService(BaseWhisperSTTService):
             "language": self._language,
         }
 
+        if self._include_prob_metrics:
+            # GPT-4o-transcribe models only support logprobs (not verbose_json)
+            if self.model_name in ("gpt-4o-transcribe", "gpt-4o-mini-transcribe"):
+                kwargs["response_format"] = "json"
+                kwargs["include"] = ["logprobs"]
+            else:
+                # Whisper models support verbose_json
+                kwargs["response_format"] = "verbose_json"
+
         if self._prompt is not None:
             kwargs["prompt"] = self._prompt
 

src/pipecat/services/sambanova/stt.py

@@ -8,6 +8,8 @@
 
 from typing import Any, Optional
 
+from loguru import logger
+
 from pipecat.services.whisper.base_stt import BaseWhisperSTTService, Transcription
 from pipecat.transcriptions.language import Language
 
@@ -54,6 +56,14 @@ class SambaNovaSTTService(BaseWhisperSTTService):  # type: ignore
     async def _transcribe(self, audio: bytes) -> Transcription:
         assert self._language is not None  # Assigned in the BaseWhisperSTTService class
 
+        if self._include_prob_metrics:
+            # https://docs.sambanova.ai/docs/en/features/audio#request-parameters
+            logger.warning(
+                "SambaNova STT does not support probability metrics "
+                "(include_prob_metrics parameter has no effect). "
+                "Check their docs: https://docs.sambanova.ai/docs/en/features/audio#request-parameters for more details."
+            )
+
         # Build kwargs dict with only set parameters
         kwargs = {
             "file": ("audio.wav", audio, "audio/wav"),

src/pipecat/services/whisper/base_stt.py

@@ -122,6 +122,7 @@ class BaseWhisperSTTService(SegmentedSTTService):
         language: Optional[Language] = Language.EN,
         prompt: Optional[str] = None,
         temperature: Optional[float] = None,
+        include_prob_metrics: bool = False,
         **kwargs,
     ):
         """Initialize the Whisper STT service.
@@ -133,6 +134,9 @@ class BaseWhisperSTTService(SegmentedSTTService):
             language: Language of the audio input. Defaults to English.
             prompt: Optional text to guide the model's style or continue a previous segment.
             temperature: Sampling temperature between 0 and 1. Defaults to 0.0.
+            include_prob_metrics: If True, enables probability metrics in API response.
+                Each service implements this differently (see child classes).
+                Defaults to False.
             **kwargs: Additional arguments passed to SegmentedSTTService.
         """
         super().__init__(**kwargs)
@@ -141,6 +145,7 @@ class BaseWhisperSTTService(SegmentedSTTService):
         self._language = self.language_to_service_language(language or Language.EN)
         self._prompt = prompt
         self._temperature = temperature
+        self._include_prob_metrics = include_prob_metrics
 
         self._settings = {
             "base_url": base_url,
@@ -223,6 +228,7 @@ class BaseWhisperSTTService(SegmentedSTTService):
                     text,
                     self._user_id,
                     time_now_iso8601(),
+                    result=response,
                 )
             else:
                 logger.warning("Received empty transcription from API")

src/pipecat/services/whisper/utils.py

@@ -0,0 +1,135 @@
+#
+# Copyright (c) 2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Utility functions for extracting probability metrics from STT services."""
+
+import math
+from typing import Optional
+
+from pipecat.frames.frames import TranscriptionFrame
+
+
+def extract_whisper_probability(frame: TranscriptionFrame) -> Optional[float]:
+    """Extract probability from Whisper-based TranscriptionFrame result.
+
+    Works with Groq, OpenAI Whisper, or other Whisper-based services that use
+    verbose_json format with segments containing avg_logprob.
+
+    Converts avg_logprob to probability.
+
+    Args:
+        frame: TranscriptionFrame with result from GroqSTTService or OpenAISTTService
+            (when include_prob_metrics=True and using Whisper models).
+
+    Returns:
+        Probability (0-1) if available, None otherwise.
+
+    Example::
+
+        from pipecat.services.groq.stt import GroqSTTService
+        from pipecat.services.whisper.utils import extract_whisper_probability
+
+        stt = GroqSTTService(include_prob_metrics=True)
+        # ... use stt in pipeline ...
+        # In your frame processor:
+        if isinstance(frame, TranscriptionFrame):
+            prob = extract_whisper_probability(frame)
+            if prob:
+                print(f"Transcription confidence: {prob:.2%}")
+    """
+    if not frame.result:
+        return None
+
+    # Whisper verbose_json format: response.segments[0].avg_logprob
+    if hasattr(frame.result, "segments") and frame.result.segments:
+        segment = frame.result.segments[0]
+        avg_logprob = getattr(segment, "avg_logprob", None)
+        if avg_logprob is not None:
+            return math.exp(avg_logprob)
+
+    return None
+
+
+def extract_openai_gpt4o_probability(frame: TranscriptionFrame) -> Optional[float]:
+    """Extract probability from OpenAI GPT-4o-transcribe TranscriptionFrame result.
+
+    Args:
+        frame: TranscriptionFrame with result from OpenAISTTService
+            using GPT-4o-transcribe model (when include_prob_metrics=True).
+
+    Returns:
+        Probability (0-1) if available, None otherwise.
+
+    Example::
+
+        from pipecat.services.openai.stt import OpenAISTTService
+        from pipecat.services.whisper.utils import extract_openai_gpt4o_probability
+
+        stt = OpenAISTTService(model="gpt-4o-transcribe", include_prob_metrics=True)
+        # ... use stt in pipeline ...
+        # In your frame processor:
+        if isinstance(frame, TranscriptionFrame):
+            prob = extract_openai_gpt4o_probability(frame)
+            if prob:
+                print(f"Transcription confidence: {prob:.2%}")
+    """
+    if not frame.result:
+        return None
+
+    # OpenAI GPT-4o-transcribe format: response.logprobs
+    if hasattr(frame.result, "logprobs"):
+        logprobs = frame.result.logprobs
+        if logprobs:
+            # Calculate average logprob and convert to probability
+            avg_logprob = sum(logprobs) / len(logprobs)
+            return math.exp(avg_logprob)
+
+    return None
+
+
+def extract_deepgram_probability(frame: TranscriptionFrame) -> Optional[float]:
+    """Extract probability from Deepgram TranscriptionFrame result.
+
+    Args:
+        frame: TranscriptionFrame with result from DeepgramSTTService.
+
+    Returns:
+        Probability (0-1) if available, None otherwise.
+        Returns alternative-level confidence if available, otherwise calculates
+        average confidence from word-level confidences.
+
+    Example::
+
+        from pipecat.services.deepgram.stt import DeepgramSTTService
+        from pipecat.services.whisper.utils import extract_deepgram_probability
+
+        stt = DeepgramSTTService()
+        # ... use stt in pipeline ...
+        # In your frame processor:
+        if isinstance(frame, TranscriptionFrame):
+            prob = extract_deepgram_probability(frame)
+            if prob:
+                print(f"Transcription confidence: {prob:.2%}")
+    """
+    if not frame.result:
+        return None
+
+    result = frame.result
+    if hasattr(result, "channel") and result.channel:
+        if hasattr(result.channel, "alternatives") and result.channel.alternatives:
+            alt = result.channel.alternatives[0]
+            conf = getattr(alt, "confidence", None)
+            if conf is not None:
+                return float(conf)
+
+            words = getattr(alt, "words", None)
+            if words:
+                word_confs = [getattr(w, "confidence", None) for w in words]
+                word_confs = [c for c in word_confs if c is not None]
+                if word_confs:
+                    return float(sum(word_confs) / len(word_confs))
+
+    return None