Update OpenAIRealtimeLLMService to work with LLMContext and LLMContextAggregatorPair

src/pipecat/adapters/services/open_ai_realtime_adapter.py

@@ -6,12 +6,18 @@
 
 """OpenAI Realtime LLM adapter for Pipecat."""
 
-from typing import Any, Dict, List, TypedDict
+import json
+from copy import copy
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional, TypedDict
+
+from loguru import logger
 
 from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
 from pipecat.adapters.schemas.function_schema import FunctionSchema
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
-from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextMessage
+from pipecat.services.openai_realtime import events
 
 
 class OpenAIRealtimeLLMInvocationParams(TypedDict):
@@ -20,7 +26,9 @@ class OpenAIRealtimeLLMInvocationParams(TypedDict):
     This is a placeholder until support for universal LLMContext machinery is added for OpenAI Realtime.
     """
 
-    pass
+    system_instruction: Optional[str]
+    messages: List[events.ConversationItem]
+    tools: List[Dict[str, Any]]
 
 
 class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
@@ -33,7 +41,7 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
     @property
     def id_for_llm_specific_messages(self) -> str:
         """Get the identifier used in LLMSpecificMessage instances for OpenAI Realtime."""
-        raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.")
+        return "openai-realtime"
 
     def get_llm_invocation_params(self, context: LLMContext) -> OpenAIRealtimeLLMInvocationParams:
         """Get OpenAI Realtime-specific LLM invocation parameters from a universal LLM context.
@@ -46,7 +54,13 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
         Returns:
             Dictionary of parameters for invoking OpenAI Realtime's API.
         """
-        raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.")
+        messages = self._from_universal_context_messages(self.get_messages(context))
+        return {
+            "system_instruction": messages.system_instruction,
+            "messages": messages.messages,
+            # NOTE: LLMContext's tools are guaranteed to be a ToolsSchema (or NOT_GIVEN)
+            "tools": self.from_standard_tools(context.tools) or [],
+        }
 
     def get_messages_for_logging(self, context) -> List[Dict[str, Any]]:
         """Get messages from a universal LLM context in a format ready for logging about OpenAI Realtime.
@@ -63,6 +77,105 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
         """
         raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.")
 
+    @dataclass
+    class ConvertedMessages:
+        """Container for OpenAI-formatted messages converted from universal context."""
+
+        messages: List[events.ConversationItem]
+        system_instruction: Optional[str] = None
+
+    def _from_universal_context_messages(
+        self, universal_context_messages: List[LLMContextMessage]
+    ) -> ConvertedMessages:
+        # We can't load a long conversation history into the openai realtime api yet. (The API/model
+        # forgets that it can do audio, if you do a series of `conversation.item.create` calls.) So
+        # our general strategy until this is fixed is just to put everything into a first "user"
+        # message as a single input.
+
+        if not universal_context_messages:
+            return self.ConvertedMessages()
+
+        messages = copy.deepcopy(universal_context_messages)
+        system_instruction = None
+
+        # If we have a "system" message as our first message, let's pull that out into session
+        # "instructions"
+        if messages[0].get("role") == "system":
+            system = messages.pop(0)
+            content = system.get("content")
+            if isinstance(content, str):
+                system_instruction = content
+            elif isinstance(content, list):
+                system_instruction = content[0].get("text")
+            if not messages:
+                return self.ConvertedMessages(messages=[], system_instruction=system_instruction)
+
+        # If we have just a single "user" item, we can just send it normally
+        if len(messages) == 1 and messages[0].get("role") == "user":
+            return self.ConvertedMessages(
+                messages=[self._from_universal_context_message(messages[0])],
+                system_instruction=system_instruction,
+            )
+
+        # Otherwise, let's pack everything into a single "user" message with a bit of
+        # explanation for the LLM
+        intro_text = """
+        This is a previously saved conversation. Please treat this conversation history as a
+        starting point for the current conversation."""
+
+        trailing_text = """
+        This is the end of the previously saved conversation. Please continue the conversation
+        from here. If the last message is a user instruction or question, act on that instruction
+        or answer the question. If the last message is an assistant response, simple say that you
+        are ready to continue the conversation."""
+
+        self.ConvertedMessages(
+            messages=[
+                {
+                    "role": "user",
+                    "type": "message",
+                    "content": [
+                        {
+                            "type": "input_text",
+                            "text": "\n\n".join(
+                                [intro_text, json.dumps(messages, indent=2), trailing_text]
+                            ),
+                        }
+                    ],
+                }
+            ],
+            system_instruction=system_instruction,
+        )
+
+    def _from_universal_context_message(
+        self, message: LLMContextMessage
+    ) -> events.ConversationItem:
+        if message.get("role") == "user":
+            content = message.get("content")
+            if isinstance(message.get("content"), list):
+                content = ""
+                for c in message.get("content"):
+                    if c.get("type") == "text":
+                        content += " " + c.get("text")
+                    else:
+                        logger.error(
+                            f"Unhandled content type in context message: {c.get('type')} - {message}"
+                        )
+            return events.ConversationItem(
+                role="user",
+                type="message",
+                content=[events.ItemContent(type="input_text", text=content)],
+            )
+        if message.get("role") == "assistant" and message.get("tool_calls"):
+            tc = message.get("tool_calls")[0]
+            return events.ConversationItem(
+                type="function_call",
+                call_id=tc["id"],
+                name=tc["function"]["name"],
+                arguments=tc["function"]["arguments"],
+            )
+        logger.error(f"Unhandled message type in _from_universal_context_message: {message}")
+
     @staticmethod
     def _to_openai_realtime_function_format(function: FunctionSchema) -> Dict[str, Any]:
         """Convert a function schema to OpenAI Realtime format.

