From 29ca1b7855a29149fa195bbaa899177816b1c3e4 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Tue, 13 Aug 2024 11:01:24 -0700 Subject: [PATCH] Anthropic tool use core Pipecat pieces refactored (#369) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * processors(rtvi): rtvi 0.1 message protocol * added a single function call handler * wip - function calling * fixup * fixup * fixup * processors(rtvi): no need for configure_on_start() * processors(rtvi): add new option values if they haven't been set yet * Add the model name to the LLM usage metrics * wip - anthropic tool calling * still wip - anthropic tool use and vision * anthropic tools and vision working * anthropic tool calling and vision * Cartesia error handling * Anthropic tool use core Pipecat pieces refactored as per plan * aleix has good ideas * Usage metrics for Anthropic LLMs * fix function call result state not getting cleared bug * Pass **kwargs through from AnthropicLLMService constructor * about to tinker with anthropic * added openai function calling * openai function calling * fixup --------- Co-authored-by: Aleix Conchillo Flaqué Co-authored-by: Chad Bailey Co-authored-by: mattie ruth backman Co-authored-by: chadbailey59 --- .../12c-describe-video-anthropic.py | 13 +- examples/foundational/14-function-calling.py | 11 +- examples/foundational/19a-tools-anthropic.py | 120 +++++ .../foundational/19b-tools-video-anthropic.py | 171 +++++++ src/pipecat/frames/frames.py | 31 +- .../processors/aggregators/llm_response.py | 41 +- .../aggregators/openai_llm_context.py | 41 +- src/pipecat/processors/frameworks/rtvi.py | 55 ++- src/pipecat/processors/logger.py | 17 +- src/pipecat/services/ai_services.py | 47 +- src/pipecat/services/anthropic.py | 439 +++++++++++++++--- src/pipecat/services/cartesia.py | 7 + src/pipecat/services/openai.py | 158 +++++-- 13 files changed, 1009 insertions(+), 142 deletions(-) create mode 100644 examples/foundational/19a-tools-anthropic.py create mode 100644 examples/foundational/19b-tools-video-anthropic.py diff --git a/examples/foundational/12c-describe-video-anthropic.py b/examples/foundational/12c-describe-video-anthropic.py index 052ffa5d4..cc1f14c92 100644 --- a/examples/foundational/12c-describe-video-anthropic.py +++ b/examples/foundational/12c-describe-video-anthropic.py @@ -16,7 +16,7 @@ from pipecat.pipeline.task import PipelineTask from pipecat.processors.aggregators.user_response import UserResponseAggregator from pipecat.processors.aggregators.vision_image_frame import VisionImageFrameAggregator from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.services.elevenlabs import ElevenLabsTTSService +from pipecat.services.cartesia import CartesiaTTSService from pipecat.services.anthropic import AnthropicLLMService from pipecat.transports.services.daily import DailyParams, DailyTransport from pipecat.vad.silero import SileroVADAnalyzer @@ -72,14 +72,13 @@ async def main(): vision_aggregator = VisionImageFrameAggregator() anthropic = AnthropicLLMService( - api_key=os.getenv("ANTHROPIC_API_KEY"), - model="claude-3-sonnet-20240229" + api_key=os.getenv("ANTHROPIC_API_KEY") ) - tts = ElevenLabsTTSService( - aiohttp_session=session, - api_key=os.getenv("ELEVENLABS_API_KEY"), - voice_id=os.getenv("ELEVENLABS_VOICE_ID"), + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + sample_rate=16000, ) @transport.event_handler("on_first_participant_joined") diff --git a/examples/foundational/14-function-calling.py b/examples/foundational/14-function-calling.py index ea6d57b78..723997e7e 100644 --- a/examples/foundational/14-function-calling.py +++ b/examples/foundational/14-function-calling.py @@ -36,11 +36,11 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def start_fetch_weather(llm): - await llm.push_frame(TextFrame("Let me think.")) +async def start_fetch_weather(llm, function_name): + await llm.push_frame(TextFrame("Let me check on that.")) -async def fetch_weather_from_api(llm, args): +async def fetch_weather_from_api(llm, function_name, args): return {"conditions": "nice", "temperature": "75"} @@ -69,8 +69,11 @@ async def main(): llm = OpenAILLMService( api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + # Register a function_name of None to get all functions + # sent to the same callback with an additional function_name parameter. llm.register_function( - "get_current_weather", + #"get_current_weather", + None, fetch_weather_from_api, start_callback=start_fetch_weather) diff --git a/examples/foundational/19a-tools-anthropic.py b/examples/foundational/19a-tools-anthropic.py new file mode 100644 index 000000000..8e0fc6bbc --- /dev/null +++ b/examples/foundational/19a-tools-anthropic.py @@ -0,0 +1,120 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import aiohttp +import os +import sys + +from pipecat.frames.frames import LLMMessagesFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.services.cartesia import CartesiaTTSService + +from pipecat.services.anthropic import AnthropicLLMService, AnthropicUserContextAggregator, AnthropicAssistantContextAggregator +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer + +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + + +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") + + +async def get_weather(function_name, tool_call_id, arguments, context, result_callback): + location = arguments["location"] + await result_callback(f"The weather in {location} is currently 72 degrees and sunny.") + + +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 + sample_rate=16000, + ) + + llm = AnthropicLLMService( + api_key=os.getenv("OPENAI_API_KEY"), + model="claude-3-5-sonnet-20240620" + ) + llm.register_function("get_weather", get_weather) + + tools = [ + { + "name": "get_weather", + "description": "Get the current weather in a given location", + "input_schema": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + } + }, + "required": ["location"], + }, + } + ] + + # todo: test with very short initial user message + + messages = [{"role": "system", + "content": "You are a helpful assistant who can report the weather in any location in the universe. Respond concisely. Your response will be turned into speech so use only simple words and punctuation."}, + {"role": "user", + "content": " Start the conversation by introducing yourself."}] + + context = OpenAILLMContext(messages, tools) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline([ + transport.input(), # Transport user input + context_aggregator.user(), # User speech to text + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses and tool context + ]) + + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True)) + + @ transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + await task.queue_frames([LLMMessagesFrame(messages)]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/foundational/19b-tools-video-anthropic.py b/examples/foundational/19b-tools-video-anthropic.py new file mode 100644 index 000000000..51a321bf9 --- /dev/null +++ b/examples/foundational/19b-tools-video-anthropic.py @@ -0,0 +1,171 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import aiohttp +import os +import sys + +from pipecat.frames.frames import LLMMessagesFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.services.cartesia import CartesiaTTSService + +from pipecat.services.anthropic import AnthropicLLMService, AnthropicUserContextAggregator, AnthropicAssistantContextAggregator +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer + +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + + +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") +# logger.add(sys.stderr, level="TRACE") + +video_participant_id = None + +# globally declare llm so that we can access it in the get_image function +llm = None + + +async def get_weather(function_name, tool_call_id, arguments, 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, context, result_callback): + global llm + question = arguments["question"] + await llm.request_image_frame(user_id=video_participant_id, text_content=question) + + +async def main(): + global llm + + 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 + sample_rate=16000, + ) + + llm = AnthropicLLMService( + api_key=os.getenv("ANTHROPIC_API_KEY"), + model="claude-3-5-sonnet-20240620" + ) + llm.register_function("get_weather", get_weather) + llm.register_function("get_image", get_image) + + tools = [ + { + "name": "get_weather", + "description": "Get the current weather in a given location", + "input_schema": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + } + }, + "required": ["location"], + }, + }, + { + "name": "get_image", + "description": "Get an image from the video stream.", + "input_schema": { + "type": "object", + "properties": { + "question": { + "type": "string", + "description": "The question that the user is asking about the image.", + } + }, + "required": ["question"], + }, + } + ] + + # todo: test with very short initial user message + + 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? + +If you need to use a tool, simply use the tool. Do not tell the user the tool you are using. Be brief and concise. + """ + + messages = [{"role": "system", + "content": system_prompt, + "content": "Start the conversation by introducing yourself."}] + + context = OpenAILLMContext(messages, tools) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline([ + transport.input(), # Transport user input + context_aggregator.user(), # User speech to text + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses and tool context + ]) + + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True)) + + @ 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(video_participant_id) + transport.capture_participant_video(video_participant_id, framerate=0) + # Kick off the conversation. + await task.queue_frames([LLMMessagesFrame(messages)]) + + runner = PipelineRunner() + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 90127492b..9efbb7595 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -4,7 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # -from typing import Any, List, Mapping, Tuple +from typing import Any, List, Mapping, Tuple, Optional from dataclasses import dataclass, field @@ -177,6 +177,15 @@ class LLMMessagesUpdateFrame(DataFrame): messages: List[dict] +@dataclass +class LLMSetToolsFrame(DataFrame): + """A frame containing a list of tools for an LLM to use for function calling. + The specific format depends on the LLM being used, but it should typically + contain JSON Schema objects. + """ + tools: List[dict] + + @dataclass class TTSSpeakFrame(DataFrame): """A frame that contains a text that should be spoken by the TTS in the @@ -389,6 +398,7 @@ class TTSStoppedFrame(ControlFrame): class UserImageRequestFrame(ControlFrame): """A frame user to request an image from the given user.""" user_id: str + context: Optional[any] def __str__(self): return f"{self.name}, user: {self.user_id}" @@ -406,3 +416,22 @@ class TTSVoiceUpdateFrame(ControlFrame): """A control frame containing a request to update to a new TTS voice. """ voice: str + + +@dataclass +class FunctionCallInProgressFrame(SystemFrame): + """A frame signaling that a function call is in progress. + """ + function_name: str + tool_call_id: str + arguments: str + + +@dataclass +class FunctionCallResultFrame(DataFrame): + """A frame containing the result of an LLM function (tool) call. + """ + function_name: str + tool_call_id: str + arguments: str + result: any diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 6939a70c4..3c0a901af 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -4,9 +4,10 @@ # SPDX-License-Identifier: BSD 2-Clause License # +import sys from typing import List -from pipecat.services.openai import OpenAILLMContextFrame, OpenAILLMContext +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame, OpenAILLMContext from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.frames.frames import ( @@ -17,6 +18,7 @@ from pipecat.frames.frames import ( LLMMessagesAppendFrame, LLMMessagesFrame, LLMMessagesUpdateFrame, + LLMSetToolsFrame, StartInterruptionFrame, TranscriptionFrame, TextFrame, @@ -80,6 +82,10 @@ class LLMResponseAggregator(FrameProcessor): # # and T2 would be dropped. + async def _set_tools(self, tools: List): + # noop in the base class + pass + async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -129,17 +135,28 @@ class LLMResponseAggregator(FrameProcessor): elif isinstance(frame, LLMMessagesUpdateFrame): # We push the frame downstream so the assistant aggregator gets # updated as well. - await self.push_frame(frame) + # TODO-CB: Now we're replacing the contents of the array so we + # don't need to push the frame here + # await self.push_frame(frame) # We can now reset this one. self._reset() - self._messages = frame.messages - messages_frame = LLMMessagesFrame(self._messages) - await self.push_frame(messages_frame) + self._set_messages(frame.messages) + # messages_frame = LLMMessagesFrame(self._messages) + # await self.push_frame(messages_frame) + await self.push_messages_frame() + elif isinstance(frame, LLMSetToolsFrame): + await self.push_frame(frame) + await self._set_tools(frame.tools) else: await self.push_frame(frame, direction) if send_aggregation: await self._push_aggregation() + + # TODO-CB: Types + def _set_messages(self, messages): + self._messages.clear() + self._messages.extend(messages) async def _push_aggregation(self): if len(self._aggregation) > 0: @@ -243,6 +260,20 @@ class LLMContextAggregator(LLMResponseAggregator): self._context = context super().__init__(**kwargs) + # TODO-CB: thanks, I hate it + self._messages = context.messages + + + async def _set_tools(self, tools: List): + # We push the frame downstream so the assistant aggregator gets + # updated as well. + self._context.tools = tools + + # TODO-CB: Types + def _set_messages(self, messages): + self._messages.clear() + self._messages.extend(messages) + async def _push_aggregation(self): if len(self._aggregation) > 0: diff --git a/src/pipecat/processors/aggregators/openai_llm_context.py b/src/pipecat/processors/aggregators/openai_llm_context.py index 65c8da6ad..bd88a9580 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/src/pipecat/processors/aggregators/openai_llm_context.py @@ -12,7 +12,9 @@ from typing import List from PIL import Image -from pipecat.frames.frames import Frame, VisionImageRawFrame +from pipecat.frames.frames import Frame, VisionImageRawFrame, FunctionCallInProgressFrame, FunctionCallResultFrame +from pipecat.processors.frame_processor import FrameProcessor + from openai._types import NOT_GIVEN, NotGiven @@ -50,12 +52,11 @@ class OpenAILLMContext: @staticmethod def from_messages(messages: List[dict]) -> "OpenAILLMContext": context = OpenAILLMContext() + for message in messages: - context.add_message({ - "content": message["content"], - "role": message["role"], - "name": message["name"] if "name" in message else message["role"] - }) + if "name" not in message: + message["name"] = message["role"] + context.add_message(message) return context @staticmethod @@ -102,6 +103,34 @@ class OpenAILLMContext: tools = NOT_GIVEN self.tools = tools + + async def call_function( + self, + f: callable, + *, + function_name: str, + tool_call_id: str, + arguments: str, + llm: FrameProcessor) -> None: + + # Push a SystemFrame downstream. This frame will let our assistant context aggregator + # know that we are in the middle of a function call. Some contexts/aggregators may + # not need this. But some definitely do (Anthropic, for example). + await llm.push_frame(FunctionCallInProgressFrame( + function_name=function_name, + tool_call_id=tool_call_id, + arguments=arguments, + )) + + # Define a callback function that pushes a FunctionCallResultFrame downstream. + async def function_call_result_callback(result): + await llm.push_frame(FunctionCallResultFrame( + function_name=function_name, + tool_call_id=tool_call_id, + arguments=arguments, + result=result)) + await f(function_name=function_name, tool_call_id=tool_call_id, arguments=arguments, + context=self, result_callback=function_call_result_callback) @dataclass diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 4643772c1..ff2055af3 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -20,6 +20,7 @@ from pipecat.frames.frames import ( TranscriptionFrame, TransportMessageFrame, UserStartedSpeakingFrame, + FunctionCallResultFrame, UserStoppedSpeakingFrame) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -176,7 +177,6 @@ class RTVIActionResponse(BaseModel): id: str data: RTVIActionResponseData - class RTVIBotReadyData(BaseModel): version: str config: List[RTVIServiceConfig] @@ -187,6 +187,34 @@ class RTVIBotReady(BaseModel): type: Literal["bot-ready"] = "bot-ready" data: RTVIBotReadyData + +class RTVILLMFunctionCallMessageData(BaseModel): + function_name: str + tool_call_id: str + args: dict + + +class RTVILLMFunctionCallMessage(BaseModel): + label: Literal["rtvi-ai"] = "rtvi-ai" + type: Literal["llm-function-call"] = "llm-function-call" + data: RTVILLMFunctionCallMessageData + + +class RTVILLMFunctionCallStartMessageData(BaseModel): + function_name: str + + +class RTVILLMFunctionCallStartMessage(BaseModel): + label: Literal["rtvi-ai"] = "rtvi-ai" + type: Literal["llm-function-call-start"] = "llm-function-call-start" + data: RTVILLMFunctionCallStartMessageData + +class RTVILLMFunctionCallResultData(BaseModel): + function_name: str + tool_call_id: str + arguments: dict + result: dict + class RTVITranscriptionMessageData(BaseModel): text: str @@ -232,6 +260,7 @@ class RTVIProcessor(FrameProcessor): def register_service(self, service: RTVIService): self._registered_services[service.name] = service + async def interrupt_bot(self): await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) @@ -273,6 +302,18 @@ class RTVIProcessor(FrameProcessor): else: await self.push_frame(frame, direction) + async def handle_function_call(self, function_name, tool_call_id, arguments, context, result_callback): + fn = RTVILLMFunctionCallMessageData(function_name=function_name, tool_call_id=tool_call_id, args=arguments) + message = RTVILLMFunctionCallMessage(data=fn) + frame = TransportMessageFrame(message=message.model_dump()) + await self.push_frame(frame) + + async def handle_function_call_start(self, function_name): + fn = RTVILLMFunctionCallStartMessageData(function_name=function_name) + message = RTVILLMFunctionCallStartMessage(data=fn) + frame = TransportMessageFrame(message=message.model_dump()) + await self.push_frame(frame) + async def cleanup(self): if self._pipeline: await self._pipeline.cleanup() @@ -362,6 +403,10 @@ class RTVIProcessor(FrameProcessor): case "action": action = RTVIActionRun.model_validate(message.data) await self._handle_action(message.id, action) + case "llm-function-call-result": + data = RTVILLMFunctionCallResultData.model_validate(message.data) + await self._handle_function_call_result(data) + case _: await self._send_error_response(message.id, f"Unsupported type {message.type}") @@ -419,6 +464,14 @@ class RTVIProcessor(FrameProcessor): await self.interrupt_bot() await self._update_config(data) await self._handle_get_config(request_id) + + async def _handle_function_call_result(self, data): + frame = FunctionCallResultFrame( + function_name=data.function_name, + tool_call_id=data.tool_call_id, + arguments=data.arguments, + result=data.result) + await self.push_frame(frame) async def _handle_action(self, request_id: str, data: RTVIActionRun): action_id = self._action_id(data.service, data.action) diff --git a/src/pipecat/processors/logger.py b/src/pipecat/processors/logger.py index 6f07548af..e32ff5945 100644 --- a/src/pipecat/processors/logger.py +++ b/src/pipecat/processors/logger.py @@ -4,7 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # -from pipecat.frames.frames import Frame +from pipecat.frames.frames import BotSpeakingFrame, Frame, AudioRawFrame, TransportMessageFrame from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from loguru import logger from typing import Optional @@ -12,16 +12,19 @@ logger = logger.opt(ansi=True) class FrameLogger(FrameProcessor): - def __init__(self, prefix="Frame", color: Optional[str] = None): + def __init__(self, prefix="Frame", color: Optional[str] = None, ignored_frame_types: Optional[list] = [BotSpeakingFrame, AudioRawFrame, TransportMessageFrame]): super().__init__() self._prefix = prefix self._color = color + self._ignored_frame_types = tuple(ignored_frame_types) if ignored_frame_types else None + async def process_frame(self, frame: Frame, direction: FrameDirection): - dir = "<" if direction is FrameDirection.UPSTREAM else ">" - msg = f"{dir} {self._prefix}: {frame}" - if self._color: - msg = f"<{self._color}>{msg}" - logger.debug(msg) + if self._ignored_frame_types and not isinstance(frame, self._ignored_frame_types): + dir = "<" if direction is FrameDirection.UPSTREAM else ">" + msg = f"{dir} {self._prefix}: {frame}" + if self._color: + msg = f"<{self._color}>{msg}" + logger.debug(msg) await self.push_frame(frame, direction) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 2b828df31..c0ceaeb70 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -25,11 +25,14 @@ from pipecat.frames.frames import ( TTSVoiceUpdateFrame, TextFrame, VisionImageRawFrame, + FunctionCallResultFrame ) from pipecat.processors.async_frame_processor import AsyncFrameProcessor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.utils.audio import calculate_audio_volume from pipecat.utils.utils import exp_smoothing +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext + import re @@ -115,27 +118,51 @@ class LLMService(AIService): self._start_callbacks = {} # TODO-CB: callback function type - def register_function(self, function_name: str, callback, start_callback=None): + def register_function(self, function_name: str | None, callback, start_callback=None): + # Registering a function with the function_name set to None will run that callback + # for all functions self._callbacks[function_name] = callback + # QUESTION FOR CB: maybe this isn't needed anymore? if start_callback: self._start_callbacks[function_name] = start_callback - def unregister_function(self, function_name: str): + def unregister_function(self, function_name: str | None): del self._callbacks[function_name] if self._start_callbacks[function_name]: del self._start_callbacks[function_name] def has_function(self, function_name: str): + if None in self._callbacks.keys(): + return True return function_name in self._callbacks.keys() - async def call_function(self, function_name: str, args): + async def call_function( + self, + *, + context: OpenAILLMContext, + tool_call_id: str, + function_name: str, + arguments: str) -> None: + f = None if function_name in self._callbacks.keys(): - return await self._callbacks[function_name](self, args) - return None + f = self._callbacks[function_name] + elif None in self._callbacks.keys(): + f = self._callbacks[None] + else: + return None + await context.call_function( + f, + function_name=function_name, + tool_call_id=tool_call_id, + arguments=arguments, + llm=self) + # QUESTION FOR CB: maybe this isn't needed anymore? async def call_start_function(self, function_name: str): if function_name in self._start_callbacks.keys(): await self._start_callbacks[function_name](self) + elif None in self._start_callbacks.keys(): + return await self._start_callbacks[None](function_name) class TTSService(AIService): @@ -151,12 +178,12 @@ class TTSService(AIService): self._push_text_frames: bool = push_text_frames self._current_sentence: str = "" - @abstractmethod + @ abstractmethod async def set_voice(self, voice: str): pass # Converts the text to audio. - @abstractmethod + @ abstractmethod async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: pass @@ -242,7 +269,7 @@ class STTService(AIService): self._smoothing_factor = 0.2 self._prev_volume = 0 - @abstractmethod + @ abstractmethod async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: """Returns transcript as a string""" pass @@ -308,7 +335,7 @@ class ImageGenService(AIService): super().__init__(**kwargs) # Renders the image. Returns an Image object. - @abstractmethod + @ abstractmethod async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]: pass @@ -331,7 +358,7 @@ class VisionService(AIService): super().__init__(**kwargs) self._describe_text = None - @abstractmethod + @ abstractmethod async def run_vision(self, frame: VisionImageRawFrame) -> AsyncGenerator[Frame, None]: pass diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 7854bb792..2c64cf167 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -5,19 +5,33 @@ # import base64 +import json +import io +import copy +from typing import List, Optional +from dataclasses import dataclass +from PIL import Image +from asyncio import CancelledError +import re from pipecat.frames.frames import ( Frame, LLMModelUpdateFrame, TextFrame, VisionImageRawFrame, + UserImageRequestFrame, + UserImageRawFrame, LLMMessagesFrame, LLMFullResponseStartFrame, - LLMFullResponseEndFrame + LLMFullResponseEndFrame, + FunctionCallResultFrame, + FunctionCallInProgressFrame, + StartInterruptionFrame ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_services import LLMService from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext, OpenAILLMContextFrame +from pipecat.processors.aggregators.llm_response import LLMUserContextAggregator, LLMAssistantContextAggregator from loguru import logger @@ -30,21 +44,36 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") +@dataclass +class AnthropicImageMessageFrame(Frame): + user_image_raw_frame: UserImageRawFrame + text: Optional[str] = None + + +@dataclass +class AnthropicContextAggregatorPair: + _user: 'AnthropicUserContextAggregator' + _assistant: 'AnthropicAssistantContextAggregator' + + def user(self) -> str: + return self._user + + def assistant(self) -> str: + return self._assistant + + class AnthropicLLMService(LLMService): """This class implements inference with Anthropic's AI models - - This service translates internally from OpenAILLMContext to the messages format - expected by the Anthropic Python SDK. We are using the OpenAILLMContext as a lingua - franca for all LLM services, so that it is easy to switch between different LLMs. """ def __init__( self, *, api_key: str, - model: str = "claude-3-opus-20240229", - max_tokens: int = 1024): - super().__init__() + model: str = "claude-3-5-sonnet-20240620", + max_tokens: int = 4096, + **kwargs): + super().__init__(**kwargs) self._client = AsyncAnthropic(api_key=api_key) self._model = model self._max_tokens = max_tokens @@ -52,89 +81,115 @@ class AnthropicLLMService(LLMService): def can_generate_metrics(self) -> bool: return True - def _get_messages_from_openai_context( - self, context: OpenAILLMContext): - openai_messages = context.get_messages() - anthropic_messages = [] - - for message in openai_messages: - role = message["role"] - text = message["content"] - if role == "system": - role = "user" - if message.get("mime_type") == "image/jpeg": - # vision frame - encoded_image = base64.b64encode(message["data"].getvalue()).decode("utf-8") - anthropic_messages.append({ - "role": role, - "content": [{ - "type": "image", - "source": { - "type": "base64", - "media_type": message.get("mime_type"), - "data": encoded_image, - } - }, { - "type": "text", - "text": text - }] - }) - else: - # Text frame. Anthropic needs the roles to alternate. This will - # cause an issue with interruptions. So, if we detect we are the - # ones asking again it probably means we were interrupted. - if role == "user" and len(anthropic_messages) > 1: - last_message = anthropic_messages[-1] - if last_message["role"] == "user": - anthropic_messages = anthropic_messages[:-1] - content = last_message["content"] - anthropic_messages.append( - {"role": "user", "content": f"Sorry, I just asked you about [{content}] but now I would like to know [{text}]."}) - else: - anthropic_messages.append({"role": role, "content": text}) - else: - anthropic_messages.append({"role": role, "content": text}) - - return anthropic_messages + @ staticmethod + def create_context_aggregator(context: OpenAILLMContext) -> AnthropicContextAggregatorPair: + user = AnthropicUserContextAggregator(context) + assistant = AnthropicAssistantContextAggregator(user) + return AnthropicContextAggregatorPair( + _user=user, + _assistant=assistant + ) async def _process_context(self, context: OpenAILLMContext): - await self.push_frame(LLMFullResponseStartFrame()) try: - logger.debug(f"Generating chat: {context.get_messages_json()}") + await self.push_frame(LLMFullResponseStartFrame()) + await self.start_processing_metrics() - messages = self._get_messages_from_openai_context(context) + logger.debug(f"Generating chat: {context.get_messages_for_logging()}") + + messages = context.messages await self.start_ttfb_metrics() response = await self._client.messages.create( messages=messages, + tools=context.tools or [], model=self._model, max_tokens=self._max_tokens, stream=True) await self.stop_ttfb_metrics() + # Tool use + tool_use_block = None + json_accumulator = '' + + # Usage tracking. We track the usage reported by Anthropic in prompt_tokens and + # completion_tokens. We also estimate the completion tokens from output text + # and use that estimate if we are interrupted, because we almost certainly won't + # get a complete usage report if the task we're running in is cancelled. + prompt_tokens = 0 + completion_tokens = 0 + completion_tokens_estimate = 0 + use_completion_tokens_estimate = False + async for event in response: # logger.debug(f"Anthropic LLM event: {event}") - if (event.type == "content_block_delta"): - await self.push_frame(TextFrame(event.delta.text)) + # Aggregate streaming content, create frames, trigger events + + if (event.type == "content_block_delta"): + if hasattr(event.delta, 'text'): + await self.push_frame(TextFrame(event.delta.text)) + completion_tokens_estimate += self._estimate_tokens(event.delta.text) + elif hasattr(event.delta, 'partial_json') and tool_use_block: + json_accumulator += event.delta.partial_json + completion_tokens_estimate += self._estimate_tokens( + event.delta.partial_json) + elif (event.type == "content_block_start"): + if event.content_block.type == "tool_use": + tool_use_block = event.content_block + json_accumulator = '' + elif (event.type == "message_delta" and + hasattr(event.delta, 'stop_reason') and event.delta.stop_reason == 'tool_use'): + if tool_use_block: + await self.call_function(context=context, + tool_call_id=tool_use_block.id, + function_name=tool_use_block.name, + arguments=json.loads(json_accumulator)) + + # Calculate usage. Do this here in its own if statement, because there may be usage data + # embedded in messages that we do other processing for, above. + if hasattr(event, "usage"): + prompt_tokens += event.usage.input_tokens if hasattr( + event.usage, "input_tokens") else 0 + completion_tokens += event.usage.output_tokens if hasattr( + event.usage, "output_tokens") else 0 + elif hasattr(event, "message") and hasattr(event.message, "usage"): + prompt_tokens += event.message.usage.input_tokens if hasattr( + event.message.usage, "input_tokens") else 0 + completion_tokens += event.message.usage.output_tokens if hasattr( + event.message.usage, "output_tokens") else 0 + + except CancelledError as e: + # If we're interrupted, we won't get a complete usage report. So set our flag to use the + # token estimate. The reraise the exception so all the processors running in this task + # also get cancelled. + use_completion_tokens_estimate = True + raise except Exception as e: logger.exception(f"{self} exception: {e}") finally: + await self.stop_processing_metrics() await self.push_frame(LLMFullResponseEndFrame()) + await self._report_usage_metrics( + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens if not use_completion_tokens_estimate else completion_tokens_estimate) async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) context = None - if isinstance(frame, OpenAILLMContextFrame): - context: OpenAILLMContext = frame.context + context = frame.context elif isinstance(frame, LLMMessagesFrame): - context = OpenAILLMContext.from_messages(frame.messages) + context = AnthropicLLMContext.from_messages(frame.messages) elif isinstance(frame, VisionImageRawFrame): - context = OpenAILLMContext.from_image_frame(frame) + # This is only useful in very simple pipelines because it creates + # a new context. Generally we want a context manager to catch + # UserImageRawFrames coming through the pipeline and add them + # to the context. + context = AnthropicLLMContext.from_image_frame(frame) elif isinstance(frame, LLMModelUpdateFrame): logger.debug(f"Switching LLM model to: [{frame.model}]") self._model = frame.model @@ -143,3 +198,265 @@ class AnthropicLLMService(LLMService): if context: await self._process_context(context) + + async def request_image_frame(self, user_id: str, *, text_content: str = None): + await self.push_frame(UserImageRequestFrame(user_id=user_id, context=text_content), FrameDirection.UPSTREAM) + + def _estimate_tokens(self, text: str) -> int: + return int(len(re.split(r'[^\w]+', text)) * 1.3) + + async def _report_usage_metrics(self, prompt_tokens: int, completion_tokens: int): + if prompt_tokens or completion_tokens: + tokens = { + "processor": self.name, + "model": self._model, + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": prompt_tokens + completion_tokens + } + await self.start_llm_usage_metrics(tokens) + + +class AnthropicLLMContext(OpenAILLMContext): + def __init__( + self, + messages: list[dict] | None = None, + tools: list[dict] | None = None, + tool_choice: dict | None = None, + *, + system: str | None = None + ): + super().__init__(messages=messages, tools=tools, tool_choice=tool_choice) + self._user_image_request_context = {} + + self.system_message = system + + @ classmethod + def from_openai_context(cls, openai_context: OpenAILLMContext): + self = cls( + messages=openai_context.messages, + tools=openai_context.tools, + tool_choice=openai_context.tool_choice, + ) + # See if we should pull the system message out of our context.messages list. (For + # compatibility with Open AI messages format.) + if self.messages and self.messages[0]["role"] == "system": + if len(self.messages) == 1: + # If we have only have a system message in the list, all we can really do + # without introducing too much magic is change the role to "user". + self.messages[0]["role"] = "user" + else: + # If we have more than one message, we'll pull the system message out of the + # list. + self.system_message = self.messages[0]["content"] + self.messages.pop(0) + return self + + @ classmethod + def from_messages(cls, messages: List[dict]) -> "AnthropicLLMContext": + return cls(messages=messages) + + @ classmethod + def from_image_frame(cls, frame: VisionImageRawFrame) -> "AnthropicLLMContext": + context = cls() + context.add_image_frame_message( + format=frame.format, + size=frame.size, + image=frame.image, + text=frame.text) + return context + + 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") + # Anthropic docs say that the image should be the first content block in the message. + content = [{"type": "image", + "source": { + "type": "base64", + "media_type": "image/jpeg", + "data": encoded_image, + }}] + if text: + content.append({"type": "text", "text": text}) + self.add_message({"role": "user", "content": content}) + + def add_message(self, message): + try: + if self.messages: + # Anthropic requires that roles alternate. If this message's role is the same as the + # last message, we should add this message's content to the last message. + if self.messages[-1]["role"] == message["role"]: + # if the last message has just a content string, convert it to a list + # in the proper format + if isinstance(self.messages[-1]["content"], str): + self.messages[-1]["content"] = [{"type": "text", + "text": self.messages[-1]["content"]}] + # if this message has just a content string, convert it to a list + # in the proper format + if isinstance(message["content"], str): + message["content"] = [{"type": "text", "text": message["content"]}] + # append the content of this message to the last message + self.messages[-1]["content"].extend(message["content"]) + else: + self.messages.append(message) + else: + self.messages.append(message) + except Exception as e: + logger.error(f"Error adding message: {e}") + + 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": + item["source"]["data"] = "..." + msgs.append(msg) + return json.dumps(msgs) + + +class AnthropicUserContextAggregator(LLMUserContextAggregator): + def __init__(self, context: OpenAILLMContext | AnthropicLLMContext): + super().__init__(context=context) + + if isinstance(context, OpenAILLMContext): + self._context = AnthropicLLMContext.from_openai_context(context) + + async def push_messages_frame(self): + frame = OpenAILLMContextFrame(self._context) + await self.push_frame(frame) + + 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 = AnthropicImageMessageFrame(user_image_raw_frame=frame, text=text) + await self.push_frame(frame) + except Exception as e: + logger.error(f"Error processing frame: {e}") + +# +# Claude returns a text content block along with a tool use content block. This works quite nicely +# with streaming. We get the text first, so we can start streaming it right away. Then we get the +# tool_use block. While the text is streaming to TTS and the transport, we can run the tool call. +# +# But Claude is verbose. It would be nice to come up with prompt language that suppresses Claude's +# chattiness about it's tool thinking. +# + + +class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator): + def __init__(self, user_context_aggregator: AnthropicUserContextAggregator): + super().__init__(context=user_context_aggregator._context) + self._user_context_aggregator = user_context_aggregator + self._function_call_in_progress = None + self._function_call_result = None + + async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + # See note above about not calling push_frame() here. + if isinstance(frame, StartInterruptionFrame): + self._function_call_in_progress = None + self._function_call_finished = None + elif isinstance(frame, FunctionCallInProgressFrame): + self._function_call_in_progress = frame + elif isinstance(frame, FunctionCallResultFrame): + if self._function_call_in_progress and self._function_call_in_progress.tool_call_id == frame.tool_call_id: + self._function_call_in_progress = None + self._function_call_result = frame + else: + logger.warning( + f"FunctionCallResultFrame tool_call_id does not match FunctionCallInProgressFrame tool_call_id") + self._function_call_in_progress = None + self._function_call_result = None + elif isinstance(frame, AnthropicImageMessageFrame): + try: + 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) + await self._user_context_aggregator.push_messages_frame() + except Exception as e: + logger.error(f"Error processing AnthropicImageMessageFrame: {e}") + + def add_message(self, message): + self._user_context_aggregator.add_message(message) + + async def _push_aggregation(self): + if not self._aggregation: + return + + run_llm = False + + aggregation = self._aggregation + self._aggregation = "" + + try: + if self._function_call_result: + frame = self._function_call_result + # TODO-khk: This was _tool_use_frame, which didn't show up anywhere else? + self._function_call_result = None + self._context.add_message({ + "role": "assistant", + "content": [ + { + "type": "text", + "text": aggregation + }, + { + "type": "tool_use", + "id": frame.tool_call_id, + "name": frame.function_name, + "input": frame.arguments + } + ] + }) + self._context.add_message({ + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": frame.tool_call_id, + "content": json.dumps(frame.result) + } + ] + }) + self._function_call_result = None + run_llm = True + else: + self._context.add_message({"role": "assistant", "content": aggregation}) + + if run_llm: + await self._user_context_aggregator.push_messages_frame() + + except Exception as e: + logger.error(f"Error processing frame: {e}") diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 5233366cd..ead8a7ac4 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -15,6 +15,7 @@ from typing import AsyncGenerator from pipecat.processors.frame_processor import FrameDirection from pipecat.frames.frames import ( CancelFrame, + ErrorFrame, Frame, AudioRawFrame, StartInterruptionFrame, @@ -173,6 +174,12 @@ class CartesiaTTSService(TTSService): num_channels=1 ) await self.push_frame(frame) + elif msg["type"] == "error": + logger.error(f"{self} error: {msg}") + await self.stop_all_metrics() + await self.push_frame(ErrorFrame(f'{self} error: {msg["error"]}')) + else: + logger.error(f"Cartesia error, unknown message type: {msg}") except asyncio.CancelledError: pass except Exception as e: diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 3dbfcfbf5..7c8525ad4 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -8,7 +8,9 @@ import aiohttp import base64 import io import json +from anthropic.types import tool_use_block import httpx +from dataclasses import dataclass from typing import AsyncGenerator, List, Literal @@ -26,8 +28,13 @@ from pipecat.frames.frames import ( MetricsFrame, TextFrame, URLImageRawFrame, - VisionImageRawFrame + VisionImageRawFrame, + FunctionCallResultFrame, + FunctionCallInProgressFrame, + StartInterruptionFrame ) +from pipecat.processors.aggregators.llm_response import LLMUserContextAggregator, LLMAssistantContextAggregator + from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContext, OpenAILLMContextFrame @@ -58,6 +65,7 @@ class OpenAIUnhandledFunctionException(Exception): pass + class BaseOpenAILLMService(LLMService): """This is the base for all services that use the AsyncOpenAI client. @@ -85,6 +93,8 @@ class BaseOpenAILLMService(LLMService): def can_generate_metrics(self) -> bool: return True + + async def get_chat_completions( self, @@ -191,44 +201,13 @@ class BaseOpenAILLMService(LLMService): arguments ): arguments = json.loads(arguments) - result = await self.call_function(function_name, arguments) - arguments = json.dumps(arguments) - if isinstance(result, (str, dict)): - # Handle it in "full magic mode" - tool_call = ChatCompletionFunctionMessageParam({ - "role": "assistant", - "tool_calls": [ - { - "id": tool_call_id, - "function": { - "arguments": arguments, - "name": function_name - }, - "type": "function" - } - ] - - }) - context.add_message(tool_call) - if isinstance(result, dict): - result = json.dumps(result) - tool_result = ChatCompletionToolParam({ - "tool_call_id": tool_call_id, - "role": "tool", - "content": result - }) - context.add_message(tool_result) - # re-prompt to get a human answer - await self._process_context(context) - elif isinstance(result, list): - # reduced magic - for msg in result: - context.add_message(msg) - await self._process_context(context) - elif isinstance(result, type(None)): - pass - else: - raise TypeError(f"Unknown return type from function callback: {type(result)}") + await self.call_function( + context=context, + tool_call_id=tool_call_id, + function_name=function_name, + arguments=arguments + ) + async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -253,13 +232,30 @@ class BaseOpenAILLMService(LLMService): await self.stop_processing_metrics() await self.push_frame(LLMFullResponseEndFrame()) +@dataclass +class OpenAIContextAggregatorPair: + _user: 'OpenAIUserContextAggregator' + _assistant: 'OpenAIAssistantContextAggregator' + + def user(self) -> str: + return self._user + + def assistant(self) -> str: + return self._assistant class OpenAILLMService(BaseOpenAILLMService): def __init__(self, *, model: str = "gpt-4o", **kwargs): super().__init__(model=model, **kwargs) - + @ staticmethod + def create_context_aggregator(context: OpenAILLMContext) -> OpenAIContextAggregatorPair: + user = OpenAIUserContextAggregator(context) + assistant = OpenAIAssistantContextAggregator(user) + return OpenAIContextAggregatorPair( + _user=user, + _assistant=assistant + ) class OpenAIImageGenService(ImageGenService): def __init__( @@ -360,3 +356,85 @@ class OpenAITTSService(TTSService): yield frame except BadRequestError as e: logger.exception(f"{self} error generating TTS: {e}") + +class OpenAIUserContextAggregator(LLMUserContextAggregator): + def __init__(self, context: OpenAILLMContext): + super().__init__(context=context) + + async def push_messages_frame(self): + frame = OpenAILLMContextFrame(self._context) + await self.push_frame(frame) + + +class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): + def __init__(self, user_context_aggregator: OpenAIUserContextAggregator): + super().__init__(context=user_context_aggregator._context) + self._user_context_aggregator = user_context_aggregator + self._function_call_in_progress = None + self._function_call_result = None + + async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + # See note above about not calling push_frame() here. + if isinstance(frame, StartInterruptionFrame): + self._function_call_in_progress = None + self._function_call_finished = None + elif isinstance(frame, FunctionCallInProgressFrame): + self._function_call_in_progress = frame + elif isinstance(frame, FunctionCallResultFrame): + if self._function_call_in_progress and self._function_call_in_progress.tool_call_id == frame.tool_call_id: + self._function_call_in_progress = None + self._function_call_result = frame + # TODO-CB: Kwin wants us to refactor this out of here but I REFUSE + await self._push_aggregation() + else: + logger.warning( + f"FunctionCallResultFrame tool_call_id does not match FunctionCallInProgressFrame tool_call_id") + self._function_call_in_progress = None + self._function_call_result = None + + + def add_message(self, message): + self._user_context_aggregator.add_message(message) + + async def _push_aggregation(self): + if not (self._aggregation or self._function_call_result): + return + + run_llm = False + + aggregation = self._aggregation + self._aggregation = "" + + try: + if self._function_call_result: + frame = self._function_call_result + self._context.add_message({ + "role": "assistant", + "tool_calls": [ + { + "id": frame.tool_call_id, + "function": { + "name": frame.function_name, + "arguments": json.dumps(frame.arguments) + }, + "type": "function" + } + ] + }) + self._context.add_message({ + "role": "tool", + "content": json.dumps(frame.result), + "tool_call_id": frame.tool_call_id + }) + self._function_call_result = None + run_llm = True + else: + self._context.add_message({"role": "assistant", "content": aggregation}) + + if run_llm: + + await self._user_context_aggregator.push_messages_frame() + + except Exception as e: + logger.error(f"Error processing frame: {e}")