Add VADUserStartedSpeakingFrame and VADUserStoppedSpeakingFrame

CHANGELOG.md

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added `VADUserStartedSpeakingFrame` and `VADUserStoppedSpeakingFrame`,
+  indicating when the VAD detected the user to start and stop speaking. These
+  events are helpful when using smart turn detection, as the user's stop time
+  can differ from when their turn ends (signified by UserStoppedSpeakingFrame).
+
 - Added `TranslationFrame`, a new frame type that contains a translated
   transcription.
 

src/pipecat/audio/vad/vad_analyzer.py

@@ -6,7 +6,7 @@
 
 from abc import ABC, abstractmethod
 from enum import Enum
-from typing import Optional
+from typing import Optional, Tuple
 
 from loguru import logger
 from pydantic import BaseModel
@@ -88,12 +88,24 @@ class VADAnalyzer(ABC):
         volume = calculate_audio_volume(audio, self.sample_rate)
         return exp_smoothing(volume, self._prev_volume, self._smoothing_factor)
 
-    def analyze_audio(self, buffer) -> VADState:
+    def analyze_audio(self, buffer) -> Tuple[VADState, Optional[str]]:
+        """Analyze audio for voice activity.
+
+        Args:
+            buffer: Audio buffer to analyze
+
+        Returns:
+            Tuple containing:
+                - VADState: Current VAD state
+                - Optional[str]: Event type if a speech event occurred ("speech_started",
+                "speech_stopped"), or None if no event occurred
+        """
         self._vad_buffer += buffer
+        event_type = None
 
         num_required_bytes = self._vad_frames_num_bytes
         if len(self._vad_buffer) < num_required_bytes:
-            return self._vad_state
+            return self._vad_state, event_type
 
         audio_frames = self._vad_buffer[:num_required_bytes]
         self._vad_buffer = self._vad_buffer[num_required_bytes:]
@@ -132,6 +144,7 @@ class VADAnalyzer(ABC):
         ):
             self._vad_state = VADState.SPEAKING
             self._vad_starting_count = 0
+            event_type = "speech_started"
 
         if (
             self._vad_state == VADState.STOPPING
@@ -139,5 +152,6 @@ class VADAnalyzer(ABC):
         ):
             self._vad_state = VADState.QUIET
             self._vad_stopping_count = 0
+            event_type = "speech_stopped"
 
-        return self._vad_state
+        return self._vad_state, event_type

src/pipecat/frames/frames.py

@@ -587,6 +587,20 @@ class EmulateUserStoppedSpeakingFrame(SystemFrame):
     pass
 
 
+@dataclass
+class VADUserStartedSpeakingFrame(SystemFrame):
+    """Frame emitted when VAD detects the user has definitively started speaking."""
+
+    pass
+
+
+@dataclass
+class VADUserStoppedSpeakingFrame(SystemFrame):
+    """Frame emitted when VAD detects the user has definitively stopped speaking."""
+
+    pass
+
+
 @dataclass
 class BotInterruptionFrame(SystemFrame):
     """Emitted by when the bot should be interrupted. This will mainly cause the

src/pipecat/transports/base_input.py

@@ -32,6 +32,8 @@ from pipecat.frames.frames import (
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
     VADParamsUpdateFrame,
+    VADUserStartedSpeakingFrame,
+    VADUserStoppedSpeakingFrame,
 )
 from pipecat.metrics.metrics import MetricsData
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@@ -228,9 +230,13 @@ class BaseInputTransport(FrameProcessor):
     async def _vad_analyze(self, audio_frame: InputAudioRawFrame) -> VADState:
         state = VADState.QUIET
         if self.vad_analyzer:
-            state = await self.get_event_loop().run_in_executor(
+            state, event_type = await self.get_event_loop().run_in_executor(
                 self._executor, self.vad_analyzer.analyze_audio, audio_frame.audio
             )
+
+            if event_type:
+                await self._handle_vad_event(event_type)
+
         return state
 
     async def _handle_vad(self, audio_frame: InputAudioRawFrame, vad_state: VADState):
@@ -260,6 +266,16 @@ class BaseInputTransport(FrameProcessor):
             vad_state = new_vad_state
         return vad_state
 
+    async def _handle_vad_event(self, event_type: str):
+        """Handle VAD speech events by creating and pushing appropriate frames."""
+        if event_type == "speech_started":
+            logger.debug("VAD detected definitive speech start")
+            await self.push_frame(VADUserStartedSpeakingFrame())
+
+        elif event_type == "speech_stopped":
+            logger.debug("VAD detected definitive speech stop")
+            await self.push_frame(VADUserStoppedSpeakingFrame())
+
     async def _handle_end_of_turn(self):
         if self.turn_analyzer:
             state, prediction = await self.turn_analyzer.analyze_end_of_turn()