Merge pull request #807 from pipecat-ai/mb/stt-mute-strategy

Add new STT mute strategy, accept a set of strategies

CHANGELOG.md

@@ -9,9 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- New `STTMuteStrategy` called `FUNCTION_CALL` which mutes the STT service
+  during LLM function calls.
+
 - `DeepgramSTTService` now exposes two event handlers `on_speech_started` and
   `on_utterance_end` that could be used to implement interruptions. See new
-  example `examples/foundational/07c-interruptible-deepgram-vad.py`
+  example `examples/foundational/07c-interruptible-deepgram-vad.py`.
 
 - Added `GroqLLMService`, `GrokLLMService`, and `NimLLMService` for Groq, Grok,
   and NVIDIA NIM API integration, with an OpenAI-compatible interface.
@@ -38,6 +41,8 @@ async def on_audio_data(processor, audio, sample_rate, num_channels):
 
 ### Changed
 
+- `STTMuteFilter` now supports multiple simultaneous muting strategies.
+
 - `XTTSService` language now defaults to `Language.EN`.
 
 - `SoundfileMixer` doesn't resample input files anymore to avoid startup

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

@@ -11,12 +11,11 @@ import sys
 import aiohttp
 from dotenv import load_dotenv
 from loguru import logger
+from openai.types.chat import ChatCompletionToolParam
 from runner import configure
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.frames.frames import (
+from pipecat.frames.frames import LLMMessagesFrame
-    LLMMessagesFrame,
-)
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -32,6 +31,18 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
+async def start_fetch_weather(function_name, llm, context):
+    logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}")
+
+
+async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
+    # Add a delay to test interruption during function calls
+    logger.info("Weather API call starting...")
+    await asyncio.sleep(5)  # 5-second delay
+    logger.info("Weather API call completed")
+    await result_callback({"conditions": "nice", "temperature": "75"})
+
+
 async def main():
     async with aiohttp.ClientSession() as session:
         (room_url, _) = await configure(session)
@@ -49,23 +60,52 @@ async def main():
         )
 
         stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
-        # Configure the mute processor to mute only during first speech
+        # Configure the mute processor with both strategies
         stt_mute_processor = STTMuteFilter(
-            stt_service=stt, config=STTMuteConfig(strategy=STTMuteStrategy.FIRST_SPEECH)
+            stt_service=stt,
+            config=STTMuteConfig(
+                strategies={STTMuteStrategy.FIRST_SPEECH, STTMuteStrategy.FUNCTION_CALL}
+            ),
         )
 
         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")
+        llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather)
+
+        tools = [
+            ChatCompletionToolParam(
+                type="function",
+                function={
+                    "name": "get_current_weather",
+                    "description": "Get the current weather",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "location": {
+                                "type": "string",
+                                "description": "The city and state, e.g. San Francisco, CA",
+                            },
+                            "format": {
+                                "type": "string",
+                                "enum": ["celsius", "fahrenheit"],
+                                "description": "The temperature unit to use. Infer this from the users location.",
+                            },
+                        },
+                        "required": ["location", "format"],
+                    },
+                },
+            )
+        ]
 
         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.",
+                "content": "You are a helpful assistant who can check the weather. Always check the weather when a location is mentioned. Respond concisely and naturally. Your output will be converted to audio so use only simple words and punctuation.",
             },
         ]
 
-        context = OpenAILLMContext(messages)
+        context = OpenAILLMContext(messages, tools)
         context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
@@ -85,8 +125,13 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
-            # Kick off the conversation.
+            # Kick off the conversation with a weather-related prompt
-            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+            messages.append(
+                {
+                    "role": "system",
+                    "content": "Ask the user what city they'd like to know the weather for.",
+                }
+            )
             await task.queue_frames([LLMMessagesFrame(messages)])
 
         runner = PipelineRunner()

src/pipecat/processors/aggregators/openai_llm_context.py

@@ -21,7 +21,7 @@ from pipecat.frames.frames import (
     FunctionCallResultFrame,
     VisionImageRawFrame,
 )
-from pipecat.processors.frame_processor import FrameProcessor
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 
 try:
     from openai._types import NOT_GIVEN, NotGiven
@@ -196,25 +196,42 @@ class OpenAILLMContext:
         # Push a SystemFrame downstream. This frame will let our assistant context aggregator
         # know that we are in the middle of a function call. Some contexts/aggregators may
         # not need this. But some definitely do (Anthropic, for example).
