LLMService: run function calls sequentially

CHANGELOG.md

@@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Function calls can now be executed sequentially (in the order received in the
+  completion) by passing `run_in_parallel=False` when creating your LLM
+  service. By default, function calls run in parallel, so if the LLM completion
+  returns 2 or more function calls they run concurrently. In both cases,
+  concurrently and sequentailly, a new LLM completion will run when the last
+  function call finishes.
+
 - Added OpenTelemetry tracing for `GeminiMultimodalLiveLLMService` and
   `OpenAIRealtimeBetaLLMService`.
 

src/pipecat/frames/frames.py

@@ -675,6 +675,7 @@ class FunctionCallInProgressFrame(SystemFrame):
     tool_call_id: str
     arguments: Any
     cancel_on_interruption: bool = False
+    run_concurrently: bool = False
 
 
 @dataclass
@@ -701,6 +702,7 @@ class FunctionCallResultFrame(SystemFrame):
     tool_call_id: str
     arguments: Any
     result: Any
+    run_llm: Optional[bool] = None
     properties: Optional[FunctionCallResultProperties] = None
 
 

src/pipecat/pipeline/runner.py

@@ -59,7 +59,7 @@ class PipelineRunner(BaseObject):
         await asyncio.gather(*[t.stop_when_done() for t in self._tasks.values()])
 
     async def cancel(self):
-        logger.debug(f"Canceling runner {self}")
+        logger.debug(f"Cancelling runner {self}")
         await asyncio.gather(*[t.cancel() for t in self._tasks.values()])
 
     def _setup_sigint(self):
@@ -72,7 +72,7 @@ class PipelineRunner(BaseObject):
             self._sig_task = asyncio.create_task(self._sig_cancel())
 
     async def _sig_cancel(self):
-        logger.warning(f"Interruption detected. Canceling runner {self}")
+        logger.warning(f"Interruption detected. Cancelling runner {self}")
         await self.cancel()
 
     def _gc_collect(self):

src/pipecat/processors/aggregators/llm_response.py

@@ -591,6 +591,8 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
             )
             return
 
+        in_progress = self._function_calls_in_progress[frame.tool_call_id]
+
         del self._function_calls_in_progress[frame.tool_call_id]
 
         properties = frame.properties
@@ -600,12 +602,14 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
         # 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
-            else:
+            elif frame.run_llm is not None:
-                # Default behavior is to run the LLM if there are no function calls in progress
+                # If the frame is indicating we should run the LLM, do it.
+                run_llm = frame.run_llm
+            elif in_progress.run_concurrently:
+                # If this was a parallel function call and there are no pending function call, run the LLM.
                 run_llm = not bool(self._function_calls_in_progress)
 
             if run_llm:

src/pipecat/services/anthropic/llm.py

@@ -202,6 +202,12 @@ class AnthropicLLMService(LLMService):
             tool_use_block = None
             json_accumulator = ""
 
+            total_func_calls = 0
+            async for event in response:
+                if event.type == "content_block_start" and event.content_block.type == "tool_use":
+                    total_func_calls += 1
+
+            current_func_call = 0
             async for event in response:
                 # logger.debug(f"Anthropic LLM event: {event}")
 
@@ -226,12 +232,15 @@ class AnthropicLLMService(LLMService):
                     and event.delta.stop_reason == "tool_use"
                 ):
                     if tool_use_block:
+                        run_llm = current_func_call == total_func_calls - 1
                         await self.call_function(
                             context=context,
                             tool_call_id=tool_use_block.id,
                             function_name=tool_use_block.name,
                             arguments=json.loads(json_accumulator) if json_accumulator else dict(),
+                            run_llm=run_llm,
                         )
+                        current_func_call += 1
 
                 # Calculate usage. Do this here in its own if statement, because there may be usage
                 # data embedded in messages that we do other processing for, above.

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -891,12 +891,15 @@ class GeminiMultimodalLiveLLMService(LLMService):
             return
         if not self._context:
             logger.error("Function calls are not supported without a context object.")
-        for call in function_calls:
+        total_items = len(function_calls)
+        for index, call in enumerate(function_calls):
+            run_llm = index == total_items - 1
             await self.call_function(
                 context=self._context,
                 tool_call_id=call.id,
                 function_name=call.name,
                 arguments=call.args,
+                run_llm=run_llm,
             )
 
     @traced_gemini_live(operation="llm_response")

src/pipecat/services/google/llm_openai.py

