diff --git a/CHANGELOG.md b/CHANGELOG.md index d2a0d6c39..391f60607 100644 --- a/CHANGELOG.md +++ b/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" diff --git a/examples/foundational/24-stt-mute-filter.py b/examples/foundational/24-stt-mute-filter.py new file mode 100644 index 000000000..d57986141 --- /dev/null +++ b/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()) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 7f7972f98..e057e97da 100644 --- a/src/pipecat/frames/frames.py +++ b/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 diff --git a/src/pipecat/processors/filters/stt_mute_filter.py b/src/pipecat/processors/filters/stt_mute_filter.py new file mode 100644 index 000000000..9ee216c7f --- /dev/null +++ b/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) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index f7a08802c..e0f16e220 100644 --- a/src/pipecat/services/ai_services.py +++ b/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)