[WIP] Universal (LLM-agnostic) context machinery to support runtime LLM switching.

- Add support for LLM-specific messages in the universal `LLMContext`, to enable using LLM-specific functionality while still using the universal LLM context
This commit is contained in:
Paul Kompfner
2025-08-14 09:31:32 -04:00
parent cfb094b3c8
commit 59ecb19000
4 changed files with 72 additions and 26 deletions

View File

@@ -57,7 +57,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
Returns: Returns:
Dictionary of parameters for Gemini's API. 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 { return {
"system_instruction": messages.system_instruction, "system_instruction": messages.system_instruction,
"messages": messages.messages, "messages": messages.messages,
@@ -97,7 +97,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
List of messages in a format ready for logging about Gemini. List of messages in a format ready for logging about Gemini.
""" """
# Get messages in Gemini's format # 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 # Sanitize messages for logging
messages_for_logging = [] messages_for_logging = []
@@ -113,6 +113,9 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
messages_for_logging.append(obj) messages_for_logging.append(obj)
return messages_for_logging return messages_for_logging
def _get_messages(self, context: LLMContext) -> List[LLMContextMessage]:
return context.get_messages("google")
@dataclass @dataclass
class ConvertedMessages: class ConvertedMessages:
"""Container for converted messages. """Container for converted messages.
@@ -123,8 +126,8 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
messages: List[Content] messages: List[Content]
system_instruction: Optional[str] = None system_instruction: Optional[str] = None
def _from_standard_messages( def _from_universal_context_messages(
self, standard_messages: List[LLMContextMessage] self, universal_context_messages: List[LLMContextMessage]
) -> ConvertedMessages: ) -> ConvertedMessages:
"""Restructures messages to ensure proper Google format and message ordering. """Restructures messages to ensure proper Google format and message ordering.
@@ -146,7 +149,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
messages = [] messages = []
# Process each message, preserving Google-formatted messages and converting others # 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): if isinstance(message, Content):
# Keep existing Google-formatted messages (e.g., function calls/responses) # Keep existing Google-formatted messages (e.g., function calls/responses)
# TODO: this branch is probably not needed anymore, since LLMContext contains a universal format # TODO: this branch is probably not needed anymore, since LLMContext contains a universal format
@@ -154,7 +157,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
continue continue
# Convert standard format to Google format # Convert standard format to Google format
converted = self._from_standard_message(message) converted = self._from_universal_context_message(message)
if isinstance(converted, Content): if isinstance(converted, Content):
# Regular (non-system) message # Regular (non-system) message
messages.append(converted) messages.append(converted)
@@ -180,15 +183,15 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
return self.ConvertedMessages(messages=messages, system_instruction=system_instruction) return self.ConvertedMessages(messages=messages, system_instruction=system_instruction)
def _from_standard_message(self, message: LLMContextMessage) -> Content | str: def _from_universal_context_message(self, message: LLMContextMessage) -> Content | str:
"""Convert standard format message to Google Content object. """Convert universal context message to Google Content object.
Handles conversion of text, images, and function calls to Google's Handles conversion of text, images, and function calls to Google's
format. format.
System instructions are returned as a plain string. System instructions are returned as a plain string.
Args: Args:
message: Message in standard format. message: Message in universal context format.
Returns: Returns:
Content object with role and parts, or a plain string for system Content object with role and parts, or a plain string for system

View File

