From 6eadad53b29d43cf00f7fd2609b899ac4af07818 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 16 Jan 2026 14:45:30 -0800 Subject: [PATCH 1/2] BaseInputTransport: throttle UserSpeakingFrame --- changelog/3483.changed.md | 1 + src/pipecat/transports/base_input.py | 15 ++++++++++++++- 2 files changed, 15 insertions(+), 1 deletion(-) create mode 100644 changelog/3483.changed.md 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. # From ac3fa7f91f42394f33036ea01ad8c0f08d2b9890 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 16 Jan 2026 14:46:25 -0800 Subject: [PATCH 2/2] BaseOuputTransport: minor cleanup --- src/pipecat/transports/base_output.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) 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()