diff --git a/src/pipecat/adapters/services/gemini_adapter.py b/src/pipecat/adapters/services/gemini_adapter.py index 2b7c03e40..e49189cf6 100644 --- a/src/pipecat/adapters/services/gemini_adapter.py +++ b/src/pipecat/adapters/services/gemini_adapter.py @@ -57,7 +57,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): Returns: Dictionary of parameters for Gemini's API. """ - messages = self._from_standard_messages(context.messages) + messages = self._from_universal_context_messages(self._get_messages(context)) return { "system_instruction": messages.system_instruction, "messages": messages.messages, @@ -97,7 +97,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): List of messages in a format ready for logging about Gemini. """ # Get messages in Gemini's format - messages = self._from_standard_messages(context.messages).messages + messages = self._from_universal_context_messages(self._get_messages(context)).messages # Sanitize messages for logging messages_for_logging = [] @@ -113,6 +113,9 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): messages_for_logging.append(obj) return messages_for_logging + def _get_messages(self, context: LLMContext) -> List[LLMContextMessage]: + return context.get_messages("google") + @dataclass class ConvertedMessages: """Container for converted messages. @@ -123,8 +126,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): messages: List[Content] system_instruction: Optional[str] = None - def _from_standard_messages( - self, standard_messages: List[LLMContextMessage] + def _from_universal_context_messages( + self, universal_context_messages: List[LLMContextMessage] ) -> ConvertedMessages: """Restructures messages to ensure proper Google format and message ordering. @@ -146,7 +149,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): messages = [] # Process each message, preserving Google-formatted messages and converting others - for message in standard_messages: + for message in universal_context_messages: if isinstance(message, Content): # Keep existing Google-formatted messages (e.g., function calls/responses) # TODO: this branch is probably not needed anymore, since LLMContext contains a universal format @@ -154,7 +157,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): continue # Convert standard format to Google format - converted = self._from_standard_message(message) + converted = self._from_universal_context_message(message) if isinstance(converted, Content): # Regular (non-system) message messages.append(converted) @@ -180,15 +183,15 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]): return self.ConvertedMessages(messages=messages, system_instruction=system_instruction) - def _from_standard_message(self, message: LLMContextMessage) -> Content | str: - """Convert standard format message to Google Content object. + def _from_universal_context_message(self, message: LLMContextMessage) -> Content | str: + """Convert universal context message to Google Content object. Handles conversion of text, images, and function calls to Google's format. System instructions are returned as a plain string. Args: - message: Message in standard format. + message: Message in universal context format. Returns: Content object with role and parts, or a plain string for system diff --git a/src/pipecat/adapters/services/open_ai_adapter.py b/src/pipecat/adapters/services/open_ai_adapter.py index f361b9edd..09979b8a7 100644 --- a/src/pipecat/adapters/services/open_ai_adapter.py +++ b/src/pipecat/adapters/services/open_ai_adapter.py @@ -56,7 +56,7 @@ class OpenAILLMAdapter(BaseLLMAdapter): Dictionary of parameters for OpenAI's ChatCompletion API. """ return { - "messages": self._from_standard_messages(context.messages), + "messages": self._from_universal_context_messages(self._get_messages(context)), # NOTE; LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN) "tools": self.from_standard_tools(context.tools), "tool_choice": context.tool_choice, @@ -90,7 +90,7 @@ class OpenAILLMAdapter(BaseLLMAdapter): List of messages in a format ready for logging about OpenAI. """ msgs = [] - for message in context.messages: + for message in self._get_messages(context): msg = copy.deepcopy(message) if "content" in msg: if isinstance(msg["content"], list): @@ -103,10 +103,13 @@ class OpenAILLMAdapter(BaseLLMAdapter): msgs.append(msg) return json.dumps(msgs, ensure_ascii=False) - def _from_standard_messages( + def _get_messages(self, context: LLMContext) -> List[LLMContextMessage]: + return context.get_messages("openai") + + def _from_universal_context_messages( self, messages: List[LLMContextMessage] ) -> List[ChatCompletionMessageParam]: - # Just a pass-through: messages is already the right type + # Just a pass-through: messages are already the right type return messages def _from_standard_tool_choice( diff --git a/src/pipecat/processors/aggregators/llm_context.py b/src/pipecat/processors/aggregators/llm_context.py index 1a44d16c4..60a78fa12 100644 --- a/src/pipecat/processors/aggregators/llm_context.py +++ b/src/pipecat/processors/aggregators/llm_context.py @@ -17,8 +17,9 @@ service-specific adapter. import base64 import io from dataclasses import dataclass -from typing import Any, List, Optional +from typing import Any, List, Optional, TypeAlias, Union +from loguru import logger from openai._types import NOT_GIVEN as OPEN_AI_NOT_GIVEN from openai._types import NotGiven as OpenAINotGiven from openai.types.chat import ( @@ -31,15 +32,33 @@ from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.frames.frames import AudioRawFrame, Frame # "Re-export" types from OpenAI that we're using as universal context types. -# NOTE: this is just for convenience, for now. As soon as the universal types -# diverge from OpenAI's, we should ditch this. In fact, audio frames already -# diverge from OpenAI's standard format...we really ought to do this. -LLMContextMessage = ChatCompletionMessageParam +# NOTE: if universal message types need to someday diverge from OpenAI's, we +# should consider managing our own definitions. But we should do so carefully, +# as the OpenAI messages are somewhat of a standard and we want to continue +# supporting them. +# TODO: "input_audio" messages already diverge slightly from OpenAI's standard +# format...but they probably don't need to? Revisit. +LLMStandardMessage = ChatCompletionMessageParam LLMContextToolChoice = ChatCompletionToolChoiceOptionParam NOT_GIVEN = OPEN_AI_NOT_GIVEN NotGiven = OpenAINotGiven +@dataclass +class LLMSpecificMessage: + """A container for a context message that is specific to a particular LLM service. + + Enables the use of service-specific message types while maintaining + compatibility with the universal LLM context format. + """ + + llm: str + message: Any + + +LLMContextMessage: TypeAlias = Union[LLMStandardMessage, LLMSpecificMessage] + + class LLMContext: """Manages conversation context for LLM interactions. @@ -66,14 +85,30 @@ class LLMContext: self._tool_choice: LLMContextToolChoice | NotGiven = tool_choice self._validate_tools() - @property - def messages(self) -> List[LLMContextMessage]: + def get_messages(self, llm_specific_filter: Optional[str] = None) -> List[LLMContextMessage]: """Get the current messages list. + Args: + llm_specific_filter: Optional filter to return LLM-specific + messages for the given LLM, in addition to the standard + messages. If messages end up being filtered, an error will be + logged. + Returns: List of conversation messages. """ - return self._messages + if llm_specific_filter is None: + return self._messages + filtered_messages = [ + msg + for msg in self._messages + if not isinstance(msg, LLMSpecificMessage) or msg.llm == llm_specific_filter + ] + if len(filtered_messages) < len(self._messages): + logger.error( + f"Attempted to use incompatible LLMSpecificMessages with LLM '{llm_specific_filter}'." + ) + return filtered_messages @property def tools(self) -> ToolsSchema | NotGiven: diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 690532835..a97e9e0f8 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -53,7 +53,11 @@ from pipecat.frames.frames import ( UserStartedSpeakingFrame, UserStoppedSpeakingFrame, ) -from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_context import ( + LLMContext, + LLMContextMessage, + LLMSpecificMessage, +) from pipecat.processors.aggregators.llm_response import ( LLMAssistantAggregatorParams, LLMUserAggregatorParams, @@ -85,13 +89,13 @@ class LLMContextAggregator(FrameProcessor): self._aggregation: str = "" @property - def messages(self) -> List[dict]: + def messages(self) -> List[LLMContextMessage]: """Get messages from the LLM context. Returns: List of message dictionaries from the context. """ - return self._context.messages + return self._context.get_messages() @property def role(self) -> str: @@ -741,9 +745,10 @@ class LLMAssistantAggregator(LLMContextAggregator): del self._function_calls_in_progress[frame.tool_call_id] def _update_function_call_result(self, function_name: str, tool_call_id: str, result: Any): - for message in self._context.messages: + for message in self._context.get_messages(): if ( - message["role"] == "tool" + not isinstance(message, LLMSpecificMessage) + and message["role"] == "tool" and message["tool_call_id"] and message["tool_call_id"] == tool_call_id ):