From bff8747e38a5a1594c2324c40fdc226219618cec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sun, 23 Nov 2025 17:54:15 -0800 Subject: [PATCH 1/3] LLMService: allow waiting for all function calls to complete --- CHANGELOG.md | 6 ++ src/pipecat/services/llm_service.py | 96 ++++++++++++++++++++--------- 2 files changed, 72 insertions(+), 30 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 24a422966..3819cb5e3 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 272df1762..9bc72b0f8 100644 --- a/src/pipecat/services/llm_service.py +++ b/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( - 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, + runner_items.append( + 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, + ) ) - if self._run_in_parallel: - 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) - 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) + if self._run_in_parallel: + await self._run_parallel_function_calls(runner_items) + else: + await self._run_sequential_function_calls(runner_items) 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. From 7648d0436c6268684951c919c8e8a62d35ba709d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 1 Dec 2025 18:30:34 -0800 Subject: [PATCH 2/3] examples(19): linting --- examples/foundational/19-openai-realtime.py | 9 +-------- examples/foundational/19a-azure-realtime.py | 1 - 2 files changed, 1 insertion(+), 9 deletions(-) diff --git a/examples/foundational/19-openai-realtime.py b/examples/foundational/19-openai-realtime.py index dca28b9fb..b7d99f525 100644 --- a/examples/foundational/19-openai-realtime.py +++ b/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 ( - LLMRunFrame, - LLMSetToolsFrame, - LLMUpdateSettingsFrame, - TranscriptionMessage, -) +from pipecat.frames.frames import LLMRunFrame, LLMSetToolsFrame, 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 diff --git a/examples/foundational/19a-azure-realtime.py b/examples/foundational/19a-azure-realtime.py index b56c05af9..7b07985be 100644 --- a/examples/foundational/19a-azure-realtime.py +++ b/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 From 9fafc1692d19517b592ec8526fd8994e53c727e8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 1 Dec 2025 18:32:00 -0800 Subject: [PATCH 3/3] update uv.lock --- uv.lock | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/uv.lock b/uv.lock index 35ae93faf..9a1ca9135 100644 --- a/uv.lock +++ b/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" },