diff --git a/CHANGELOG.md b/CHANGELOG.md index 51eb668ac..529ece30f 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/examples/foundational/24-stt-mute-filter.py b/examples/foundational/24-stt-mute-filter.py index d57986141..c1cb33aed 100644 --- a/examples/foundational/24-stt-mute-filter.py +++ b/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 ( - LLMMessagesFrame, -) +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 @@ -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. - messages.append({"role": "system", "content": "Please introduce yourself to the user."}) + # 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([LLMMessagesFrame(messages)]) runner = PipelineRunner() diff --git a/src/pipecat/processors/aggregators/openai_llm_context.py b/src/pipecat/processors/aggregators/openai_llm_context.py index 5e4f44093..f647936c9 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/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( - FunctionCallInProgressFrame( + # Also push a SystemFrame upstream for use by other processors, like STTMuteFilter. + 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. - async def function_call_result_callback(result): - await llm.push_frame( - FunctionCallResultFrame( - function_name=function_name, - tool_call_id=tool_call_id, - arguments=arguments, - result=result, - run_llm=run_llm, - ) - ) + # Push frame both downstream and upstream + await llm.push_frame(result_frame_downstream, FrameDirection.DOWNSTREAM) + await llm.push_frame(result_frame_upstream, FrameDirection.UPSTREAM) await f(function_name, tool_call_id, arguments, llm, self, function_call_result_callback) diff --git a/src/pipecat/processors/filters/stt_mute_filter.py b/src/pipecat/processors/filters/stt_mute_filter.py index 9ee216c7f..914198e5a 100644 --- a/src/pipecat/processors/filters/stt_mute_filter.py +++ b/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, - treating them as a single coordinated feature. When STT is muted, interruptions - are automatically disabled. + This processor combines STT muting and interruption control as a coordinated + feature. When STT is muted, interruptions 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: - return False + for strategy in self._config.strategies: + match strategy: + case STTMuteStrategy.FUNCTION_CALL: + if self._function_call_in_progress: + return True - 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) + case STTMuteStrategy.ALWAYS: + if self._bot_is_speaking: + return True + + case STTMuteStrategy.FIRST_SPEECH: + if self._bot_is_speaking and not self._first_speech_handled: + self._first_speech_handled = True + return True + + 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):