Push FunctionCall Frames upstream and downstream; update example

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)