From 3e9678db8430101a7964d9f3e246a6db48355c9d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 19 Mar 2025 13:12:42 -0700 Subject: [PATCH] user image requests can now be related to function calls --- .../14b-function-calling-anthropic-video.py | 7 ++- .../14d-function-calling-video.py | 9 ++- .../14e-function-calling-gemini.py | 7 ++- .../20d-persistent-context-gemini.py | 7 ++- examples/moondream-chatbot/bot.py | 1 - src/pipecat/frames/frames.py | 25 ++++---- .../processors/aggregators/llm_response.py | 12 ++-- .../aggregators/openai_llm_context.py | 1 - src/pipecat/services/ai_services.py | 17 +++++- src/pipecat/services/anthropic.py | 57 ++++--------------- .../services/gemini_multimodal_live/gemini.py | 4 +- src/pipecat/services/google/google.py | 15 +++-- src/pipecat/services/openai.py | 55 ++++-------------- src/pipecat/transports/services/daily.py | 21 ++++--- 14 files changed, 101 insertions(+), 137 deletions(-) diff --git a/examples/foundational/14b-function-calling-anthropic-video.py b/examples/foundational/14b-function-calling-anthropic-video.py index 302722be2..cd9777b72 100644 --- a/examples/foundational/14b-function-calling-anthropic-video.py +++ b/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(): diff --git a/examples/foundational/14d-function-calling-video.py b/examples/foundational/14d-function-calling-video.py index 6e290d55f..a7e997f9b 100644 --- a/examples/foundational/14d-function-calling-video.py +++ b/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() diff --git a/examples/foundational/14e-function-calling-gemini.py b/examples/foundational/14e-function-calling-gemini.py index f0003fbab..8448643b4 100644 --- a/examples/foundational/14e-function-calling-gemini.py +++ b/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(): diff --git a/examples/foundational/20d-persistent-context-gemini.py b/examples/foundational/20d-persistent-context-gemini.py index 740bdc68f..2381ea131 100644 --- a/examples/foundational/20d-persistent-context-gemini.py +++ b/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( diff --git a/examples/moondream-chatbot/bot.py b/examples/moondream-chatbot/bot.py index 5dd88f4d1..6639c08c1 100644 --- a/examples/moondream-chatbot/bot.py +++ b/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 diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index e589e9480..c2a79461f 100644 --- a/src/pipecat/frames/frames.py +++ b/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 # diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 494d1de38..ff1b6e1be 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/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): - await self._handle_image_frame_message(frame) + elif isinstance(frame, UserImageRawFrame) and frame.request and frame.request.tool_call_id: + 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): - await self.handle_image_frame_message(frame) + async def _handle_user_image_frame(self, frame: UserImageRawFrame): + 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/openai_llm_context.py b/src/pipecat/processors/aggregators/openai_llm_context.py index 93c2875be..2e5ade0a0 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/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]: diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 2f30f505f..678132cb6 100644 --- a/src/pipecat/services/ai_services.py +++ b/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( diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 8e06f4558..a80e2154c 100644 --- a/src/pipecat/services/anthropic.py +++ b/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): - 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}") + pass # @@ -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): - self._context.add_image_frame_message( - format=frame.user_image_raw_frame.format, - size=frame.user_image_raw_frame.size, - image=frame.user_image_raw_frame.image, - text=frame.text, + 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.format, + size=frame.size, + image=frame.image, + text=frame.request.context, ) diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index da1697da6..801af46b4 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/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 diff --git a/src/pipecat/services/google/google.py b/src/pipecat/services/google/google.py index 3db2044db..128eb43b6 100644 --- a/src/pipecat/services/google/google.py +++ b/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, - size=frame.user_image_raw_frame.size, - image=frame.user_image_raw_frame.image, - text=frame.text, + format=frame.format, + size=frame.size, + image=frame.image, + text=frame.request.context, ) diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 700f1c0b5..8318739de 100644 --- a/src/pipecat/services/openai.py +++ b/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): - 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}") + pass 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): - self._context.add_image_frame_message( - format=frame.user_image_raw_frame.format, - size=frame.user_image_raw_frame.size, - image=frame.user_image_raw_frame.image, - text=frame.text, + 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.format, + size=frame.size, + image=frame.image, + text=frame.request.context, ) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index f4b83dfa7..af7d2308c 100644 --- a/src/pipecat/transports/services/daily.py +++ b/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): - if participant_id in self._video_renderers: - self._video_renderers[participant_id]["render_next_frame"] = True + async def request_participant_image(self, frame: UserImageRequestFrame): + if frame.user_id in self._video_renderers: + 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