fix VAD stuck in SPEAKING state when audio stops mid-speech

src/pipecat/audio/vad/vad_controller.py

@@ -10,8 +10,11 @@ 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 asyncio
 import time
-from typing import Type
+from typing import Optional, Type
+
+from loguru import logger
 
 from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADState
 from pipecat.frames.frames import (
@@ -22,6 +25,7 @@ from pipecat.frames.frames import (
     VADParamsUpdateFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
+from pipecat.utils.asyncio.task_manager import BaseTaskManager
 from pipecat.utils.base_object import BaseObject
 
 
@@ -35,7 +39,8 @@ class VADController(BaseObject):
     Event handlers available:
 
     - on_speech_started: Called when speech begins.
-    - on_speech_stopped: Called when speech ends.
+    - on_speech_stopped: Called when speech ends, including forced stop on
+      audio starvation (no frames received while speaking).
     - 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.
@@ -63,13 +68,23 @@ class VADController(BaseObject):
             ...
     """
 
-    def __init__(self, vad_analyzer: VADAnalyzer, *, speech_activity_period: float = 0.2):
+    def __init__(
+        self,
+        vad_analyzer: VADAnalyzer,
+        *,
+        speech_activity_period: float = 0.2,
+        audio_starvation_timeout: float = 1.0,
+    ):
         """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.
+            audio_starvation_timeout: Timeout in seconds to force speech stop
+                when no audio frames are received while in SPEAKING state.
+                This handles cases like mic mute mid-speech.
+                Set to 0 to disable. Defaults to 1.0.
         """
         super().__init__()
         self._vad_analyzer = vad_analyzer
@@ -81,12 +96,33 @@ class VADController(BaseObject):
         # be greater than the audio chunks to have any effect).
         self._speech_activity_period = speech_activity_period
 
+        # Audio starvation detection: force speech stop when no audio arrives
+        # while in SPEAKING state (e.g. user mutes mic mid-speech).
+        self._audio_starvation_timeout = audio_starvation_timeout
+        self._task_manager: Optional[BaseTaskManager] = None
+        self._audio_received_event = asyncio.Event()
+        self._starvation_task: Optional[asyncio.Task] = None
+
         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 setup(self, task_manager: BaseTaskManager):
+        """Initialize the controller with the given task manager.
+
+        Args:
+            task_manager: The task manager to be associated with this instance.
+        """
+        self._task_manager = task_manager
+        self._audio_received_event.clear()
+        if self._audio_starvation_timeout > 0 and not self._starvation_task:
+            self._starvation_task = self._task_manager.create_task(
+                self._audio_starvation_handler(),
+                f"{self}::_audio_starvation_handler",
+            )
+
     async def process_frame(self, frame: Frame):
         """Process a frame and handle VAD-related events.
 
@@ -116,6 +152,10 @@ class VADController(BaseObject):
         It waits for all currently executing event handler tasks to finish
         before returning.
         """
+        await super().cleanup()
+        if self._starvation_task and self._task_manager:
+            await self._task_manager.cancel_task(self._starvation_task)
+            self._starvation_task = None
         if self._vad_analyzer:
             await self._vad_analyzer.cleanup()
 
@@ -128,6 +168,8 @@ class VADController(BaseObject):
         Args:
             frame: Audio frame to process.
         """
+        self._audio_received_event.set()
+
         self._vad_state = await self._handle_vad(frame.audio, self._vad_state)
 
         if self._vad_state == VADState.SPEAKING:
@@ -149,6 +191,26 @@ class VADController(BaseObject):
             vad_state = new_vad_state
         return vad_state
 
+    async def _audio_starvation_handler(self):
+        """Monitor for audio starvation while in SPEAKING state.
+
+        When no audio frames arrive for `audio_starvation_timeout` seconds
+        (e.g. user mutes mic mid-speech), forces a transition to QUIET and
+        emits `on_speech_stopped`.
+        """
+        while True:
+            try:
+                await asyncio.wait_for(
+                    self._audio_received_event.wait(),
+                    timeout=self._audio_starvation_timeout,
+                )
+                self._audio_received_event.clear()
+            except asyncio.TimeoutError:
+                if self._vad_state == VADState.SPEAKING:
+                    logger.warning(f"{self}: no audio received while speaking, forcing speech stop")
+                    self._vad_state = VADState.QUIET
+                    await self._call_event_handler("on_speech_stopped")
+
     async def _maybe_speech_activity(self):
         """Handle user speaking frame."""
         diff_time = time.time() - self._speech_activity_time

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -107,6 +107,9 @@ class LLMUserAggregatorParams:
             has been idle (not speaking) for this duration. Set to 0 to disable
             idle detection.
         vad_analyzer: Voice Activity Detection analyzer instance.
+        audio_starvation_timeout: Timeout in seconds to force speech stop when
+            no audio frames are received while in SPEAKING state (e.g. user mutes
+            mic mid-speech). Set to 0 to disable. Defaults to 1.0.
         filter_incomplete_user_turns: Whether to filter out incomplete user turns.
             When enabled, the LLM outputs a turn completion marker at the start of
             each response: ✓ (complete), ○ (incomplete short), or ◐ (incomplete long).
@@ -121,6 +124,7 @@ class LLMUserAggregatorParams:
     user_turn_stop_timeout: float = 5.0
     user_idle_timeout: float = 0
     vad_analyzer: Optional[VADAnalyzer] = None
+    audio_starvation_timeout: float = 1.0
     filter_incomplete_user_turns: bool = False
     user_turn_completion_config: Optional[UserTurnCompletionConfig] = None
 
@@ -467,7 +471,10 @@ class LLMUserAggregator(LLMContextAggregator):
         # VAD controller
         self._vad_controller: Optional[VADController] = None
         if self._params.vad_analyzer:
-            self._vad_controller = VADController(self._params.vad_analyzer)
+            self._vad_controller = VADController(
+                self._params.vad_analyzer,
+                audio_starvation_timeout=self._params.audio_starvation_timeout,
+            )
             self._vad_controller.add_event_handler("on_speech_started", self._on_vad_speech_started)
             self._vad_controller.add_event_handler("on_speech_stopped", self._on_vad_speech_stopped)
             self._vad_controller.add_event_handler(
@@ -553,6 +560,9 @@ class LLMUserAggregator(LLMContextAggregator):
         return aggregation
 
     async def _start(self, frame: StartFrame):
+        if self._vad_controller:
+            await self._vad_controller.setup(self.task_manager)
+
         await self._user_turn_controller.setup(self.task_manager)
 
         await self._user_idle_controller.setup(self.task_manager)
@@ -584,6 +594,8 @@ class LLMUserAggregator(LLMContextAggregator):
         await self._cleanup()
 
     async def _cleanup(self):
+        if self._vad_controller:
+            await self._vad_controller.cleanup()
         await self._user_turn_controller.cleanup()
         await self._user_idle_controller.cleanup()
 

src/pipecat/processors/audio/vad_processor.py

@@ -18,6 +18,7 @@ from pipecat.audio.vad.vad_analyzer import VADAnalyzer
 from pipecat.audio.vad.vad_controller import VADController
 from pipecat.frames.frames import (
     Frame,
+    StartFrame,
     UserSpeakingFrame,
     VADUserStartedSpeakingFrame,
     VADUserStoppedSpeakingFrame,
@@ -45,6 +46,7 @@ class VADProcessor(FrameProcessor):
         *,
         vad_analyzer: VADAnalyzer,
         speech_activity_period: float = 0.2,
+        audio_starvation_timeout: float = 1.0,
         **kwargs,
     ):
         """Initialize the VAD processor.
@@ -53,11 +55,16 @@ class VADProcessor(FrameProcessor):
             vad_analyzer: The VADAnalyzer instance for processing audio.
             speech_activity_period: Minimum interval in seconds between
                 UserSpeakingFrame pushes. Defaults to 0.2.
+            audio_starvation_timeout: Timeout in seconds to force speech stop
+                when no audio frames are received while in SPEAKING state.
+                Set to 0 to disable. Defaults to 1.0.
             **kwargs: Additional arguments passed to parent class.
         """
         super().__init__(**kwargs)
         self._vad_controller = VADController(
-            vad_analyzer, speech_activity_period=speech_activity_period
+            vad_analyzer,
+            speech_activity_period=speech_activity_period,
+            audio_starvation_timeout=audio_starvation_timeout,
         )
 
         # Push VAD frames when speech events are detected
@@ -90,6 +97,11 @@ class VADProcessor(FrameProcessor):
         async def on_broadcast_frame(_controller, frame_cls: Type[Frame], **kwargs):
             await self.broadcast_frame(frame_cls, **kwargs)
 
+    async def cleanup(self):
+        """Clean up VAD controller resources."""
+        await super().cleanup()
+        await self._vad_controller.cleanup()
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process a frame through VAD and forward it.
 
@@ -104,5 +116,8 @@ class VADProcessor(FrameProcessor):
         # 2. Audio flows through immediately while VAD detection happens after
         await self.push_frame(frame, direction)
 
+        if isinstance(frame, StartFrame):
+            await self._vad_controller.setup(self.task_manager)
+
         # Let the VAD controller handle the frame
         await self._vad_controller.process_frame(frame)

tests/test_vad_controller.py

@@ -4,6 +4,7 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+import asyncio
 import unittest
 from typing import List
 
@@ -11,6 +12,7 @@ from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADParams, VADState
 from pipecat.audio.vad.vad_controller import VADController
 from pipecat.frames.frames import Frame, InputAudioRawFrame, SpeechControlParamsFrame, StartFrame
 from pipecat.processors.frame_processor import FrameDirection
+from pipecat.utils.asyncio.task_manager import TaskManager, TaskManagerParams
 
 
 class MockVADAnalyzer(VADAnalyzer):
@@ -206,5 +208,64 @@ class TestVADController(unittest.IsolatedAsyncioTestCase):
         self.assertIsInstance(broadcast_calls[0][1]["vad_params"], VADParams)
 
 
+AUDIO_STARVATION_TIMEOUT = 0.1
+
+
+class TestVADControllerStarvation(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.task_manager = TaskManager()
+        self.task_manager.setup(TaskManagerParams(loop=asyncio.get_running_loop()))
+
+    async def test_audio_starvation_forces_speech_stop(self):
+        """Test that on_speech_stopped fires when no audio arrives while SPEAKING."""
+        analyzer = MockVADAnalyzer()
+        controller = VADController(analyzer, audio_starvation_timeout=AUDIO_STARVATION_TIMEOUT)
+
+        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)
+        await controller.setup(self.task_manager)
+
+        # Enter SPEAKING state
+        audio_frame = InputAudioRawFrame(audio=b"\x00" * 1024, sample_rate=16000, num_channels=1)
+        analyzer.set_next_state(VADState.SPEAKING)
+        await controller.process_frame(audio_frame)
+        self.assertFalse(speech_stopped)
+
+        # Stop sending audio, wait for starvation timeout
+        await asyncio.sleep(AUDIO_STARVATION_TIMEOUT + 0.1)
+        self.assertTrue(speech_stopped)
+
+        await controller.cleanup()
+
+    async def test_audio_starvation_does_not_fire_when_quiet(self):
+        """Test that starvation timeout does NOT fire when VAD is in QUIET state."""
+        analyzer = MockVADAnalyzer()
+        controller = VADController(analyzer, audio_starvation_timeout=AUDIO_STARVATION_TIMEOUT)
+
+        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)
+        await controller.setup(self.task_manager)
+
+        # Stay in QUIET state, wait past starvation timeout
+        await asyncio.sleep(AUDIO_STARVATION_TIMEOUT + 0.1)
+        self.assertFalse(speech_stopped)
+
+        await controller.cleanup()
+
+
 if __name__ == "__main__":
     unittest.main()