@@ -56,7 +56,7 @@ class OpenAILLMAdapter(BaseLLMAdapter):
Dictionary of parameters for OpenAI's ChatCompletion API. Dictionary of parameters for OpenAI's ChatCompletion API.
""" """
return { 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) # NOTE; LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
"tools": self.from_standard_tools(context.tools), "tools": self.from_standard_tools(context.tools),
"tool_choice": context.tool_choice, "tool_choice": context.tool_choice,
@@ -90,7 +90,7 @@ class OpenAILLMAdapter(BaseLLMAdapter):
List of messages in a format ready for logging about OpenAI. List of messages in a format ready for logging about OpenAI.
""" """
msgs = [] msgs = []
for message in context.messages: for message in self._get_messages(context):
msg = copy.deepcopy(message) msg = copy.deepcopy(message)
if "content" in msg: if "content" in msg:
if isinstance(msg["content"], list): if isinstance(msg["content"], list):
@@ -103,10 +103,13 @@ class OpenAILLMAdapter(BaseLLMAdapter):
msgs.append(msg) msgs.append(msg)
return json.dumps(msgs, ensure_ascii=False) 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] self, messages: List[LLMContextMessage]
) -> List[ChatCompletionMessageParam]: ) -> List[ChatCompletionMessageParam]:
# Just a pass-through: messages is already the right type # Just a pass-through: messages are already the right type
return messages return messages
def _from_standard_tool_choice( def _from_standard_tool_choice(

View File

@@ -17,8 +17,9 @@ service-specific adapter.
import base64 import base64
import io import io
from dataclasses import dataclass 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 NOT_GIVEN as OPEN_AI_NOT_GIVEN
from openai._types import NotGiven as OpenAINotGiven from openai._types import NotGiven as OpenAINotGiven
from openai.types.chat import ( from openai.types.chat import (
@@ -31,15 +32,33 @@ from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.frames.frames import AudioRawFrame, Frame from pipecat.frames.frames import AudioRawFrame, Frame
# "Re-export" types from OpenAI that we're using as universal context types. # "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 # NOTE: if universal message types need to someday diverge from OpenAI's, we
# diverge from OpenAI's, we should ditch this. In fact, audio frames already # should consider managing our own definitions. But we should do so carefully,
# diverge from OpenAI's standard format...we really ought to do this. # as the OpenAI messages are somewhat of a standard and we want to continue
LLMContextMessage = ChatCompletionMessageParam # 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 LLMContextToolChoice = ChatCompletionToolChoiceOptionParam
NOT_GIVEN = OPEN_AI_NOT_GIVEN NOT_GIVEN = OPEN_AI_NOT_GIVEN
NotGiven = OpenAINotGiven 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: class LLMContext:
"""Manages conversation context for LLM interactions. """Manages conversation context for LLM interactions.
@@ -66,14 +85,30 @@ class LLMContext:
self._tool_choice: LLMContextToolChoice | NotGiven = tool_choice self._tool_choice: LLMContextToolChoice | NotGiven = tool_choice
self._validate_tools() self._validate_tools()
@property def get_messages(self, llm_specific_filter: Optional[str] = None) -> List[LLMContextMessage]:
def messages(self) -> List[LLMContextMessage]:
"""Get the current messages list. """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: Returns:
List of conversation messages. 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 @property
def tools(self) -> ToolsSchema | NotGiven: def tools(self) -> ToolsSchema | NotGiven:

View File

@@ -53,7 +53,11 @@ from pipecat.frames.frames import (
UserStartedSpeakingFrame, UserStartedSpeakingFrame,
UserStoppedSpeakingFrame, 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 ( from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams, LLMAssistantAggregatorParams,
LLMUserAggregatorParams, LLMUserAggregatorParams,
@@ -85,13 +89,13 @@ class LLMContextAggregator(FrameProcessor):
self._aggregation: str = "" self._aggregation: str = ""
@property @property
def messages(self) -> List[dict]: def messages(self) -> List[LLMContextMessage]:
"""Get messages from the LLM context. """Get messages from the LLM context.
Returns: Returns:
List of message dictionaries from the context. List of message dictionaries from the context.
""" """
return self._context.messages return self._context.get_messages()
@property @property
def role(self) -> str: def role(self) -> str:
@@ -741,9 +745,10 @@ class LLMAssistantAggregator(LLMContextAggregator):
del self._function_calls_in_progress[frame.tool_call_id] 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): 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 ( if (
message["role"] == "tool" not isinstance(message, LLMSpecificMessage)
and message["role"] == "tool"
and message["tool_call_id"] and message["tool_call_id"]
and message["tool_call_id"] == tool_call_id and message["tool_call_id"] == tool_call_id
): ):