Add LLMSwitcher.register_direct_function()

CHANGELOG.md

@@ -39,6 +39,10 @@ reason")`.
   extract probability metrics from `TranscriptionFrame` objects for Whisper-based,
   OpenAI GPT-4o-transcribe, and Deepgram STT services respectively.
 
+- Added `LLMSwitcher.register_direct_function()`. It works much like
+  `LLMSwitcher.register_function()` in that it's a shorthand for registering
+  functions on all LLMs in the switcher, but for direct functions.
+
 ### Changed
 
 - Bumped the `fastapi` dependency's upperbound to `<0.122.0`.

examples/foundational/48-service-switcher.py

@@ -40,10 +40,22 @@ from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
 load_dotenv(override=True)
 
 
+# "Classic" function
 async def fetch_weather_from_api(params: FunctionCallParams):
     await params.result_callback({"conditions": "nice", "temperature": "75"})
 
 
+# "Direct" function
+async def get_restaurant_recommendation(params: FunctionCallParams, location: str):
+    """
+    Get a restaurant recommendation.
+
+    Args:
+        location (str): The city and state, e.g. "San Francisco, CA".
+    """
+    await params.result_callback({"name": "The Golden Dragon"})
+
+
 # We store functions so objects (e.g. SileroVADAnalyzer) don't get
 # instantiated. The function will be called when the desired transport gets
 # selected.
@@ -109,7 +121,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     llm_switcher = LLMSwitcher(
         llms=[llm_openai, llm_google], strategy_type=ServiceSwitcherStrategyManual
     )
+    # Register a "classic" function
     llm_switcher.register_function("get_current_weather", fetch_weather_from_api)
+    # Register a "direct" function
+    llm_switcher.register_direct_function(get_restaurant_recommendation)
 
     messages = [
         {
@@ -117,7 +132,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
             "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.",
         },
     ]
-    tools = ToolsSchema(standard_tools=[weather_function])
+    tools = ToolsSchema(standard_tools=[weather_function, get_restaurant_recommendation])
 
     context = LLMContext(messages, tools)
     context_aggregator = LLMContextAggregatorPair(context)

src/pipecat/pipeline/llm_switcher.py

@@ -8,6 +8,7 @@
 
 from typing import Any, List, Optional, Type
 
+from pipecat.adapters.schemas.direct_function import DirectFunction
 from pipecat.pipeline.service_switcher import ServiceSwitcher, StrategyType
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.services.llm_service import LLMService
@@ -95,3 +96,22 @@ class LLMSwitcher(ServiceSwitcher[StrategyType]):
                 start_callback=start_callback,
                 cancel_on_interruption=cancel_on_interruption,
             )
+
+    def register_direct_function(
+        self,
+        handler: DirectFunction,
+        *,
+        cancel_on_interruption: bool = True,
+    ):
+        """Register a direct function handler for LLM function calls, on all LLMs, active or not.
+
+        Args:
+            handler: The direct function to register. Must follow DirectFunction protocol.
+            cancel_on_interruption: Whether to cancel this function call when an
+                interruption occurs. Defaults to True.
+        """
+        for llm in self.llms:
+            llm.register_direct_function(
+                handler=handler,
+                cancel_on_interruption=cancel_on_interruption,
+            )