From 5fe679039c47f30b1d7c7c362106c92c623e3d0a Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 14 Nov 2024 12:50:26 -0500 Subject: [PATCH 1/5] Add STTMuteProcessor to un/mute the STT --- .../foundational/24-stt-mute-processor.py | 98 +++++++++++++++++++ src/pipecat/frames/frames.py | 7 ++ src/pipecat/processors/filters/stt_mute.py | 96 ++++++++++++++++++ src/pipecat/services/ai_services.py | 13 ++- 4 files changed, 213 insertions(+), 1 deletion(-) create mode 100644 examples/foundational/24-stt-mute-processor.py create mode 100644 src/pipecat/processors/filters/stt_mute.py diff --git a/examples/foundational/24-stt-mute-processor.py b/examples/foundational/24-stt-mute-processor.py new file mode 100644 index 000000000..2353b057d --- /dev/null +++ b/examples/foundational/24-stt-mute-processor.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 import STTMuteConfig, STTMuteProcessor, 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 = STTMuteProcessor( + stt_service=stt, config=STTMuteConfig(strategy=STTMuteStrategy.ALWAYS) + ) + + 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..77aadb397 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.""" + + muted: bool + + @dataclass class STTUpdateSettingsFrame(ServiceUpdateSettingsFrame): pass diff --git a/src/pipecat/processors/filters/stt_mute.py b/src/pipecat/processors/filters/stt_mute.py new file mode 100644 index 000000000..9b7444eb2 --- /dev/null +++ b/src/pipecat/processors/filters/stt_mute.py @@ -0,0 +1,96 @@ +from dataclasses import dataclass +from enum import Enum +from typing import Callable, Optional + +from loguru import logger + +from pipecat.frames.frames import ( + BotStartedSpeakingFrame, + BotStoppedSpeakingFrame, + Frame, + StartInterruptionFrame, + StopInterruptionFrame, + STTMuteFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.services.ai_services import STTService + + +class STTMuteStrategy(Enum): + NEVER = "never" # Never mute + 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 STTMuteProcessor""" + + strategy: STTMuteStrategy = STTMuteStrategy.NEVER + # Optional callback for custom muting logic + should_mute_callback: Optional[Callable[["STTMuteProcessor"], bool]] = None + + +class STTMuteProcessor(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 = 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.info(f"STT {'muting' if should_mute else 'unmuting'}") + await self.push_frame(STTMuteFrame(muted=should_mute)) + + 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 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(self._should_mute()) + elif isinstance(frame, BotStoppedSpeakingFrame): + self._bot_is_speaking = False + await self._handle_mute_state(self._should_mute()) + + # Handle frame propagation + if isinstance(frame, (StartInterruptionFrame, StopInterruptionFrame)): + # Only pass interruption frames when not muted + if not self.is_muted: + await self.push_frame(frame, direction) + else: + logger.debug("Interruption frame 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..fbb495b47 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.muted + logger.debug(f"STT service {'muted' if frame.muted else 'unmuted'}") else: await self.push_frame(frame, direction) From 52de825af87abc33bb00b63f5f2987bc8e503266 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 14 Nov 2024 13:44:19 -0500 Subject: [PATCH 2/5] Update CHANGELOG --- CHANGELOG.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index d2a0d6c39..c698cb25f 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 `STTMuteProcessor`, 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: `NEVER` (default), `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" From 33108f5798a2da3028136cb5af89a7d3d8a74e76 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 14 Nov 2024 16:42:01 -0500 Subject: [PATCH 3/5] Code review feedback --- CHANGELOG.md | 4 ++-- src/pipecat/frames/frames.py | 2 +- src/pipecat/processors/filters/stt_mute.py | 25 +++++++++++++--------- src/pipecat/services/ai_services.py | 4 ++-- 4 files changed, 20 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c698cb25f..1c1e1cbc0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,10 +14,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added `STTMuteProcessor`, 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: `NEVER` (default), `FIRST_SPEECH` (mute only during bot's first + 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 +- Added `STTMuteFrame`, a control frame that enables/disables speech transcription in STT services. ## [0.0.48] - 2024-11-10 "Antonio release" diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 77aadb397..e057e97da 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -574,7 +574,7 @@ class TTSUpdateSettingsFrame(ServiceUpdateSettingsFrame): class STTMuteFrame(ControlFrame): """Control frame to mute/unmute the STT service.""" - muted: bool + mute: bool @dataclass diff --git a/src/pipecat/processors/filters/stt_mute.py b/src/pipecat/processors/filters/stt_mute.py index 9b7444eb2..5fae94fb7 100644 --- a/src/pipecat/processors/filters/stt_mute.py +++ b/src/pipecat/processors/filters/stt_mute.py @@ -1,6 +1,12 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + from dataclasses import dataclass from enum import Enum -from typing import Callable, Optional +from typing import Awaitable, Callable, Optional from loguru import logger @@ -17,7 +23,6 @@ from pipecat.services.ai_services import STTService class STTMuteStrategy(Enum): - NEVER = "never" # Never mute FIRST_SPEECH = "first_speech" # Mute only during first bot speech ALWAYS = "always" # Mute during all bot speech CUSTOM = "custom" # Allow custom logic via callback @@ -27,9 +32,9 @@ class STTMuteStrategy(Enum): class STTMuteConfig: """Configuration for STTMuteProcessor""" - strategy: STTMuteStrategy = STTMuteStrategy.NEVER + strategy: STTMuteStrategy # Optional callback for custom muting logic - should_mute_callback: Optional[Callable[["STTMuteProcessor"], bool]] = None + should_mute_callback: Optional[Callable[["STTMuteProcessor"], Awaitable[bool]]] = None class STTMuteProcessor(FrameProcessor): @@ -40,7 +45,7 @@ class STTMuteProcessor(FrameProcessor): are automatically disabled. """ - def __init__(self, stt_service: STTService, config: STTMuteConfig = STTMuteConfig(), **kwargs): + def __init__(self, stt_service: STTService, config: STTMuteConfig, **kwargs): super().__init__(**kwargs) self._stt_service = stt_service self._config = config @@ -56,9 +61,9 @@ class STTMuteProcessor(FrameProcessor): """Handles both STT muting and interruption control.""" if should_mute != self.is_muted: logger.info(f"STT {'muting' if should_mute else 'unmuting'}") - await self.push_frame(STTMuteFrame(muted=should_mute)) + await self.push_frame(STTMuteFrame(mute=should_mute)) - def _should_mute(self) -> bool: + 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 @@ -71,7 +76,7 @@ class STTMuteProcessor(FrameProcessor): self._first_speech_handled = True return True elif self._config.strategy == STTMuteStrategy.CUSTOM and self._config.should_mute_callback: - return self._config.should_mute_callback(self) + return await self._config.should_mute_callback(self) return False @@ -79,10 +84,10 @@ class STTMuteProcessor(FrameProcessor): # Handle bot speaking state changes if isinstance(frame, BotStartedSpeakingFrame): self._bot_is_speaking = True - await self._handle_mute_state(self._should_mute()) + await self._handle_mute_state(await self._should_mute()) elif isinstance(frame, BotStoppedSpeakingFrame): self._bot_is_speaking = False - await self._handle_mute_state(self._should_mute()) + await self._handle_mute_state(await self._should_mute()) # Handle frame propagation if isinstance(frame, (StartInterruptionFrame, StopInterruptionFrame)): diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index fbb495b47..e0f16e220 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -506,8 +506,8 @@ class STTService(AIService): elif isinstance(frame, STTUpdateSettingsFrame): await self._update_settings(frame.settings) elif isinstance(frame, STTMuteFrame): - self._muted = frame.muted - logger.debug(f"STT service {'muted' if frame.muted else 'unmuted'}") + self._muted = frame.mute + logger.debug(f"STT service {'muted' if frame.mute else 'unmuted'}") else: await self.push_frame(frame, direction) From f807f233bd9b1df0651936a2702bdb6499b572c1 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 14 Nov 2024 17:11:51 -0500 Subject: [PATCH 4/5] Suppress UserStartedSpeakingFrame and UserStoppedSpeakingFrame when muted --- examples/foundational/24-stt-mute-processor.py | 2 +- src/pipecat/processors/filters/stt_mute.py | 18 ++++++++++++++---- 2 files changed, 15 insertions(+), 5 deletions(-) diff --git a/examples/foundational/24-stt-mute-processor.py b/examples/foundational/24-stt-mute-processor.py index 2353b057d..4259a9558 100644 --- a/examples/foundational/24-stt-mute-processor.py +++ b/examples/foundational/24-stt-mute-processor.py @@ -51,7 +51,7 @@ async def main(): stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) # Configure the mute processor to mute only during first speech stt_mute_processor = STTMuteProcessor( - stt_service=stt, config=STTMuteConfig(strategy=STTMuteStrategy.ALWAYS) + stt_service=stt, config=STTMuteConfig(strategy=STTMuteStrategy.FIRST_SPEECH) ) tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-helios-en") diff --git a/src/pipecat/processors/filters/stt_mute.py b/src/pipecat/processors/filters/stt_mute.py index 5fae94fb7..60315094b 100644 --- a/src/pipecat/processors/filters/stt_mute.py +++ b/src/pipecat/processors/filters/stt_mute.py @@ -17,6 +17,8 @@ from pipecat.frames.frames import ( StartInterruptionFrame, StopInterruptionFrame, STTMuteFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.ai_services import STTService @@ -60,7 +62,7 @@ class STTMuteProcessor(FrameProcessor): async def _handle_mute_state(self, should_mute: bool): """Handles both STT muting and interruption control.""" if should_mute != self.is_muted: - logger.info(f"STT {'muting' if should_mute else 'unmuting'}") + logger.debug(f"STT {'muting' if should_mute else 'unmuting'}") await self.push_frame(STTMuteFrame(mute=should_mute)) async def _should_mute(self) -> bool: @@ -90,12 +92,20 @@ class STTMuteProcessor(FrameProcessor): await self._handle_mute_state(await self._should_mute()) # Handle frame propagation - if isinstance(frame, (StartInterruptionFrame, StopInterruptionFrame)): - # Only pass interruption frames when not muted + 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("Interruption frame suppressed - STT currently muted") + logger.debug(f"{frame.__class__.__name__} suppressed - STT currently muted") else: # Pass all other frames through await self.push_frame(frame, direction) From 966974bfc6e01729ff0869a9e2aded547c05f306 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 14 Nov 2024 19:45:41 -0500 Subject: [PATCH 5/5] Change STTMuteProcessor to STTMuteFilter --- CHANGELOG.md | 2 +- .../{24-stt-mute-processor.py => 24-stt-mute-filter.py} | 4 ++-- .../processors/filters/{stt_mute.py => stt_mute_filter.py} | 6 +++--- 3 files changed, 6 insertions(+), 6 deletions(-) rename examples/foundational/{24-stt-mute-processor.py => 24-stt-mute-filter.py} (95%) rename src/pipecat/processors/filters/{stt_mute.py => stt_mute_filter.py} (95%) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1c1e1cbc0..391f60607 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,7 +11,7 @@ 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 `STTMuteProcessor`, a general-purpose processor that combines STT +- 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 diff --git a/examples/foundational/24-stt-mute-processor.py b/examples/foundational/24-stt-mute-filter.py similarity index 95% rename from examples/foundational/24-stt-mute-processor.py rename to examples/foundational/24-stt-mute-filter.py index 4259a9558..d57986141 100644 --- a/examples/foundational/24-stt-mute-processor.py +++ b/examples/foundational/24-stt-mute-filter.py @@ -21,7 +21,7 @@ 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 import STTMuteConfig, STTMuteProcessor, STTMuteStrategy +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 @@ -50,7 +50,7 @@ async def main(): stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) # Configure the mute processor to mute only during first speech - stt_mute_processor = STTMuteProcessor( + stt_mute_processor = STTMuteFilter( stt_service=stt, config=STTMuteConfig(strategy=STTMuteStrategy.FIRST_SPEECH) ) diff --git a/src/pipecat/processors/filters/stt_mute.py b/src/pipecat/processors/filters/stt_mute_filter.py similarity index 95% rename from src/pipecat/processors/filters/stt_mute.py rename to src/pipecat/processors/filters/stt_mute_filter.py index 60315094b..9ee216c7f 100644 --- a/src/pipecat/processors/filters/stt_mute.py +++ b/src/pipecat/processors/filters/stt_mute_filter.py @@ -32,14 +32,14 @@ class STTMuteStrategy(Enum): @dataclass class STTMuteConfig: - """Configuration for STTMuteProcessor""" + """Configuration for STTMuteFilter""" strategy: STTMuteStrategy # Optional callback for custom muting logic - should_mute_callback: Optional[Callable[["STTMuteProcessor"], Awaitable[bool]]] = None + should_mute_callback: Optional[Callable[["STTMuteFilter"], Awaitable[bool]]] = None -class STTMuteProcessor(FrameProcessor): +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,