LLMAssistantAggregator: don't mark function calls as completed when receiving user image

Before, when requesting a user image from a function call we had to wait for a
random time before we could indicate the function call was done. This was to
given time to the aggregator to process the image before marking the function
call as completed.

To avoid this, we now wait for the requested image to be received by the LLM
assistant agrgegator (using an asyncio event). Then, we can successfully mark
the function call as completed.

src/pipecat/frames/frames.py

@@ -11,6 +11,7 @@ including data frames, system frames, and control frames for audio, video, text,
 and LLM processing.
 """
 
+import asyncio
 from dataclasses import dataclass, field
 from typing import (
     TYPE_CHECKING,
@@ -1218,6 +1219,7 @@ class UserImageRequestFrame(SystemFrame):
     function_name: Optional[str] = None
     tool_call_id: Optional[str] = None
     video_source: Optional[str] = None
+    request_event: Optional[asyncio.Event] = None
 
     def __str__(self):
         return f"{self.name}(user: {self.user_id}, video_source: {self.video_source}, function: {self.function_name}, request: {self.tool_call_id})"

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -777,12 +777,6 @@ class LLMAssistantAggregator(LLMContextAggregator):
             )
             return
 
-        del self._function_calls_in_progress[frame.request.tool_call_id]
-
-        # Update context with the image frame
-        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,
@@ -793,6 +787,10 @@ class LLMAssistantAggregator(LLMContextAggregator):
         await self.push_aggregation()
         await self.push_context_frame(FrameDirection.UPSTREAM)
 
+        # Notify who ever requested the image that we have added it to the context.
+        if frame.request and frame.request.request_event:
+            frame.request.request_event.set()
+
     async def _handle_llm_start(self, _: LLMFullResponseStartFrame):
         self._started += 1
 

src/pipecat/services/llm_service.py

@@ -492,11 +492,16 @@ class LLMService(AIService):
         tool_call_id: Optional[str] = None,
         text_content: Optional[str] = None,
         video_source: Optional[str] = None,
+        timeout: Optional[float] = 10.0,
     ):
         """Request an image from a user.
 
         Pushes a UserImageRequestFrame upstream to request an image from the
-        specified user.
+        specified user. The user image can then be processed by the LLM.
+
+        Use this function from a function call if you want the LLM to process
+        the image. If you expect the image to be processed by a vision service,
+        you might want to push a UserImageRequestFrame upstream directly.
 
         Args:
             user_id: The ID of the user to request an image from.
@@ -504,7 +509,11 @@ class LLMService(AIService):
             tool_call_id: Optional tool call ID associated with the request.
             text_content: Optional text content/context for the image request.
             video_source: Optional video source identifier.
+            timeout: Optional timeout for the requested image to be added to the LLM context.
+
         """
+        request_event = asyncio.Event() if timeout else None
+
         await self.push_frame(
             UserImageRequestFrame(
                 user_id=user_id,
@@ -512,10 +521,15 @@ class LLMService(AIService):
                 tool_call_id=tool_call_id,
                 context=text_content,
                 video_source=video_source,
+                request_event=request_event,
             ),
             FrameDirection.UPSTREAM,
         )
 
+        # Wait for the requested image to be added to the context.
+        if request_event:
+            await asyncio.wait_for(request_event.wait(), timeout=timeout)
+
     async def _create_sequential_runner_task(self):
         if not self._sequential_runner_task:
             self._sequential_runner_queue = asyncio.Queue()