Feat: allow accessing prob metrics for Whisper STT services with include_prob_metrics param

This commit is contained in:
Thu Nguyen
2025-11-05 13:24:49 +07:00
parent c20aa78648
commit 0b779a880b
4 changed files with 106 additions and 1 deletions

View File

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

View File

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

View File

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

View File

@@ -0,0 +1,85 @@
#
# Copyright (c) 2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
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_logprobs(frame: TranscriptionFrame) -> Optional[list]:
"""Extract logprobs 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:
List of logprobs if available, None otherwise.
Example:
>>> from pipecat.services.openai.stt import OpenAISTTService
>>> from pipecat.services.whisper.utils import extract_openai_gpt4o_logprobs
>>>
>>> stt = OpenAISTTService(model="gpt-4o-transcribe", include_prob_metrics=True)
>>> # ... use stt in pipeline ...
>>> # In your frame processor:
>>> if isinstance(frame, TranscriptionFrame):
>>> logprobs = extract_openai_gpt4o_logprobs(frame)
>>> if logprobs:
>>> # Calculate average logprob
>>> avg_logprob = sum(logprobs) / len(logprobs)
>>> prob = math.exp(avg_logprob)
>>> print(f"Transcription confidence: {prob:.2%}")
"""
if not frame.result:
return None
# OpenAI GPT-4o-transcribe format: response.logprobs
if hasattr(frame.result, "logprobs"):
return frame.result.logprobs
return None