src/pipecat/services/openai_realtime/context.py

@@ -39,24 +39,6 @@ class OpenAIRealtimeLLMContext(OpenAILLMContext):
     realtime conversation items.
     """
 
-    def __init__(self, messages=None, tools=None, **kwargs):
-        """Initialize the OpenAIRealtimeLLMContext.
-
-        Args:
-            messages: Initial conversation messages. Defaults to None.
-            tools: Available function tools. Defaults to None.
-            **kwargs: Additional arguments passed to parent OpenAILLMContext.
-        """
-        super().__init__(messages=messages, tools=tools, **kwargs)
-        self.__setup_local()
-
-    def __setup_local(self):
-        self.llm_needs_settings_update = True
-        self.llm_needs_initial_messages = True
-        self._session_instructions = ""
-
-        return
-
     @staticmethod
     def upgrade_to_realtime(obj: OpenAILLMContext) -> "OpenAIRealtimeLLMContext":
         """Upgrade a standard OpenAI LLM context to a realtime context.
@@ -72,106 +54,6 @@ class OpenAIRealtimeLLMContext(OpenAILLMContext):
             obj.__setup_local()
         return obj
 
-    # todo
-    #   - finish implementing all frames
-
-    def from_standard_message(self, message):
-        """Convert a standard message format to a realtime conversation item.
-
-        Args:
-            message: The standard message dictionary to convert.
-
-        Returns:
-            A ConversationItem instance for the realtime API.
-        """
-        if message.get("role") == "user":
-            content = message.get("content")
-            if isinstance(message.get("content"), list):
-                content = ""
-                for c in message.get("content"):
-                    if c.get("type") == "text":
-                        content += " " + c.get("text")
-                    else:
-                        logger.error(
-                            f"Unhandled content type in context message: {c.get('type')} - {message}"
-                        )
-            return events.ConversationItem(
-                role="user",
-                type="message",
-                content=[events.ItemContent(type="input_text", text=content)],
-            )
-        if message.get("role") == "assistant" and message.get("tool_calls"):
-            tc = message.get("tool_calls")[0]
-            return events.ConversationItem(
-                type="function_call",
-                call_id=tc["id"],
-                name=tc["function"]["name"],
-                arguments=tc["function"]["arguments"],
-            )
-        logger.error(f"Unhandled message type in from_standard_message: {message}")
-
-    def get_messages_for_initializing_history(self):
-        """Get conversation items for initializing the realtime session history.
-
-        Converts the context's messages to a format suitable for the realtime API,
-        handling system instructions and conversation history packaging.
-
-        Returns:
-            List of conversation items for session initialization.
-        """
-        # We can't load a long conversation history into the openai realtime api yet. (The API/model
-        # forgets that it can do audio, if you do a series of `conversation.item.create` calls.) So
-        # our general strategy until this is fixed is just to put everything into a first "user"
-        # message as a single input.
-        if not self.messages:
-            return []
-
-        messages = copy.deepcopy(self.messages)
-
-        # If we have a "system" message as our first message, let's pull that out into session
-        # "instructions"
-        if messages[0].get("role") == "system":
-            self.llm_needs_settings_update = True
-            system = messages.pop(0)
-            content = system.get("content")
-            if isinstance(content, str):
-                self._session_instructions = content
-            elif isinstance(content, list):
-                self._session_instructions = content[0].get("text")
-            if not messages:
-                return []
-
-        # If we have just a single "user" item, we can just send it normally
-        if len(messages) == 1 and messages[0].get("role") == "user":
-            return [self.from_standard_message(messages[0])]
-
-        # Otherwise, let's pack everything into a single "user" message with a bit of
-        # explanation for the LLM
-        intro_text = """
-        This is a previously saved conversation. Please treat this conversation history as a
-        starting point for the current conversation."""
-
-        trailing_text = """
-        This is the end of the previously saved conversation. Please continue the conversation
-        from here. If the last message is a user instruction or question, act on that instruction
-        or answer the question. If the last message is an assistant response, simple say that you
-        are ready to continue the conversation."""
-
-        return [
-            {
-                "role": "user",
-                "type": "message",
-                "content": [
-                    {
-                        "type": "input_text",
-                        "text": "\n\n".join(
-                            [intro_text, json.dumps(messages, indent=2), trailing_text]
-                        ),
-                    }
-                ],
-            }
-        ]
-
     def add_user_content_item_as_message(self, item):
         """Add a user content item as a standard message to the context.
 

src/pipecat/services/openai_realtime/openai.py

@@ -10,11 +10,14 @@ import base64
 import json
 import time
 from dataclasses import dataclass
-from typing import Optional
+from typing import Any, Dict, List, Optional
 
 from loguru import logger
 
-from pipecat.adapters.services.open_ai_realtime_adapter import OpenAIRealtimeLLMAdapter
+from pipecat.adapters.services.open_ai_realtime_adapter import (
+    OpenAIRealtimeLLMAdapter,
+    OpenAIRealtimeLLMInvocationParams,
+)
 from pipecat.frames.frames import (
     BotStoppedSpeakingFrame,
     CancelFrame,
@@ -149,6 +152,8 @@ class OpenAIRealtimeLLMService(LLMService):
         self._context: LLMContext = None
         self._last_received_context: OpenAILLMContext | LLMContext = None
 
+        self._llm_needs_conversation_setup = True
+
         self._disconnecting = False
         self._api_session_ready = False
         self._run_llm_when_api_session_ready = False
@@ -386,6 +391,7 @@ class OpenAIRealtimeLLMService(LLMService):
         elif isinstance(frame, LLMMessagesAppendFrame):
             await self._handle_messages_append(frame)
         elif isinstance(frame, RealtimeMessagesUpdateFrame):
+            # TODO: we don't need RealtimeMessagesUpdateFrame, I think...?
             self._context = frame.context
         elif isinstance(frame, LLMUpdateSettingsFrame):
             self._session_properties = events.SessionProperties(**frame.settings)
@@ -468,13 +474,19 @@ class OpenAIRealtimeLLMService(LLMService):
 
     async def _update_settings(self):
         settings = self._session_properties
+
+        adapter: OpenAIRealtimeLLMAdapter = self.get_llm_adapter()
+        llm_invocation_params = adapter.get_llm_invocation_params(self._context)
+
         # tools given in the context override the tools in the session properties
-        if self._context and self._context.tools:
+        if llm_invocation_params["tools"]:
-            settings.tools = self._context.tools
+            settings.tools = llm_invocation_params["tools"]
+
         # instructions in the context come from an initial "system" message in the
         # messages list, and override instructions in the session properties
-        if self._context and self._context._session_instructions:
+        if llm_invocation_params["system_instruction"]:
-            settings.instructions = self._context._session_instructions
+            settings.instructions = llm_invocation_params["system_instruction"]
+
         await self.send_client_event(events.SessionUpdateEvent(session=settings))
 
     #
@@ -769,9 +781,7 @@ class OpenAIRealtimeLLMService(LLMService):
         """
         logger.debug("Resetting conversation")
         await self._disconnect()
