Merge pull request #3120 from pipecat-ai/aleix/function-calls-wait-for-all

allow waiting for all function calls to complete

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Added `wait_for_all` argument to the base `LLMService`. When enabled, this
+  ensures all function calls complete before returning results to the LLM (i.e.,
+  before running a new inference with those results).
+
 ### Changed
 
 - Updated `AICFilter` to use Quail STT as the default model

examples/foundational/19-openai-realtime.py

@@ -14,20 +14,13 @@ from loguru import logger
 
 from pipecat.adapters.schemas.function_schema import FunctionSchema
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
-from pipecat.adapters.services.open_ai_realtime_adapter import OpenAIRealtimeLLMAdapter
 from pipecat.audio.vad.silero import SileroVADAnalyzer
-from pipecat.frames.frames import (
+from pipecat.frames.frames import LLMRunFrame, LLMSetToolsFrame, TranscriptionMessage
-    LLMRunFrame,
-    LLMSetToolsFrame,
-    LLMUpdateSettingsFrame,
-    TranscriptionMessage,
-)
 from pipecat.observers.loggers.transcription_log_observer import TranscriptionLogObserver
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.llm_context import LLMContext
-from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
 from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.transcript_processor import TranscriptProcessor
 from pipecat.runner.types import RunnerArguments

examples/foundational/19a-azure-realtime.py

@@ -19,7 +19,6 @@ from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.llm_context import LLMContext
-from pipecat.processors.aggregators.llm_response import LLMAssistantAggregatorParams
 from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport

src/pipecat/services/llm_service.py

@@ -166,20 +166,24 @@ class LLMService(AIService):
     # However, subclasses should override this with a more specific adapter when necessary.
     adapter_class: Type[BaseLLMAdapter] = OpenAILLMAdapter
 
-    def __init__(self, run_in_parallel: bool = True, **kwargs):
+    def __init__(self, run_in_parallel: bool = True, wait_for_all: bool = False, **kwargs):
         """Initialize the LLM service.
 
         Args:
             run_in_parallel: Whether to run function calls in parallel or sequentially.
                 Defaults to True.
+            wait_for_all: Whether to wait for all function calls (parallel or
+                sequential) to complete. Defaults to False.
             **kwargs: Additional arguments passed to the parent AIService.
+
         """
         super().__init__(**kwargs)
         self._run_in_parallel = run_in_parallel
+        self._wait_for_all = wait_for_all
         self._start_callbacks = {}
         self._adapter = self.adapter_class()
         self._functions: Dict[Optional[str], FunctionCallRegistryItem] = {}
-        self._function_call_tasks: Dict[asyncio.Task, FunctionCallRunnerItem] = {}
+        self._function_call_tasks: Dict[Optional[asyncio.Task], FunctionCallRunnerItem] = {}
         self._sequential_runner_task: Optional[asyncio.Task] = None
         self._tracing_enabled: bool = False
         self._skip_tts: bool = False
@@ -435,6 +439,7 @@ class LLMService(AIService):
 
         await self.broadcast_frame(FunctionCallsStartedFrame, function_calls=function_calls)
 
+        runner_items = []
         for function_call in function_calls:
             if function_call.function_name in self._functions.keys():
                 item = self._functions[function_call.function_name]
@@ -446,28 +451,20 @@ class LLMService(AIService):
                 )
                 continue
 
