From ddfedaf478bac23b5548dba4bc5482f7633d3478 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 28 Jan 2026 10:13:34 -0800 Subject: [PATCH] audio(vad): add new VADController --- src/pipecat/audio/vad/vad_controller.py | 169 +++++++++++++++++++++ tests/test_vad_controller.py | 190 ++++++++++++++++++++++++ 2 files changed, 359 insertions(+) create mode 100644 src/pipecat/audio/vad/vad_controller.py create mode 100644 tests/test_vad_controller.py diff --git a/src/pipecat/audio/vad/vad_controller.py b/src/pipecat/audio/vad/vad_controller.py new file mode 100644 index 000000000..66f7199e5 --- /dev/null +++ b/src/pipecat/audio/vad/vad_controller.py @@ -0,0 +1,169 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Voice Activity Detection controller for managing speech state transitions. + +This module provides a controller that wraps a VADAnalyzer to track speech state +and emit events when speech starts, stops, or is actively detected. +""" + +import time +from typing import Type + +from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADState +from pipecat.frames.frames import ( + Frame, + InputAudioRawFrame, + SpeechControlParamsFrame, + StartFrame, + VADParamsUpdateFrame, +) +from pipecat.processors.frame_processor import FrameDirection +from pipecat.utils.base_object import BaseObject + + +class VADController(BaseObject): + """Manages voice activity detection state and emits speech events. + + Wraps a `VADAnalyzer` to process audio and trigger events based on speech + state transitions. Tracks whether the user is speaking, quiet, or + transitioning between states. + + Event handlers available: + + - on_speech_started: Called when speech begins. + - on_speech_stopped: Called when speech ends. + - on_speech_activity: Called periodically while speech is detected. + - on_push_frame: Called when the controller wants to push a frame. + - on_broadcast_frame: Called when the controller wants to broadcast a frame. + + Example:: + + @vad_controller.event_handler("on_speech_started") + async def on_speech_started(controller): + ... + + @vad_controller.event_handler("on_speech_stopped") + async def on_speech_stopped(controller): + ... + + @vad_controller.event_handler("on_speech_activity") + async def on_speech_activity(controller): + ... + + @vad_controller.event_handler("on_push_frame") + async def on_push_frame(controller, frame: Frame, direction: FrameDirection): + ... + + @vad_controller.event_handler("on_broadcast_frame") + async def on_broadcast_frame(controller, frame_cls: Type[Frame], **kwargs): + ... + """ + + def __init__(self, vad_analyzer: VADAnalyzer, *, speech_activity_period: float = 0.2): + """Initialize the VAD controller. + + Args: + vad_analyzer: The `VADAnalyzer` instance for processing audio. + speech_activity_period: Minimum interval in seconds between + `on_speech_activity` events. Defaults to 0.2. + """ + super().__init__() + self._vad_analyzer = vad_analyzer + self._vad_state: VADState = VADState.QUIET + + # Last time a on_speech_activity was triggered. + self._speech_activity_time = 0 + # How often a on_speech_activity event should be triggered (value should + # be greater than the audio chunks to have any effect). + self._speech_activity_period = speech_activity_period + + self._register_event_handler("on_speech_started", sync=True) + self._register_event_handler("on_speech_stopped", sync=True) + self._register_event_handler("on_speech_activity", sync=True) + self._register_event_handler("on_push_frame", sync=True) + self._register_event_handler("on_broadcast_frame", sync=True) + + async def process_frame(self, frame: Frame): + """Process a frame and handle VAD-related events. + + Handles `StartFrame` to initialize the sample rate and `InputAudioRawFrame` + to analyze audio for voice activity. + + Args: + frame: The frame to process. + """ + if isinstance(frame, StartFrame): + await self._start(frame) + elif isinstance(frame, InputAudioRawFrame): + await self._handle_audio(frame) + elif isinstance(frame, VADParamsUpdateFrame): + self._vad_analyzer.set_params(frame.params) + await self.broadcast_frame(SpeechControlParamsFrame, vad_params=frame.params) + + async def _start(self, frame: StartFrame): + self._vad_analyzer.set_sample_rate(frame.audio_in_sample_rate) + + async def _handle_audio(self, frame: InputAudioRawFrame): + """Process an audio chunk and emit speech events as needed. + + Analyzes the audio for voice activity and triggers `on_speech_started`, + `on_speech_stopped`, or `on_speech_activity` events based on state changes. + + Args: + frame: Audio frame to process. + """ + self._vad_state = await self._handle_vad(frame.audio, self._vad_state) + + if self._vad_state == VADState.SPEAKING: + await self._call_event_handler("on_speech_activity") + + async def _handle_vad(self, audio: bytes, vad_state: VADState) -> VADState: + """Handle Voice Activity Detection results and trigger appropriate events.""" + new_vad_state = await self._vad_analyzer.analyze_audio(audio) + if ( + new_vad_state != vad_state + and new_vad_state != VADState.STARTING + and new_vad_state != VADState.STOPPING + ): + if new_vad_state == VADState.SPEAKING: + await self._call_event_handler("on_speech_started") + elif new_vad_state == VADState.QUIET: + await self._call_event_handler("on_speech_stopped") + + vad_state = new_vad_state + return vad_state + + async def _maybe_speech_activity(self): + """Handle user speaking frame.""" + diff_time = time.time() - self._speech_activity_time + if diff_time >= self._speech_activity_period: + self._speech_activity_time = time.time() + await self._call_event_handler("on_speech_activity") + + async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): + """Request a frame to be pushed through the pipeline. + + This emits an on_push_frame event that must be handled by a processor + to actually push the frame into the pipeline. + + Args: + frame: The frame to push. + direction: The direction to push the frame. + """ + await self._call_event_handler("on_push_frame", frame, direction) + + async def broadcast_frame(self, frame_cls: Type[Frame], **kwargs): + """Request a frame to be broadcast upstream and downstream. + + This emits an on_broadcast_frame event that must be handled by a processor + to actually broadcast the frame in the pipeline. + + Args: + frame_cls: The class of the frame to broadcast. + **kwargs: Arguments to pass to the frame constructor. + """ + await self._call_event_handler("on_broadcast_frame", frame_cls, **kwargs) diff --git a/tests/test_vad_controller.py b/tests/test_vad_controller.py new file mode 100644 index 000000000..b649a906f --- /dev/null +++ b/tests/test_vad_controller.py @@ -0,0 +1,190 @@ +# +# 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.audio.vad.vad_controller import VADController +from pipecat.frames.frames import Frame, InputAudioRawFrame, StartFrame +from pipecat.processors.frame_processor import FrameDirection + + +class MockVADAnalyzer(VADAnalyzer): + """A mock VAD analyzer that returns a configurable state.""" + + def __init__(self): + """Initialize with default QUIET state.""" + super().__init__(sample_rate=16000) + self._next_state = VADState.QUIET + + def set_next_state(self, state: VADState): + """Set the state to return on the next analyze_audio call. + + Args: + state: The VADState to return. + """ + self._next_state = state + + 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: + """Return the configured state.""" + return self._next_state + + +class TestVADController(unittest.IsolatedAsyncioTestCase): + async def test_speech_started_event(self): + """Test that on_speech_started event is triggered when speech begins.""" + analyzer = MockVADAnalyzer() + controller = VADController(analyzer) + + speech_started = False + + @controller.event_handler("on_speech_started") + async def on_speech_started(_controller): + nonlocal speech_started + speech_started = True + + start_frame = StartFrame(audio_in_sample_rate=16000, audio_out_sample_rate=16000) + await controller.process_frame(start_frame) + + audio_frame = InputAudioRawFrame(audio=b"\x00" * 1024, sample_rate=16000, num_channels=1) + + # Process with QUIET state - no event should fire + analyzer.set_next_state(VADState.QUIET) + await controller.process_frame(audio_frame) + self.assertFalse(speech_started) + + # Process with SPEAKING state - event should fire + analyzer.set_next_state(VADState.SPEAKING) + await controller.process_frame(audio_frame) + self.assertTrue(speech_started) + + async def test_speech_stopped_event(self): + """Test that on_speech_stopped event is triggered when speech ends.""" + analyzer = MockVADAnalyzer() + controller = VADController(analyzer) + + speech_stopped = False + + @controller.event_handler("on_speech_stopped") + async def on_speech_stopped(_controller): + nonlocal speech_stopped + speech_stopped = True + + start_frame = StartFrame(audio_in_sample_rate=16000, audio_out_sample_rate=16000) + await controller.process_frame(start_frame) + + audio_frame = InputAudioRawFrame(audio=b"\x00" * 1024, sample_rate=16000, num_channels=1) + + # Start speaking + analyzer.set_next_state(VADState.SPEAKING) + await controller.process_frame(audio_frame) + self.assertFalse(speech_stopped) + + # Stop speaking - event should fire + analyzer.set_next_state(VADState.QUIET) + await controller.process_frame(audio_frame) + self.assertTrue(speech_stopped) + + async def test_speech_activity_event(self): + """Test that on_speech_activity event is triggered while speaking.""" + analyzer = MockVADAnalyzer() + controller = VADController(analyzer) + + activity_count = 0 + + @controller.event_handler("on_speech_activity") + async def on_speech_activity(_controller): + nonlocal activity_count + activity_count += 1 + + start_frame = StartFrame(audio_in_sample_rate=16000, audio_out_sample_rate=16000) + await controller.process_frame(start_frame) + + audio_frame = InputAudioRawFrame(audio=b"\x00" * 1024, sample_rate=16000, num_channels=1) + + # Activity events fire while in SPEAKING state + analyzer.set_next_state(VADState.SPEAKING) + await controller.process_frame(audio_frame) + await controller.process_frame(audio_frame) + self.assertEqual(activity_count, 2) + + async def test_push_frame_event(self): + """Test that push_frame emits on_push_frame event.""" + analyzer = MockVADAnalyzer() + controller = VADController(analyzer) + + pushed_frames: List[tuple] = [] + + @controller.event_handler("on_push_frame") + async def on_push_frame(_controller, frame: Frame, direction: FrameDirection): + pushed_frames.append((frame, direction)) + + test_frame = InputAudioRawFrame(audio=b"\x00" * 1024, sample_rate=16000, num_channels=1) + await controller.push_frame(test_frame, FrameDirection.DOWNSTREAM) + + self.assertEqual(len(pushed_frames), 1) + self.assertEqual(pushed_frames[0][0], test_frame) + self.assertEqual(pushed_frames[0][1], FrameDirection.DOWNSTREAM) + + async def test_broadcast_frame_event(self): + """Test that broadcast_frame emits on_broadcast_frame event.""" + analyzer = MockVADAnalyzer() + controller = VADController(analyzer) + + broadcast_calls: List[tuple] = [] + + @controller.event_handler("on_broadcast_frame") + async def on_broadcast_frame(_controller, frame_cls, **kwargs): + broadcast_calls.append((frame_cls, kwargs)) + + await controller.broadcast_frame( + InputAudioRawFrame, audio=b"\x00", sample_rate=16000, num_channels=1 + ) + + self.assertEqual(len(broadcast_calls), 1) + self.assertEqual(broadcast_calls[0][0], InputAudioRawFrame) + self.assertEqual(broadcast_calls[0][1]["sample_rate"], 16000) + + async def test_no_event_on_transitional_states(self): + """Test that STARTING and STOPPING states don't trigger events.""" + analyzer = MockVADAnalyzer() + controller = VADController(analyzer) + + events_triggered = [] + + @controller.event_handler("on_speech_started") + async def on_speech_started(_controller): + events_triggered.append("started") + + @controller.event_handler("on_speech_stopped") + async def on_speech_stopped(_controller): + events_triggered.append("stopped") + + start_frame = StartFrame(audio_in_sample_rate=16000, audio_out_sample_rate=16000) + await controller.process_frame(start_frame) + + audio_frame = InputAudioRawFrame(audio=b"\x00" * 1024, sample_rate=16000, num_channels=1) + + # STARTING is a transitional state - no event + analyzer.set_next_state(VADState.STARTING) + await controller.process_frame(audio_frame) + self.assertEqual(events_triggered, []) + + # STOPPING is a transitional state - no event + analyzer.set_next_state(VADState.STOPPING) + await controller.process_frame(audio_frame) + self.assertEqual(events_triggered, []) + + +if __name__ == "__main__": + unittest.main()