Merge pull request #3292 from pipecat-ai/aleix/initial-user-mute-strategies

initial user mute strategies

changelog/3292.added.md

@@ -0,0 +1,29 @@
+- Introducing user mute strategies. User mute strategies indicate when user input should be muted based on the current system state.
+
+  In conversational agents, user mute strategies are used to prevent user input from interrupting bot speech, tool execution, or other critical system operations.
+
+  A list of strategies can be specified; all strategies are evaluated for every frame so that each strategy can maintain its internal state. A user frame is muted if any of the configured strategies indicates it should be muted.
+
+  Available user mute strategies:
+
+  * `FirstSpeechUserMuteStrategy`
+  * `MuteUntilFirstBotCompleteUserMuteStrategy`
+  * `AlwaysUserMuteStrategy`
+  * `FunctionCallUserMuteStrategy`
+
+  User mute strategies replace the legacy `STTMuteFilter` and provide a more flexible and composable approach to muting user input.
+
+  User mute strategies are configured when setting up the `LLMContextAggregatorPair`. For example:
+
+  ```python
+  context_aggregator = LLMContextAggregatorPair(
+      context,
+      user_params=LLMUserAggregatorParams(
+          user_mute_strategies=[
+              FirstSpeechUserMuteStrategy(),
+          ]
+      ),
+  )
+  ```
+
+  In order to use user mute strategies you should update to the new universal `LLMContext` and `LLMContextAggregatorPair`.

changelog/3292.deprecated.md

@@ -0,0 +1 @@
+- `STTMuteFilter` is deprecated and will be removed in a future version. Use `LLMUserAggregator`'s new `user_mute_strategies` instead.

changelog/3292.fixed.md

@@ -0,0 +1 @@
+- Fixed a bug in `STTMuteFilter` where the user was not always muted during function calls, especially when there were multiple simultaneous calls.

examples/foundational/24-user-mute-strategy.py

