From b9da241b7149981a9c50b718d3278f46a951e6e4 Mon Sep 17 00:00:00 2001 From: angelos-p Date: Fri, 17 Apr 2026 15:29:18 +0100 Subject: [PATCH 1/3] [Ground Truth + fixes] Adding Scribe v2 as an additional ground truth option and some ElevenLabs fixes --- README.md | 21 +- src/stt_benchmark/cli/ground_truth.py | 61 +++-- src/stt_benchmark/cli/main.py | 2 +- src/stt_benchmark/cli/wer.py | 64 +++-- src/stt_benchmark/evaluation/semantic_wer.py | 25 +- src/stt_benchmark/ground_truth/__init__.py | 3 +- .../ground_truth/gemini_transcriber.py | 5 +- .../ground_truth/scribe_transcriber.py | 194 ++++++++++++++++ src/stt_benchmark/storage/database.py | 219 +++++++++++++----- 9 files changed, 484 insertions(+), 110 deletions(-) create mode 100644 src/stt_benchmark/ground_truth/scribe_transcriber.py diff --git a/README.md b/README.md index 60ab22f..571916d 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,8 @@ A framework for benchmarking Speech-to-Text services with TTFS (Time To Final Segment) latency and Semantic WER (Word Error Rate) accuracy measurement. +> **Note:** Ground truth transcriptions are generated using batch transcription (Gemini Flash or ElevenLabs Scribe v2), not manual human transcription. The reference text may contain inaccuracies that affect WER scores. Results should be interpreted as relative comparisons between services rather than absolute accuracy metrics. + ## Results Summary Benchmark results on 1000 samples from the `pipecat-ai/smart-turn-data-v3.1-train` dataset. @@ -65,8 +67,9 @@ uv run stt-benchmark download --num-samples 100 # Run benchmarks uv run stt-benchmark run --services deepgram,openai -# Generate ground truth (Gemini) -uv run stt-benchmark ground-truth +# Generate ground truth +uv run stt-benchmark ground-truth # Gemini (default) +uv run stt-benchmark ground-truth -p scribe # ElevenLabs Scribe v2 # Calculate semantic WER (Claude) uv run stt-benchmark wer @@ -166,9 +169,14 @@ uv run stt-benchmark run --services deepgram --limit 50 --vad-stop-secs 0.3 ### Generating Ground Truth ```bash -# Generate ground truth for all samples +# Generate ground truth using Gemini (default) uv run stt-benchmark ground-truth +# Generate ground truth using ElevenLabs Scribe v2 +uv run stt-benchmark ground-truth -p scribe + +# Both providers can coexist — they are stored independently per model + # Interactive review with audio playback uv run stt-benchmark ground-truth review ``` @@ -181,6 +189,9 @@ uv run stt-benchmark wer # Force recalculate uv run stt-benchmark wer --services deepgram --force-recalculate + +# Evaluate against a specific ground truth model and store separately +uv run stt-benchmark wer -s elevenlabs -m v2 --gt-model scribe_v2 --wer-label v2-scribe-gt ``` ### Viewing Reports @@ -215,7 +226,7 @@ stt_benchmark_data/ |-------|-------------| | `samples` | Audio sample metadata | | `benchmark_results` | TTFS and transcription results | -| `ground_truths` | Reference transcriptions (Gemini) | +| `ground_truth` | Reference transcriptions (keyed by sample + model) | | `wer_metrics` | Semantic WER calculations | | `semantic_wer_traces` | Full Claude reasoning traces | @@ -246,7 +257,7 @@ The benchmark dataset (audio samples and ground truth transcriptions) is publicl **[pipecat-ai/stt-benchmark-data](https://huggingface.co/datasets/pipecat-ai/stt-benchmark-data)** -Audio samples are sourced from the `pipecat-ai/smart-turn-data-v3.1-train` dataset. Ground truth transcriptions are generated with Gemini and human-reviewed. +Audio samples are sourced from the `pipecat-ai/smart-turn-data-v3.1-train` dataset. Ground truth transcriptions are generated via batch transcription using Gemini or ElevenLabs Scribe v2. ## Documentation diff --git a/src/stt_benchmark/cli/ground_truth.py b/src/stt_benchmark/cli/ground_truth.py index 00a05b8..a04ec3b 100644 --- a/src/stt_benchmark/cli/ground_truth.py +++ b/src/stt_benchmark/cli/ground_truth.py @@ -9,6 +9,7 @@ from stt_benchmark.config import get_config from stt_benchmark.ground_truth.gemini_transcriber import GeminiTranscriber +from stt_benchmark.ground_truth.scribe_transcriber import ScribeTranscriber from stt_benchmark.storage.database import Database app = typer.Typer() @@ -24,11 +25,17 @@ def generate_ground_truth( "-n", help="Limit number of samples to transcribe", ), - model: str = typer.Option( - "gemini-3-flash-preview", + provider: str = typer.Option( + "gemini", + "--provider", + "-p", + help="Transcription provider: 'gemini' or 'scribe'", + ), + model: str | None = typer.Option( + None, "--model", "-m", - help="Gemini model to use for transcription", + help="Model to use (default: gemini-3-flash-preview for gemini, scribe_v2 for scribe)", ), force: bool = typer.Option( False, @@ -37,32 +44,45 @@ def generate_ground_truth( help="Re-transcribe samples that already have ground truth", ), ): - """Generate ground truth transcriptions using Gemini. + """Generate ground truth transcriptions using Gemini or ElevenLabs Scribe. - Uses Google's Gemini model to transcribe audio samples, - creating reference transcriptions for WER calculation. + Uses Google's Gemini model or ElevenLabs Scribe v2 to transcribe audio + samples, creating reference transcriptions for WER calculation. - Requires GOOGLE_API_KEY environment variable. + Requires GOOGLE_API_KEY (for gemini) or ELEVENLABS_API_KEY (for scribe). Subcommands: iterate - Run a repeatable transcription iteration (saves to JSONL) list - List available transcription runs review - Interactive review of transcription runs with audio playback """ - # If a subcommand was invoked, don't run the default behavior if ctx.invoked_subcommand is not None: return config = get_config() + provider = provider.lower() - if not config.google_api_key: - console.print("[red]Error: GOOGLE_API_KEY not set[/red]") - console.print("\nSet the environment variable:") - console.print(" export GOOGLE_API_KEY='your-api-key'") + if provider == "scribe": + if not config.elevenlabs_api_key: + console.print("[red]Error: ELEVENLABS_API_KEY not set[/red]") + console.print("\nSet the environment variable:") + console.print(" export ELEVENLABS_API_KEY='your-api-key'") + raise typer.Exit(1) + effective_model = model or "scribe_v2" + elif provider == "gemini": + if not config.google_api_key: + console.print("[red]Error: GOOGLE_API_KEY not set[/red]") + console.print("\nSet the environment variable:") + console.print(" export GOOGLE_API_KEY='your-api-key'") + raise typer.Exit(1) + effective_model = model or "gemini-3-flash-preview" + else: + console.print(f"[red]Error: Unknown provider '{provider}'. Use 'gemini' or 'scribe'.[/red]") raise typer.Exit(1) console.print("\n[bold blue]STT Benchmark - Generate Ground Truth[/bold blue]\n") - console.print(f"Model: {model}") + console.print(f"Provider: {provider}") + console.print(f"Model: {effective_model}") if limit: console.print(f"Limit: {limit}") console.print(f"Force re-transcribe: {force}") @@ -71,11 +91,10 @@ async def run(): db = Database() await db.initialize() - # Get samples if force: samples = await db.get_all_samples() else: - samples = await db.get_samples_without_ground_truth() + samples = await db.get_samples_without_ground_truth(model_used=effective_model) if not samples: console.print("\n[green]All samples already have ground truth![/green]") @@ -86,8 +105,10 @@ async def run(): console.print(f"Samples to transcribe: {len(samples)}\n") - # Create transcriber - transcriber = GeminiTranscriber(model_name=model) + if provider == "scribe": + transcriber = ScribeTranscriber(model_name=effective_model) + else: + transcriber = GeminiTranscriber(model_name=effective_model) with Progress( TextColumn("[progress.description]{task.description}"), @@ -110,13 +131,11 @@ def callback(current, total, sample_id): progress.update(task, completed=len(samples)) - # Summary console.print(f"\n[green]✓ Generated {len(results)} ground truth transcriptions[/green]") - # Show stats - gt_count = await db.get_ground_truth_count() + gt_count = await db.get_ground_truth_count(model_used=effective_model) sample_count = await db.get_sample_count() - console.print(f"\nGround truth coverage: {gt_count}/{sample_count} samples") + console.print(f"\nGround truth coverage ({effective_model}): {gt_count}/{sample_count} samples") await db.close() diff --git a/src/stt_benchmark/cli/main.py b/src/stt_benchmark/cli/main.py index 4479ef8..00c52a9 100644 --- a/src/stt_benchmark/cli/main.py +++ b/src/stt_benchmark/cli/main.py @@ -18,7 +18,7 @@ # Add subcommands app.add_typer(download_app, name="download", help="Download and prepare audio samples") app.add_typer(benchmark_app, name="run", help="Run STT benchmarks") -app.add_typer(ground_truth_app, name="ground-truth", help="Generate ground truth using Gemini") +app.add_typer(ground_truth_app, name="ground-truth", help="Generate ground truth transcriptions") app.add_typer(wer_app, name="wer", help="Calculate semantic WER metrics") app.add_typer(report_app, name="report", help="Generate reports and compare services") app.add_typer(export_app, name="export", help="Export data for a specific service") diff --git a/src/stt_benchmark/cli/wer.py b/src/stt_benchmark/cli/wer.py index 1d22576..53080f2 100644 --- a/src/stt_benchmark/cli/wer.py +++ b/src/stt_benchmark/cli/wer.py @@ -45,7 +45,18 @@ def calculate_wer( None, "--model", "-m", - help="Model name filter", + help="Model name filter (for the STT service result)", + ), + gt_model: str | None = typer.Option( + None, + "--gt-model", + help="Ground truth model to evaluate against (e.g. 'scribe_v2', 'gemini-3-flash-preview')", + ), + wer_label: str | None = typer.Option( + None, + "--wer-label", + help="Label for WER results (stored as model_name in wer_metrics). " + "Use to avoid overwriting existing WER results.", ), force_recalculate: bool = typer.Option( False, @@ -103,6 +114,10 @@ def calculate_wer( console.print(f"Services: {', '.join(s.value for s in service_list)}") if model: console.print(f"Model filter: {model}") + if gt_model: + console.print(f"Ground truth model: {gt_model}") + if wer_label: + console.print(f"WER label: {wer_label}") if force_recalculate: console.print("[yellow]Force recalculate: ON[/yellow]") @@ -142,24 +157,31 @@ async def run(): all_stats = [] + # When wer_label is set but model is not, default to empty string + # so we only process the default model results, not all models + effective_model = model if model is not None else ("" if wer_label else None) + effective_wer_label = wer_label or effective_model + for service_name in service_list: console.print(f"\n[bold]Evaluating semantic WER for {service_name.value}...[/bold]") # Delete existing WER metrics if force recalculate if force_recalculate: - await db.delete_wer_metrics_for_service(service_name, model) - await db.delete_semantic_wer_traces_for_service(service_name, model) + await db.delete_wer_metrics_for_service(service_name, effective_wer_label) + await db.delete_semantic_wer_traces_for_service(service_name, effective_wer_label) console.print(" [yellow]Deleted existing WER metrics and traces[/yellow]") # Get samples that need WER calculation - pending = await db.get_samples_without_wer(service_name, model) + pending = await db.get_samples_without_wer( + service_name, effective_model, wer_label=effective_wer_label + ) if not pending: console.print(" All samples already have WER metrics") # Still show existing stats - metrics = await db.get_wer_metrics_for_service(service_name, model) + metrics = await db.get_wer_metrics_for_service(service_name, effective_wer_label) if metrics: - stats = compute_wer_stats(service_name, metrics) + stats = compute_wer_stats(service_name, metrics, effective_wer_label) all_stats.append(stats) console.print(f" Mean WER: {stats['wer_mean']:.2%}") continue @@ -180,16 +202,18 @@ def callback(current, total, sample_id, task_id=progress_task): metrics = await evaluator.evaluate_service( service_name, - model_name=model, + model_name=effective_model, + gt_model=gt_model, + wer_label=wer_label, progress_callback=callback, ) progress.update(progress_task, completed=len(pending)) # Get all metrics for stats - all_metrics = await db.get_wer_metrics_for_service(service_name, model) + all_metrics = await db.get_wer_metrics_for_service(service_name, effective_wer_label) if all_metrics: - stats = compute_wer_stats(service_name, all_metrics) + stats = compute_wer_stats(service_name, all_metrics, effective_wer_label) all_stats.append(stats) console.print(f" [green]Completed: {len(metrics)} samples[/green]") console.print(f" Mean Semantic WER: {stats['wer_mean']:.2%}") @@ -205,17 +229,19 @@ def callback(current, total, sample_id, task_id=progress_task): asyncio.run(run()) -def compute_wer_stats(service_name: ServiceName, metrics: list) -> dict: +def compute_wer_stats( + service_name: ServiceName, metrics: list, model_name: str | None = None +) -> dict: """Compute aggregate semantic WER statistics.""" wer_values = [m.wer for m in metrics if m.wer < float("inf")] - # Compute pooled WER total_errors = sum(m.substitutions + m.deletions + m.insertions for m in metrics) total_ref_words = sum(m.reference_words for m in metrics) pooled_wer = total_errors / total_ref_words if total_ref_words > 0 else 0.0 return { "service_name": service_name, + "model_name": model_name or "", "num_samples": len(metrics), "wer_mean": statistics.mean(wer_values) if wer_values else 0.0, "wer_median": statistics.median(wer_values) if wer_values else 0.0, @@ -226,11 +252,21 @@ def compute_wer_stats(service_name: ServiceName, metrics: list) -> dict: } +def _format_service_label(stats: dict) -> str: + """Format service name with model for display.""" + name = stats["service_name"].value + model = stats.get("model_name", "") + if model: + return f"{name} ({model})" + return name + + def print_wer_summary(stats_list: list[dict]): """Print semantic WER summary table.""" table = Table(title="Semantic WER Summary") table.add_column("Service", style="cyan", no_wrap=True) + table.add_column("Model", style="dim") table.add_column("Samples", justify="right") table.add_column("WER Mean", justify="right") table.add_column("WER Median", justify="right") @@ -238,12 +274,12 @@ def print_wer_summary(stats_list: list[dict]): table.add_column("WER Max", justify="right") table.add_column("Pooled WER", justify="right") - # Sort by mean WER sorted_stats = sorted(stats_list, key=lambda x: x["wer_mean"]) for stats in sorted_stats: table.add_row( stats["service_name"].value, + stats.get("model_name", ""), str(stats["num_samples"]), f"{stats['wer_mean']:.2%}", f"{stats['wer_median']:.2%}", @@ -254,8 +290,8 @@ def print_wer_summary(stats_list: list[dict]): console.print(table) - # Rankings console.print("\n[bold]Rankings (by mean semantic WER, lower is better):[/bold]") for i, stats in enumerate(sorted_stats, 1): + label = _format_service_label(stats) medal = "🥇" if i == 1 else "🥈" if i == 2 else "🥉" if i == 3 else f"{i}." - console.print(f" {medal} {stats['service_name'].value}: {stats['wer_mean']:.2%}") + console.print(f" {medal} {label}: {stats['wer_mean']:.2%}") diff --git a/src/stt_benchmark/evaluation/semantic_wer.py b/src/stt_benchmark/evaluation/semantic_wer.py index f4a5983..e624b62 100644 --- a/src/stt_benchmark/evaluation/semantic_wer.py +++ b/src/stt_benchmark/evaluation/semantic_wer.py @@ -435,9 +435,9 @@ def _calculate_wer( ) -> dict: """Programmatic WER calculation - the only non-LLM logic.""" if reference_words == 0: - wer = 0.0 if (substitutions + deletions + insertions) == 0 else float("inf") + wer = 0.0 if (substitutions + deletions + insertions) == 0 else 1.0 else: - wer = (substitutions + deletions + insertions) / reference_words + wer = min((substitutions + deletions + insertions) / reference_words, 1.0) return { "wer": wer, @@ -800,6 +800,8 @@ async def evaluate_service( self, service_name: ServiceName, model_name: str | None = None, + gt_model: str | None = None, + wer_label: str | None = None, progress_callback: Callable | None = None, ) -> list[WERMetrics]: """Evaluate all transcriptions for a service. @@ -809,7 +811,12 @@ async def evaluate_service( Args: service_name: Service to evaluate - model_name: Optional model name filter + model_name: Optional model name filter (for the STT service result) + gt_model: Optional ground truth model to use (e.g. 'scribe_v2'). + If None, uses the default GT selection (human-verified first). + wer_label: Optional label for storing WER results. When set, WER + metrics are stored with this as model_name instead of the + source result's model_name. Allows multiple WER runs per service. progress_callback: Optional callback(current, total, sample_id) Returns: @@ -817,8 +824,12 @@ async def evaluate_service( """ await self.db.initialize() + effective_wer_label = wer_label or model_name + # Get samples that need WER calculation - samples = await self.db.get_samples_without_wer(service_name, model_name) + samples = await self.db.get_samples_without_wer( + service_name, model_name, wer_label=effective_wer_label + ) if not samples: logger.info(f"All samples already have WER metrics for {service_name.value}") return [] @@ -832,7 +843,7 @@ async def _eval_sample(sample): async with self._api_semaphore: # Get result and ground truth result, gt = await self.db.get_result_with_ground_truth( - sample.sample_id, service_name, model_name + sample.sample_id, service_name, model_name, gt_model=gt_model ) if not result or not result.transcription: @@ -857,7 +868,7 @@ async def _eval_sample(sample): # Update trace with sample info trace.sample_id = sample.sample_id trace.service_name = service_name - trace.model_name = model_name + trace.model_name = effective_wer_label # Store the trace await self.db.insert_semantic_wer_trace(trace) @@ -866,7 +877,7 @@ async def _eval_sample(sample): metrics = WERMetrics( sample_id=sample.sample_id, service_name=service_name, - model_name=model_name, + model_name=effective_wer_label, wer=eval_result["wer"], substitutions=eval_result["substitutions"], deletions=eval_result["deletions"], diff --git a/src/stt_benchmark/ground_truth/__init__.py b/src/stt_benchmark/ground_truth/__init__.py index c3f9b41..8a53325 100644 --- a/src/stt_benchmark/ground_truth/__init__.py +++ b/src/stt_benchmark/ground_truth/__init__.py @@ -1,5 +1,6 @@ """Ground truth generation for WER calculation.""" from stt_benchmark.ground_truth.gemini_transcriber import GeminiTranscriber +from stt_benchmark.ground_truth.scribe_transcriber import ScribeTranscriber -__all__ = ["GeminiTranscriber"] +__all__ = ["GeminiTranscriber", "ScribeTranscriber"] diff --git a/src/stt_benchmark/ground_truth/gemini_transcriber.py b/src/stt_benchmark/ground_truth/gemini_transcriber.py index b89005a..b0dbb74 100644 --- a/src/stt_benchmark/ground_truth/gemini_transcriber.py +++ b/src/stt_benchmark/ground_truth/gemini_transcriber.py @@ -183,9 +183,10 @@ async def transcribe_batch( if progress_callback: progress_callback(i, len(samples), sample.sample_id) - # Check if already transcribed (unless force is True) if not force: - existing = await self.db.get_ground_truth(sample.sample_id) + existing = await self.db.get_ground_truth( + sample.sample_id, model_used=self.model_name + ) if existing: logger.debug(f"Sample {sample.sample_id} already transcribed, skipping") results.append(existing) diff --git a/src/stt_benchmark/ground_truth/scribe_transcriber.py b/src/stt_benchmark/ground_truth/scribe_transcriber.py new file mode 100644 index 0000000..c6f06cc --- /dev/null +++ b/src/stt_benchmark/ground_truth/scribe_transcriber.py @@ -0,0 +1,194 @@ +"""ElevenLabs Scribe v2 transcription for ground truth generation. + +Uses the ElevenLabs Speech-to-Text HTTP API (batch mode) to transcribe audio samples. +""" + +import asyncio +import time +from collections.abc import Callable +from datetime import datetime, timezone +from pathlib import Path + +import aiohttp +from loguru import logger + +from stt_benchmark.config import BenchmarkConfig, get_config +from stt_benchmark.ground_truth.gemini_transcriber import pcm_to_wav +from stt_benchmark.models import AudioSample, GroundTruth +from stt_benchmark.storage.database import Database + +SCRIBE_API_URL = "https://api.elevenlabs.io/v1/speech-to-text" + + +class ScribeTranscriber: + """Generates ground truth transcriptions using ElevenLabs Scribe v2.""" + + def __init__( + self, + model_name: str = "scribe_v2", + config: BenchmarkConfig | None = None, + ): + self.config = config or get_config() + self.model_name = model_name + self.db = Database() + + if not self.config.elevenlabs_api_key: + raise ValueError("ELEVENLABS_API_KEY not set in environment") + + self.api_key = self.config.elevenlabs_api_key + + # Rate limiting + self.requests_per_minute = 60 + self.request_times: list[float] = [] + + async def _rate_limit(self) -> None: + """Enforce rate limiting.""" + now = time.time() + self.request_times = [t for t in self.request_times if now - t < 60] + + if len(self.request_times) >= self.requests_per_minute: + sleep_time = 60 - (now - self.request_times[0]) + if sleep_time > 0: + logger.debug(f"Rate limiting: sleeping {sleep_time:.1f}s") + await asyncio.sleep(sleep_time) + + self.request_times.append(time.time()) + + async def _transcribe(self, wav_bytes: bytes, session: aiohttp.ClientSession) -> str | None: + """Send audio to Scribe API and return transcription text.""" + data = aiohttp.FormData() + data.add_field( + "file", + wav_bytes, + filename="audio.wav", + content_type="audio/wav", + ) + data.add_field("model_id", self.model_name) + data.add_field("language_code", "eng") + data.add_field("tag_audio_events", "false") + data.add_field("timestamps_granularity", "none") + + headers = {"xi-api-key": self.api_key} + + async with session.post(SCRIBE_API_URL, data=data, headers=headers) as response: + if response.status != 200: + error_text = await response.text() + raise RuntimeError(f"status={response.status} {error_text}") + + result = await response.json() + return result.get("text", "").strip() or None + + async def transcribe_sample( + self, sample: AudioSample, session: aiohttp.ClientSession | None = None + ) -> GroundTruth | None: + """Transcribe a single audio sample. + + Args: + sample: AudioSample to transcribe + session: Optional aiohttp session to reuse + + Returns: + GroundTruth if successful, None if failed + """ + await self._rate_limit() + + try: + audio_path = Path(sample.audio_path) + if not audio_path.exists(): + logger.error(f"Audio file not found: {audio_path}") + return None + + pcm_bytes = audio_path.read_bytes() + wav_bytes = pcm_to_wav(pcm_bytes, sample_rate=16000, channels=1) + + if session: + transcription = await self._transcribe(wav_bytes, session) + else: + async with aiohttp.ClientSession() as new_session: + transcription = await self._transcribe(wav_bytes, new_session) + + if not transcription: + logger.warning(f"Empty response for sample {sample.sample_id}") + return None + + return GroundTruth( + sample_id=sample.sample_id, + text=transcription, + model_used=self.model_name, + generated_at=datetime.now(timezone.utc), + ) + + except Exception as e: + logger.error(f"Error transcribing sample {sample.sample_id}: {e}") + return None + + async def transcribe_batch( + self, + samples: list[AudioSample], + progress_callback: Callable | None = None, + save_incrementally: bool = True, + force: bool = False, + ) -> list[GroundTruth]: + """Transcribe a batch of samples. + + Args: + samples: List of AudioSample to transcribe + progress_callback: Optional callback(current, total, sample_id) + save_incrementally: Save each result to DB as it completes + force: Re-transcribe even if ground truth already exists + + Returns: + List of GroundTruth objects + """ + await self.db.initialize() + results = [] + + async with aiohttp.ClientSession() as session: + for i, sample in enumerate(samples): + if progress_callback: + progress_callback(i, len(samples), sample.sample_id) + + if not force: + existing = await self.db.get_ground_truth( + sample.sample_id, model_used=self.model_name + ) + if existing: + logger.debug(f"Sample {sample.sample_id} already transcribed, skipping") + results.append(existing) + continue + + gt = await self.transcribe_sample(sample, session=session) + if gt: + results.append(gt) + if save_incrementally: + await self.db.insert_ground_truth(gt) + text_preview = gt.text[:50] + "..." if len(gt.text) > 50 else gt.text + logger.info(f"[{i+1}/{len(samples)}] Transcribed: {text_preview}") + else: + logger.warning( + f"[{i+1}/{len(samples)}] Failed to transcribe sample {sample.sample_id}" + ) + + return results + + async def generate_all_ground_truth( + self, + progress_callback: Callable | None = None, + ) -> list[GroundTruth]: + """Generate ground truth for all samples that don't have it yet. + + Args: + progress_callback: Optional progress callback + + Returns: + List of newly generated GroundTruth objects + """ + await self.db.initialize() + + samples = await self.db.get_samples_without_ground_truth(model_used=self.model_name) + if not samples: + logger.info("All samples already have ground truth") + return [] + + logger.info(f"Generating ground truth for {len(samples)} samples") + return await self.transcribe_batch(samples, progress_callback=progress_callback) diff --git a/src/stt_benchmark/storage/database.py b/src/stt_benchmark/storage/database.py index cc6c5c6..c5c9f8f 100644 --- a/src/stt_benchmark/storage/database.py +++ b/src/stt_benchmark/storage/database.py @@ -78,15 +78,16 @@ async def _create_tables(self) -> None: config_snapshot TEXT ); - -- Ground truth table + -- Ground truth table (keyed by sample + model to allow multiple providers) CREATE TABLE IF NOT EXISTS ground_truth ( - sample_id TEXT PRIMARY KEY REFERENCES samples(sample_id), + sample_id TEXT NOT NULL REFERENCES samples(sample_id), text TEXT NOT NULL, model_used TEXT NOT NULL DEFAULT 'gemini-3-flash-preview', generated_at TEXT NOT NULL, verified_by TEXT, verified_at TEXT, - original_text TEXT + original_text TEXT, + PRIMARY KEY (sample_id, model_used) ); -- Semantic WER metrics table @@ -166,6 +167,37 @@ async def _run_migrations(self) -> None: except Exception: pass # Column already exists + # Migration: Change ground_truth PK from sample_id to (sample_id, model_used) + try: + cursor = await self._conn.execute( + "SELECT sql FROM sqlite_master WHERE type='table' AND name='ground_truth'" + ) + row = await cursor.fetchone() + if row and "PRIMARY KEY (sample_id, model_used)" not in row[0]: + await self._conn.executescript( + """ + CREATE TABLE ground_truth_new ( + sample_id TEXT NOT NULL REFERENCES samples(sample_id), + text TEXT NOT NULL, + model_used TEXT NOT NULL DEFAULT 'gemini-3-flash-preview', + generated_at TEXT NOT NULL, + verified_by TEXT, + verified_at TEXT, + original_text TEXT, + PRIMARY KEY (sample_id, model_used) + ); + INSERT INTO ground_truth_new + SELECT sample_id, text, model_used, generated_at, + verified_by, verified_at, original_text + FROM ground_truth; + DROP TABLE ground_truth; + ALTER TABLE ground_truth_new RENAME TO ground_truth; + """ + ) + logger.info("Migrated ground_truth table to composite PK (sample_id, model_used)") + except Exception as e: + logger.debug(f"ground_truth PK migration skipped: {e}") + await self._conn.commit() async def close(self) -> None: @@ -475,6 +507,7 @@ async def update_ground_truth_text( sample_id: str, new_text: str, verified_by: str = "human", + model_used: str | None = None, ) -> bool: """Update ground truth text with human correction. @@ -484,23 +517,23 @@ async def update_ground_truth_text( sample_id: Sample to update new_text: The corrected transcription verified_by: Who made the correction + model_used: If provided, update GT for this specific model. + If None, updates the best-match GT (human-verified first, then most recent). Returns: True if updated, False if sample not found """ - # Get existing ground truth - existing = await self.get_ground_truth(sample_id) + existing = await self.get_ground_truth(sample_id, model_used=model_used) if not existing: return False - # Store original text if this is the first correction original_text = existing.original_text or existing.text await self._conn.execute( """ UPDATE ground_truth SET text = ?, verified_by = ?, verified_at = ?, original_text = ? - WHERE sample_id = ? + WHERE sample_id = ? AND model_used = ? """, ( new_text, @@ -508,16 +541,34 @@ async def update_ground_truth_text( datetime.now(timezone.utc).isoformat(), original_text, sample_id, + existing.model_used, ), ) await self._conn.commit() return True - async def get_ground_truth(self, sample_id: str) -> GroundTruth | None: - """Get ground truth for a sample.""" - cursor = await self._conn.execute( - "SELECT * FROM ground_truth WHERE sample_id = ?", (sample_id,) - ) + async def get_ground_truth( + self, sample_id: str, model_used: str | None = None + ) -> GroundTruth | None: + """Get ground truth for a sample. + + Args: + sample_id: The sample to look up. + model_used: If provided, get GT for this specific model. + If None, prefer human-verified, then most recent. + """ + if model_used: + cursor = await self._conn.execute( + "SELECT * FROM ground_truth WHERE sample_id = ? AND model_used = ?", + (sample_id, model_used), + ) + else: + cursor = await self._conn.execute( + """SELECT * FROM ground_truth WHERE sample_id = ? + ORDER BY verified_by IS NOT NULL DESC, generated_at DESC + LIMIT 1""", + (sample_id,), + ) row = await cursor.fetchone() if row: return GroundTruth( @@ -533,16 +584,35 @@ async def get_ground_truth(self, sample_id: str) -> GroundTruth | None: ) return None - async def get_samples_without_ground_truth(self) -> list[AudioSample]: - """Get samples that don't have ground truth.""" - cursor = await self._conn.execute( - """ - SELECT s.* FROM samples s - LEFT JOIN ground_truth gt ON s.sample_id = gt.sample_id - WHERE gt.sample_id IS NULL - ORDER BY s.dataset_index - """ - ) + async def get_samples_without_ground_truth( + self, model_used: str | None = None + ) -> list[AudioSample]: + """Get samples that don't have ground truth. + + Args: + model_used: If provided, get samples missing GT for this specific model. + If None, get samples with no GT at all. + """ + if model_used: + cursor = await self._conn.execute( + """ + SELECT s.* FROM samples s + LEFT JOIN ground_truth gt ON s.sample_id = gt.sample_id + AND gt.model_used = ? + WHERE gt.sample_id IS NULL + ORDER BY s.dataset_index + """, + (model_used,), + ) + else: + cursor = await self._conn.execute( + """ + SELECT s.* FROM samples s + LEFT JOIN ground_truth gt ON s.sample_id = gt.sample_id + WHERE gt.sample_id IS NULL + ORDER BY s.dataset_index + """ + ) rows = await cursor.fetchall() return [ AudioSample( @@ -555,9 +625,22 @@ async def get_samples_without_ground_truth(self) -> list[AudioSample]: for row in rows ] - async def get_ground_truth_count(self) -> int: - """Get number of ground truth entries.""" - cursor = await self._conn.execute("SELECT COUNT(*) FROM ground_truth") + async def get_ground_truth_count(self, model_used: str | None = None) -> int: + """Get number of ground truth entries. + + Args: + model_used: If provided, count GT entries for this specific model. + If None, count distinct samples that have any GT. + """ + if model_used: + cursor = await self._conn.execute( + "SELECT COUNT(*) FROM ground_truth WHERE model_used = ?", + (model_used,), + ) + else: + cursor = await self._conn.execute( + "SELECT COUNT(DISTINCT sample_id) FROM ground_truth" + ) row = await cursor.fetchone() return row[0] @@ -636,37 +719,47 @@ async def delete_wer_metrics_for_service( return cursor.rowcount async def get_samples_without_wer( - self, service_name: ServiceName, model_name: str | None = None + self, + service_name: ServiceName, + model_name: str | None = None, + wer_label: str | None = None, ) -> list[AudioSample]: - """Get samples that have results but no WER metrics for a service.""" - if model_name: - cursor = await self._conn.execute( - """ - SELECT s.* FROM samples s - INNER JOIN results r ON s.sample_id = r.sample_id - AND r.service_name = ? AND r.model_name = ? - INNER JOIN ground_truth gt ON s.sample_id = gt.sample_id - LEFT JOIN wer_metrics w ON s.sample_id = w.sample_id - AND w.service_name = ? AND w.model_name = ? - WHERE w.id IS NULL AND r.transcription IS NOT NULL - ORDER BY s.dataset_index - """, - (service_name.value, model_name, service_name.value, model_name), - ) + """Get samples that have results but no WER metrics for a service. + + Args: + service_name: STT service to check. + model_name: Filter results by this model name (None = all models). + wer_label: Check WER existence under this label instead of model_name. + Useful when storing WER with a different key than the source results. + """ + wer_model = wer_label if wer_label is not None else model_name + + # Build query parts based on which filters are active + if model_name is not None: + results_join = "INNER JOIN results r ON s.sample_id = r.sample_id AND r.service_name = ? AND r.model_name = ?" + results_params = [service_name.value, model_name] else: - cursor = await self._conn.execute( - """ - SELECT s.* FROM samples s - INNER JOIN results r ON s.sample_id = r.sample_id - AND r.service_name = ? - INNER JOIN ground_truth gt ON s.sample_id = gt.sample_id - LEFT JOIN wer_metrics w ON s.sample_id = w.sample_id - AND w.service_name = ? - WHERE w.id IS NULL AND r.transcription IS NOT NULL - ORDER BY s.dataset_index - """, - (service_name.value, service_name.value), - ) + results_join = "INNER JOIN results r ON s.sample_id = r.sample_id AND r.service_name = ?" + results_params = [service_name.value] + + if wer_model is not None: + wer_join = "LEFT JOIN wer_metrics w ON s.sample_id = w.sample_id AND w.service_name = ? AND w.model_name = ?" + wer_params = [service_name.value, wer_model] + else: + wer_join = "LEFT JOIN wer_metrics w ON s.sample_id = w.sample_id AND w.service_name = ?" + wer_params = [service_name.value] + + cursor = await self._conn.execute( + f""" + SELECT s.* FROM samples s + {results_join} + {wer_join} + WHERE w.id IS NULL AND r.transcription IS NOT NULL + AND EXISTS (SELECT 1 FROM ground_truth gt WHERE gt.sample_id = s.sample_id) + ORDER BY s.dataset_index + """, + results_params + wer_params, + ) rows = await cursor.fetchall() return [ AudioSample( @@ -808,11 +901,14 @@ async def delete_semantic_wer_traces_for_service( return cursor.rowcount async def get_result_with_ground_truth( - self, sample_id: str, service_name: ServiceName, model_name: str | None = None + self, + sample_id: str, + service_name: ServiceName, + model_name: str | None = None, + gt_model: str | None = None, ) -> tuple[BenchmarkResult | None, GroundTruth | None]: """Get a result and its ground truth for WER calculation.""" - # Get result - if model_name: + if model_name is not None: cursor = await self._conn.execute( "SELECT * FROM results WHERE sample_id = ? AND service_name = ? AND model_name = ?", (sample_id, service_name.value, model_name), @@ -825,8 +921,7 @@ async def get_result_with_ground_truth( row = await cursor.fetchone() result = self._row_to_result(row) if row else None - # Get ground truth - gt = await self.get_ground_truth(sample_id) + gt = await self.get_ground_truth(sample_id, model_used=gt_model) return result, gt @@ -1064,6 +1159,12 @@ async def get_report_data( r.transcription FROM wer_metrics w JOIN ground_truth g ON w.sample_id = g.sample_id + AND g.model_used = ( + SELECT model_used FROM ground_truth g2 + WHERE g2.sample_id = g.sample_id + ORDER BY g2.verified_by IS NOT NULL DESC, g2.generated_at DESC + LIMIT 1 + ) JOIN results r ON w.sample_id = r.sample_id AND w.service_name = r.service_name AND w.model_name = r.model_name From e5e0b09ad0b7e6b71bb98c9e4e94e6b2d0bcf620 Mon Sep 17 00:00:00 2001 From: angelos-p Date: Fri, 17 Apr 2026 15:44:27 +0100 Subject: [PATCH 2/3] Make Scribe the default GT model --- src/stt_benchmark/cli/ground_truth.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/stt_benchmark/cli/ground_truth.py b/src/stt_benchmark/cli/ground_truth.py index a04ec3b..39cf111 100644 --- a/src/stt_benchmark/cli/ground_truth.py +++ b/src/stt_benchmark/cli/ground_truth.py @@ -26,10 +26,10 @@ def generate_ground_truth( help="Limit number of samples to transcribe", ), provider: str = typer.Option( - "gemini", + "scribe", "--provider", "-p", - help="Transcription provider: 'gemini' or 'scribe'", + help="Transcription provider: 'scribe' or 'gemini'", ), model: str | None = typer.Option( None, From 7972b7356b230eb440749c9900e31f7dda9ae852 Mon Sep 17 00:00:00 2001 From: angelos-p Date: Fri, 17 Apr 2026 22:48:17 +0100 Subject: [PATCH 3/3] Clarify --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 571916d..abd95fe 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ A framework for benchmarking Speech-to-Text services with TTFS (Time To Final Segment) latency and Semantic WER (Word Error Rate) accuracy measurement. -> **Note:** Ground truth transcriptions are generated using batch transcription (Gemini Flash or ElevenLabs Scribe v2), not manual human transcription. The reference text may contain inaccuracies that affect WER scores. Results should be interpreted as relative comparisons between services rather than absolute accuracy metrics. +> **Note:** The results below use ground truth that was human-reviewed after batch transcription. If you reproduce the benchmark yourself, the generated ground truth (via Gemini Flash or ElevenLabs Scribe v2) will not have human review and may contain inaccuracies that affect WER scores. Results should be interpreted as relative comparisons between services rather than absolute accuracy metrics. ## Results Summary