Progress on LLM failover support

src/pipecat/adapters/base_llm_adapter.py

@@ -16,16 +16,36 @@ from typing import Any, List, Union, cast
 from loguru import logger
 
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.processors.aggregators.llm_context import LLMContext
 
 
 class BaseLLMAdapter(ABC):
     """Abstract base class for LLM provider adapters.
 
-    Provides a standard interface for converting between Pipecat's standardized
+    Provides a standard interface for converting to provider-specific formats.
-    tool schemas and provider-specific tool formats. Subclasses must implement
+
-    provider-specific conversion logic.
+    Handles:
+    - Converting universal LLM context to provider-specific parameters for LLM
+      invocation.
+    - Converting standardized tools schema to provider-specific tool formats.
+    - Extracting messages from the LLM context for the purposes of logging
+      about the specific provider.
+
+    Subclasses must implement provider-specific conversion logic.
     """
 
+    @abstractmethod
+    def get_llm_invocation_params(self, context: LLMContext) -> dict[str, Any]:
+        """Get provider-specific LLM invocation parameters from a universal LLM context.
+
+        Args:
+            context: The LLM context containing messages, tools, etc.
+
+        Returns:
+            Provider-specific parameters for invoking the LLM.
+        """
+        pass
+
     @abstractmethod
     def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Any]:
         """Convert tools schema to the provider's specific format.
@@ -38,6 +58,19 @@ class BaseLLMAdapter(ABC):
         """
         pass
 
+    @abstractmethod
+    def get_messages_for_logging(self, context: LLMContext) -> List[dict[str, Any]]:
+        """Get messages from the LLM context in a format ready for logging about this provider.
+
+        Args:
+            context: The LLM context containing messages.
+
+        Returns:
+            List of messages in a format ready for logging about this
+            provider.
+        """
+        pass
+
     def from_standard_tools(self, tools: Any) -> List[Any]:
         """Convert tools from standard format to provider format.
 

src/pipecat/adapters/services/open_ai_adapter.py

@@ -6,12 +6,15 @@
 
 """OpenAI LLM adapter for Pipecat."""
 
-from typing import List
+import copy
+import json
+from typing import Any, List
 
 from openai.types.chat import ChatCompletionToolParam
 
 from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.processors.aggregators.llm_context import LLMContext
 
 
 class OpenAILLMAdapter(BaseLLMAdapter):
@@ -22,6 +25,22 @@ class OpenAILLMAdapter(BaseLLMAdapter):
     function calling capabilities.
     """
 
+    def get_llm_invocation_params(self, context: LLMContext) -> dict[str, Any]:
+        """Get OpenAI-specific LLM invocation parameters from a universal LLM context.
+
+        Args:
+            context: The LLM context containing messages, tools, etc.
+
+        Returns:
+            Dictionary of parameters for OpenAI's chat completion API.
+        """
+        return {
+            "messages": context.messages,
+            # TODO: doesn't seem right that we may or may not need to convert tools here; they should already be guaranteed to exist in a universal format in the LLMContext, right?
+            "tools": self.from_standard_tools(context.tools),
+            "tool_choice": context.tool_choice,
+        }
+
     def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[ChatCompletionToolParam]:
         """Convert function schemas to OpenAI's function-calling format.
 
@@ -37,3 +56,28 @@ class OpenAILLMAdapter(BaseLLMAdapter):
             ChatCompletionToolParam(type="function", function=func.to_default_dict())
             for func in functions_schema
         ]
+
+    def get_messages_for_logging(self, context: LLMContext) -> List[dict[str, Any]]:
+        """Get messages from the LLM context in a format ready for logging about OpenAI.
+
+        Removes or truncates sensitive data like image content for safe logging.
+
+        Args:
+            context: The LLM context containing messages.
+
+        Returns:
+            List of messages in a format ready for logging about OpenAI.
+        """
+        msgs = []
+        for message in context.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/..."
+            if "mime_type" in msg and msg["mime_type"].startswith("image/"):
+                msg["data"] = "..."
+            msgs.append(msg)
+        return json.dumps(msgs, ensure_ascii=False)

src/pipecat/processors/aggregators/llm_context.py

@@ -17,7 +17,7 @@ service-specific adapter.
 import base64
 import io
 from dataclasses import dataclass
-from typing import List, Optional
+from typing import Any, List, Optional
 
 from openai._types import NOT_GIVEN as OPEN_AI_NOT_GIVEN
 from openai._types import NotGiven as OpenAINotGiven
@@ -122,6 +122,7 @@ class LLMContext:
             tools: List of tools available to the LLM, a ToolsSchema, or NOT_GIVEN to disable tools.
         """
         # TODO: convert empty ToolsSchema to NOT_GIVEN if needed