@@ -112,8 +112,10 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService):
             logger.debug(
                 f"Function list: {functions_list}, Arguments list: {arguments_list}, Tool ID list: {tool_id_list}"
             )
+
+            total_func_calls = len(functions_list)
             for index, (function_name, arguments, tool_id) in enumerate(
-                zip(functions_list, arguments_list, tool_id_list), start=1
+                zip(functions_list, arguments_list, tool_id_list)
             ):
                 if function_name == "":
                     # TODO: Remove the _process_context method once Google resolves the bug
@@ -121,8 +123,8 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService):
                     # which currently results in an empty function name('').
                     continue
                 if self.has_function(function_name):
-                    run_llm = False
                     arguments = json.loads(arguments)
+                    run_llm = index == total_func_calls - 1
                     await self.call_function(
                         context=context,
                         function_name=function_name,

src/pipecat/services/llm_service.py

@@ -7,18 +7,21 @@
 import asyncio
 import inspect
 from dataclasses import dataclass
-from typing import Any, Awaitable, Callable, Mapping, Optional, Protocol, Set, Tuple, Type
+from typing import Any, Awaitable, Callable, Dict, Mapping, Optional, Protocol, Type
 
 from loguru import logger
 
 from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
 from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter
 from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
     Frame,
     FunctionCallCancelFrame,
     FunctionCallInProgressFrame,
     FunctionCallResultFrame,
     FunctionCallResultProperties,
+    StartFrame,
     StartInterruptionFrame,
     UserImageRequestFrame,
 )
@@ -42,19 +45,43 @@ class FunctionCallResultCallback(Protocol):
 
 
 @dataclass
-class FunctionCallEntry:
+class FunctionCallItem:
-    """Represents an internal entry for a function call.
+    """Represents an entry of our function call registry.
 
     Attributes:
         function_name (Optional[str]): The name of the function.
         handler (FunctionCallHandler): The handler for processing function call parameters.
         cancel_on_interruption (bool): Flag indicating whether to cancel the call on interruption.
