From d3f4cbb6201bf980b93e12df32ae69f9a0984ed7 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Wed, 28 Jan 2026 15:39:06 -0300 Subject: [PATCH] Providing a way to defer the function call results. --- src/pipecat/frames/frames.py | 2 ++ .../processors/aggregators/llm_response.py | 5 ++++ .../aggregators/llm_response_universal.py | 10 ++++--- src/pipecat/services/llm_service.py | 28 ++++++++++++++----- 4 files changed, 34 insertions(+), 11 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 26f9af9d2..8df97d313 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -1466,6 +1466,7 @@ class UserImageRequestFrame(SystemFrame): video_source: Specific video source to capture from. function_name: Name of function that generated this request (if any). tool_call_id: Tool call ID if generated by function call (if any). + result_callback: Optional callback to invoke when the image is retrieved. context: [DEPRECATED] Optional context for the image request. """ @@ -1475,6 +1476,7 @@ class UserImageRequestFrame(SystemFrame): video_source: Optional[str] = None function_name: Optional[str] = None tool_call_id: Optional[str] = None + result_callback: Optional[Any] = None context: Optional[Any] = None def __post_init__(self): diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 1b53e9934..e11ededbf 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -1042,6 +1042,11 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): del self._function_calls_in_progress[frame.request.tool_call_id] + # Call the result_callback if provided. This signals that the image + # has been retrieved and the function call can now complete. + if frame.request and frame.request.result_callback: + await frame.request.result_callback(None) + await self.handle_user_image_frame(frame) await self.push_aggregation() await self.push_context_frame(FrameDirection.UPSTREAM) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 3a8bbb542..200010591 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -941,16 +941,18 @@ class LLMAssistantAggregator(LLMContextAggregator): async def _handle_user_image_frame(self, frame: UserImageRawFrame): image_appended = False - # Check if this image is a result of a function call if so, let's cache. - # TODO(aleix): The function call might have already been executed - # because FunctionCallResultFrame was just faster, in that case we just - # push the context frame now. + # Check if this image is a result of a function call. if ( frame.request and frame.request.tool_call_id and frame.request.tool_call_id in self._function_calls_in_progress ): self._function_calls_image_results[frame.request.tool_call_id] = frame + + # Call the result_callback if provided. This signals that the image + # has been retrieved and the function call can now complete. + if frame.request.result_callback: + await frame.request.result_callback(None) else: image_appended = await self._maybe_append_image_to_context(frame) diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 0368c8feb..035f1b2f2 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -170,17 +170,22 @@ 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, function_call_timeout_secs: float = 10.0, **kwargs + ): """Initialize the LLM service. Args: run_in_parallel: Whether to run function calls in parallel or sequentially. Defaults to True. + function_call_timeout_secs: Timeout in seconds for deferred function calls. + Defaults to 10.0 seconds. **kwargs: Additional arguments passed to the parent AIService. """ super().__init__(**kwargs) self._run_in_parallel = run_in_parallel + self._function_call_timeout_secs = function_call_timeout_secs self._start_callbacks = {} self._adapter = self.adapter_class() self._functions: Dict[Optional[str], FunctionCallRegistryItem] = {} @@ -596,14 +601,26 @@ class LLMService(AIService): cancel_on_interruption=item.cancel_on_interruption, ) - callback_executed = False + # Start a timeout task for deferred function calls + async def timeout_handler(): + await asyncio.sleep(self._function_call_timeout_secs) + logger.warning( + f"{self} Function call [{runner_item.function_name}:{runner_item.tool_call_id}] timed out after {self._function_call_timeout_secs} seconds" + ) + await function_call_result_callback(None) + + timeout_task = self.create_task(timeout_handler()) # Define a callback function that pushes a FunctionCallResultFrame upstream & downstream. async def function_call_result_callback( result: Any, *, properties: Optional[FunctionCallResultProperties] = None ): - nonlocal callback_executed - callback_executed = True + nonlocal timeout_task + + # Cancel timeout task if it exists + if timeout_task and not timeout_task.done(): + await self.cancel_task(timeout_task) + await self.broadcast_frame( FunctionCallResultFrame, function_name=runner_item.function_name, @@ -653,9 +670,6 @@ class LLMService(AIService): error_message = f"Error executing function call [{runner_item.function_name}]: {e}" logger.error(f"{self} {error_message}") await self.push_error(error_msg=error_message, exception=e, fatal=False) - finally: - if not callback_executed: - await function_call_result_callback(None) async def _cancel_function_call(self, function_name: Optional[str]): cancelled_tasks = set()