-        if self._context:
+        self._llm_needs_conversation_setup = True
-            self._context.llm_needs_settings_update = True
-            self._context.llm_needs_initial_messages = True
         await self._connect()
 
     @traced_openai_realtime(operation="llm_request")
@@ -780,17 +790,22 @@ class OpenAIRealtimeLLMService(LLMService):
             self._run_llm_when_api_session_ready = True
             return
 
-        if self._context.llm_needs_initial_messages:
+        # Configure the LLM for this session if needed
-            messages = self._context.get_messages_for_initializing_history()
+        if self._llm_needs_conversation_setup:
+            # Send initial messages
+            adapter: OpenAIRealtimeLLMAdapter = self.get_llm_adapter()
+            llm_invocation_params = adapter.get_llm_invocation_params(self._context)
+            messages = llm_invocation_params["messages"]
             for item in messages:
                 evt = events.ConversationItemCreateEvent(item=item)
                 self._messages_added_manually[evt.item.id] = True
                 await self.send_client_event(evt)
-            self._context.llm_needs_initial_messages = False
 
-        if self._context.llm_needs_settings_update:
+            # Send new settings if needed
             await self._update_settings()
-            self._context.llm_needs_settings_update = False
+
+            # We're done configuring the LLM for this session
+            self._llm_needs_conversation_setup = False
 
         logger.debug(f"Creating response: {self._context.get_messages_for_logging()}")