+        run_concurrently (bool): Flag to indicate if this function call should run concurrently or not.
 
     """
 
     function_name: Optional[str]
     handler: FunctionCallHandler
     cancel_on_interruption: bool
+    run_concurrently: bool
+
+
+@dataclass
+class FunctionCallRunnerItem:
+    """Represents a function call entry for our function call runner. The runner
+    executes function calls in order.
+
+    Attributes:
+        registry_name (Optional[str]): The function call name registration (could be None).
+        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.
+
+    """
+
+    registry_name: Optional[str]
+    function_name: str
+    tool_call_id: str
+    arguments: Mapping[str, Any]
+    context: OpenAILLMContext
+    run_llm: bool
 
 
 @dataclass
@@ -88,10 +115,10 @@ class LLMService(AIService):
 
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
-        self._functions = {}
         self._start_callbacks = {}
         self._adapter = self.adapter_class()
-        self._function_call_tasks: Set[Tuple[asyncio.Task, str, str]] = set()
+        self._functions: Dict[Optional[str], FunctionCallItem] = {}
+        self._function_call_tasks: Dict[asyncio.Task, FunctionCallRunnerItem] = {}
 
         self._register_event_handler("on_completion_timeout")
 
@@ -107,13 +134,28 @@ class LLMService(AIService):
     ) -> Any:
         pass
 
+    async def start(self, frame: StartFrame):
+        await super().start(frame)
+        self._function_call_runner_queue = asyncio.Queue()
+        self._function_call_runner_task = self.create_task(self._function_call_runner_handler())
+
+    async def stop(self, frame: EndFrame):
+        await super().stop(frame)
+        await self._cancel_function_call(None)
+        await self.cancel_task(self._function_call_runner_task)
+
+    async def cancel(self, frame: CancelFrame):
+        await super().cancel(frame)
+        await self._cancel_function_call(None)
+        await self.cancel_task(self._function_call_runner_task)
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
         if isinstance(frame, StartInterruptionFrame):
             await self._handle_interruptions(frame)
 
-    async def _handle_interruptions(self, frame: StartInterruptionFrame):
+    async def _handle_interruptions(self, _: StartInterruptionFrame):
         for function_name, entry in self._functions.items():
             if entry.cancel_on_interruption:
                 await self._cancel_function_call(function_name)
@@ -125,13 +167,15 @@ class LLMService(AIService):
         start_callback=None,
         *,
         cancel_on_interruption: bool = False,
+        run_concurrently: bool = False,
     ):
         # Registering a function with the function_name set to None will run
         # that handler for all functions
-        self._functions[function_name] = FunctionCallEntry(
+        self._functions[function_name] = FunctionCallItem(
             function_name=function_name,
             handler=handler,
             cancel_on_interruption=cancel_on_interruption,
+            run_concurrently=run_concurrently,
         )
 
         # Start callbacks are now deprecated.
@@ -166,16 +210,29 @@ class LLMService(AIService):
         arguments: Mapping[str, Any],
         run_llm: bool = True,
     ):
-        if not function_name in self._functions.keys() and not None in self._functions.keys():
+        if function_name in self._functions.keys():
+            item = self._functions[function_name]
+            registry_name = function_name
+        elif None in self._functions.keys():
+            item = self._functions[None]
+            registry_name = None
+        else:
             return
 
-        task = self.create_task(
+        runner_item = FunctionCallRunnerItem(
-            self._run_function_call(context, tool_call_id, function_name, arguments, run_llm)
+            registry_name=registry_name,
+            function_name=function_name,
+            tool_call_id=tool_call_id,
+            arguments=arguments,
+            context=context,
+            run_llm=run_llm,
         )
-
+        if item.run_concurrently:
-        self._function_call_tasks.add((task, tool_call_id, function_name))
+            task = self.create_task(self._run_function_call(runner_item))
-
+            self._function_call_tasks[task] = runner_item
-        task.add_done_callback(self._function_call_task_finished)
+            task.add_done_callback(self._function_call_task_finished)
+        else:
+            await self._function_call_runner_queue.put(runner_item)
 
     async def call_start_function(self, context: OpenAILLMContext, function_name: str):
         if function_name in self._start_callbacks.keys():
@@ -203,43 +260,45 @@ class LLMService(AIService):
             FrameDirection.UPSTREAM,
         )
 
-    async def _run_function_call(
+    async def _function_call_runner_handler(self):
-        self,
+        while True:
-        context: OpenAILLMContext,
+            runner_item = await self._function_call_runner_queue.get()
-        tool_call_id: str,
+            task = self.create_task(self._run_function_call(runner_item))
-        function_name: str,
+            await self.wait_for_task(task)
-        arguments: Mapping[str, Any],
+
-        run_llm: bool = True,
+    async def _run_function_call(self, runner_item: FunctionCallRunnerItem):
-    ):
+        if runner_item.function_name in self._functions.keys():
-        if function_name in self._functions.keys():
+            item = self._functions[runner_item.function_name]
-            entry = self._functions[function_name]
         elif None in self._functions.keys():
-            entry = self._functions[None]
+            item = self._functions[None]
         else:
             return
 
         logger.debug(
-            f"{self} Calling function [{function_name}:{tool_call_id}] with arguments {arguments}"
+            f"{self} Calling function [{runner_item.function_name}:{runner_item.tool_call_id}] with arguments {runner_item.arguments}"
         )
 
         # NOTE(aleix): This needs to be removed after we remove the deprecation.
-        await self.call_start_function(context, function_name)
+        await self.call_start_function(runner_item.context, runner_item.function_name)
 
-        # Push a SystemFrame downstream. This frame will let our assistant context aggregator
+        # Push a function call in-progress downstream. This frame will let our
-        # know that we are in the middle of a function call. Some contexts/aggregators may
+        # assistant context aggregator know that we are in the middle of a
-        # not need this. But some definitely do (Anthropic, for example).
+        # function call. Some contexts/aggregators may not need this. But some
-        # Also push a SystemFrame upstream for use by other processors, like STTMuteFilter.
+        # definitely do (Anthropic, for example).  Also push it upstream for use
+        # by other processors, like STTMuteFilter.
         progress_frame_downstream = FunctionCallInProgressFrame(
-            function_name=function_name,
+            function_name=runner_item.function_name,
-            tool_call_id=tool_call_id,
+            tool_call_id=runner_item.tool_call_id,
-            arguments=arguments,
+            arguments=runner_item.arguments,
-            cancel_on_interruption=entry.cancel_on_interruption,
+            cancel_on_interruption=item.cancel_on_interruption,
+            run_concurrently=item.run_concurrently,
         )
         progress_frame_upstream = FunctionCallInProgressFrame(
-            function_name=function_name,
+            function_name=runner_item.function_name,
-            tool_call_id=tool_call_id,
+            tool_call_id=runner_item.tool_call_id,
-            arguments=arguments,
+            arguments=runner_item.arguments,
-            cancel_on_interruption=entry.cancel_on_interruption,
+            cancel_on_interruption=item.cancel_on_interruption,
+            run_concurrently=item.run_concurrently,
         )
 
         # Push frame both downstream and upstream
@@ -251,24 +310,26 @@ class LLMService(AIService):
             result: Any, *, properties: Optional[FunctionCallResultProperties] = None
         ):
             result_frame_downstream = FunctionCallResultFrame(
-                function_name=function_name,
+                function_name=runner_item.function_name,
-                tool_call_id=tool_call_id,
+                tool_call_id=runner_item.tool_call_id,
-                arguments=arguments,
+                arguments=runner_item.arguments,
                 result=result,
+                run_llm=runner_item.run_llm,
                 properties=properties,
             )
             result_frame_upstream = FunctionCallResultFrame(
-                function_name=function_name,
+                function_name=runner_item.function_name,
-                tool_call_id=tool_call_id,
+                tool_call_id=runner_item.tool_call_id,
-                arguments=arguments,
+                arguments=runner_item.arguments,
                 result=result,
+                run_llm=runner_item.run_llm,
                 properties=properties,
             )
 
             await self.push_frame(result_frame_downstream, FrameDirection.DOWNSTREAM)
             await self.push_frame(result_frame_upstream, FrameDirection.UPSTREAM)
 
-        signature = inspect.signature(entry.handler)
+        signature = inspect.signature(item.handler)
         if len(signature.parameters) > 1:
             import warnings
 
@@ -279,24 +340,32 @@ class LLMService(AIService):
                     DeprecationWarning,
                 )
 
-            await entry.handler(
+            await item.handler(
-                function_name, tool_call_id, arguments, self, context, function_call_result_callback
+                runner_item.function_name,
+                runner_item.tool_call_id,
+                runner_item.arguments,
+                self,
+                runner_item.context,
+                function_call_result_callback,
             )
         else:
             params = FunctionCallParams(
-                function_name=function_name,
+                function_name=runner_item.function_name,
-                tool_call_id=tool_call_id,
+                tool_call_id=runner_item.tool_call_id,
-                arguments=arguments,
+                arguments=runner_item.arguments,
                 llm=self,
-                context=context,
+                context=runner_item.context,
                 result_callback=function_call_result_callback,
             )
-            await entry.handler(params)
+            await item.handler(params)
 
-    async def _cancel_function_call(self, function_name: str):
+    async def _cancel_function_call(self, function_name: Optional[str]):
         cancelled_tasks = set()
-        for task, tool_call_id, name in self._function_call_tasks:
+        for task, runner_item in self._function_call_tasks.items():
-            if name == function_name:
+            if runner_item.registry_name == function_name:
+                name = runner_item.function_name
+                tool_call_id = runner_item.tool_call_id
+
                 # We remove the callback because we are going to cancel the task
                 # now, otherwise we will be removing it from the set while we
                 # are iterating.
@@ -306,9 +375,7 @@ class LLMService(AIService):
 
                 await self.cancel_task(task)
 
-                frame = FunctionCallCancelFrame(
+                frame = FunctionCallCancelFrame(function_name=name, tool_call_id=tool_call_id)
-                    function_name=function_name, tool_call_id=tool_call_id
-                )
                 await self.push_frame(frame)
 
                 logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled")
@@ -320,9 +387,8 @@ class LLMService(AIService):
             self._function_call_task_finished(task)
 
     def _function_call_task_finished(self, task: asyncio.Task):
-        tuple_to_remove = next((t for t in self._function_call_tasks if t[0] == task), None)
+        if task in self._function_call_tasks:
-        if tuple_to_remove:
+            del self._function_call_tasks[task]
-            self._function_call_tasks.discard(tuple_to_remove)
             # The task is finished so this should exit immediately. We need to
             # do this because otherwise the task manager would report a dangling
             # task if we don't remove it.

src/pipecat/services/openai/base_llm.py

@@ -260,11 +260,12 @@ class BaseOpenAILLMService(LLMService):
             arguments_list.append(arguments)
             tool_id_list.append(tool_call_id)
 
+            total_func_calls = len(functions_list)
             for index, (function_name, arguments, tool_id) in enumerate(
-                zip(functions_list, arguments_list, tool_id_list), start=1
+                zip(functions_list, arguments_list, tool_id_list)
             ):
                 if self.has_function(function_name):
-                    run_llm = False
+                    run_llm = index == total_func_calls - 1
                     arguments = json.loads(arguments)
                     await self.call_function(
                         context=context,