user image requests can now be related to function calls

examples/foundational/14b-function-calling-anthropic-video.py

@@ -39,7 +39,12 @@ async def get_weather(function_name, tool_call_id, arguments, llm, context, resu
 
 async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback):
     question = arguments["question"]
-    await llm.request_image_frame(user_id=video_participant_id, text_content=question)
+    await llm.request_image_frame(
+        user_id=video_participant_id,
+        function_name=function_name,
+        tool_call_id=tool_call_id,
+        text_content=question,
+    )
 
 
 async def main():

examples/foundational/14d-function-calling-video.py

@@ -39,7 +39,12 @@ async def get_weather(function_name, tool_call_id, arguments, llm, context, resu
 async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback):
     logger.debug(f"!!! IN get_image {video_participant_id}, {arguments}")
     question = arguments["question"]
-    await llm.request_image_frame(user_id=video_participant_id, text_content=question)
+    await llm.request_image_frame(
+        user_id=video_participant_id,
+        function_name=function_name,
+        tool_call_id=tool_call_id,
+        text_content=question,
+    )
 
 
 async def main():
@@ -141,7 +146,7 @@ indicate you should use the get_image tool are:
             await transport.capture_participant_transcription(participant["id"])
             await transport.capture_participant_video(video_participant_id, framerate=0)
             # Kick off the conversation.
-            await tts.say("Hi! Ask me about the weather in San Francisco.")
+            await task.queue_frames([context_aggregator.user().get_context_frame()])
 
         runner = PipelineRunner()
 

examples/foundational/14e-function-calling-gemini.py

@@ -42,7 +42,12 @@ async def get_weather(function_name, tool_call_id, arguments, llm, context, resu
 async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback):
     logger.debug(f"!!! IN get_image {video_participant_id}, {arguments}")
     question = arguments["question"]
-    await llm.request_image_frame(user_id=video_participant_id, text_content=question)
+    await llm.request_image_frame(
+        user_id=video_participant_id,
+        function_name=function_name,
+        tool_call_id=tool_call_id,
+        text_content=question,
+    )
 
 
 async def main():

examples/foundational/20d-persistent-context-gemini.py

@@ -54,7 +54,12 @@ async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context
 
 async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback):
     question = arguments["question"]