-        await llm.push_frame(
+        # Also push a SystemFrame upstream for use by other processors, like STTMuteFilter.
-            FunctionCallInProgressFrame(
+        progress_frame_downstream = FunctionCallInProgressFrame(
+            function_name=function_name,
+            tool_call_id=tool_call_id,
+            arguments=arguments,
+        )
+        progress_frame_upstream = FunctionCallInProgressFrame(
+            function_name=function_name,
+            tool_call_id=tool_call_id,
+            arguments=arguments,
+        )
+
+        # Push frame both downstream and upstream
+        await llm.push_frame(progress_frame_downstream, FrameDirection.DOWNSTREAM)
+        await llm.push_frame(progress_frame_upstream, FrameDirection.UPSTREAM)
+
+        # Define a callback function that pushes a FunctionCallResultFrame upstream & downstream.
+        async def function_call_result_callback(result):
+            result_frame_downstream = FunctionCallResultFrame(
                 function_name=function_name,
                 tool_call_id=tool_call_id,
                 arguments=arguments,
+                result=result,
+                run_llm=run_llm,
+            )
+            result_frame_upstream = FunctionCallResultFrame(
+                function_name=function_name,
+                tool_call_id=tool_call_id,
+                arguments=arguments,
+                result=result,
+                run_llm=run_llm,
             )
-        )
 
-        # Define a callback function that pushes a FunctionCallResultFrame downstream.
+            # Push frame both downstream and upstream
-        async def function_call_result_callback(result):
+            await llm.push_frame(result_frame_downstream, FrameDirection.DOWNSTREAM)
-            await llm.push_frame(
+            await llm.push_frame(result_frame_upstream, FrameDirection.UPSTREAM)
-                FunctionCallResultFrame(
-                    function_name=function_name,
-                    tool_call_id=tool_call_id,
-                    arguments=arguments,
-                    result=result,
-                    run_llm=run_llm,
-                )
-            )
 
         await f(function_name, tool_call_id, arguments, llm, self, function_call_result_callback)
 

src/pipecat/processors/filters/stt_mute_filter.py

@@ -4,6 +4,13 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+"""Speech-to-text (STT) muting control module.
+
+This module provides functionality to control STT muting based on different strategies,
+such as during function calls, bot speech, or custom conditions. It helps manage when
+the STT service should be active or inactive during a conversation.
+"""
+
 from dataclasses import dataclass
 from enum import Enum
 from typing import Awaitable, Callable, Optional
@@ -14,6 +21,8 @@ from pipecat.frames.frames import (
     BotStartedSpeakingFrame,
     BotStoppedSpeakingFrame,
     Frame,
+    FunctionCallInProgressFrame,
+    FunctionCallResultFrame,
     StartInterruptionFrame,
     StopInterruptionFrame,
     STTMuteFrame,
@@ -25,26 +34,46 @@ from pipecat.services.ai_services import STTService
 
 
 class STTMuteStrategy(Enum):
+    """Strategies determining when STT should be muted.
+
+    Attributes:
+        FIRST_SPEECH: Mute only during first bot speech
+        FUNCTION_CALL: Mute during function calls
+        ALWAYS: Mute during all bot speech
+        CUSTOM: Allow custom logic via callback
+    """
+
     FIRST_SPEECH = "first_speech"  # Mute only during first bot speech
+    FUNCTION_CALL = "function_call"  # Mute during function calls
     ALWAYS = "always"  # Mute during all bot speech
     CUSTOM = "custom"  # Allow custom logic via callback
 
 
 @dataclass
 class STTMuteConfig:
-    """Configuration for STTMuteFilter"""
+    """Configuration for STT muting behavior.
 
-    strategy: STTMuteStrategy
+    Args:
+        strategies: Set of muting strategies to apply
+        should_mute_callback: Optional callback for custom muting logic.
+            Only required when using STTMuteStrategy.CUSTOM
+    """
+
+    strategies: set[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.
+    """A processor that handles STT muting and interruption control.
 
-    This processor combines the concepts of STT muting and interruption control,
+    This processor combines STT muting and interruption control as a coordinated
-    treating them as a single coordinated feature. When STT is muted, interruptions
+    feature. When STT is muted, interruptions are automatically disabled.
-    are automatically disabled.
+
+    Args:
+        stt_service: Service handling speech-to-text functionality
+        config: Configuration specifying muting strategies
+        **kwargs: Additional arguments passed to parent class
     """
 
     def __init__(self, stt_service: STTService, config: STTMuteConfig, **kwargs):
@@ -53,6 +82,7 @@ class STTMuteFilter(FrameProcessor):
         self._config = config
         self._first_speech_handled = False
         self._bot_is_speaking = False
+        self._function_call_in_progress = False
 
     @property
     def is_muted(self) -> bool:
@@ -67,24 +97,40 @@ class STTMuteFilter(FrameProcessor):
 
     async def _should_mute(self) -> bool:
         """Determines if STT should be muted based on current state and strategy."""
-        if not self._bot_is_speaking:
+        for strategy in self._config.strategies:
-            return False
+            match strategy:
+                case STTMuteStrategy.FUNCTION_CALL:
+                    if self._function_call_in_progress:
+                        return True
 
-        if self._config.strategy == STTMuteStrategy.ALWAYS:
+                case STTMuteStrategy.ALWAYS:
-            return True
+                    if self._bot_is_speaking:
-        elif (
+                        return True
-            self._config.strategy == STTMuteStrategy.FIRST_SPEECH and not self._first_speech_handled
+
-        ):
+                case STTMuteStrategy.FIRST_SPEECH:
-            self._first_speech_handled = True
+                    if self._bot_is_speaking and not self._first_speech_handled:
-            return True
+                        self._first_speech_handled = True
-        elif self._config.strategy == STTMuteStrategy.CUSTOM and self._config.should_mute_callback:
+                        return True
-            return await self._config.should_mute_callback(self)
+
+                case STTMuteStrategy.CUSTOM:
+                    if self._bot_is_speaking and self._config.should_mute_callback:
+                        should_mute = await self._config.should_mute_callback(self)
+                        if should_mute:
+                            return True
 
         return False
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        """Processes incoming frames and manages muting state."""
+        # Handle function call state changes
+        if isinstance(frame, FunctionCallInProgressFrame):
+            self._function_call_in_progress = True
+            await self._handle_mute_state(await self._should_mute())
+        elif isinstance(frame, FunctionCallResultFrame):
+            self._function_call_in_progress = False
+            await self._handle_mute_state(await self._should_mute())
         # Handle bot speaking state changes
-        if isinstance(frame, BotStartedSpeakingFrame):
+        elif isinstance(frame, BotStartedSpeakingFrame):
             self._bot_is_speaking = True
             await self._handle_mute_state(await self._should_mute())
         elif isinstance(frame, BotStoppedSpeakingFrame):