LLMService: add new FunctionCallsStartedFrame

CHANGELOG.md

@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added a new frame `FunctionCallsStartedFrame`. This frame is pushed both
+  upstream and downstream from the LLM service to indicate that one or more
+  function calls are going to be executed.
+
 - Added LLM services `on_function_calls_started` event. This event will be
   triggered when the LLM service receives function calls from the model and is
   going to start executing them.

src/pipecat/frames/frames.py

@@ -667,6 +667,32 @@ class MetricsFrame(SystemFrame):
     data: List[MetricsData]
 
 
+@dataclass
+class FunctionCallFromLLM:
+    """Represents a function call returned by the LLM to be registered for execution.
+
+    Attributes:
+        function_name (str): The name of the function.
+        tool_call_id (str): A unique identifier for the function call.
+        arguments (Mapping[str, Any]): The arguments for the function.
+        context (OpenAILLMContext): The LLM context.
+
+    """
+
+    function_name: str
+    tool_call_id: str
+    arguments: Mapping[str, Any]
+    context: Any
+
+
+@dataclass
+class FunctionCallsStartedFrame(SystemFrame):
+    """A frame signaling that one or more function call execution is going to
+    start."""
+
+    function_calls: Sequence[FunctionCallFromLLM]
+
+
 @dataclass
 class FunctionCallInProgressFrame(SystemFrame):
     """A frame signaling that a function call is in progress."""

src/pipecat/processors/aggregators/llm_response.py

@@ -23,6 +23,7 @@ from pipecat.frames.frames import (
     FunctionCallCancelFrame,
     FunctionCallInProgressFrame,
     FunctionCallResultFrame,
+    FunctionCallsStartedFrame,
     InterimTranscriptionFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
@@ -500,7 +501,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
             self._params.expect_stripped_words = kwargs["expect_stripped_words"]
 
         self._started = 0
-        self._function_calls_in_progress: Dict[str, FunctionCallInProgressFrame] = {}
+        self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {}
         self._context_updated_tasks: Set[asyncio.Task] = set()
 
     async def handle_aggregation(self, aggregation: str):
@@ -538,6 +539,8 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
             self.set_tools(frame.tools)
         elif isinstance(frame, LLMSetToolChoiceFrame):
             self.set_tool_choice(frame.tool_choice)
+        elif isinstance(frame, FunctionCallsStartedFrame):
+            await self._handle_function_calls_started(frame)
         elif isinstance(frame, FunctionCallInProgressFrame):
             await self._handle_function_call_in_progress(frame)
         elif isinstance(frame, FunctionCallResultFrame):
@@ -574,6 +577,12 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
         self._started = 0
         self.reset()
 
+    async def _handle_function_calls_started(self, frame: FunctionCallsStartedFrame):
+        function_names = [f"{f.function_name}:{f.tool_call_id}" for f in frame.function_calls]
+        logger.debug(f"{self} FunctionCallsStartedFrame: {function_names}")
+        for function_call in frame.function_calls:
+            self._function_calls_in_progress[function_call.tool_call_id] = None
+
     async def _handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
         logger.debug(
             f"{self} FunctionCallInProgressFrame: [{frame.function_name}:{frame.tool_call_id}]"
@@ -597,9 +606,10 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
 
         await self.handle_function_call_result(frame)
 
+        run_llm = False
+
         # Run inference if the function call result requires it.
         if frame.result:
-            run_llm = False
             if properties and properties.run_llm is not None:
                 # If the tool call result has a run_llm property, use it.
                 run_llm = properties.run_llm
@@ -610,8 +620,8 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
                 # If this is the last function call in progress, run the LLM.
                 run_llm = not bool(self._function_calls_in_progress)
 
-            if run_llm:
+        if run_llm:
-                await self.push_context_frame(FrameDirection.UPSTREAM)
+            await self.push_context_frame(FrameDirection.UPSTREAM)
 
         # Call the `on_context_updated` callback once the function call result
         # is added to the context. Also, run this in a separate task to make

src/pipecat/services/llm_service.py

@@ -18,9 +18,11 @@ from pipecat.frames.frames import (
     EndFrame,
     Frame,
     FunctionCallCancelFrame,
+    FunctionCallFromLLM,
     FunctionCallInProgressFrame,
     FunctionCallResultFrame,
     FunctionCallResultProperties,
+    FunctionCallsStartedFrame,
     StartFrame,
     StartInterruptionFrame,
     UserImageRequestFrame,
@@ -66,24 +68,6 @@ class FunctionCallParams:
     result_callback: FunctionCallResultCallback
 
 
-@dataclass
-class FunctionCallFromLLM:
-    """Represents a function call returned by the LLM to be registered for execution.
-
-    Attributes:
-        function_name (str): The name of the function.
-        tool_call_id (str): A unique identifier for the function call.
-        arguments (Mapping[str, Any]): The arguments for the function.
-        context (OpenAILLMContext): The LLM context.
-
-    """
-
-    function_name: str
-    tool_call_id: str
-    arguments: Mapping[str, Any]
-    context: OpenAILLMContext
-
-
 @dataclass
 class FunctionCallRegistryItem:
     """Represents an entry in our function call registry. This is what the user
@@ -238,8 +222,13 @@ class LLMService(AIService):
     async def run_function_calls(self, function_calls: Sequence[FunctionCallFromLLM]):
         await self._call_event_handler("on_function_calls_started", function_calls)
 
-        total_function_calls = len(function_calls)
+        # Push frame both downstream and upstream
-        for index, function_call in enumerate(function_calls):
+        started_frame_downstream = FunctionCallsStartedFrame(function_calls=function_calls)
+        started_frame_upstream = FunctionCallsStartedFrame(function_calls=function_calls)
+        await self.push_frame(started_frame_downstream, FrameDirection.DOWNSTREAM)
+        await self.push_frame(started_frame_upstream, FrameDirection.UPSTREAM)
+
+        for function_call in function_calls:
             if function_call.function_name in self._functions.keys():
                 item = self._functions[function_call.function_name]
             elif None in self._functions.keys():
@@ -250,20 +239,12 @@ class LLMService(AIService):
                 )
                 continue
 
-            # If we are not running in parallel, run inference on the last
-            # function call. Otherwise, the last function call to finish is the
-            # one that will run the inference.
-            run_llm = None
-            if not self._run_in_parallel:
-                run_llm = index == total_function_calls - 1
-
             runner_item = FunctionCallRunnerItem(
                 registry_item=item,
                 function_name=function_call.function_name,
                 tool_call_id=function_call.tool_call_id,
                 arguments=function_call.arguments,
                 context=function_call.context,
-                run_llm=run_llm,
             )
 
             if self._run_in_parallel: