From 602724b9841273423183805c6cbc1a5c339c0aee Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 24 Jul 2025 16:35:51 -0400 Subject: [PATCH] Progress on LLM failover support --- src/pipecat/adapters/base_llm_adapter.py | 39 +++- .../adapters/services/open_ai_adapter.py | 46 ++++- .../processors/aggregators/llm_context.py | 3 +- src/pipecat/services/llm_service.py | 9 +- src/pipecat/services/openai/base_llm.py | 60 +++--- src/pipecat/services/openai/llm.py | 179 +----------------- 6 files changed, 119 insertions(+), 217 deletions(-) diff --git a/src/pipecat/adapters/base_llm_adapter.py b/src/pipecat/adapters/base_llm_adapter.py index 6a957c267..31f6206a0 100644 --- a/src/pipecat/adapters/base_llm_adapter.py +++ b/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 - tool schemas and provider-specific tool formats. Subclasses must implement - provider-specific conversion logic. + Provides a standard interface for converting to provider-specific formats. + + 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. diff --git a/src/pipecat/adapters/services/open_ai_adapter.py b/src/pipecat/adapters/services/open_ai_adapter.py index 59d70aa1e..ca3c3ea73 100644 --- a/src/pipecat/adapters/services/open_ai_adapter.py +++ b/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) diff --git a/src/pipecat/processors/aggregators/llm_context.py b/src/pipecat/processors/aggregators/llm_context.py index 6cfdd3e9d..0e63349e8 100644 --- a/src/pipecat/processors/aggregators/llm_context.py +++ b/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 diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index f3e2b843a..aa1b7d2dd 100644 --- a/src/pipecat/services/llm_service.py +++ b/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(): diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index f88134fa2..606dfdc89 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/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 ( - OpenAILLMContext, - OpenAILLMContextFrame, +from pipecat.processors.aggregators.llm_context import ( + LLMContext, + 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 - to an OpenAILLMContext object. The context defines what is sent to the LLM for + This service consumes LLMContextFrame frames, which contain a reference to + 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. - messages: List of chat completion messages to send. + params_from_context: Parameters, derived from the LLM context, to + 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 - 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) + chunks = await self.get_chat_completions(params) 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): - context: OpenAILLMContext = frame.context + if isinstance(frame, LLMContextFrame): + 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 ) diff --git a/src/pipecat/services/openai/llm.py b/src/pipecat/services/openai/llm.py index 7919dd159..85dab1b95 100644 --- a/src/pipecat/services/openai/llm.py +++ b/src/pipecat/services/openai/llm.py @@ -16,45 +16,16 @@ from pipecat.frames.frames import ( FunctionCallResultFrame, UserImageRawFrame, ) -from pipecat.processors.aggregators.llm_response import ( - LLMAssistantAggregatorParams, - LLMAssistantContextAggregator, - LLMUserAggregatorParams, - LLMUserContextAggregator, +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import ( + LLMAssistantContextAggregator_Universal, + LLMAssistantContextAggregatorParams, + 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, - )