diff --git a/changelog/3483.changed.md b/changelog/3483.changed.md new file mode 100644 index 000000000..97bb10371 --- /dev/null +++ b/changelog/3483.changed.md @@ -0,0 +1 @@ +- Throttle `UserSpeakingFrame` to broadcast at most every 200ms instead of on every audio chunk, reducing frame processing overhead during user speech. diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 31bc3e035..b5f5a17c7 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -11,6 +11,7 @@ input processing, including VAD, turn analysis, and interruption management. """ import asyncio +import time from typing import Optional from loguru import logger @@ -77,6 +78,11 @@ class BaseInputTransport(FrameProcessor): # Track user speaking state for interruption logic self._user_speaking = False + # Last time a UserSpeakingFrame was pushed. + self._user_speaking_frame_time = 0 + # How often a UserSpeakingFrame should be pushed (value should be + # greater than the audio chunks to have any effect). + self._user_speaking_frame_period = 0.2 # Task to process incoming audio (VAD) and push audio frames downstream # if passthrough is enabled. @@ -423,7 +429,7 @@ class BaseInputTransport(FrameProcessor): await self._deprecated_run_turn_analyzer(frame, vad_state, previous_vad_state) if vad_state == VADState.SPEAKING: - await self.broadcast_frame(UserSpeakingFrame) + await self._user_currently_speaking() # Push audio downstream if passthrough is set. if self._params.audio_in_passthrough: @@ -444,6 +450,13 @@ class BaseInputTransport(FrameProcessor): else: await self.push_frame(VADUserStoppedSpeakingFrame()) + async def _user_currently_speaking(self): + """Handle user speaking frame.""" + diff_time = time.time() - self._user_speaking_frame_time + if diff_time >= self._user_speaking_frame_period: + await self.broadcast_frame(UserSpeakingFrame) + self._user_speaking_frame_time = time.time() + # # DEPRECATED. #