From b95a6afe77dcbb875bf40a1b2509aa00421c116b Mon Sep 17 00:00:00 2001 From: poseneror Date: Sun, 11 Jan 2026 09:43:02 +0200 Subject: [PATCH 1/2] feat(gladia): add VAD events support Add support for Gladia's speech_start/speech_end events to emit UserStartedSpeakingFrame and UserStoppedSpeakingFrame frames. When enable_vad=True in GladiaInputParams: - speech_start triggers interruption and pushes UserStartedSpeakingFrame - speech_end pushes UserStoppedSpeakingFrame - Tracks speaking state to prevent duplicate events This allows using Gladia's built-in VAD instead of a separate VAD in the pipeline. --- src/pipecat/services/gladia/config.py | 4 ++++ src/pipecat/services/gladia/stt.py | 33 +++++++++++++++++++++++++++ 2 files changed, 37 insertions(+) diff --git a/src/pipecat/services/gladia/config.py b/src/pipecat/services/gladia/config.py index ed160a36e..5fd5bfdac 100644 --- a/src/pipecat/services/gladia/config.py +++ b/src/pipecat/services/gladia/config.py @@ -169,6 +169,9 @@ class GladiaInputParams(BaseModel): pre_processing: Audio pre-processing options realtime_processing: Real-time processing features messages_config: WebSocket message filtering options + enable_vad: Enable VAD to trigger end of utterance detection. This should be used + without any other VAD enabled in the agent and will emit the speaker started + and stopped frames. Defaults to False. """ encoding: Optional[str] = "wav/pcm" @@ -182,3 +185,4 @@ class GladiaInputParams(BaseModel): pre_processing: Optional[PreProcessingConfig] = None realtime_processing: Optional[RealtimeProcessingConfig] = None messages_config: Optional[MessagesConfig] = None + enable_vad: bool = False diff --git a/src/pipecat/services/gladia/stt.py b/src/pipecat/services/gladia/stt.py index 4ba0a2ffd..9a20d6dcb 100644 --- a/src/pipecat/services/gladia/stt.py +++ b/src/pipecat/services/gladia/stt.py @@ -28,6 +28,8 @@ from pipecat.frames.frames import ( StartFrame, TranscriptionFrame, TranslationFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, ) from pipecat.services.gladia.config import GladiaInputParams from pipecat.services.stt_service import WebsocketSTTService @@ -266,6 +268,9 @@ class GladiaSTTService(WebsocketSTTService): self._max_buffer_size = max_buffer_size self._buffer_lock = asyncio.Lock() + # VAD state tracking + self._is_speaking = False + def __str__(self): return f"{self.name} [{self._session_id}]" @@ -507,6 +512,30 @@ class GladiaSTTService(WebsocketSTTService): await self.stop_ttfb_metrics() await self.stop_processing_metrics() + async def _on_speech_started(self): + """Handle speech start event from Gladia. + + Triggers interruption and emits UserStartedSpeakingFrame when VAD is enabled. + """ + if not self._params.enable_vad or self._is_speaking: + return + logger.debug(f"{self} User started speaking") + self._is_speaking = True + # Push interruption first to stop the bot, then notify about user speaking + await self.push_interruption_task_frame_and_wait() + await self.push_frame(UserStartedSpeakingFrame()) + + async def _on_speech_ended(self): + """Handle speech end event from Gladia. + + Emits UserStoppedSpeakingFrame when VAD is enabled. + """ + if not self._params.enable_vad or not self._is_speaking: + return + self._is_speaking = False + await self.push_frame(UserStoppedSpeakingFrame()) + logger.debug(f"{self} User stopped speaking") + async def _send_audio(self, audio: bytes): """Send audio chunk with proper message format.""" if self._websocket and self._websocket.state is State.OPEN: @@ -599,6 +628,10 @@ class GladiaSTTService(WebsocketSTTService): translation, "", time_now_iso8601(), translated_language ) ) + elif content["type"] == "speech_start": + await self._on_speech_started() + elif content["type"] == "speech_end": + await self._on_speech_ended() except json.JSONDecodeError: logger.warning(f"{self} Received non-JSON message: {message}") From 3304b18ac2ecfbbafe8d06bcad3d4b56ceb38557 Mon Sep 17 00:00:00 2001 From: poseneror Date: Tue, 13 Jan 2026 14:19:50 +0200 Subject: [PATCH 2/2] Add should_interrupt + broadcast user events --- src/pipecat/services/gladia/stt.py | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/src/pipecat/services/gladia/stt.py b/src/pipecat/services/gladia/stt.py index 9a20d6dcb..8c5ec7aa4 100644 --- a/src/pipecat/services/gladia/stt.py +++ b/src/pipecat/services/gladia/stt.py @@ -204,6 +204,7 @@ class GladiaSTTService(WebsocketSTTService): model: str = "solaria-1", params: Optional[GladiaInputParams] = None, max_buffer_size: int = 1024 * 1024 * 20, # 20MB default buffer + should_interrupt: bool = True, **kwargs, ): """Initialize the Gladia STT service. @@ -222,6 +223,8 @@ class GladiaSTTService(WebsocketSTTService): model: Model to use for transcription. Defaults to "solaria-1". params: Additional configuration parameters for Gladia service. max_buffer_size: Maximum size of audio buffer in bytes. Defaults to 20MB. + should_interrupt: Determine whether the bot should be interrupted when + Gladia VAD detects user speech. Defaults to True. **kwargs: Additional arguments passed to the STTService parent class. """ super().__init__(sample_rate=sample_rate, **kwargs) @@ -270,6 +273,7 @@ class GladiaSTTService(WebsocketSTTService): # VAD state tracking self._is_speaking = False + self._should_interrupt = should_interrupt def __str__(self): return f"{self.name} [{self._session_id}]" @@ -515,25 +519,28 @@ class GladiaSTTService(WebsocketSTTService): async def _on_speech_started(self): """Handle speech start event from Gladia. - Triggers interruption and emits UserStartedSpeakingFrame when VAD is enabled. + Broadcasts UserStartedSpeakingFrame and optionally triggers interruption + when VAD is enabled. """ if not self._params.enable_vad or self._is_speaking: return + logger.debug(f"{self} User started speaking") self._is_speaking = True - # Push interruption first to stop the bot, then notify about user speaking - await self.push_interruption_task_frame_and_wait() - await self.push_frame(UserStartedSpeakingFrame()) + + await self.broadcast_frame(UserStartedSpeakingFrame) + if self._should_interrupt: + await self.push_interruption_task_frame_and_wait() async def _on_speech_ended(self): """Handle speech end event from Gladia. - Emits UserStoppedSpeakingFrame when VAD is enabled. + Broadcasts UserStoppedSpeakingFrame when VAD is enabled. """ if not self._params.enable_vad or not self._is_speaking: return self._is_speaking = False - await self.push_frame(UserStoppedSpeakingFrame()) + await self.broadcast_frame(UserStoppedSpeakingFrame) logger.debug(f"{self} User stopped speaking") async def _send_audio(self, audio: bytes):