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.
This commit is contained in:
poseneror
2026-01-11 09:43:02 +02:00
parent f6ed7d7582
commit b95a6afe77
2 changed files with 37 additions and 0 deletions

View File

@@ -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

View File

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