+        # TODO: maybe also convert non-ToolsSchema tools to ToolsSchema? See open_ai_adapter.py for related comment
         if isinstance(tools, list) and len(tools) == 0:
             tools = NOT_GIVEN
         self._tools = tools

src/pipecat/services/llm_service.py

@@ -41,6 +41,7 @@ from pipecat.frames.frames import (
     StartInterruptionFrame,
     UserImageRequestFrame,
 )
+from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantAggregatorParams,
     LLMUserAggregatorParams,
@@ -89,7 +90,8 @@ class FunctionCallParams:
     tool_call_id: str
     arguments: Mapping[str, Any]
     llm: "LLMService"
-    context: OpenAILLMContext
+    # TODO: after migration of all services to universal LLMContext, OpenAILLMContext can be removed
+    context: LLMContext | OpenAILLMContext
     result_callback: FunctionCallResultCallback
 
 
@@ -418,7 +420,10 @@ class LLMService(AIService):
             else:
                 await self._sequential_runner_queue.put(runner_item)
 
-    async def _call_start_function(self, context: OpenAILLMContext, function_name: str):
+    # TODO: after migration of all services to universal LLMContext, OpenAILLMContext can be removed
+    async def _call_start_function(
+        self, context: LLMContext | OpenAILLMContext, function_name: str
+    ):
         if function_name in self._start_callbacks.keys():
             await self._start_callbacks[function_name](function_name, self, context)
         elif None in self._start_callbacks.keys():

src/pipecat/services/openai/base_llm.py

@@ -4,7 +4,7 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-"""Base OpenAI LLM service implementation."""
+"""Base LLM service implementation for services that use the AsyncOpenAI client."""
 
 import base64
 import json
