Merge pull request #665 from pipecat-ai/aleix/fix-bot-started-stopped-speaking

transports(base_output): fix constant bot started/stopped speaking fr…

CHANGELOG.md

@@ -24,6 +24,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an issue that was generating constant bot started/stopped speaking
+  frames for HTTP TTS services.
+
 - Fixed an issue that was causing stuttering with AWS TTS service.
 
 ## [0.0.47] - 2024-10-22

src/pipecat/audio/vad/vad_analyzer.py

@@ -12,6 +12,11 @@ from pydantic.main import BaseModel
 
 from pipecat.audio.utils import calculate_audio_volume, exp_smoothing
 
+VAD_CONFIDENCE = 0.7
+VAD_START_SECS = 0.2
+VAD_STOP_SECS = 0.8
+VAD_MIN_VOLUME = 0.6
+
 
 class VADState(Enum):
     QUIET = 1
@@ -21,10 +26,10 @@ class VADState(Enum):
 
 
 class VADParams(BaseModel):
-    confidence: float = 0.7
-    start_secs: float = 0.2
-    stop_secs: float = 0.8
-    min_volume: float = 0.6
+    confidence: float = VAD_CONFIDENCE
+    start_secs: float = VAD_START_SECS
+    stop_secs: float = VAD_STOP_SECS
+    min_volume: float = VAD_MIN_VOLUME
 
 
 class VADAnalyzer:
@@ -41,13 +46,17 @@ class VADAnalyzer:
         self._prev_volume = 0
 
     @property
-    def sample_rate(self):
+    def sample_rate(self) -> int:
         return self._sample_rate
 
     @property
-    def num_channels(self):
+    def num_channels(self) -> int:
         return self._num_channels
 
+    @property
+    def params(self) -> VADParams:
+        return self._params
+
     @abstractmethod
     def num_frames_required(self) -> int:
         pass

src/pipecat/transports/base_output.py

@@ -13,6 +13,7 @@ from typing import List
 from loguru import logger
 from PIL import Image
 
+from pipecat.audio.vad.vad_analyzer import VAD_STOP_SECS
 from pipecat.frames.frames import (
     BotSpeakingFrame,
     BotStartedSpeakingFrame,
@@ -70,9 +71,14 @@ class BaseOutputTransport(FrameProcessor):
         self._stopped_event = asyncio.Event()
 
         # Indicates if the bot is currently speaking. This is useful when we
-        # have an interruption since all the queued messages will be thrown
-        # away and we would lose the TTSStoppedFrame.
+        # have an interruption since all the queued messages will be thrown away
+        # and we would lose the TTSStoppedFrame. We also keep a scheduler handle
+        # which is used to schedule pushing bot stopped speaking frames. With
+        # some services (HTTP TTS services) we will get a bunch of
+        # TTSStartedFrame and TTSStoppedFrame and we don't want to constantly
+        # generate bot started/stopped speaking frames.
         self._bot_speaking = False
+        self._bot_stopped_speaking_handle = None
 
         # Create sink frame task. This is the task that will actually write
         # audio or video frames. We write audio/video in a task so we can keep
@@ -320,14 +326,40 @@ class BaseOutputTransport(FrameProcessor):
                 logger.exception(f"{self} error processing sink clock queue: {e}")
 
     async def _bot_started_speaking(self):
-        logger.debug("Bot started speaking")
-        self._bot_speaking = True
-        await self.push_frame(BotStartedSpeakingFrame(), FrameDirection.UPSTREAM)
+        # If we scheduled bot stopped speaking, cancel it.
+        if self._bot_stopped_speaking_handle:
+            self._bot_stopped_speaking_handle.cancel()
+            self._bot_stopped_speaking_handle = None
+
+        if not self._bot_speaking:
+            logger.debug("Bot started speaking")
+            await self.push_frame(BotStartedSpeakingFrame(), FrameDirection.UPSTREAM)
+            self._bot_speaking = True
 
     async def _bot_stopped_speaking(self):
-        logger.debug("Bot stopped speaking")
-        self._bot_speaking = False
-        await self.push_frame(BotStoppedSpeakingFrame(), FrameDirection.UPSTREAM)
+        async def push_bot_stopped_speaking_async():
+            try:
+                logger.debug("Bot stopped speaking")
+                await self.push_frame(BotStoppedSpeakingFrame(), FrameDirection.UPSTREAM)
+                self._bot_speaking = False
+            except asyncio.CancelledError:
+                logger.debug("We've been cancelled")
+                pass
+
+        def push_bot_stopped_speaking():
+            self.get_event_loop().create_task(push_bot_stopped_speaking_async())
+
+        # Schedule a bot stopped speaking to be pushed after VAD stop secs
+        # (unless we get another bot started speaking).
+        wait_time = (
+            self._params.vad_analyzer.params.stop_secs
+            if self._params.vad_analyzer
+            else VAD_STOP_SECS
+        )
+        if not self._bot_stopped_speaking_handle:
+            self._bot_stopped_speaking_handle = self.get_event_loop().call_later(
+                wait_time, push_bot_stopped_speaking
+            )
 
     #
     # Camera out