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. # diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index a19d65b75..c4f59a61d 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -403,7 +403,7 @@ class BaseOutputTransport(FrameProcessor): # Last time a BotSpeakingFrame was pushed. self._bot_speaking_frame_time = 0 # How often a BotSpeakingFrame should be pushed (value should be - # lower than the audio chunks). + # greater than the audio chunks to have any effect). self._bot_speaking_frame_period = 0.2 # Last time the bot actually spoke. self._bot_speech_last_time = 0 @@ -644,8 +644,7 @@ class BaseOutputTransport(FrameProcessor): diff_time = time.time() - self._bot_speaking_frame_time if diff_time >= self._bot_speaking_frame_period: - await self._transport.push_frame(BotSpeakingFrame()) - await self._transport.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM) + await self._transport.broadcast_frame(BotSpeakingFrame) self._bot_speaking_frame_time = time.time() self._bot_speech_last_time = time.time()