@@ -31,9 +31,9 @@ from pipecat.frames.frames import (
     VisionImageRawFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
-from pipecat.processors.aggregators.openai_llm_context import (
+from pipecat.processors.aggregators.llm_context import (
-    OpenAILLMContext,
+    LLMContext,
-    OpenAILLMContextFrame,
+    LLMContextFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
@@ -44,8 +44,8 @@ from pipecat.utils.tracing.service_decorators import traced_llm
 class BaseOpenAILLMService(LLMService):
     """Base class for all services that use the AsyncOpenAI client.
 
-    This service consumes OpenAILLMContextFrame frames, which contain a reference
+    This service consumes LLMContextFrame frames, which contain a reference to
-    to an OpenAILLMContext object. The context defines what is sent to the LLM for
+    an LLMContext object. The context defines what is sent to the LLM for
     completion, including user, assistant, and system messages, as well as tool
     choices and function call configurations.
     """
@@ -173,13 +173,13 @@ class BaseOpenAILLMService(LLMService):
         return True
 
     async def get_chat_completions(
-        self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
+        self, params_from_context: dict[str, Any]
     ) -> AsyncStream[ChatCompletionChunk]:
         """Get streaming chat completions from OpenAI API.
 
         Args:
-            context: The LLM context containing tools and configuration.
+            params_from_context: Parameters, derived from the LLM context, to
-            messages: List of chat completion messages to send.
+                use for the chat completion. Contains messages, tools, and tool choice.
 
         Returns:
             Async stream of chat completion chunks.
@@ -187,9 +187,6 @@ class BaseOpenAILLMService(LLMService):
         params = {
             "model": self.model_name,
             "stream": True,
-            "messages": messages,
-            "tools": context.tools,
-            "tool_choice": context.tool_choice,
             "stream_options": {"include_usage": True},
             "frequency_penalty": self._settings["frequency_penalty"],
             "presence_penalty": self._settings["presence_penalty"],
@@ -200,39 +197,28 @@ class BaseOpenAILLMService(LLMService):
             "max_completion_tokens": self._settings["max_completion_tokens"],
         }
 
+        # Messages, tools, tool_choice
+        params.update(params_from_context)
+
         params.update(self._settings["extra"])
 
         chunks = await self._client.chat.completions.create(**params)
         return chunks
 
     async def _stream_chat_completions(
-        self, context: OpenAILLMContext
+        self, context: LLMContext
     ) -> AsyncStream[ChatCompletionChunk]:
-        logger.debug(f"{self}: Generating chat [{context.get_messages_for_logging()}]")
+        adapter = self.get_llm_adapter()
+        logger.debug(f"{self}: Generating chat [{adapter.get_messages_for_logging(context)}]")
 
-        messages: List[ChatCompletionMessageParam] = context.get_messages()
+        params = adapter.get_llm_invocation_params(context)
 
-        # base64 encode any images
+        chunks = await self.get_chat_completions(params)
-        for message in messages:
-            if message.get("mime_type") == "image/jpeg":
-                encoded_image = base64.b64encode(message["data"].getvalue()).decode("utf-8")
-                text = message["content"]
-                message["content"] = [
-                    {"type": "text", "text": text},
-                    {
-                        "type": "image_url",
-                        "image_url": {"url": f"data:image/jpeg;base64,{encoded_image}"},
-                    },
-                ]
-                del message["data"]
-                del message["mime_type"]
-
-        chunks = await self.get_chat_completions(context, messages)
 
         return chunks
 
     @traced_llm
-    async def _process_context(self, context: OpenAILLMContext):
+    async def _process_context(self, context: LLMContext):
         functions_list = []
         arguments_list = []
         tool_id_list = []
@@ -331,7 +317,7 @@ class BaseOpenAILLMService(LLMService):
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process frames for LLM completion requests.
 
-        Handles OpenAILLMContextFrame, LLMMessagesFrame, VisionImageRawFrame,
+        Handles LLMContextFrame, LLMMessagesFrame, VisionImageRawFrame,
         and LLMUpdateSettingsFrame to trigger LLM completions and manage settings.
 
         Args:
@@ -341,12 +327,12 @@ class BaseOpenAILLMService(LLMService):
         await super().process_frame(frame, direction)
 
         context = None
-        if isinstance(frame, OpenAILLMContextFrame):
+        if isinstance(frame, LLMContextFrame):
-            context: OpenAILLMContext = frame.context
+            context: LLMContext = frame.context
         elif isinstance(frame, LLMMessagesFrame):
-            context = OpenAILLMContext.from_messages(frame.messages)
+            context = LLMContext(messages=frame.messages)
         elif isinstance(frame, VisionImageRawFrame):
-            context = OpenAILLMContext()
+            context = LLMContext()
             context.add_image_frame_message(
                 format=frame.format, size=frame.size, image=frame.image, text=frame.text
             )

src/pipecat/services/openai/llm.py

@@ -16,45 +16,16 @@ from pipecat.frames.frames import (
     FunctionCallResultFrame,
     UserImageRawFrame,
 )
-from pipecat.processors.aggregators.llm_response import (
+from pipecat.processors.aggregators.llm_context import LLMContext
-    LLMAssistantAggregatorParams,
+from pipecat.processors.aggregators.llm_response_universal import (
-    LLMAssistantContextAggregator,
+    LLMAssistantContextAggregator_Universal,
-    LLMUserAggregatorParams,
+    LLMAssistantContextAggregatorParams,
-    LLMUserContextAggregator,
+    LLMUserContextAggregator_Universal,
+    LLMUserContextAggregatorParams,
 )
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.openai.base_llm import BaseOpenAILLMService
 
 
-@dataclass
-class OpenAIContextAggregatorPair:
-    """Pair of OpenAI context aggregators for user and assistant messages.
-
-    Parameters:
-        _user: User context aggregator for processing user messages.
-        _assistant: Assistant context aggregator for processing assistant messages.
-    """
-
-    _user: "OpenAIUserContextAggregator"
-    _assistant: "OpenAIAssistantContextAggregator"
-
-    def user(self) -> "OpenAIUserContextAggregator":
-        """Get the user context aggregator.
-
-        Returns:
-            The user context aggregator instance.
-        """
-        return self._user
-
-    def assistant(self) -> "OpenAIAssistantContextAggregator":
-        """Get the assistant context aggregator.
-
-        Returns:
-            The assistant context aggregator instance.
-        """
-        return self._assistant
-
-
 class OpenAILLMService(BaseOpenAILLMService):
     """OpenAI LLM service implementation.
 
@@ -78,141 +49,3 @@ class OpenAILLMService(BaseOpenAILLMService):
             **kwargs: Additional arguments passed to the parent BaseOpenAILLMService.
         """
         super().__init__(model=model, params=params, **kwargs)
-
-    def create_context_aggregator(
-        self,
-        context: OpenAILLMContext,
-        *,
-        user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
-        assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
-    ) -> OpenAIContextAggregatorPair:
-        """Create OpenAI-specific context aggregators.
-
-        Creates a pair of context aggregators optimized for OpenAI's message format,
-        including support for function calls, tool usage, and image handling.
-
-        Args:
-            context: The LLM context to create aggregators for.
-            user_params: Parameters for user message aggregation.
-            assistant_params: Parameters for assistant message aggregation.
-
-        Returns:
-            OpenAIContextAggregatorPair: A pair of context aggregators, one for
-            the user and one for the assistant, encapsulated in an
-            OpenAIContextAggregatorPair.
-
-        """
-        context.set_llm_adapter(self.get_llm_adapter())
-        user = OpenAIUserContextAggregator(context, params=user_params)
-        assistant = OpenAIAssistantContextAggregator(context, params=assistant_params)
-        return OpenAIContextAggregatorPair(_user=user, _assistant=assistant)
-
-
-class OpenAIUserContextAggregator(LLMUserContextAggregator):
-    """OpenAI-specific user context aggregator.
-
-    Handles aggregation of user messages for OpenAI LLM services.
-    Inherits all functionality from the base LLMUserContextAggregator.
-    """
-
-    pass
-
-
-class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
-    """OpenAI-specific assistant context aggregator.
-
-    Handles aggregation of assistant messages for OpenAI LLM services,
-    with specialized support for OpenAI's function calling format,
-    tool usage tracking, and image message handling.
-    """
-
-    async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
-        """Handle a function call in progress.
-
-        Adds the function call to the context with an IN_PROGRESS status
-        to track ongoing function execution.
-
-        Args:
-            frame: Frame containing function call progress information.
-        """
-        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": "IN_PROGRESS",
-                "tool_call_id": frame.tool_call_id,
-            }
-        )
-
-    async def handle_function_call_result(self, frame: FunctionCallResultFrame):
-        """Handle the result of a function call.
-
-        Updates the context with the function call result, replacing any
-        previous IN_PROGRESS status.
-
-        Args:
-            frame: Frame containing the function call result.
-        """
-        if frame.result:
-            result = json.dumps(frame.result)
-            await self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
-        else:
-            await self._update_function_call_result(
-                frame.function_name, frame.tool_call_id, "COMPLETED"
-            )
-
-    async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
-        """Handle a cancelled function call.
-
-        Updates the context to mark the function call as cancelled.
-
-        Args:
-            frame: Frame containing the function call cancellation information.
-        """
-        await self._update_function_call_result(
-            frame.function_name, frame.tool_call_id, "CANCELLED"
-        )
-
-    async def _update_function_call_result(
-        self, function_name: str, tool_call_id: str, result: Any
-    ):
-        for message in self._context.messages:
-            if (
-                message["role"] == "tool"
-                and message["tool_call_id"]
-                and message["tool_call_id"] == tool_call_id
-            ):
-                message["content"] = result
-
-    async def handle_user_image_frame(self, frame: UserImageRawFrame):
-        """Handle a user image frame from a function call request.
-
-        Marks the associated function call as completed and adds the image
-        to the context for processing.
-
-        Args:
-            frame: Frame containing the user image and request context.
-        """
-        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,
-        )