Merge pull request #716 from pipecat-ai/mb/mute-stt-service

Add STTMuteFilter to un/mute the STT

CHANGELOG.md

@@ -11,6 +11,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Added `RimeHttpTTSService` and the `07q-interruptible-rime.py` foundational
   example.
+- Added `STTMuteFilter`, a general-purpose processor that combines STT
+  muting and interruption control. When active, it prevents both transcription
+  and interruptions during bot speech. The processor supports multiple
+  strategies: `FIRST_SPEECH` (mute only during bot's first
+  speech), `ALWAYS` (mute during all bot speech), or `CUSTOM` (using provided
+  callback).
+- Added `STTMuteFrame`, a control frame that enables/disables speech
+  transcription in STT services.
 
 ## [0.0.48] - 2024-11-10 "Antonio release"
 

examples/foundational/24-stt-mute-filter.py

@@ -0,0 +1,98 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+from runner import configure
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import (
+    LLMMessagesFrame,
+)
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.filters.stt_mute_filter import STTMuteConfig, STTMuteFilter, STTMuteStrategy
+from pipecat.services.deepgram import DeepgramSTTService, DeepgramTTSService
+from pipecat.services.openai import OpenAILLMService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        (room_url, _) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            None,
+            "Respond bot",
+            DailyParams(
+                audio_out_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+                vad_audio_passthrough=True,
+            ),
+        )
+
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+        # Configure the mute processor to mute only during first speech
+        stt_mute_processor = STTMuteFilter(
+            stt_service=stt, config=STTMuteConfig(strategy=STTMuteStrategy.FIRST_SPEECH)
+        )
+
+        tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-helios-en")
+
+        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
+
+        messages = [
+            {
+                "role": "system",
+                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+            },
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        pipeline = Pipeline(
+            [
+                transport.input(),  # Transport user input
+                stt_mute_processor,  # Add the mute processor before STT
+                stt,  # STT
+                context_aggregator.user(),  # User responses
+                llm,  # LLM
+                tts,  # TTS
+                transport.output(),  # Transport bot output
+                context_aggregator.assistant(),  # Assistant spoken responses
+            ]
+        )
+
+        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            # Kick off the conversation.
+            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+            await task.queue_frames([LLMMessagesFrame(messages)])
+
+        runner = PipelineRunner()
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/pipecat/frames/frames.py

@@ -570,6 +570,13 @@ class TTSUpdateSettingsFrame(ServiceUpdateSettingsFrame):
     pass
 
 
+@dataclass
+class STTMuteFrame(ControlFrame):
+    """Control frame to mute/unmute the STT service."""
+
+    mute: bool
+
+
 @dataclass
 class STTUpdateSettingsFrame(ServiceUpdateSettingsFrame):
     pass

src/pipecat/processors/filters/stt_mute_filter.py

@@ -0,0 +1,111 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Awaitable, Callable, Optional
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    BotStartedSpeakingFrame,
+    BotStoppedSpeakingFrame,
+    Frame,
+    StartInterruptionFrame,
+    StopInterruptionFrame,
+    STTMuteFrame,
+    UserStartedSpeakingFrame,
+    UserStoppedSpeakingFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.services.ai_services import STTService
+
+
+class STTMuteStrategy(Enum):
+    FIRST_SPEECH = "first_speech"  # Mute only during first bot speech
+    ALWAYS = "always"  # Mute during all bot speech
+    CUSTOM = "custom"  # Allow custom logic via callback
+
+
+@dataclass
+class STTMuteConfig:
+    """Configuration for STTMuteFilter"""
+
+    strategy: STTMuteStrategy
+    # Optional callback for custom muting logic
+    should_mute_callback: Optional[Callable[["STTMuteFilter"], Awaitable[bool]]] = None
+
+
+class STTMuteFilter(FrameProcessor):
+    """A general-purpose processor that handles STT muting and interruption control.
+
+    This processor combines the concepts of STT muting and interruption control,
+    treating them as a single coordinated feature. When STT is muted, interruptions
+    are automatically disabled.
+    """
+
+    def __init__(self, stt_service: STTService, config: STTMuteConfig, **kwargs):
+        super().__init__(**kwargs)
+        self._stt_service = stt_service
+        self._config = config
+        self._first_speech_handled = False
+        self._bot_is_speaking = False
+
+    @property
+    def is_muted(self) -> bool:
+        """Returns whether STT is currently muted."""
+        return self._stt_service.is_muted
+
+    async def _handle_mute_state(self, should_mute: bool):
+        """Handles both STT muting and interruption control."""
+        if should_mute != self.is_muted:
+            logger.debug(f"STT {'muting' if should_mute else 'unmuting'}")
+            await self.push_frame(STTMuteFrame(mute=should_mute))
+
+    async def _should_mute(self) -> bool:
+        """Determines if STT should be muted based on current state and strategy."""
+        if not self._bot_is_speaking:
+            return False
+
+        if self._config.strategy == STTMuteStrategy.ALWAYS:
+            return True
+        elif (
+            self._config.strategy == STTMuteStrategy.FIRST_SPEECH and not self._first_speech_handled
+        ):
+            self._first_speech_handled = True
+            return True
+        elif self._config.strategy == STTMuteStrategy.CUSTOM and self._config.should_mute_callback:
+            return await self._config.should_mute_callback(self)
+
+        return False
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        # Handle bot speaking state changes
+        if isinstance(frame, BotStartedSpeakingFrame):
+            self._bot_is_speaking = True
+            await self._handle_mute_state(await self._should_mute())
+        elif isinstance(frame, BotStoppedSpeakingFrame):
+            self._bot_is_speaking = False
+            await self._handle_mute_state(await self._should_mute())
+
+        # Handle frame propagation
+        if isinstance(
+            frame,
+            (
+                StartInterruptionFrame,
+                StopInterruptionFrame,
+                UserStartedSpeakingFrame,
+                UserStoppedSpeakingFrame,
+            ),
+        ):
+            # Only pass VAD-related frames when not muted
+            if not self.is_muted:
+                await self.push_frame(frame, direction)
+            else:
+                logger.debug(f"{frame.__class__.__name__} suppressed - STT currently muted")
+        else:
+            # Pass all other frames through
+            await self.push_frame(frame, direction)

src/pipecat/services/ai_services.py

@@ -22,6 +22,7 @@ from pipecat.frames.frames import (
     LLMFullResponseEndFrame,
     StartFrame,
     StartInterruptionFrame,
+    STTMuteFrame,
     STTUpdateSettingsFrame,
     TextFrame,
     TTSAudioRawFrame,
@@ -454,6 +455,12 @@ class STTService(AIService):
         super().__init__(**kwargs)
         self._audio_passthrough = audio_passthrough
         self._settings: Dict[str, Any] = {}
+        self._muted: bool = False
+
+    @property
+    def is_muted(self) -> bool:
+        """Returns whether the STT service is currently muted."""
+        return self._muted
 
     @abstractmethod
     async def set_model(self, model: str):
@@ -482,7 +489,8 @@ class STTService(AIService):
                 logger.warning(f"Unknown setting for STT service: {key}")
 
     async def process_audio_frame(self, frame: AudioRawFrame):
-        await self.process_generator(self.run_stt(frame.audio))
+        if not self._muted:
+            await self.process_generator(self.run_stt(frame.audio))
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Processes a frame of audio data, either buffering or transcribing it."""
@@ -497,6 +505,9 @@ class STTService(AIService):
                 await self.push_frame(frame, direction)
         elif isinstance(frame, STTUpdateSettingsFrame):
             await self._update_settings(frame.settings)
+        elif isinstance(frame, STTMuteFrame):
+            self._muted = frame.mute
+            logger.debug(f"STT service {'muted' if frame.mute else 'unmuted'}")
         else:
             await self.push_frame(frame, direction)