From b298376766ad2b5150fdcc2107c9dbb45aa9e82b Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 25 Apr 2025 09:02:04 -0400 Subject: [PATCH 1/2] Add VADUserStartedSpeakingFrame and VADUserStoppedSpeakingFrame --- CHANGELOG.md | 5 +++++ src/pipecat/audio/vad/vad_analyzer.py | 22 ++++++++++++++++++---- src/pipecat/frames/frames.py | 14 ++++++++++++++ src/pipecat/transports/base_input.py | 18 +++++++++++++++++- 4 files changed, 54 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 251539ba7..741d86c4b 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/src/pipecat/audio/vad/vad_analyzer.py b/src/pipecat/audio/vad/vad_analyzer.py index 3ca21a208..a57181008 100644 --- a/src/pipecat/audio/vad/vad_analyzer.py +++ b/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 diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 073112a9f..0d7d90d78 100644 --- a/src/pipecat/frames/frames.py +++ b/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 diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 497b21cda..57511a96a 100644 --- a/src/pipecat/transports/base_input.py +++ b/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() From dfa10af6edf1f8a3f1448a6dd2c829de21b02a88 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 25 Apr 2025 10:28:44 -0400 Subject: [PATCH 2/2] Simplify VAD events to be detected and emitted from BaseInputTransport --- src/pipecat/audio/vad/vad_analyzer.py | 22 ++++------------------ src/pipecat/transports/base_input.py | 25 +++++++------------------ 2 files changed, 11 insertions(+), 36 deletions(-) diff --git a/src/pipecat/audio/vad/vad_analyzer.py b/src/pipecat/audio/vad/vad_analyzer.py index a57181008..3ca21a208 100644 --- a/src/pipecat/audio/vad/vad_analyzer.py +++ b/src/pipecat/audio/vad/vad_analyzer.py @@ -6,7 +6,7 @@ from abc import ABC, abstractmethod from enum import Enum -from typing import Optional, Tuple +from typing import Optional from loguru import logger from pydantic import BaseModel @@ -88,24 +88,12 @@ 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) -> 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 - """ + def analyze_audio(self, buffer) -> VADState: 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, event_type + return self._vad_state audio_frames = self._vad_buffer[:num_required_bytes] self._vad_buffer = self._vad_buffer[num_required_bytes:] @@ -144,7 +132,6 @@ class VADAnalyzer(ABC): ): self._vad_state = VADState.SPEAKING self._vad_starting_count = 0 - event_type = "speech_started" if ( self._vad_state == VADState.STOPPING @@ -152,6 +139,5 @@ class VADAnalyzer(ABC): ): self._vad_state = VADState.QUIET self._vad_stopping_count = 0 - event_type = "speech_stopped" - return self._vad_state, event_type + return self._vad_state diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 57511a96a..acb0058c6 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -230,13 +230,9 @@ class BaseInputTransport(FrameProcessor): async def _vad_analyze(self, audio_frame: InputAudioRawFrame) -> VADState: state = VADState.QUIET if self.vad_analyzer: - state, event_type = await self.get_event_loop().run_in_executor( + state = 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): @@ -254,10 +250,13 @@ class BaseInputTransport(FrameProcessor): self._params.turn_analyzer is None or not self._params.turn_analyzer.speech_triggered ) - if can_create_user_frames: - if new_vad_state == VADState.SPEAKING: + if new_vad_state == VADState.SPEAKING: + await self.push_frame(VADUserStartedSpeakingFrame()) + if can_create_user_frames: frame = UserStartedSpeakingFrame() - elif new_vad_state == VADState.QUIET: + elif new_vad_state == VADState.QUIET: + await self.push_frame(VADUserStoppedSpeakingFrame()) + if can_create_user_frames: frame = UserStoppedSpeakingFrame() if frame: @@ -266,16 +265,6 @@ 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()