diff --git a/README.md b/README.md index e9293f9..b351062 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,7 @@ Benchmark results on 1000 samples from the `pipecat-ai/smart-turn-data-v3.1-trai | NVIDIA | Nemotron 3.0 ASR (en) | 100.0% | 76.1% | 1.90% | 1.95% | 221ms | 238ms | 252ms | | NVIDIA | Nemotron 3.5 ASR (multilingual) | 99.6% | 62.0% | 4.54% | 4.58% | 236ms | 253ms | 266ms | | OpenAI | gpt-4o-transcribe | 99.3% | 75.9% | 3.24% | 3.06% | 637ms | 965ms | 1655ms | +| Reson8 | Resonant-1 Turns | 99.9% | 85.6% | 1.39% | 1.10% | 326ms | 639ms | 1156ms | | Smallest AI | pulse | 100.0% | 72.4% | 2.30% | 2.37% | 398ms | 533ms | 1593ms | | Soniox | stt-rt-v4 | 99.8% | 84.1% | 1.25% | 1.29% | 249ms | 281ms | 310ms | | Speechmatics | N/A | 99.7% | 83.2% | 1.40% | 1.07% | 495ms | 676ms | 736ms | diff --git a/assets/stt_pareto_frontier.png b/assets/stt_pareto_frontier.png index 1b4f24e..522bd05 100644 Binary files a/assets/stt_pareto_frontier.png and b/assets/stt_pareto_frontier.png differ diff --git a/assets/stt_pareto_frontier_p95.png b/assets/stt_pareto_frontier_p95.png index f6da775..f7e2180 100644 Binary files a/assets/stt_pareto_frontier_p95.png and b/assets/stt_pareto_frontier_p95.png differ diff --git a/env.example b/env.example index ba52e01..1f4ee85 100644 --- a/env.example +++ b/env.example @@ -52,6 +52,9 @@ SAGEMAKER_ASR_ENDPOINT_NAME= # OpenAI OPENAI_API_KEY= +# Reson8 +RESON8_API_KEY= + # Sarvam SARVAM_API_KEY= diff --git a/src/stt_benchmark/cli/wer.py b/src/stt_benchmark/cli/wer.py index 1d22576..53e80ea 100644 --- a/src/stt_benchmark/cli/wer.py +++ b/src/stt_benchmark/cli/wer.py @@ -59,6 +59,11 @@ def calculate_wer( "-t", help="Use test database (test_results.db) instead of main database", ), + passes: int = typer.Option( + 1, + "--passes", + help="Score each sample this many times and keep the median (suppresses scorer non-determinism). Use an odd value.", + ), ): """Calculate semantic WER metrics for transcription results. @@ -131,7 +136,7 @@ async def run(): console.print(f"Ground truth coverage: {gt_count}/{sample_count} samples\n") - evaluator = SemanticWEREvaluator(db_path=db_path) + evaluator = SemanticWEREvaluator(db_path=db_path, num_passes=passes) console.print("Warming prompt cache...", end=" ") try: diff --git a/src/stt_benchmark/config.py b/src/stt_benchmark/config.py index 85a64c2..3cedf56 100644 --- a/src/stt_benchmark/config.py +++ b/src/stt_benchmark/config.py @@ -41,6 +41,7 @@ class BenchmarkConfig(BaseSettings): mistral_api_key: str = Field(default="", alias="MISTRAL_API_KEY") nvidia_api_key: str = Field(default="", alias="NVIDIA_API_KEY") openai_api_key: str = Field(default="", alias="OPENAI_API_KEY") + reson8_api_key: str = Field(default="", alias="RESON8_API_KEY") sagemaker_asr_endpoint_name: str = Field(default="", alias="SAGEMAKER_ASR_ENDPOINT_NAME") sarvam_api_key: str = Field(default="", alias="SARVAM_API_KEY") smallest_api_key: str = Field(default="", alias="SMALLEST_API_KEY") diff --git a/src/stt_benchmark/evaluation/semantic_wer.py b/src/stt_benchmark/evaluation/semantic_wer.py index 028d866..05ed032 100644 --- a/src/stt_benchmark/evaluation/semantic_wer.py +++ b/src/stt_benchmark/evaluation/semantic_wer.py @@ -364,11 +364,17 @@ def __init__( model: str = "claude-sonnet-4-5-20250929", db_path: Path | None = None, max_concurrency: int = 50, + num_passes: int = 1, ): self.config = get_config() self.model = model self.db = Database(db_path=db_path) self._api_semaphore = asyncio.Semaphore(max_concurrency) + # The scorer is non-deterministic even at temperature 0 (Claude self-reports the + # counts, and its normalization occasionally drops/duplicates a word). With + # num_passes > 1 each sample is scored that many times and the median-by-total-errors + # pass is kept, suppressing those outlier passes. Use an odd value for a clean median. + self.num_passes = max(1, num_passes) if not self.config.anthropic_api_key: raise ValueError("ANTHROPIC_API_KEY not set in environment") @@ -800,6 +806,34 @@ async def evaluate_with_retry( return None + async def evaluate_median( + self, + reference: str, + hypothesis: str, + filename: str = "", + ) -> tuple[dict, SemanticWERTrace] | None: + """Score ``self.num_passes`` times and return the median-by-total-errors pass. + + Passes run sequentially so each still holds a single API-semaphore slot (the + caller acquires it once per sample). The returned (result, trace) is a real, + self-consistent pass — the one whose total error count is the median — rather + than a synthetic blend of per-component medians. + """ + if self.num_passes <= 1: + return await self.evaluate_with_retry(reference, hypothesis, filename=filename) + + evals: list[tuple[dict, SemanticWERTrace]] = [] + for _ in range(self.num_passes): + pair = await self.evaluate_with_retry(reference, hypothesis, filename=filename) + if pair is not None: + evals.append(pair) + + if not evals: + return None + + evals.sort(key=lambda er: er[0]["substitutions"] + er[0]["deletions"] + er[0]["insertions"]) + return evals[(len(evals) - 1) // 2] + async def evaluate_service( self, service_name: ServiceName, @@ -847,8 +881,8 @@ async def _eval_sample(sample): logger.warning(f"No ground truth for sample {sample.sample_id}") return - # Evaluate with Claude (with retry and timeout) - eval_pair = await self.evaluate_with_retry( + # Evaluate with Claude (median of num_passes, each with retry and timeout) + eval_pair = await self.evaluate_median( gt.text, result.transcription, filename=sample.sample_id ) diff --git a/src/stt_benchmark/models.py b/src/stt_benchmark/models.py index 0d81326..a168f61 100644 --- a/src/stt_benchmark/models.py +++ b/src/stt_benchmark/models.py @@ -34,6 +34,7 @@ class ServiceName(str, Enum): NVIDIA_SAGEMAKER = "nvidia_sagemaker" OPENAI = "openai" OPENAI_REALTIME = "openai_realtime" + RESON8 = "reson8" SARVAM = "sarvam" SARVAM_SAARAS_V3 = "sarvam_saaras_v3" SMALLEST = "smallest" diff --git a/src/stt_benchmark/services.py b/src/stt_benchmark/services.py index 90ac6ca..7c6dc7d 100644 --- a/src/stt_benchmark/services.py +++ b/src/stt_benchmark/services.py @@ -357,6 +357,36 @@ def create_openai_realtime() -> FrameProcessor: ) +def create_reson8() -> FrameProcessor: + # Reson8 is not bundled with Pipecat; the service lives in this repo. + from stt_benchmark.services_custom.reson8_stt import DEFAULT_URL, Reson8STTService + + # When targeting a local piano build (RESON8_URL set to ws://localhost:...), + # there is no gateway to validate the API key and inject the Tuba auth + # headers, so pass them directly. Any UUIDs work — billing/vocab degrade + # gracefully on the dev server. + url = os.getenv("RESON8_URL", DEFAULT_URL) + extra_headers: dict[str, str] = {} + customer_id = os.getenv("RESON8_CUSTOMER_ID") + if customer_id: + extra_headers = { + "X-Customer-Id": customer_id, + "X-Organization-Id": os.getenv("RESON8_ORGANIZATION_ID", customer_id), + "X-Client-Id": os.getenv("RESON8_CLIENT_ID", customer_id), + } + + return Reson8STTService( + api_key=_get_env("RESON8_API_KEY"), + url=url, + extra_headers=extra_headers, + min_patience_seconds=0.2, + max_patience_seconds=0.2, + settings=Reson8STTService.Settings( + language=Language.EN, + ), + ) + + def create_sarvam() -> FrameProcessor: from pipecat.services.sarvam.stt import SarvamSTTService @@ -565,6 +595,12 @@ def create_xai() -> FrameProcessor: model_label="gpt-4o-transcribe", required_env_vars=["OPENAI_API_KEY"], ), + "reson8": ServiceDefinition( + factory=create_reson8, + vendor="Reson8", + model_label="Resonant-1 Turns", + required_env_vars=["RESON8_API_KEY"], + ), "sarvam": ServiceDefinition( factory=create_sarvam, vendor="Sarvam", diff --git a/src/stt_benchmark/services_custom/__init__.py b/src/stt_benchmark/services_custom/__init__.py new file mode 100644 index 0000000..cfa8900 --- /dev/null +++ b/src/stt_benchmark/services_custom/__init__.py @@ -0,0 +1 @@ +"""Custom Pipecat STT services for vendors not yet bundled with Pipecat.""" diff --git a/src/stt_benchmark/services_custom/reson8_stt.py b/src/stt_benchmark/services_custom/reson8_stt.py new file mode 100644 index 0000000..acb1712 --- /dev/null +++ b/src/stt_benchmark/services_custom/reson8_stt.py @@ -0,0 +1,346 @@ +"""Reson8 speech-to-text service implementation. + +Reson8 is not (yet) bundled with Pipecat, so this is a local Pipecat +``WebsocketSTTService`` implementation of its WSS "Turns" API: +https://docs.reson8.dev/api/speech-to-text/turns/ + +The Turns API is turn-based: the server performs its own endpoint/turn +detection and streams a small set of JSON control messages over the websocket +while we feed it raw binary PCM audio: + +- ``{"type": "turn_start"}`` — speaker began a new turn +- ``{"type": "turn_end_candidate", — detected silence; ``text`` is the + "text": "..."}`` (running) best transcript for the turn +- ``{"type": "turn_end"}`` — confirms the previous candidate is final +- ``{"type": "turn_continuation"}`` — cancels the candidate; speaker resumed, + the next candidate carries the full + accumulated text + +We push each ``turn_end_candidate`` as an ``InterimTranscriptionFrame`` and emit +the final ``TranscriptionFrame`` on ``turn_end`` (using the latest candidate's +text, since ``turn_end`` itself carries none). TTFS is measured to the last +``turn_end_candidate`` — when the final text first became available. + +Endpointing is driven entirely by the server; aggressiveness is tuned with the +``patience`` query parameters (see :func:`stt_benchmark.services.create_reson8`). +""" + +import json +import time +from collections.abc import AsyncGenerator +from dataclasses import dataclass +from urllib.parse import urlencode + +from loguru import logger +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + Frame, + InterimTranscriptionFrame, + StartFrame, + TranscriptionFrame, +) +from pipecat.services.settings import STTSettings +from pipecat.services.stt_latency import DEFAULT_TTFS_P99 +from pipecat.services.stt_service import WebsocketSTTService +from pipecat.transcriptions.language import Language, resolve_language +from pipecat.utils.time import time_now_iso8601 + +try: + from websockets.asyncio.client import connect as websocket_connect + from websockets.protocol import State +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error("In order to use Reson8, you need to `pip install websockets`.") + raise ImportError(f"Missing module: {e}") from e + + +DEFAULT_URL = "wss://api.reson8.dev/v1/speech-to-text/turns" + + +def language_to_reson8_language(language: Language) -> str | None: + """Convert a Pipecat ``Language`` to a Reson8 language code. + + Reson8 takes an optional ISO language code as a transcription bias, so we + map to the base two-letter code (e.g. ``Language.EN_US`` -> ``"en"``). + """ + return resolve_language(language, {}, use_base_code=True) + + +@dataclass +class Reson8STTSettings(STTSettings): + """Settings for :class:`Reson8STTService`.""" + + +class Reson8STTService(WebsocketSTTService): + """Speech-to-Text service using Reson8's WSS Turns API. + + Streams raw PCM audio over a websocket and turns Reson8's turn-based control + messages into Pipecat ``InterimTranscriptionFrame`` / ``TranscriptionFrame``. + + For complete API documentation, see: + https://docs.reson8.dev/api/speech-to-text/turns/ + """ + + Settings = Reson8STTSettings + _settings: Settings + + def __init__( + self, + *, + api_key: str, + url: str = DEFAULT_URL, + sample_rate: int | None = None, + encoding: str = "pcm_s16le", + num_channels: int = 1, + custom_model_id: str | None = None, + extra_headers: dict[str, str] | None = None, + patience: float | None = None, + min_patience_seconds: float | None = None, + max_patience_seconds: float | None = None, + settings: Settings | None = None, + ttfs_p99_latency: float | None = DEFAULT_TTFS_P99, + **kwargs, + ): + """Initialize the Reson8 STT service. + + Args: + api_key: Reson8 API key. + url: Reson8 WSS Turns API URL. + sample_rate: Audio sample rate in Hz. If None, taken from the pipeline. + encoding: Audio encoding query param (``pcm_s16le`` or ``auto``). + num_channels: Number of audio channels. + custom_model_id: Optional custom model identifier. + extra_headers: Additional websocket headers. Used to target a local + piano build that authenticates via Tuba headers (``X-Customer-Id`` + etc.) injected by the gateway in production instead of the + ``Authorization: ApiKey`` header. + patience: End-of-turn patience scalar in [0, 1] (server query param). + min_patience_seconds: Lower bound on the end-of-turn delay (query param). + max_patience_seconds: Upper bound on the end-of-turn delay (query param). + settings: Runtime-updatable settings (e.g. ``language``). + ttfs_p99_latency: P99 latency from speech end to final transcript in seconds. + **kwargs: Additional arguments passed to the STTService. + """ + # reson8 has no selectable model (model is unset; a custom model is + # selected via the custom_model_id query param), so model=None. + default_settings = self.Settings(model=None, language=Language.EN) + if settings is not None: + default_settings.apply_update(settings) + + super().__init__( + sample_rate=sample_rate, + ttfs_p99_latency=ttfs_p99_latency, + settings=default_settings, + **kwargs, + ) + + self._api_key = api_key + self._url = url + self._encoding = encoding + self._num_channels = num_channels + self._custom_model_id = custom_model_id + self._extra_headers = extra_headers or {} + self._patience = patience + self._min_patience_seconds = min_patience_seconds + self._max_patience_seconds = max_patience_seconds + + # Text of the most recent turn_end_candidate; emitted as the final + # transcript on turn_end (which itself carries no text). + self._candidate_text: str = "" + + # Wall-clock time of the most recent (non-empty) turn_end_candidate in the + # current turn — when the final text first became available. TTFS is + # measured to this rather than to turn_end. + self._last_candidate_time: float = 0.0 + + self._receive_task = None + + def can_generate_metrics(self) -> bool: + """Reson8 supports TTFB/TTFS metrics generation.""" + return True + + def language_to_service_language(self, language: Language) -> str | None: + """Convert a Pipecat ``Language`` to a Reson8 language code.""" + return language_to_reson8_language(language) + + async def start(self, frame: StartFrame): + """Start the Reson8 websocket connection.""" + await super().start(frame) + await self._connect() + + async def stop(self, frame: EndFrame): + """Stop the Reson8 websocket connection.""" + await super().stop(frame) + await self._disconnect() + + async def cancel(self, frame: CancelFrame): + """Cancel the Reson8 websocket connection immediately.""" + await super().cancel(frame) + await self._disconnect() + + async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame | None, None]: + """Send audio data to Reson8. + + Args: + audio: Raw PCM audio bytes to transcribe. + + Yields: + None — transcription results arrive asynchronously over the websocket. + """ + if self._websocket and self._websocket.state is State.OPEN: + try: + await self._websocket.send(audio) + except Exception as e: + logger.warning(f"{self}: send failed: {e}") + yield None + + def _build_url(self) -> str: + """Build the websocket URL with Reson8 query parameters.""" + params = { + "encoding": self._encoding, + "sample_rate": self.sample_rate, + "channels": self._num_channels, + } + language = self._settings.language + if isinstance(language, Language): + language = language_to_reson8_language(language) + if language: + params["language"] = language + if self._custom_model_id: + params["custom_model_id"] = self._custom_model_id + if self._patience is not None: + params["patience"] = self._patience + if self._min_patience_seconds is not None: + params["min_patience_seconds"] = self._min_patience_seconds + if self._max_patience_seconds is not None: + params["max_patience_seconds"] = self._max_patience_seconds + return f"{self._url}?{urlencode(params)}" + + async def _connect(self): + """Connect to Reson8 and start the receive task.""" + await self._connect_websocket() + await super()._connect() + if self._websocket and not self._receive_task: + self._receive_task = self.create_task(self._receive_task_handler(self._report_error)) + + async def _disconnect(self): + """Disconnect from Reson8 and tear down the receive task.""" + await super()._disconnect() + if self._receive_task: + await self.cancel_task(self._receive_task) + self._receive_task = None + await self._disconnect_websocket() + + async def _connect_websocket(self): + """Establish the websocket connection to Reson8.""" + try: + if self._websocket and self._websocket.state is State.OPEN: + return + + url = self._build_url() + logger.debug(f"Connecting to Reson8 STT at {url}") + headers = {"Authorization": f"ApiKey {self._api_key}", **self._extra_headers} + self._websocket = await websocket_connect( + url, + additional_headers=headers, + ) + await self._call_event_handler("on_connected") + logger.debug("Connected to Reson8 STT") + except Exception as e: + self._websocket = None + await self.push_error(error_msg=f"Unable to connect to Reson8: {e}", exception=e) + + async def _disconnect_websocket(self): + """Close the websocket connection to Reson8.""" + try: + if self._websocket: + logger.debug("Disconnecting from Reson8 STT") + await self._websocket.close() + except Exception as e: + await self.push_error(error_msg=f"Error closing websocket: {e}", exception=e) + finally: + self._websocket = None + await self._call_event_handler("on_disconnected") + + def _get_websocket(self): + """Return the current websocket connection.""" + if self._websocket: + return self._websocket + raise Exception("Websocket not connected") + + async def _emit_final_transcript(self): + """Emit the latest candidate text as a final TranscriptionFrame. + + Latency is measured to the *last turn_end_candidate* time (when the final + text first became available), not to turn_end. We report TTFB explicitly + with that end time and cancel the pending timeout, then push the transcript + as non-finalized so the base class does not re-report TTFB at push time + (which would overwrite our earlier timestamp). + """ + text = self._candidate_text.strip() + if not text: + return + end_time = self._last_candidate_time if self._last_candidate_time > 0 else time.time() + await self.stop_ttfb_metrics(end_time=end_time) + await self._cancel_ttfb_timeout() + await self.push_frame( + TranscriptionFrame( + text=text, + user_id=self._user_id, + timestamp=time_now_iso8601(), + language=self._settings.language + if isinstance(self._settings.language, Language) + else None, + finalized=False, + ) + ) + await self.stop_processing_metrics() + self._candidate_text = "" + self._last_candidate_time = 0.0 + + async def _receive_messages(self): + """Receive and process Reson8 turn control messages.""" + self._candidate_text = "" + + async for message in self._get_websocket(): + try: + content = json.loads(message) + except (json.JSONDecodeError, TypeError): + logger.warning(f"{self}: received non-JSON message: {message!r}") + continue + + msg_type = content.get("type") + + if msg_type == "turn_start": + await self.start_processing_metrics() + self._candidate_text = "" + self._last_candidate_time = 0.0 + + elif msg_type == "turn_end_candidate": + # Running best transcript for the turn — surfaced as interim, and + # the latency endpoint (the time the final text became available). + self._candidate_text = content.get("text", "") or "" + if self._candidate_text: + self._last_candidate_time = time.time() + await self.push_frame( + InterimTranscriptionFrame( + text=self._candidate_text, + user_id=self._user_id, + timestamp=time_now_iso8601(), + ) + ) + + elif msg_type == "turn_continuation": + # Speaker resumed; the prior candidate was premature. The next + # turn_end_candidate carries the full accumulated text, so discard + # the cancelled candidate's timestamp too. + self._candidate_text = "" + self._last_candidate_time = 0.0 + + elif msg_type == "turn_end": + # Turn confirmed — emit the latest candidate as the final segment. + await self._emit_final_transcript() + + elif content.get("error") or content.get("code"): + logger.error(f"{self}: Reson8 error: {content}")