-    await llm.request_image_frame(user_id=video_participant_id, text_content=question)
+    await llm.request_image_frame(
+        user_id=video_participant_id,
+        function_name=function_name,
+        tool_call_id=tool_call_id,
+        text_content=question,
+    )
 
 
 async def get_saved_conversation_filenames(

examples/moondream-chatbot/bot.py

@@ -23,7 +23,6 @@ from pipecat.frames.frames import (
     OutputImageRawFrame,
     SpriteFrame,
     TextFrame,
-    UserImageRawFrame,
     UserImageRequestFrame,
 )
 from pipecat.pipeline.parallel_pipeline import ParallelPipeline

src/pipecat/frames/frames.py

@@ -662,13 +662,19 @@ class TransportMessageUrgentFrame(SystemFrame):
 
 @dataclass
 class UserImageRequestFrame(SystemFrame):
-    """A frame user to request an image from the given user."""
+    """A frame to request an image from the given user. The frame might be
+    generated by a function call in which case the corresponding fields will be
+    properly set.
+
+    """
 
     user_id: str
     context: Optional[Any] = None
+    function_name: Optional[str] = None
+    tool_call_id: Optional[str] = None
 
     def __str__(self):
-        return f"{self.name}, user: {self.user_id}"
+        return f"{self.name}(user: {self.user_id}, function: {self.function_name}, request: {self.tool_call_id})"
 
 
 @dataclass
@@ -698,10 +704,11 @@ class UserImageRawFrame(InputImageRawFrame):
     """An image associated to a user."""
 
     user_id: str
+    request: Optional[UserImageRequestFrame] = None
 
     def __str__(self):
         pts = format_pts(self.pts)
-        return f"{self.name}(pts: {pts}, user: {self.user_id}, size: {self.size}, format: {self.format})"
+        return f"{self.name}(pts: {pts}, user: {self.user_id}, size: {self.size}, format: {self.format}, request: {self.request})"
 
 
 @dataclass
@@ -715,18 +722,6 @@ class VisionImageRawFrame(InputImageRawFrame):
         return f"{self.name}(pts: {pts}, text: [{self.text}], size: {self.size}, format: {self.format})"
 
 
-@dataclass
-class UserImageMessageFrame(SystemFrame):
-    """An image associated to a user."""
-
-    user_image_raw_frame: UserImageRawFrame
-    text: Optional[str] = None
-
-    def __str__(self):
-        pts = format_pts(self.pts)
-        return f"{self.name}(pts: {pts}, image: {self.user_image_raw_frame}, text: {self.text})"
-
-
 #
 # Control frames
 #

src/pipecat/processors/aggregators/llm_response.py

@@ -34,7 +34,7 @@ from pipecat.frames.frames import (
     StartInterruptionFrame,
     TextFrame,
     TranscriptionFrame,
-    UserImageMessageFrame,
+    UserImageRawFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
@@ -401,7 +401,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
     async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
         pass
 
-    async def handle_image_frame_message(self, frame: UserImageMessageFrame):
+    async def handle_user_image_frame(self, frame: UserImageRawFrame):
         pass
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -428,8 +428,8 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
             await self._handle_function_call_result(frame)
         elif isinstance(frame, FunctionCallCancelFrame):
             await self._handle_function_call_cancel(frame)
-        elif isinstance(frame, UserImageMessageFrame):
+        elif isinstance(frame, UserImageRawFrame) and frame.request and frame.request.tool_call_id:
-            await self._handle_image_frame_message(frame)
+            await self._handle_user_image_frame(frame)
         elif isinstance(frame, BotStoppedSpeakingFrame):
             await self.push_aggregation()
         else:
@@ -510,8 +510,8 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
             await self.handle_function_call_cancel(frame)
             del self._function_calls_in_progress[frame.tool_call_id]
 
-    async def _handle_image_frame_message(self, frame: UserImageMessageFrame):
+    async def _handle_user_image_frame(self, frame: UserImageRawFrame):
-        await self.handle_image_frame_message(frame)
+        await self.handle_user_image_frame(frame)
         await self.push_aggregation()
         await self.push_context_frame(FrameDirection.UPSTREAM)
 

src/pipecat/processors/aggregators/openai_llm_context.py

@@ -46,7 +46,6 @@ class OpenAILLMContext:
         self._messages: List[ChatCompletionMessageParam] = messages if messages else []
         self._tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = tool_choice
         self._tools: List[ChatCompletionToolParam] | NotGiven | ToolsSchema = tools
-        self._user_image_request_context = {}
         self._llm_adapter: Optional[BaseLLMAdapter] = None
 
     def get_llm_adapter(self) -> Optional[BaseLLMAdapter]:

src/pipecat/services/ai_services.py

@@ -253,9 +253,22 @@ class LLMService(AIService):
         elif None in self._start_callbacks.keys():
             return await self._start_callbacks[None](function_name, self, context)
 
-    async def request_image_frame(self, user_id: str, *, text_content: Optional[str] = None):
+    async def request_image_frame(
+        self,
+        user_id: str,
+        *,
+        function_name: Optional[str] = None,
+        tool_call_id: Optional[str] = None,
+        text_content: Optional[str] = None,
+    ):
         await self.push_frame(
-            UserImageRequestFrame(user_id=user_id, context=text_content), FrameDirection.UPSTREAM
+            UserImageRequestFrame(
+                user_id=user_id,
+                function_name=function_name,
+                tool_call_id=tool_call_id,
+                context=text_content,
+            ),
+            FrameDirection.UPSTREAM,
         )
 
     async def _run_function_call(

src/pipecat/services/anthropic.py

@@ -30,9 +30,7 @@ from pipecat.frames.frames import (
     LLMMessagesFrame,
     LLMTextFrame,
     LLMUpdateSettingsFrame,
-    UserImageMessageFrame,
     UserImageRawFrame,
-    UserImageRequestFrame,
     VisionImageRawFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
@@ -674,42 +672,7 @@ class AnthropicLLMContext(OpenAILLMContext):
 
 
 class AnthropicUserContextAggregator(LLMUserContextAggregator):
-    def __init__(self, context: OpenAILLMContext | AnthropicLLMContext, **kwargs):
+    pass
-        super().__init__(context=context, **kwargs)
-
-    async def process_frame(self, frame, direction):
-        await super().process_frame(frame, direction)
-        # Our parent method has already called push_frame(). So we can't interrupt the
-        # flow here and we don't need to call push_frame() ourselves. Possibly something
-        # to talk through (tagging @aleix). At some point we might need to refactor these
-        # context aggregators.
-        try:
-            if isinstance(frame, UserImageRequestFrame):
-                # The LLM sends a UserImageRequestFrame upstream. Cache any context provided with
-                # that frame so we can use it when we assemble the image message in the assistant
-                # context aggregator.
-                if frame.context:
-                    if isinstance(frame.context, str):
-                        self._context._user_image_request_context[frame.user_id] = frame.context
-                    else:
-                        logger.error(
-                            f"Unexpected UserImageRequestFrame context type: {type(frame.context)}"
-                        )
-                        del self._context._user_image_request_context[frame.user_id]
-                else:
-                    if frame.user_id in self._context._user_image_request_context:
-                        del self._context._user_image_request_context[frame.user_id]
-            elif isinstance(frame, UserImageRawFrame):
-                # Push a new AnthropicImageMessageFrame with the text context we cached
-                # downstream to be handled by our assistant context aggregator. This is
-                # necessary so that we add the message to the context in the right order.
-                text = self._context._user_image_request_context.get(frame.user_id) or ""
-                if text:
-                    del self._context._user_image_request_context[frame.user_id]
-                frame = UserImageMessageFrame(user_image_raw_frame=frame, text=text)
-                await self.push_frame(frame)
-        except Exception as e:
-            logger.error(f"Error processing frame: {e}")
 
 
 #
@@ -723,9 +686,6 @@ class AnthropicUserContextAggregator(LLMUserContextAggregator):
 
 
 class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
-    def __init__(self, context: OpenAILLMContext | AnthropicLLMContext, **kwargs):
-        super().__init__(context=context, **kwargs)
-
     async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
         assistant_message = {"role": "assistant", "content": []}
         assistant_message["content"].append(
@@ -776,10 +736,13 @@ class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
                     ):
                         content["content"] = result
 
-    async def handle_image_frame_message(self, frame: UserImageMessageFrame):
+    async def handle_user_image_frame(self, frame: UserImageRawFrame):
-        self._context.add_image_frame_message(
+        await self._update_function_call_result(
-            format=frame.user_image_raw_frame.format,
+            frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
-            size=frame.user_image_raw_frame.size,
+        )
-            image=frame.user_image_raw_frame.image,
+        self._context.add_image_frame_message(
-            text=frame.text,
+            format=frame.format,
+            size=frame.size,
+            image=frame.image,
+            text=frame.request.context,
         )

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -39,7 +39,7 @@ from pipecat.frames.frames import (
     TTSStartedFrame,
     TTSStoppedFrame,
     TTSTextFrame,
-    UserImageMessageFrame,
+    UserImageRawFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
@@ -119,7 +119,7 @@ class GeminiMultimodalLiveUserContextAggregator(OpenAIUserContextAggregator):
 
 
 class GeminiMultimodalLiveAssistantContextAggregator(OpenAIAssistantContextAggregator):
-    async def handle_image_frame_message(self, frame: UserImageMessageFrame):
+    async def handle_user_image_frame(self, frame: UserImageRawFrame):
         # We don't want to store any images in the context. Revisit this later
         # when the API evolves.
         pass

src/pipecat/services/google/google.py

@@ -49,7 +49,7 @@ from pipecat.frames.frames import (
     TTSStartedFrame,
     TTSStoppedFrame,
     URLImageRawFrame,
-    UserImageMessageFrame,
+    UserImageRawFrame,
     VisionImageRawFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
@@ -624,12 +624,15 @@ class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
                     if part.function_response and part.function_response.id == tool_call_id:
                         part.function_response.response = result
 
-    async def handle_image_frame_message(self, frame: UserImageMessageFrame):
+    async def handle_user_image_frame(self, frame: UserImageRawFrame):
+        await self._update_function_call_result(
+            frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
+        )
         self._context.add_image_frame_message(
-            format=frame.user_image_raw_frame.format,
+            format=frame.format,
-            size=frame.user_image_raw_frame.size,
+            size=frame.size,
-            image=frame.user_image_raw_frame.image,
+            image=frame.image,
-            text=frame.text,
+            text=frame.request.context,
         )
 
 

src/pipecat/services/openai.py

@@ -40,9 +40,7 @@ from pipecat.frames.frames import (
     TTSStartedFrame,
     TTSStoppedFrame,
     URLImageRawFrame,
-    UserImageMessageFrame,
     UserImageRawFrame,
-    UserImageRequestFrame,
     VisionImageRawFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
@@ -557,46 +555,10 @@ class OpenAITTSService(TTSService):
 
 
 class OpenAIUserContextAggregator(LLMUserContextAggregator):
-    def __init__(self, context: OpenAILLMContext, **kwargs):
+    pass
-        super().__init__(context=context, **kwargs)
-
-    async def process_frame(self, frame, direction):
-        await super().process_frame(frame, direction)
-        # Our parent method has already called push_frame(). So we can't interrupt the
-        # flow here and we don't need to call push_frame() ourselves.
-        try:
-            if isinstance(frame, UserImageRequestFrame):
-                # The LLM sends a UserImageRequestFrame upstream. Cache any context provided with
-                # that frame so we can use it when we assemble the image message in the assistant
-                # context aggregator.
-                if frame.context:
-                    if isinstance(frame.context, str):
-                        self._context._user_image_request_context[frame.user_id] = frame.context
-                    else:
-                        logger.error(
-                            f"Unexpected UserImageRequestFrame context type: {type(frame.context)}"
-                        )
-                        del self._context._user_image_request_context[frame.user_id]
-                else:
-                    if frame.user_id in self._context._user_image_request_context:
-                        del self._context._user_image_request_context[frame.user_id]
-            elif isinstance(frame, UserImageRawFrame):
-                # Push a new OpenAIImageMessageFrame with the text context we cached
-                # downstream to be handled by our assistant context aggregator. This is
-                # necessary so that we add the message to the context in the right order.
-                text = self._context._user_image_request_context.get(frame.user_id) or ""
-                if text:
-                    del self._context._user_image_request_context[frame.user_id]
-                frame = UserImageMessageFrame(user_image_raw_frame=frame, text=text)
-                await self.push_frame(frame)
-        except Exception as e:
-            logger.error(f"Error processing frame: {e}")
 
 
 class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
-    def __init__(self, context: OpenAILLMContext, **kwargs):
-        super().__init__(context=context, **kwargs)
-
     async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
         self._context.add_message(
             {
@@ -645,10 +607,13 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
             ):
                 message["content"] = result
 
-    async def handle_image_frame_message(self, frame: UserImageMessageFrame):
+    async def handle_user_image_frame(self, frame: UserImageRawFrame):
-        self._context.add_image_frame_message(
+        await self._update_function_call_result(
-            format=frame.user_image_raw_frame.format,
+            frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
-            size=frame.user_image_raw_frame.size,
+        )
-            image=frame.user_image_raw_frame.image,
+        self._context.add_image_frame_message(
-            text=frame.text,
+            format=frame.format,
+            size=frame.size,
+            image=frame.image,
+            text=frame.request.context,
         )

src/pipecat/transports/services/daily.py

@@ -898,7 +898,7 @@ class DailyInputTransport(BaseInputTransport):
         await super().process_frame(frame, direction)
 
         if isinstance(frame, UserImageRequestFrame):
-            await self.request_participant_image(frame.user_id)
+            await self.request_participant_image(frame)
 
     #
     # Frames
@@ -935,16 +935,16 @@ class DailyInputTransport(BaseInputTransport):
         self._video_renderers[participant_id] = {
             "framerate": framerate,
             "timestamp": 0,
-            "render_next_frame": False,
+            "render_next_frame": [],
         }
 
         await self._client.capture_participant_video(
             participant_id, self._on_participant_video_frame, framerate, video_source, color_format
         )
 
-    async def request_participant_image(self, participant_id: str):
+    async def request_participant_image(self, frame: UserImageRequestFrame):
-        if participant_id in self._video_renderers:
+        if frame.user_id in self._video_renderers:
-            self._video_renderers[participant_id]["render_next_frame"] = True
+            self._video_renderers[frame.user_id]["render_next_frame"].append(frame)
 
     async def _on_participant_video_frame(self, participant_id: str, buffer, size, format):
         render_frame = False
@@ -953,17 +953,24 @@ class DailyInputTransport(BaseInputTransport):
         prev_time = self._video_renderers[participant_id]["timestamp"]
         framerate = self._video_renderers[participant_id]["framerate"]
 
+        # Some times we render frames because of a request.
+        request_frame = None
+
         if framerate > 0:
             next_time = prev_time + 1 / framerate
             render_frame = (next_time - curr_time) < 0.1
 
         elif self._video_renderers[participant_id]["render_next_frame"]:
-            self._video_renderers[participant_id]["render_next_frame"] = False
+            request_frame = self._video_renderers[participant_id]["render_next_frame"].pop(0)
             render_frame = True
 
         if render_frame:
             frame = UserImageRawFrame(
-                user_id=participant_id, image=buffer, size=size, format=format
+                user_id=participant_id,
+                request=request_frame,
+                image=buffer,
+                size=size,
+                format=format,
             )
             await self.push_frame(frame)
             self._video_renderers[participant_id]["timestamp"] = curr_time