GradiumSTTService now flushes pending transcripts on VAD stopped detection

changelog/3587.changed.md

@@ -0,0 +1 @@
+- `GradiumSTTService` now flushes pending transcriptions when VAD detects the user stopped speaking, improving response latency.

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

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()

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()