Merge pull request #3571 from pipecat-ai/filipi/funcion_call_improvements

Function call improvements

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):

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)

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)
 

src/pipecat/services/google/llm.py

@@ -40,7 +40,6 @@ from pipecat.frames.frames import (
     LLMThoughtStartFrame,
     LLMThoughtTextFrame,
     LLMUpdateSettingsFrame,
-    UserImageRawFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
 from pipecat.processors.aggregators.llm_context import LLMContext
@@ -199,22 +198,6 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
                     if part.function_response and part.function_response.id == tool_call_id:
                         part.function_response.response = {"value": json.dumps(result)}
 
-    async def handle_user_image_frame(self, frame: UserImageRawFrame):
-        """Handle user image frame.
-
-        Args:
-            frame: Frame containing user image data and request context.
-        """
-        await self._update_function_call_result(
-            frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
-        )
-        self._context.add_image_frame_message(
-            format=frame.format,
-            size=frame.size,
-            image=frame.image,
-            text=frame.request.context,
-        )
-
 
 @dataclass
 class GoogleContextAggregatorPair:

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()