From b486f35c70398b93e152e6b75cfa76b610238199 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 28 Jan 2026 10:58:58 -0800 Subject: [PATCH] audio: add new VADProcessor --- src/pipecat/processors/audio/vad_processor.py | 100 ++++++++++++ tests/test_vad_processor.py | 143 ++++++++++++++++++ 2 files changed, 243 insertions(+) create mode 100644 src/pipecat/processors/audio/vad_processor.py create mode 100644 tests/test_vad_processor.py diff --git a/src/pipecat/processors/audio/vad_processor.py b/src/pipecat/processors/audio/vad_processor.py new file mode 100644 index 000000000..2076eeeb3 --- /dev/null +++ b/src/pipecat/processors/audio/vad_processor.py @@ -0,0 +1,100 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Voice Activity Detection processor for detecting speech in audio streams. + +This module provides a VADProcessor that wraps a VADController to process +audio frames and push VAD-related frames into the pipeline. +""" + +from typing import Type + +from loguru import logger + +from pipecat.audio.vad.vad_analyzer import VADAnalyzer +from pipecat.audio.vad.vad_controller import VADController +from pipecat.frames.frames import ( + Frame, + UserSpeakingFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + + +class VADProcessor(FrameProcessor): + """Processes audio frames through voice activity detection. + + This processor wraps a VADController to detect speech in audio streams + and push VAD frames into the pipeline: + + - ``VADUserStartedSpeakingFrame``: Pushed when speech begins. + - ``VADUserStoppedSpeakingFrame``: Pushed when speech ends. + - ``UserSpeakingFrame``: Pushed periodically while speech is detected. + + Example:: + + vad_processor = VADProcessor(vad_analyzer=SileroVADAnalyzer()) + """ + + def __init__( + self, + *, + vad_analyzer: VADAnalyzer, + speech_activity_period: float = 0.2, + **kwargs, + ): + """Initialize the VAD processor. + + Args: + vad_analyzer: The VADAnalyzer instance for processing audio. + speech_activity_period: Minimum interval in seconds between + UserSpeakingFrame pushes. Defaults to 0.2. + **kwargs: Additional arguments passed to parent class. + """ + super().__init__(**kwargs) + self._vad_controller = VADController( + vad_analyzer, speech_activity_period=speech_activity_period + ) + + # Push VAD frames when speech events are detected + @self._vad_controller.event_handler("on_speech_started") + async def on_speech_started(_controller): + logger.debug(f"{self}: User started speaking") + await self.push_frame(VADUserStartedSpeakingFrame()) + + @self._vad_controller.event_handler("on_speech_stopped") + async def on_speech_stopped(_controller): + logger.debug(f"{self}: User stopped speaking") + await self.push_frame(VADUserStoppedSpeakingFrame()) + + @self._vad_controller.event_handler("on_speech_activity") + async def on_speech_activity(_controller): + await self.push_frame(UserSpeakingFrame()) + + # Wire up frame pushing from controller to processor + @self._vad_controller.event_handler("on_push_frame") + async def on_push_frame(_controller, frame: Frame, direction: FrameDirection): + await self.push_frame(frame, direction) + + @self._vad_controller.event_handler("on_broadcast_frame") + async def on_broadcast_frame(_controller, frame_cls: Type[Frame], **kwargs): + await self.broadcast_frame(frame_cls, **kwargs) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process a frame through VAD and forward it. + + Args: + frame: The frame to process. + direction: The direction of frame flow in the pipeline. + """ + await super().process_frame(frame, direction) + + # Let the VAD controller handle the frame + await self._vad_controller.process_frame(frame) + + # Always forward the frame + await self.push_frame(frame, direction) diff --git a/tests/test_vad_processor.py b/tests/test_vad_processor.py new file mode 100644 index 000000000..2acb4f29d --- /dev/null +++ b/tests/test_vad_processor.py @@ -0,0 +1,143 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import unittest +from typing import List + +from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADState +from pipecat.frames.frames import ( + InputAudioRawFrame, + UserSpeakingFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, +) +from pipecat.processors.audio.vad_processor import VADProcessor +from pipecat.tests.utils import run_test + + +class MockVADAnalyzer(VADAnalyzer): + """A mock VAD analyzer that returns states from a predefined sequence.""" + + def __init__(self, states: List[VADState]): + super().__init__(sample_rate=16000) + self._states = list(states) + self._call_index = 0 + + def num_frames_required(self) -> int: + return 512 + + def voice_confidence(self, buffer: bytes) -> float: + return 0.9 + + async def analyze_audio(self, buffer: bytes) -> VADState: + if self._call_index < len(self._states): + state = self._states[self._call_index] + self._call_index += 1 + return state + return VADState.QUIET + + +class TestVADProcessor(unittest.IsolatedAsyncioTestCase): + def _make_audio_frame(self): + return InputAudioRawFrame(audio=b"\x00" * 1024, sample_rate=16000, num_channels=1) + + async def test_forwards_audio_frames(self): + """Test that audio frames are forwarded downstream.""" + analyzer = MockVADAnalyzer([VADState.QUIET]) + processor = VADProcessor(vad_analyzer=analyzer) + + await run_test( + processor, + frames_to_send=[self._make_audio_frame()], + expected_down_frames=[InputAudioRawFrame], + ) + + async def test_pushes_started_speaking_frame(self): + """Test that VADUserStartedSpeakingFrame is pushed when speech starts.""" + analyzer = MockVADAnalyzer([VADState.QUIET, VADState.SPEAKING]) + processor = VADProcessor(vad_analyzer=analyzer) + + await run_test( + processor, + frames_to_send=[self._make_audio_frame(), self._make_audio_frame()], + expected_down_frames=[ + InputAudioRawFrame, + VADUserStartedSpeakingFrame, + UserSpeakingFrame, + InputAudioRawFrame, + ], + ) + + async def test_pushes_stopped_speaking_frame(self): + """Test that VADUserStoppedSpeakingFrame is pushed when speech stops.""" + analyzer = MockVADAnalyzer([VADState.SPEAKING, VADState.QUIET]) + processor = VADProcessor(vad_analyzer=analyzer) + + await run_test( + processor, + frames_to_send=[self._make_audio_frame(), self._make_audio_frame()], + expected_down_frames=[ + VADUserStartedSpeakingFrame, + UserSpeakingFrame, + InputAudioRawFrame, + VADUserStoppedSpeakingFrame, + InputAudioRawFrame, + ], + ) + + async def test_pushes_user_speaking_frame(self): + """Test that UserSpeakingFrame is pushed while speaking.""" + analyzer = MockVADAnalyzer([VADState.SPEAKING, VADState.SPEAKING]) + processor = VADProcessor(vad_analyzer=analyzer) + + await run_test( + processor, + frames_to_send=[self._make_audio_frame(), self._make_audio_frame()], + expected_down_frames=[ + VADUserStartedSpeakingFrame, + UserSpeakingFrame, + InputAudioRawFrame, + UserSpeakingFrame, + InputAudioRawFrame, + ], + ) + + async def test_no_vad_frames_on_starting_state(self): + """Test that STARTING state doesn't push VAD frames.""" + analyzer = MockVADAnalyzer([VADState.STARTING]) + processor = VADProcessor(vad_analyzer=analyzer) + + await run_test( + processor, + frames_to_send=[self._make_audio_frame()], + expected_down_frames=[InputAudioRawFrame], + ) + + async def test_no_vad_frames_on_stopping_state(self): + """Test that STOPPING state doesn't push VAD frames.""" + analyzer = MockVADAnalyzer([VADState.STOPPING]) + processor = VADProcessor(vad_analyzer=analyzer) + + await run_test( + processor, + frames_to_send=[self._make_audio_frame()], + expected_down_frames=[InputAudioRawFrame], + ) + + async def test_no_vad_frames_when_quiet(self): + """Test that no VAD frames are pushed when staying quiet.""" + analyzer = MockVADAnalyzer([VADState.QUIET, VADState.QUIET]) + processor = VADProcessor(vad_analyzer=analyzer) + + await run_test( + processor, + frames_to_send=[self._make_audio_frame(), self._make_audio_frame()], + expected_down_frames=[InputAudioRawFrame, InputAudioRawFrame], + ) + + +if __name__ == "__main__": + unittest.main()