@@ -0,0 +1,178 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+
+import asyncio
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.audio.vad.vad_analyzer import VADParams
+from pipecat.frames.frames import LLMRunFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import (
+    LLMContextAggregatorPair,
+    LLMUserAggregatorParams,
+)
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.deepgram.tts import DeepgramTTSService
+from pipecat.services.llm_service import FunctionCallParams
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+from pipecat.turns.bot.turn_analyzer_bot_turn_start_strategy import TurnAnalyzerBotTurnStartStrategy
+from pipecat.turns.mute import (
+    FunctionCallUserMuteStrategy,
+    MuteUntilFirstBotCompleteUserMuteStrategy,
+)
+from pipecat.turns.turn_start_strategies import TurnStartStrategies
+
+load_dotenv(override=True)
+
+
+async def fetch_weather_from_api(params: FunctionCallParams):
+    # 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 params.result_callback({"conditions": "nice", "temperature": "75"})
+
+
+# We store functions so objects (e.g. SileroVADAnalyzer) don't get
+# instantiated. The function will be called when the desired transport gets
+# selected.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-helios-en")
+
+    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+    llm.register_function("get_current_weather", fetch_weather_from_api)
+
+    weather_function = FunctionSchema(
+        name="get_current_weather",
+        description="Get the current weather",
+        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 user's location.",
+            },
+        },
+        required=["location", "format"],
+    )
+    tools = ToolsSchema(standard_tools=[weather_function])
+
+    messages = [
+        {
+            "role": "system",
+            "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 spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points.",
+        },
+    ]
+
+    context = LLMContext(messages, tools)
+    context_aggregator = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(
+            user_mute_strategies=[
+                MuteUntilFirstBotCompleteUserMuteStrategy(),
+                FunctionCallUserMuteStrategy(),
+            ]
+        ),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            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,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+            turn_start_strategies=TurnStartStrategies(
+                bot=[TurnAnalyzerBotTurnStartStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())]
+            ),
+        ),
+        idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Kick off the conversation with a weather-related prompt
+        messages.append(
+            {
+                "role": "system",
+                "content": "Ask the user what city they'd like to know the weather for.",
+            }
+        )
+        await task.queue_frames([LLMRunFrame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+
+    await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()

examples/foundational/README.md

@@ -80,7 +80,7 @@ uv run 07-interruptible.py -t twilio -x NGROK_HOST_NAME
 ### Common Utilities
 
 - **[17-detect-user-idle.py](./17-detect-user-idle.py)**: Handle inactive users (UserIdleProcessor)
-- **[24-stt-mute-filter.py](./24-stt-mute-filter.py)**: Selectively mute user input (STTMuteFilter)
+- **[24-user-mute-strategy.py](./24-user-mute-strategy.py)**: Selectively mute user input (LLMUserAggregator user mute strategies)
 - **[28-transcription-processor.py](./28-transcription-processor.py)**: Record conversation text (TranscriptProcessor)
 - **[30-observer.py](./30-observer.py)**: Access frame data (Custom observers)
 - **[31-heartbeats.py](./31-heartbeats.py)**: Detect idle pipelines (Pipeline monitoring)

src/pipecat/frames/frames.py

@@ -956,6 +956,10 @@ class StartFrame(SystemFrame):
         enable_tracing: Whether to enable OpenTelemetry tracing.
         enable_usage_metrics: Whether to enable usage metrics collection.
         interruption_strategies: List of interruption handling strategies.
+
+            .. deprecated:: 0.0.99
+                Use the `turn_start_strategies` instead.
+
         report_only_initial_ttfb: Whether to report only initial time-to-first-byte.
     """
 

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -15,7 +15,7 @@ import asyncio
 import json
 import warnings
 from abc import abstractmethod
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Any, Dict, List, Literal, Optional, Set, Type
 
 from loguru import logger
@@ -31,6 +31,8 @@ from pipecat.frames.frames import (
     FunctionCallInProgressFrame,
     FunctionCallResultFrame,
     FunctionCallsStartedFrame,
+    InputAudioRawFrame,
+    InterimTranscriptionFrame,
     InterruptionFrame,
     LLMContextAssistantTimestampFrame,
     LLMContextFrame,
@@ -62,6 +64,7 @@ from pipecat.processors.aggregators.llm_context import (
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.turns.bot.base_bot_turn_start_strategy import BaseBotTurnStartStrategy
+from pipecat.turns.mute.base_user_mute_strategy import BaseUserMuteStrategy
 from pipecat.turns.user.base_user_turn_start_strategy import BaseUserTurnStartStrategy
 from pipecat.utils.string import TextPartForConcatenation, concatenate_aggregated_text
 from pipecat.utils.time import time_now_iso8601
@@ -77,11 +80,13 @@ class LLMUserAggregatorParams:
             interruption frames. This is enabled by default, but you may want
             to disable it if another component (e.g., an STT service) is already
             generating these frames.
+        user_mute_strategies: List of user mute strategies.
         user_turn_end_timeout: Time in seconds to wait before considering the
             user's turn finished and starting the bot turn.
     """
 
     enable_user_speaking_frames: bool = True
+    user_mute_strategies: List[BaseUserMuteStrategy] = field(default_factory=list)
     user_turn_end_timeout: float = 5.0
 
 
@@ -269,6 +274,7 @@ class LLMUserAggregator(LLMContextAggregator):
         self._vad_user_speaking = False
 
         self._user_turn = False
+        self._user_is_muted = False
         self._user_turn_end_timeout_event = asyncio.Event()
         self._user_turn_end_timeout_task: Optional[asyncio.Task] = None
 
@@ -281,18 +287,6 @@ class LLMUserAggregator(LLMContextAggregator):
         await super().cleanup()
         await self._cleanup()
 
-    async def reset(self):
-        """Reset the aggregation state and turn start strategies."""
-        await super().reset()
-
-        if self.turn_start_strategies and self.turn_start_strategies.user:
-            for s in self.turn_start_strategies.user:
-                await s.reset()
-
-        if self.turn_start_strategies and self.turn_start_strategies.bot:
-            for s in self.turn_start_strategies.bot:
-                await s.reset()
-
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process frames for user speech aggregation and context management.
 
@@ -302,6 +296,9 @@ class LLMUserAggregator(LLMContextAggregator):
         """
         await super().process_frame(frame, direction)
 
+        if await self._maybe_mute_frame(frame):
+            return
+
         if isinstance(frame, StartFrame):
             # Push StartFrame before start(), because we want StartFrame to be
             # processed by every processor before any other frame is processed.
@@ -362,6 +359,9 @@ class LLMUserAggregator(LLMContextAggregator):
                 self._user_turn_end_timeout_task_handler()
             )
 
+        for s in self._params.user_mute_strategies:
+            await s.setup(self.task_manager)
+
         if self.turn_start_strategies and self.turn_start_strategies.user:
             for s in self.turn_start_strategies.user:
                 await s.setup(self.task_manager)
@@ -387,6 +387,9 @@ class LLMUserAggregator(LLMContextAggregator):
             await self.cancel_task(self._user_turn_end_timeout_task)
             self._user_turn_end_timeout_task = None
 
+        for s in self._params.user_mute_strategies:
+            await s.cleanup()
+
         if self.turn_start_strategies and self.turn_start_strategies.user:
             for s in self.turn_start_strategies.user:
                 await s.cleanup()
@@ -395,6 +398,34 @@ class LLMUserAggregator(LLMContextAggregator):
             for s in self.turn_start_strategies.bot:
                 await s.cleanup()
 
+    async def _maybe_mute_frame(self, frame: Frame):
+        should_mute_frame = self._user_is_muted and isinstance(
+            frame,
+            (
+                InterruptionFrame,
+                VADUserStartedSpeakingFrame,
+                VADUserStoppedSpeakingFrame,
+                UserStartedSpeakingFrame,
+                UserStoppedSpeakingFrame,
+                InputAudioRawFrame,
+                InterimTranscriptionFrame,
+                TranscriptionFrame,
+            ),
+        )
+
+        if should_mute_frame:
+            logger.trace(f"{frame.name} suppressed - user currently muted")
+
+        should_mute_next_time = False
+        for s in self._params.user_mute_strategies:
+            should_mute_next_time |= await s.process_frame(frame)
+
+        if should_mute_next_time != self._user_is_muted:
+            logger.debug(f"{self}: user is now {'muted' if should_mute_next_time else 'unmuted'}")
+            self._user_is_muted = should_mute_next_time
+
+        return should_mute_frame
+
     async def _turn_start_strategies_process_frame(self, frame: Frame):
         if self.turn_start_strategies and self.turn_start_strategies.user:
             for strategy in self.turn_start_strategies.user:

src/pipecat/processors/filters/stt_mute_filter.py

@@ -21,8 +21,9 @@ from pipecat.frames.frames import (
     BotStartedSpeakingFrame,
     BotStoppedSpeakingFrame,
     Frame,
-    FunctionCallInProgressFrame,
+    FunctionCallCancelFrame,
     FunctionCallResultFrame,
+    FunctionCallsStartedFrame,
     InputAudioRawFrame,
     InterimTranscriptionFrame,
     InterruptionFrame,
@@ -50,6 +51,10 @@ class STTMuteStrategy(Enum):
         FUNCTION_CALL: Mute STT during function calls to prevent interruptions.
         ALWAYS: Always mute STT when the bot is speaking.
         CUSTOM: Use a custom callback to determine muting logic dynamically.
+
+    .. deprecated:: 0.0.99
+        `STTMuteStrategy` is deprecated and will be removed in a future version.
+        Use `LLMUserAggregator`'s new `user_mute_strategies` instead.
     """
 
     FIRST_SPEECH = "first_speech"
@@ -76,6 +81,10 @@ class STTMuteConfig:
     Note:
         MUTE_UNTIL_FIRST_BOT_COMPLETE and FIRST_SPEECH strategies should not be used together
         as they handle the first bot speech differently.
+
+    .. deprecated:: 0.0.99
+        `STTMuteConfig` is deprecated and will be removed in a future version.
+        Use `LLMUserAggregator`'s new `user_mute_strategies` instead.
     """
 
     strategies: set[STTMuteStrategy]
@@ -103,6 +112,10 @@ class STTMuteFilter(FrameProcessor):
     feature. When STT is muted, interruptions are automatically disabled by
     suppressing VAD-related frames. This prevents unwanted speech detection
     during bot speech, function calls, or other specified conditions.
+
+    .. deprecated:: 0.0.99
+        `STTMuteFilter` is deprecated and will be removed in a future version.
+        Use `LLMUserAggregator`'s new `user_mute_strategies` instead.
     """
 
     def __init__(self, *, config: STTMuteConfig, **kwargs):
@@ -116,9 +129,19 @@ class STTMuteFilter(FrameProcessor):
         self._config = config
         self._first_speech_handled = False
         self._bot_is_speaking = False
-        self._function_call_in_progress = False
+        self._function_call_in_progress = set()
         self._is_muted = False
 
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "`STTMuteFilter` is deprecated and will be removed in a future version. "
+                "Use `LLMUserAggregator`'s new `user_mute_strategies` instead.",
+                DeprecationWarning,
+            )
+
     async def _handle_mute_state(self, should_mute: bool):
         """Handle STT muting and interruption control state changes."""
         if should_mute != self._is_muted:
@@ -176,11 +199,12 @@ class STTMuteFilter(FrameProcessor):
         # Process frames to determine mute state
         if isinstance(frame, StartFrame):
             should_mute = await self._should_mute()
-        elif isinstance(frame, FunctionCallInProgressFrame):
+        elif isinstance(frame, FunctionCallsStartedFrame):
-            self._function_call_in_progress = True
+            for f in frame.function_calls:
+                self._function_call_in_progress.add(f.tool_call_id)
             should_mute = await self._should_mute()
-        elif isinstance(frame, FunctionCallResultFrame):
+        elif isinstance(frame, (FunctionCallCancelFrame, FunctionCallResultFrame)):
-            self._function_call_in_progress = False
+            self._function_call_in_progress.remove(frame.tool_call_id)
             should_mute = await self._should_mute()
         elif isinstance(frame, BotStartedSpeakingFrame):
             self._bot_is_speaking = True

src/pipecat/processors/frame_processor.py

@@ -359,7 +359,7 @@ class FrameProcessor(BaseObject):
     def interruption_strategies(self) -> Sequence[BaseInterruptionStrategy]:
         """Get the interruption strategies for this processor.
 
-        .. deprecated:: 0.0.98
+        .. deprecated:: 0.0.99
             This function is deprecated, use the new user and bot turn start
             strategies insted.
 

src/pipecat/turns/bot/base_bot_turn_start_strategy.py

@@ -43,10 +43,6 @@ class BaseBotTurnStartStrategy(BaseObject):
             raise RuntimeError(f"{self} bot turn start strategy was not properly setup")
         return self._task_manager
 
-    async def reset(self):
-        """Reset the strategy to its initial state."""
-        pass
-
     async def setup(self, task_manager: BaseTaskManager):
         """Initialize the strategy with the given task manager.
 
@@ -59,6 +55,10 @@ class BaseBotTurnStartStrategy(BaseObject):
         """Cleanup the strategy."""
         pass
 
+    async def reset(self):
+        """Reset the strategy to its initial state."""
+        pass
+
     async def process_frame(self, frame: Frame):
         """Process an incoming frame to decide whether the bot should speak.
 
@@ -67,7 +67,6 @@ class BaseBotTurnStartStrategy(BaseObject):
 
         Args:
             frame: The frame to be analyzed.
-
         """
         pass
 

src/pipecat/turns/mute/__init__.py

@@ -0,0 +1,13 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from pipecat.turns.mute.always_user_mute_strategy import AlwaysUserMuteStrategy
+from pipecat.turns.mute.base_user_mute_strategy import BaseUserMuteStrategy
+from pipecat.turns.mute.first_speech_user_mute_strategy import FirstSpeechUserMuteStrategy
+from pipecat.turns.mute.function_call_user_mute_strategy import FunctionCallUserMuteStrategy
+from pipecat.turns.mute.mute_until_first_bot_complete_user_mute_strategy import (
+    MuteUntilFirstBotCompleteUserMuteStrategy,
+)

src/pipecat/turns/mute/always_user_mute_strategy.py

@@ -0,0 +1,41 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""User mute strategy that always mutes the user while the bot is speaking."""
+
+from pipecat.frames.frames import BotStartedSpeakingFrame, BotStoppedSpeakingFrame, Frame
+from pipecat.turns.mute.base_user_mute_strategy import BaseUserMuteStrategy
+
+
+class AlwaysUserMuteStrategy(BaseUserMuteStrategy):
+    """User mute strategy that always mutes the user while the bot is speaking."""
+
+    def __init__(self):
+        """Initialize the always user mute strategy."""
+        super().__init__()
+        self._bot_speaking = False
+
+    async def reset(self):
+        """Reset the strategy to its initial state."""
+        self._bot_speaking = False
+
+    async def process_frame(self, frame: Frame) -> bool:
+        """Process an incoming frame.
+
+        Args:
+            frame: The frame to be processed.
+
+        Returns:
+            Whether the strategy is muted.
+        """
+        await super().process_frame(frame)
+
+        if isinstance(frame, BotStartedSpeakingFrame):
+            self._bot_speaking = True
+        elif isinstance(frame, BotStoppedSpeakingFrame):
+            self._bot_speaking = False
+
+        return self._bot_speaking

src/pipecat/turns/mute/base_user_mute_strategy.py

@@ -0,0 +1,69 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Base strategy for deciding whether user frames should be muted."""
+
+from typing import Optional
+
+from pipecat.frames.frames import Frame
+from pipecat.utils.asyncio.task_manager import BaseTaskManager
+from pipecat.utils.base_object import BaseObject
+
+
+class BaseUserMuteStrategy(BaseObject):
+    """Base class for strategies that decide whether user frames should be muted.
+
+    A user mute strategy determines whether incoming user frames should be
+    suppressed based on the *current system state*.
+
+    Typical heuristics include:
+    - The bot is currently speaking, so user should be muted
+    - A function call or tool execution is in progress
+    - The system is otherwise not ready to accept user input
+
+    The strategy is evaluated per frame and returns a boolean indicating whether
+    the user should be muted.
+
+    """
+
+    def __init__(self, **kwargs):
+        """Initialize the base user mute strategy."""
+        super().__init__(**kwargs)
+        self._task_manager: Optional[BaseTaskManager] = None
+
+    @property
+    def task_manager(self) -> BaseTaskManager:
+        """Returns the configured task manager."""
+        if not self._task_manager:
+            raise RuntimeError(f"{self} user mute strategy was not properly setup")
+        return self._task_manager
+
+    async def setup(self, task_manager: BaseTaskManager):
+        """Initialize the strategy with the given task manager.
+
+        Args:
+            task_manager: The task manager to be associated with this instance.
+        """
+        self._task_manager = task_manager
+
+    async def cleanup(self):
+        """Cleanup the strategy."""
+        pass
+
+    async def reset(self):
+        """Reset the strategy to its initial state."""
+        pass
+
+    async def process_frame(self, frame: Frame) -> bool:
+        """Process an incoming frame.
+
+        Args:
+            frame: The frame to be processed.
+
+        Returns:
+            Whether the strategy is muted.
+        """
+        return False

src/pipecat/turns/mute/first_speech_user_mute_strategy.py

@@ -0,0 +1,64 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""User mute strategy that mutes the user only during the bot’s first speech."""
+
+from pipecat.frames.frames import BotStartedSpeakingFrame, BotStoppedSpeakingFrame, Frame
+from pipecat.turns.mute.base_user_mute_strategy import BaseUserMuteStrategy
+
+
+class FirstSpeechUserMuteStrategy(BaseUserMuteStrategy):
+    """User mute strategy that mutes the user only during the bot’s first speech.
+
+    This strategy allows user input before the bot starts speaking. Once the bot
+    begins its first speaking turn, user frames are muted until the bot finishes
+    that speech. After the bot completes its first speaking turn, user input is
+    no longer muted by this strategy.
+
+    Use this strategy when early user input is acceptable, but interruptions
+    during the bot’s initial response should be prevented.
+
+    """
+
+    def __init__(self):
+        """Initialize the first-bot-speech user mute strategy."""
+        super().__init__()
+        self._bot_speaking = False
+        self._first_speech_handled = False
+
+    async def reset(self):
+        """Reset the strategy to its initial state."""
+        self._bot_speaking = False
+        self._first_speech_handled = False
+
+    async def process_frame(self, frame: Frame) -> bool:
+        """Process an incoming frame.
+
+        Args:
+            frame: The frame to be processed.
+
+        Returns:
+            Whether the strategy is muted.
+        """
+        await super().process_frame(frame)
+
+        if isinstance(frame, BotStartedSpeakingFrame):
+            await self._handle_bot_started_speaking(frame)
+        elif isinstance(frame, BotStoppedSpeakingFrame):
+            await self._handle_bot_stopped_speaking(frame)
+
+        if self._bot_speaking and not self._first_speech_handled:
+            return True
+
+        return False
+
+    async def _handle_bot_started_speaking(self, frame: BotStartedSpeakingFrame):
+        self._bot_speaking = True
+
+    async def _handle_bot_stopped_speaking(self, frame: BotStoppedSpeakingFrame):
+        self._bot_speaking = False
+        if not self._first_speech_handled:
+            self._first_speech_handled = True

src/pipecat/turns/mute/function_call_user_mute_strategy.py

@@ -0,0 +1,59 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""User mute strategy that mutes the user while a function call is executing."""
+
+from typing import Set
+
+from pipecat.frames.frames import (
+    Frame,
+    FunctionCallCancelFrame,
+    FunctionCallResultFrame,
+    FunctionCallsStartedFrame,
+)
+from pipecat.turns.mute.base_user_mute_strategy import BaseUserMuteStrategy
+
+
+class FunctionCallUserMuteStrategy(BaseUserMuteStrategy):
+    """User mute strategy that mutes the user while a function call is executing.
+
+    This strategy ensures that user input does not interfere with ongoing
+    function execution. While a function call is active, all user frames are
+    muted. Once the function call completes or is canceled, user input is
+    allowed again.
+
+    """
+
+    def __init__(self):
+        """Initialize the function call user mute strategy."""
+        super().__init__()
+        self._function_call_in_progress: Set[str] = set()
+
+    async def reset(self):
+        """Reset the strategy to its initial state."""
+        self._function_call_in_progress = set()
+
+    async def process_frame(self, frame: Frame) -> bool:
+        """Process an incoming frame.
+
+        Args:
+            frame: The frame to be processed.
+
+        Returns:
+            Whether the strategy is muted.
+        """
+        await super().process_frame(frame)
+
+        if isinstance(frame, FunctionCallsStartedFrame):
+            await self._handle_function_calls_started(frame)
+        elif isinstance(frame, (FunctionCallCancelFrame, FunctionCallResultFrame)):
+            self._function_call_in_progress.remove(frame.tool_call_id)
+
+        return bool(self._function_call_in_progress)
+
+    async def _handle_function_calls_started(self, frame: FunctionCallsStartedFrame):
+        for f in frame.function_calls:
+            self._function_call_in_progress.add(f.tool_call_id)

src/pipecat/turns/mute/mute_until_first_bot_complete_user_mute_strategy.py

@@ -0,0 +1,56 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""User mute strategy that mutes the user until the bot completes its first speech."""
+
+from pipecat.frames.frames import BotStoppedSpeakingFrame, Frame
+from pipecat.turns.mute.base_user_mute_strategy import BaseUserMuteStrategy
+
+
+class MuteUntilFirstBotCompleteUserMuteStrategy(BaseUserMuteStrategy):
+    """User mute strategy that mutes the user until the bot completes its first speech.
+
+    This strategy mutes user frames immediately from the start of the
+    interaction, even if the bot has not started speaking yet. User input
+    remains muted until the bot finishes its first speaking turn.
+
+    After the bot completes its initial speech, all subsequent user frames are
+    allowed to pass through without muting.
+
+    Use this strategy when the bot must fully control the beginning of the
+    interaction and deliver its first response without any user interruption.
+
+    """
+
+    def __init__(self):
+        """Initialize the mute-until-first-bot-complete user mute strategy."""
+        super().__init__()
+        self._first_speech_handled = False
+
+    async def reset(self):
+        """Reset the strategy to its initial state."""
+        self._first_speech_handled = False
+
+    async def process_frame(self, frame: Frame) -> bool:
+        """Process an incoming frame.
+
+        Args:
+            frame: The frame to be processed.
+
+        Returns:
+            Whether the strategy is muted.
+        """
+        await super().process_frame(frame)
+
+        if isinstance(frame, BotStoppedSpeakingFrame):
+            await self._handle_bot_stopped_speaking(frame)
+
+        return not self._first_speech_handled
+
+    async def _handle_bot_stopped_speaking(self, frame: BotStoppedSpeakingFrame):
+        self._bot_speaking = False
+        if not self._first_speech_handled:
+            self._first_speech_handled = True

src/pipecat/turns/user/base_user_turn_start_strategy.py

@@ -43,10 +43,6 @@ class BaseUserTurnStartStrategy(BaseObject):
             raise RuntimeError(f"{self} user turn start strategy was not properly setup")
         return self._task_manager
 
-    async def reset(self):
-        """Reset the strategy to its initial state."""
-        pass
-
     async def setup(self, task_manager: BaseTaskManager):
         """Initialize the strategy with the given task manager.
 
@@ -59,6 +55,10 @@ class BaseUserTurnStartStrategy(BaseObject):
         """Cleanup the strategy."""
         pass
 
+    async def reset(self):
+        """Reset the strategy to its initial state."""
+        pass
+
     async def process_frame(self, frame: Frame):
         """Process an incoming frame.
 
@@ -67,7 +67,6 @@ class BaseUserTurnStartStrategy(BaseObject):
 
         Args:
             frame: The frame to be processed.
-
         """
         pass
 

tests/test_context_aggregators_universal.py

@@ -7,6 +7,12 @@
 import unittest
 
 from pipecat.frames.frames import (
+    BotStartedSpeakingFrame,
+    BotStoppedSpeakingFrame,
+    FunctionCallFromLLM,
+    FunctionCallInProgressFrame,
+    FunctionCallResultFrame,
+    FunctionCallsStartedFrame,
     InterruptionFrame,
     LLMContextFrame,
     LLMMessagesAppendFrame,
@@ -29,6 +35,8 @@ from pipecat.tests.utils import SleepFrame, run_test
 from pipecat.turns.bot.transcription_bot_turn_start_strategy import (
     TranscriptionBotTurnStartStrategy,
 )
+from pipecat.turns.mute.first_speech_user_mute_strategy import FirstSpeechUserMuteStrategy
+from pipecat.turns.mute.function_call_user_mute_strategy import FunctionCallUserMuteStrategy
 from pipecat.turns.turn_start_strategies import TurnStartStrategies
 
 USER_TURN_END_TIMEOUT = 0.2
@@ -233,3 +241,58 @@ class TestUserAggregator(unittest.IsolatedAsyncioTestCase):
         # The transcription strategy should kick-in before the user turn end timeout.
         self.assertTrue(bot_turn)
         self.assertFalse(timeout)
+
+    async def test_user_mute_strategies(self):
+        context = LLMContext()
+
+        user_aggregator = LLMUserAggregator(
+            context,
+            params=LLMUserAggregatorParams(
+                user_mute_strategies=[
+                    FirstSpeechUserMuteStrategy(),
+                    FunctionCallUserMuteStrategy(),
+                ]
+            ),
+        )
+
+        user_turn = False
+
+        @user_aggregator.event_handler("on_user_turn_started")
+        async def on_user_turn_started(aggregator, strategy):
+            nonlocal user_turn
+            user_turn = True
+
+        pipeline = Pipeline([user_aggregator])
+
+        frames_to_send = [
+            # Bot is speaking, user should be muted.
+            BotStartedSpeakingFrame(),
+            VADUserStartedSpeakingFrame(),
+            VADUserStoppedSpeakingFrame(),
+            TranscriptionFrame(text="Hello!", user_id="", timestamp="now"),
+            SleepFrame(),
+            BotStoppedSpeakingFrame(),
+            # Function call is executing, user should be muted.
+            FunctionCallsStartedFrame(
+                function_calls=[
+                    FunctionCallFromLLM(
+                        function_name="fn_1", tool_call_id="1", arguments={}, context=None
+                    )
+                ]
+            ),
+            SleepFrame(),
+            VADUserStartedSpeakingFrame(),
+            VADUserStoppedSpeakingFrame(),
+            TranscriptionFrame(text="Hello!", user_id="", timestamp="now"),
+            FunctionCallResultFrame(
+                function_name="fn_1", tool_call_id="1", arguments={}, result={}
+            ),
+            SleepFrame(),
+        ]
+        await run_test(
+            pipeline,
+            frames_to_send=frames_to_send,
+        )
+
+        # The user mute strategies should have muted the user.
+        self.assertFalse(user_turn)

tests/test_stt_mute_filter.py

@@ -9,8 +9,10 @@ import unittest
 from pipecat.frames.frames import (
     BotStartedSpeakingFrame,
     BotStoppedSpeakingFrame,
+    FunctionCallFromLLM,
     FunctionCallInProgressFrame,
     FunctionCallResultFrame,
+    FunctionCallsStartedFrame,
     InputAudioRawFrame,
     InterimTranscriptionFrame,
     TranscriptionFrame,
@@ -148,54 +150,59 @@ class TestSTTMuteFilter(unittest.IsolatedAsyncioTestCase):
             expected_down_frames=expected_returned_frames,
         )
 
-    # TODO: Revisit once we figure out how to test SystemFrames and DataFrames
+    async def test_function_call_strategy(self):
-    # async def test_function_call_strategy(self):
+        filter = STTMuteFilter(config=STTMuteConfig(strategies={STTMuteStrategy.FUNCTION_CALL}))
-    #     filter = STTMuteFilter(config=STTMuteConfig(strategies={STTMuteStrategy.FUNCTION_CALL}))
 
-    #     frames_to_send = [
+        frames_to_send = [
-    #         VADUserStartedSpeakingFrame(),  # Should pass through initially
+            VADUserStartedSpeakingFrame(),  # Should pass through initially
-    #         UserStartedSpeakingFrame(),  # Should pass through initially
+            UserStartedSpeakingFrame(),  # Should pass through initially
-    #         VADUserStoppedSpeakingFrame(),
+            VADUserStoppedSpeakingFrame(),  # Should pass through initially
-    #         UserStoppedSpeakingFrame(),
+            UserStoppedSpeakingFrame(),  # Should pass through initially
-    #         FunctionCallInProgressFrame(
+            FunctionCallsStartedFrame(
-    #             function_name="get_weather",
+                function_calls=[
-    #             tool_call_id="call_123",
+                    FunctionCallFromLLM(
-    #             arguments='{"location": "San Francisco"}',
+                        function_name="get_weather",
-    #         ),  # Start function call
+                        tool_call_id="call_123",
-    #         VADUserStartedSpeakingFrame(),  # Should be suppressed
+                        arguments='{"location": "San Francisco"}',
-    #         UserStartedSpeakingFrame(),  # Should be suppressed
+                        context=None,
-    #         VADUserStoppedSpeakingFrame(),  # Should be suppressed
+                    )
-    #         UserStoppedSpeakingFrame(),  # Should be suppressed
+                ]
-    #         FunctionCallResultFrame(
+            ),  # Start function call
-    #             function_name="get_weather",
+            VADUserStartedSpeakingFrame(),  # Should be suppressed
-    #             tool_call_id="call_123",
+            UserStartedSpeakingFrame(),  # Should be suppressed
-    #             arguments='{"location": "San Francisco"}',
+            VADUserStoppedSpeakingFrame(),  # Should be suppressed
-    #             result={"temperature": 22},
+            UserStoppedSpeakingFrame(),  # Should be suppressed
-    #         ),  # End function call
+            FunctionCallResultFrame(
-    #         VADUserStartedSpeakingFrame(),  # Should pass through again
+                function_name="get_weather",
-    #         UserStartedSpeakingFrame(),  # Should pass through again
+                tool_call_id="call_123",
-    #         VADUserStoppedSpeakingFrame(),
+                arguments='{"location": "San Francisco"}',
-    #         UserStoppedSpeakingFrame(),
+                result={"temperature": 22},
-    #     ]
+            ),  # End function call
+            SleepFrame(),
+            VADUserStartedSpeakingFrame(),  # Should pass through again
+            UserStartedSpeakingFrame(),  # Should pass through again
+            VADUserStoppedSpeakingFrame(),
+            UserStoppedSpeakingFrame(),
+        ]
 
-    #     expected_returned_frames = [
+        expected_returned_frames = [
-    #         VADUserStartedSpeakingFrame,
+            VADUserStartedSpeakingFrame,
-    #         UserStartedSpeakingFrame,
+            UserStartedSpeakingFrame,
-    #         VADUserStoppedSpeakingFrame,
+            VADUserStoppedSpeakingFrame,
-    #         UserStoppedSpeakingFrame,
+            UserStoppedSpeakingFrame,
-    #         FunctionCallInProgressFrame,
+            FunctionCallsStartedFrame,
-    #         FunctionCallResultFrame,
+            FunctionCallResultFrame,
-    #         VADUserStartedSpeakingFrame,
+            VADUserStartedSpeakingFrame,
-    #         UserStartedSpeakingFrame,
+            UserStartedSpeakingFrame,
-    #         VADUserStoppedSpeakingFrame,
+            VADUserStoppedSpeakingFrame,
-    #         UserStoppedSpeakingFrame,
+            UserStoppedSpeakingFrame,
-    #     ]
+        ]
 
-    #     await run_test(
+        await run_test(
-    #         filter,
+            filter,
-    #         frames_to_send=frames_to_send,
+            frames_to_send=frames_to_send,
-    #         expected_down_frames=expected_returned_frames,
+            expected_down_frames=expected_returned_frames,
-    #     )
+        )
 
     async def test_mute_until_first_bot_complete_strategy(self):
         filter = STTMuteFilter(

tests/test_user_mute_strategy.py

@@ -0,0 +1,139 @@
+#
+# Copyright (c) 2024-2025 Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import unittest
+
+from pipecat.frames.frames import (
+    BotStartedSpeakingFrame,
+    BotStoppedSpeakingFrame,
+    FunctionCallCancelFrame,
+    FunctionCallFromLLM,
+    FunctionCallResultFrame,
+    FunctionCallsStartedFrame,
+    InterruptionFrame,
+)
+from pipecat.turns.mute.always_user_mute_strategy import AlwaysUserMuteStrategy
+from pipecat.turns.mute.first_speech_user_mute_strategy import FirstSpeechUserMuteStrategy
+from pipecat.turns.mute.function_call_user_mute_strategy import FunctionCallUserMuteStrategy
+from pipecat.turns.mute.mute_until_first_bot_complete_user_mute_strategy import (
+    MuteUntilFirstBotCompleteUserMuteStrategy,
+)
+
+
+class TestAlwaysUserMuteStrategy(unittest.IsolatedAsyncioTestCase):
+    async def test_user_mute_strategy(self):
+        strategy = AlwaysUserMuteStrategy()
+
+        self.assertTrue(await strategy.process_frame(BotStartedSpeakingFrame()))
+        self.assertTrue(await strategy.process_frame(InterruptionFrame()))
+        self.assertFalse(await strategy.process_frame(BotStoppedSpeakingFrame()))
+        self.assertFalse(await strategy.process_frame(InterruptionFrame()))
+
+
+class TestFirstSpeechUserMuteStrategy(unittest.IsolatedAsyncioTestCase):
+    async def test_user_mute_strategy(self):
+        strategy = FirstSpeechUserMuteStrategy()
+
+        self.assertFalse(await strategy.process_frame(InterruptionFrame()))
+        self.assertTrue(await strategy.process_frame(BotStartedSpeakingFrame()))
+        self.assertTrue(await strategy.process_frame(InterruptionFrame()))
+        self.assertFalse(await strategy.process_frame(BotStoppedSpeakingFrame()))
+        self.assertFalse(await strategy.process_frame(InterruptionFrame()))
+
+
+class TestMuteUntilFirstBotCompleteUserMuteStrategy(unittest.IsolatedAsyncioTestCase):
+    async def test_user_mute_strategy(self):
+        strategy = MuteUntilFirstBotCompleteUserMuteStrategy()
+
+        self.assertTrue(await strategy.process_frame(InterruptionFrame()))
+        self.assertTrue(await strategy.process_frame(BotStartedSpeakingFrame()))
+        self.assertTrue(await strategy.process_frame(InterruptionFrame()))
+        self.assertFalse(await strategy.process_frame(BotStoppedSpeakingFrame()))
+        self.assertFalse(await strategy.process_frame(InterruptionFrame()))
+
+
+class TestFunctionCallUserMuteStrategy(unittest.IsolatedAsyncioTestCase):
+    async def test_user_mute_strategy(self):
+        strategy = FunctionCallUserMuteStrategy()
+
+        self.assertFalse(await strategy.process_frame(InterruptionFrame()))
+        # First function call (cancelled)
+        self.assertTrue(
+            await strategy.process_frame(
+                FunctionCallsStartedFrame(
+                    function_calls=[
+                        FunctionCallFromLLM(
+                            function_name="fn_1", tool_call_id="1", arguments={}, context=None
+                        )
+                    ]
+                )
+            )
+        )
+        self.assertTrue(await strategy.process_frame(InterruptionFrame()))
+        self.assertFalse(
+            await strategy.process_frame(
+                FunctionCallCancelFrame(function_name="fn_1", tool_call_id="1")
+            )
+        )
+        self.assertFalse(await strategy.process_frame(InterruptionFrame()))
+
+        # Second function call (finished)
+        self.assertTrue(
+            await strategy.process_frame(
+                FunctionCallsStartedFrame(
+                    function_calls=[
+                        FunctionCallFromLLM(
+                            function_name="fn_2", tool_call_id="2", arguments={}, context=None
+                        )
+                    ]
+                )
+            )
+        )
+        self.assertTrue(await strategy.process_frame(InterruptionFrame()))
+        self.assertFalse(
+            await strategy.process_frame(
+                FunctionCallResultFrame(
+                    function_name="fn_2", tool_call_id="2", arguments={}, result={}
+                )
+            )
+        )
+        self.assertFalse(await strategy.process_frame(InterruptionFrame()))
+
+        # Multiple function calls
+        self.assertTrue(
+            await strategy.process_frame(
+                FunctionCallsStartedFrame(
+                    function_calls=[
+                        FunctionCallFromLLM(
+                            function_name="fn_3", tool_call_id="3", arguments={}, context=None
+                        ),
+                        FunctionCallFromLLM(
+                            function_name="fn_4", tool_call_id="4", arguments={}, context=None
+                        ),
+                    ]
+                )
+            )
+        )
+        self.assertTrue(await strategy.process_frame(InterruptionFrame()))
+        # First function call is done, we still should be muted since there's
+        # another one ongoing.
+        self.assertTrue(
+            await strategy.process_frame(
+                FunctionCallResultFrame(
+                    function_name="fn_3", tool_call_id="3", arguments={}, result={}
+                )
+            )
+        )
+        self.assertTrue(await strategy.process_frame(InterruptionFrame()))
+        # Last function call finishes.
+        self.assertFalse(
+            await strategy.process_frame(
+                FunctionCallResultFrame(
+                    function_name="fn_4", tool_call_id="4", arguments={}, result={}
+                )
+            )
+        )
+        self.assertFalse(await strategy.process_frame(InterruptionFrame()))