From 92d8b37229741fba217ec456f5dadb06c8e15ca8 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Mon, 30 Sep 2024 21:49:29 -0700 Subject: [PATCH] implement vision for openai --- .../14d-function-calling-video.py | 167 ++++++++++++++++++ .../aggregators/openai_llm_context.py | 31 ++++ src/pipecat/services/anthropic.py | 2 +- src/pipecat/services/openai.py | 62 ++++++- 4 files changed, 259 insertions(+), 3 deletions(-) create mode 100644 examples/foundational/14d-function-calling-video.py diff --git a/examples/foundational/14d-function-calling-video.py b/examples/foundational/14d-function-calling-video.py new file mode 100644 index 000000000..f42665d5b --- /dev/null +++ b/examples/foundational/14d-function-calling-video.py @@ -0,0 +1,167 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import aiohttp +import os +import sys + +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineTask +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.openai import OpenAILLMContext, OpenAILLMService +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer + +from openai.types.chat import ChatCompletionToolParam + +from runner import configure + +from loguru import logger + +from dotenv import load_dotenv + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + +video_participant_id = None + + +async def get_weather(function_name, tool_call_id, arguments, llm, context, result_callback): + location = arguments["location"] + await result_callback(f"The weather in {location} is currently 72 degrees and sunny.") + + +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) + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), + ) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + ) + + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm.register_function("get_weather", get_weather) + llm.register_function("get_image", get_image) + + tools = [ + ChatCompletionToolParam( + type="function", + function={ + "name": "get_weather", + "description": "Get the current weather", + "parameters": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use. Infer this from the users location.", + }, + }, + "required": ["location", "format"], + }, + }, + ), + ChatCompletionToolParam( + type="function", + function={ + "name": "get_image", + "description": "Get an image from the video stream.", + "parameters": { + "type": "object", + "properties": { + "question": { + "type": "string", + "description": "The question to ask the AI to generate an image of", + }, + }, + "required": ["question"], + }, + }, + ), + ] + + system_prompt = """\ +You are a helpful assistant who converses with a user and answers questions. Respond concisely to general questions. + +Your response will be turned into speech so use only simple words and punctuation. + +You have access to two tools: get_weather and get_image. + +You can respond to questions about the weather using the get_weather tool. + +You can answer questions about the user's video stream using the get_image tool. Some examples of phrases that \ +indicate you should use the get_image tool are: + - What do you see? + - What's in the video? + - Can you describe the video? + - Tell me about what you see. + - Tell me something interesting about what you see. + - What's happening in the video? +""" + messages = [ + {"role": "system", "content": system_prompt}, + ] + + context = OpenAILLMContext(messages, tools) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), + context_aggregator.user(), + llm, + tts, + transport.output(), + context_aggregator.assistant(), + ] + ) + + task = PipelineTask(pipeline) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + global video_participant_id + video_participant_id = participant["id"] + transport.capture_participant_transcription(participant["id"]) + 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.") + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/pipecat/processors/aggregators/openai_llm_context.py b/src/pipecat/processors/aggregators/openai_llm_context.py index 4bf3f042c..c86045fab 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/src/pipecat/processors/aggregators/openai_llm_context.py @@ -4,6 +4,8 @@ # SPDX-License-Identifier: BSD 2-Clause License # +import base64 +import copy import io import json @@ -60,6 +62,7 @@ class OpenAILLMContext: self._messages: List[ChatCompletionMessageParam] = messages if messages else [] self._tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = tool_choice self._tools: List[ChatCompletionToolParam] | NotGiven = tools + self._user_image_request_context = {} @staticmethod def from_messages(messages: List[dict]) -> "OpenAILLMContext": @@ -114,6 +117,19 @@ class OpenAILLMContext: def get_messages_json(self) -> str: return json.dumps(self._messages, cls=CustomEncoder) + def get_messages_for_logging(self) -> str: + msgs = [] + for message in self.messages: + msg = copy.deepcopy(message) + if "content" in msg: + if isinstance(msg["content"], list): + for item in msg["content"]: + if item["type"] == "image_url": + if item["image_url"]["url"].startswith("data:image/"): + item["image_url"]["url"] = "data:image/..." + msgs.append(msg) + return json.dumps(msgs) + def set_tool_choice(self, tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven): self._tool_choice = tool_choice @@ -122,6 +138,21 @@ class OpenAILLMContext: tools = NOT_GIVEN self._tools = tools + def add_image_frame_message( + self, *, format: str, size: tuple[int, int], image: bytes, text: str = None + ): + buffer = io.BytesIO() + Image.frombytes(format, size, image).save(buffer, format="JPEG") + encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8") + + content = [ + {"type": "text", "text": text}, + {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{encoded_image}"}}, + ] + if text: + content.append({"type": "text", "text": text}) + self.add_message({"role": "user", "content": content}) + async def call_function( self, f: Callable[ diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 86e1e3726..639a922e6 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -55,6 +55,7 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") +# internal use only -- todo: refactor @dataclass class AnthropicImageMessageFrame(Frame): user_image_raw_frame: UserImageRawFrame @@ -359,7 +360,6 @@ class AnthropicLLMContext(OpenAILLMContext): system: str | NotGiven = NOT_GIVEN, ): super().__init__(messages=messages, tools=tools, tool_choice=tool_choice) - self._user_image_request_context = {} # For beta prompt caching. This is a counter that tracks the number of turns # we've seen above the cache threshold. We reset this when we reset the diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 49fd04371..9a7cc9023 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -31,6 +31,8 @@ from pipecat.frames.frames import ( TTSStartedFrame, TTSStoppedFrame, URLImageRawFrame, + UserImageRawFrame, + UserImageRequestFrame, VisionImageRawFrame, ) from pipecat.metrics.metrics import LLMTokenUsage @@ -181,7 +183,7 @@ class BaseOpenAILLMService(LLMService): async def _stream_chat_completions( self, context: OpenAILLMContext ) -> AsyncStream[ChatCompletionChunk]: - logger.debug(f"Generating chat: {context.get_messages_json()}") + logger.debug(f"Generating chat: {context.get_messages_for_logging()}") messages: List[ChatCompletionMessageParam] = context.get_messages() @@ -476,10 +478,49 @@ class OpenAITTSService(TTSService): logger.exception(f"{self} error generating TTS: {e}") +# internal use only -- todo: refactor +@dataclass +class OpenAIImageMessageFrame(Frame): + user_image_raw_frame: UserImageRawFrame + text: Optional[str] = None + + class OpenAIUserContextAggregator(LLMUserContextAggregator): def __init__(self, context: OpenAILLMContext): super().__init__(context=context) + 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 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 = OpenAIImageMessageFrame(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, user_context_aggregator: OpenAIUserContextAggregator, **kwargs): @@ -487,6 +528,7 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): self._user_context_aggregator = user_context_aggregator self._function_calls_in_progress = {} self._function_call_result = None + self._pending_image_frame_message = None async def process_frame(self, frame, direction): await super().process_frame(frame, direction) @@ -507,9 +549,14 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): "FunctionCallResultFrame tool_call_id does not match any function call in progress" ) self._function_call_result = None + elif isinstance(frame, OpenAIImageMessageFrame): + self._pending_image_frame_message = frame + await self._push_aggregation() async def _push_aggregation(self): - if not (self._aggregation or self._function_call_result): + if not ( + self._aggregation or self._function_call_result or self._pending_image_frame_message + ): return run_llm = False @@ -548,6 +595,17 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): else: self._context.add_message({"role": "assistant", "content": aggregation}) + if self._pending_image_frame_message: + frame = self._pending_image_frame_message + self._pending_image_frame_message = None + 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, + ) + run_llm = True + if run_llm: await self._user_context_aggregator.push_context_frame()