diff --git a/examples/foundational/19-openai-realtime.py b/examples/foundational/19-openai-realtime.py index f182d7c8c..685070d5e 100644 --- a/examples/foundational/19-openai-realtime.py +++ b/examples/foundational/19-openai-realtime.py @@ -19,6 +19,8 @@ from pipecat.observers.loggers.transcription_log_observer import TranscriptionLo from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.transcript_processor import TranscriptProcessor from pipecat.runner.types import RunnerArguments @@ -163,12 +165,12 @@ Remember, your responses should be short. Just one or two sentences, usually. Re # Create a standard OpenAI LLM context object using the normal messages format. The # OpenAIRealtimeLLMService will convert this internally to messages that the # openai WebSocket API can understand. - context = OpenAILLMContext( + context = LLMContext( [{"role": "user", "content": "Say hello!"}], tools, ) - context_aggregator = llm.create_context_aggregator(context) + context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline( [ diff --git a/src/pipecat/adapters/services/open_ai_realtime_adapter.py b/src/pipecat/adapters/services/open_ai_realtime_adapter.py index 2ff629e2e..67bdfb6ae 100644 --- a/src/pipecat/adapters/services/open_ai_realtime_adapter.py +++ b/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 copy +import json +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. @@ -61,7 +75,106 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): Returns: List of messages in a format ready for logging about OpenAI Realtime. """ - raise NotImplementedError("Universal LLMContext is not yet supported for OpenAI Realtime.") + return self._from_universal_context_messages(self.get_messages(context)).messages + + @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]: diff --git a/src/pipecat/services/openai/realtime/context.py b/src/pipecat/services/openai/realtime/context.py index cb1c0a9f5..da08f78ad 100644 --- a/src/pipecat/services/openai/realtime/context.py +++ b/src/pipecat/services/openai/realtime/context.py @@ -31,160 +31,6 @@ from . import events from .frames import RealtimeFunctionCallResultFrame, RealtimeMessagesUpdateFrame -class OpenAIRealtimeLLMContext(OpenAILLMContext): - """OpenAI Realtime LLM context with session management and message conversion. - - Extends the standard OpenAI LLM context to support real-time session properties, - instruction management, and conversion between standard message formats and - 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. - - Args: - obj: The OpenAILLMContext instance to upgrade. - - Returns: - The upgraded OpenAIRealtimeLLMContext instance. - """ - if isinstance(obj, OpenAILLMContext) and not isinstance(obj, OpenAIRealtimeLLMContext): - obj.__class__ = OpenAIRealtimeLLMContext - 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. - - Args: - item: The conversation item to add as a user message. - """ - message = { - "role": "user", - "content": [{"type": "text", "text": item.content[0].transcript}], - } - self.add_message(message) - - class OpenAIRealtimeUserContextAggregator(OpenAIUserContextAggregator): """User context aggregator for OpenAI Realtime API. diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index 8b3d500eb..f33b664f7 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -14,7 +14,9 @@ from typing import 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, +) from pipecat.frames.frames import ( BotStoppedSpeakingFrame, CancelFrame, @@ -41,6 +43,7 @@ from pipecat.frames.frames import ( UserStoppedSpeakingFrame, ) from pipecat.metrics.metrics import LLMTokenUsage +from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response import ( LLMAssistantAggregatorParams, LLMUserAggregatorParams, @@ -59,7 +62,6 @@ from pipecat.utils.tracing.service_decorators import traced_openai_realtime, tra from . import events from .context import ( OpenAIRealtimeAssistantContextAggregator, - OpenAIRealtimeLLMContext, OpenAIRealtimeUserContextAggregator, ) from .frames import RealtimeFunctionCallResultFrame, RealtimeMessagesUpdateFrame @@ -138,7 +140,17 @@ class OpenAIRealtimeLLMService(LLMService): self._send_transcription_frames = send_transcription_frames self._websocket = None self._receive_task = None - self._context = None + # "Last received context" is only needed while we still support + # OpenAILLMContextFrame. The "last received context" is the context received + # in the most recent OpenAILLMContextFrame or LLMContextFrame, *before* + # it's converted to an LLMContext if needed. Storing the "last received + # context" lets us determine whether the context has changed. (We can't + # compare contexts after conversion because conversion creates a new + # object.) + 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 @@ -347,22 +359,22 @@ class OpenAIRealtimeLLMService(LLMService): if isinstance(frame, TranscriptionFrame): pass - elif isinstance(frame, OpenAILLMContextFrame): - context: OpenAIRealtimeLLMContext = OpenAIRealtimeLLMContext.upgrade_to_realtime( + elif isinstance(frame, (LLMContextFrame, OpenAILLMContextFrame)): + context = ( frame.context + if isinstance(frame, LLMContextFrame) + else LLMContext.from_openai_context(frame.context) ) if not self._context: + self._last_received_context = frame.context self._context = context - elif frame.context is not self._context: + elif frame.context is not self._last_received_context: # If the context has changed, reset the conversation + self._last_received_context = frame.context self._context = context await self.reset_conversation() # Run the LLM at next opportunity await self._create_response() - elif isinstance(frame, LLMContextFrame): - raise NotImplementedError( - "Universal LLMContext is not yet supported for OpenAI Realtime." - ) elif isinstance(frame, InputAudioRawFrame): if not self._audio_input_paused: await self._send_user_audio(frame) @@ -377,6 +389,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) @@ -459,13 +472,20 @@ class OpenAIRealtimeLLMService(LLMService): async def _update_settings(self): settings = self._session_properties - # tools given in the context override the tools in the session properties - if self._context and self._context.tools: - settings.tools = self._context.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: - settings.instructions = self._context._session_instructions + + if self._context: + 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 llm_invocation_params["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 llm_invocation_params["system_instruction"]: + settings.instructions = llm_invocation_params["system_instruction"] + await self.send_client_event(events.SessionUpdateEvent(session=settings)) # @@ -760,9 +780,7 @@ class OpenAIRealtimeLLMService(LLMService): """ logger.debug("Resetting conversation") await self._disconnect() - if self._context: - self._context.llm_needs_settings_update = True - self._context.llm_needs_initial_messages = True + self._llm_needs_conversation_setup = True await self._connect() @traced_openai_realtime(operation="llm_request") @@ -771,19 +789,25 @@ class OpenAIRealtimeLLMService(LLMService): self._run_llm_when_api_session_ready = True return - if self._context.llm_needs_initial_messages: - messages = self._context.get_messages_for_initializing_history() + adapter: OpenAIRealtimeLLMAdapter = self.get_llm_adapter() + + # Configure the LLM for this session if needed + if self._llm_needs_conversation_setup: + # Send initial messages + 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 - logger.debug(f"Creating response: {self._context.get_messages_for_logging()}") + # We're done configuring the LLM for this session + self._llm_needs_conversation_setup = False + + logger.debug(f"Creating response: {adapter.get_messages_for_logging(self._context)}") await self.push_frame(LLMFullResponseStartFrame()) await self.start_processing_metrics() diff --git a/src/pipecat/services/openai_realtime/context.py b/src/pipecat/services/openai_realtime/context.py deleted file mode 100644 index 58f1cfe75..000000000 --- a/src/pipecat/services/openai_realtime/context.py +++ /dev/null @@ -1,21 +0,0 @@ -# -# Copyright (c) 2025, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""OpenAI Realtime LLM context and aggregator implementations.""" - -import warnings - -from pipecat.services.openai.realtime.context import * - -with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "Types in pipecat.services.openai_realtime.context are deprecated. " - "Please use the equivalent types from " - "pipecat.services.openai.realtime.context instead.", - DeprecationWarning, - stacklevel=2, - )