From 717e1ccc0108a2e10dc62044aae907885f36d095 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 29 Jan 2026 09:47:19 -0500 Subject: [PATCH] GradiumSTTService now flushes pending transcripts on VAD stopped detection --- changelog/3587.changed.md | 1 + .../07zf-interruptible-gradium.py | 6 +- .../foundational/13l-gradium-transcription.py | 84 +++++++++++++++++++ src/pipecat/services/gradium/stt.py | 70 +++++++++++++++- 4 files changed, 159 insertions(+), 2 deletions(-) create mode 100644 changelog/3587.changed.md create mode 100644 examples/foundational/13l-gradium-transcription.py diff --git a/changelog/3587.changed.md b/changelog/3587.changed.md new file mode 100644 index 000000000..f3901c724 --- /dev/null +++ b/changelog/3587.changed.md @@ -0,0 +1 @@ +- `GradiumSTTService` now flushes pending transcriptions when VAD detects the user stopped speaking, improving response latency. diff --git a/examples/foundational/07zf-interruptible-gradium.py b/examples/foundational/07zf-interruptible-gradium.py index b6207d9f4..14275345f 100644 --- a/examples/foundational/07zf-interruptible-gradium.py +++ b/examples/foundational/07zf-interruptible-gradium.py @@ -59,11 +59,15 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") - stt = GradiumSTTService(api_key=os.getenv("GRADIUM_API_KEY")) + stt = GradiumSTTService( + api_key=os.getenv("GRADIUM_API_KEY"), + api_endpoint_base_url="wss://us.api.gradium.ai/api/speech/asr", + ) tts = GradiumTTSService( api_key=os.getenv("GRADIUM_API_KEY"), voice_id="YTpq7expH9539ERJ", + url="wss://us.api.gradium.ai/api/speech/tts", ) llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) diff --git a/examples/foundational/13l-gradium-transcription.py b/examples/foundational/13l-gradium-transcription.py new file mode 100644 index 000000000..6803d21c7 --- /dev/null +++ b/examples/foundational/13l-gradium-transcription.py @@ -0,0 +1,84 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import os + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.frames.frames import Frame, TranscriptionFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineTask +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.gradium.stt import GradiumSTTService +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.daily.transport import DailyParams +from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams + +load_dotenv(override=True) + + +class TranscriptionLogger(FrameProcessor): + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, TranscriptionFrame): + print(f"Transcription: {frame.text}") + + # Push all frames through + await self.push_frame(frame, direction) + + +# We store functions so objects (e.g. SileroVADAnalyzer) don't get +# instantiated. The function will be called when the desired transport gets +# selected. +transport_params = { + "daily": lambda: DailyParams(audio_in_enabled=True), + "twilio": lambda: FastAPIWebsocketParams(audio_in_enabled=True), + "webrtc": lambda: TransportParams(audio_in_enabled=True), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + stt = GradiumSTTService( + api_key=os.getenv("GRADIUM_API_KEY"), + api_endpoint_base_url="wss://us.api.gradium.ai/api/speech/asr", + ) + + tl = TranscriptionLogger() + + pipeline = Pipeline([transport.input(), stt, tl]) + + task = PipelineTask( + pipeline, + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() + + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + + await runner.run(task) + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud.""" + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +if __name__ == "__main__": + from pipecat.runner.run import main + + main() diff --git a/src/pipecat/services/gradium/stt.py b/src/pipecat/services/gradium/stt.py index 1c378ff9f..32c1d48c7 100644 --- a/src/pipecat/services/gradium/stt.py +++ b/src/pipecat/services/gradium/stt.py @@ -22,7 +22,10 @@ from pipecat.frames.frames import ( Frame, StartFrame, TranscriptionFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, ) +from pipecat.processors.frame_processor import FrameDirection from pipecat.services.stt_service import WebsocketSTTService from pipecat.transcriptions.language import Language from pipecat.utils.time import time_now_iso8601 @@ -76,6 +79,11 @@ class GradiumSTTService(WebsocketSTTService): self._chunk_size_ms = 80 self._chunk_size_bytes = 0 + # Set from the ready message when connecting to the service. + # These values are used for flushing transcription. + self._delay_in_frames = 0 + self._frame_size = 0 + def can_generate_metrics(self) -> bool: """Check if the service can generate metrics. @@ -112,6 +120,57 @@ class GradiumSTTService(WebsocketSTTService): await super().cancel(frame) await self._disconnect() + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process frames with VAD-specific handling. + + When VAD detects the user has stopped speaking, we flush the transcription + by sending silence frames. This makes the system more reactive by getting + the final transcription faster without closing the connection. + + Args: + frame: The frame to process. + direction: The direction of frame processing. + """ + await super().process_frame(frame, direction) + + if isinstance(frame, VADUserStartedSpeakingFrame): + await self.start_processing_metrics() + elif isinstance(frame, VADUserStoppedSpeakingFrame): + await self._flush_transcription() + + async def _flush_transcription(self): + """Flush the transcription by sending silence frames. + + When VAD detects the user stopped speaking, we send delay_in_frames + chunks of silence (zeros) to flush the remaining audio from the model's + buffer. This allows for faster turn-around without closing the connection. + + From Gradium docs: "feed in delay_in_frames chunks of silence (vectors + of zeros). If those are fed in faster than realtime, the API also has + a possibility to process them faster." + """ + if not self._websocket or self._websocket.state is not State.OPEN: + return + + if self._delay_in_frames <= 0: + logger.debug("No delay_in_frames set, skipping flush") + return + + # Create a silence chunk (zeros) of frame_size samples + # Each sample is 2 bytes (16-bit PCM) + silence_bytes = bytes(self._frame_size * 2) + silence_b64 = base64.b64encode(silence_bytes).decode("utf-8") + + logger.debug(f"Flushing Gradium STT with {self._delay_in_frames} silence frames") + + for _ in range(self._delay_in_frames): + msg = {"type": "audio", "audio": silence_b64} + try: + await self._websocket.send(json.dumps(msg)) + except Exception as e: + logger.warning(f"Failed to send silence frame: {e}") + break + async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: """Process audio data for speech-to-text conversion. @@ -122,7 +181,6 @@ class GradiumSTTService(WebsocketSTTService): None (processing handled via WebSocket messages). """ self._audio_buffer.extend(audio) - await self.start_processing_metrics() while len(self._audio_buffer) >= self._chunk_size_bytes: chunk = bytes(self._audio_buffer[: self._chunk_size_bytes]) @@ -151,6 +209,9 @@ class GradiumSTTService(WebsocketSTTService): try: if self._websocket and self._websocket.state is State.OPEN: return + + logger.debug("Connecting to Gradium STT") + ws_url = self._api_endpoint_base_url headers = { "x-api-key": self._api_key, @@ -175,6 +236,11 @@ class GradiumSTTService(WebsocketSTTService): if ready_msg["type"] != "ready": raise Exception(f"unexpected first message type {ready_msg['type']}") + # Store delay_in_frames and frame_size for silence flushing + self._delay_in_frames = ready_msg.get("delay_in_frames", 0) + self._frame_size = ready_msg.get("frame_size", 1920) + logger.debug(f"Connected to Gradium STT") + except Exception as e: await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) raise @@ -240,3 +306,5 @@ class GradiumSTTService(WebsocketSTTService): time_now_iso8601(), ) ) + await self._trace_transcription(text, is_final=True, language=None) + await self.stop_processing_metrics()