-            runner_item = FunctionCallRunnerItem(
+            runner_items.append(
-                registry_item=item,
+                FunctionCallRunnerItem(
-                function_name=function_call.function_name,
+                    registry_item=item,
-                tool_call_id=function_call.tool_call_id,
+                    function_name=function_call.function_name,
-                arguments=function_call.arguments,
+                    tool_call_id=function_call.tool_call_id,
-                context=function_call.context,
+                    arguments=function_call.arguments,
+                    context=function_call.context,
+                )
             )
 
-            if self._run_in_parallel:
+        if self._run_in_parallel:
-                task = self.create_task(self._run_function_call(runner_item))
+            await self._run_parallel_function_calls(runner_items)
-                self._function_call_tasks[task] = runner_item
+        else:
-                task.add_done_callback(self._function_call_task_finished)
+            await self._run_sequential_function_calls(runner_items)
-            else:
-                await self._sequential_runner_queue.put(runner_item)
-
-    async def _call_start_function(
-        self, context: OpenAILLMContext | LLMContext, function_name: str
-    ):
-        if function_name in self._start_callbacks.keys():
-            await self._start_callbacks[function_name](function_name, self, context)
-        elif None in self._start_callbacks.keys():
-            return await self._start_callbacks[None](function_name, self, context)
 
     async def request_image_frame(
         self,
@@ -540,6 +537,46 @@ class LLMService(AIService):
             await task
             del self._function_call_tasks[task]
 
+    async def _run_parallel_function_calls(self, runner_items: Sequence[FunctionCallRunnerItem]):
+        tasks = []
+        for runner_item in runner_items:
+            task = self.create_task(self._run_function_call(runner_item))
+            tasks.append(task)
+            self._function_call_tasks[task] = runner_item
+            task.add_done_callback(self._function_call_task_finished)
+
+        if self._wait_for_all:
+            # Protect gather from being cancelled. This will protect all tasks
+            # form being cancelled. That is fine, because we cancel them
+            # explicitly when handling the interruption (InterruptionFrame). We
+            # need to set `return_exceptions=True` because `asyncio.shield()`
+            # will get cancelled (from FrameProcessor process task), then
+            # `asyncio.gather()` will keep running (because it was protected by
+            # the shield). Then, individiaul function call tasks will be
+            # cancelled by us and we don't need to propagate those
+            # CancelledErrors at that point.
+            await asyncio.shield(asyncio.gather(*tasks, return_exceptions=True))
+
+    async def _run_sequential_function_calls(self, runner_items: Sequence[FunctionCallRunnerItem]):
+        if self._wait_for_all:
+            # Run each function call sequentially, waiting for each to complete.
+            for runner_item in runner_items:
+                self._function_call_tasks[None] = runner_item
+                await self._run_function_call(runner_item)
+                del self._function_call_tasks[None]
+        else:
+            # Enqueue all function calls for background execution.
+            for runner_item in runner_items:
+                await self._sequential_runner_queue.put(runner_item)
+
+    async def _call_start_function(
+        self, context: OpenAILLMContext | LLMContext, function_name: str
+    ):
+        if function_name in self._start_callbacks.keys():
+            await self._start_callbacks[function_name](function_name, self, context)
+        elif None in self._start_callbacks.keys():
+            return await self._start_callbacks[None](function_name, self, context)
+
     async def _run_function_call(self, runner_item: FunctionCallRunnerItem):
         if runner_item.function_name in self._functions.keys():
             item = self._functions[runner_item.function_name]
@@ -623,20 +660,19 @@ class LLMService(AIService):
                 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.
-                task.remove_done_callback(self._function_call_task_finished)
-
                 logger.debug(f"{self} Cancelling function call [{name}:{tool_call_id}]...")
 
-                await self.cancel_task(task)
+                if task:
+                    # We remove the callback because we are going to cancel the
+                    # task next, otherwise we will be removing it from the set
+                    # while we are iterating.
+                    task.remove_done_callback(self._function_call_task_finished)
+                    await self.cancel_task(task)
+                    cancelled_tasks.add(task)
 
                 frame = FunctionCallCancelFrame(function_name=name, tool_call_id=tool_call_id)
                 await self.push_frame(frame)
 
-                cancelled_tasks.add(task)
-
                 logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled")
 
         # Remove all cancelled tasks from our set.

uv.lock

@@ -36,12 +36,12 @@ wheels = [
 
 [[package]]
 name = "aic-sdk"
-version = "1.1.0"
+version = "1.2.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "numpy" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/99/83/bf38b95d98c67b8ebc574fb4a4f23c07a3740b51992d7522976173d30b98/aic_sdk-1.1.0.tar.gz", hash = "sha256:04e08df695581c8cb4db8acca20e73815e9f449e7bd08e0162fd55518c727963", size = 34954, upload-time = "2025-11-11T20:45:24.25Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/f9/ba/3ebe31b91e03d42437ec864e9d2af3a52b7ccc73a1a0c1026275956270b0/aic_sdk-1.2.0.tar.gz", hash = "sha256:eeda9a181c679f175dbe6f0efc0c67ec98ff3d84cfe01541fef7fa12ecd505ca", size = 35606, upload-time = "2025-11-20T14:42:14.333Z" }
 
 [[package]]
 name = "aioboto3"
@@ -4666,7 +4666,7 @@ docs = [
 [package.metadata]
 requires-dist = [
     { name = "accelerate", marker = "extra == 'moondream'", specifier = "~=1.10.0" },
-    { name = "aic-sdk", marker = "extra == 'aic'", specifier = "~=1.1.0" },
+    { name = "aic-sdk", marker = "extra == 'aic'", specifier = "~=1.2.0" },
     { name = "aioboto3", marker = "extra == 'aws'", specifier = "~=15.5.0" },
     { name = "aiofiles", specifier = ">=24.1.0,<25" },
     { name = "aiohttp", specifier = ">=3.11.12,<4" },