From 3ea1e357f28c5b0ab35153e941c986a72cbaab15 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 16 Oct 2025 15:59:50 -0400 Subject: [PATCH 01/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (initial part of work) --- examples/foundational/19-openai-realtime.py | 6 +- .../services/open_ai_realtime_adapter.py | 125 +++++++++++++- .../services/openai/realtime/context.py | 154 ------------------ src/pipecat/services/openai/realtime/llm.py | 76 ++++++--- .../services/openai_realtime/context.py | 21 --- 5 files changed, 173 insertions(+), 209 deletions(-) delete mode 100644 src/pipecat/services/openai_realtime/context.py 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, - ) From 29fd17b9ff57b50442544c3908e21cb461d8ca15 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 20 Oct 2025 13:29:27 -0400 Subject: [PATCH 02/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). Avoid pushing `LLMTextFrame` when `OpenAIRealtimeLLMService` is configured to output audio. This avoids duplicate text in assistant messages in context. Conceptually, a speech-to-speech service encapsulates TTS behavior; in a "traditional" pipeline, `LLMTextFrames` are swallowed by the TTS service, so they should similarly not be pushed by a speech-to-speech service. Only. `TTSTextFrame`s should be pushed. --- src/pipecat/services/openai/realtime/llm.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index f33b664f7..ace8be655 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -687,12 +687,15 @@ class OpenAIRealtimeLLMService(LLMService): logger.debug(f"Handling standalone response: {evt.response.id}") async def _handle_evt_text_delta(self, evt): + # We receive text deltas (as opposed to audio transcript deltas) when + # the output modality is "text" if evt.delta: await self.push_frame(LLMTextFrame(evt.delta)) async def _handle_evt_audio_transcript_delta(self, evt): + # We receive audio transcript deltas (as opposed to text deltas) when + # the output modality is "audio" (the default) if evt.delta: - await self.push_frame(LLMTextFrame(evt.delta)) await self.push_frame(TTSTextFrame(evt.delta)) async def _handle_evt_function_call_arguments_done(self, evt): From ec42f8c24e034316f605f03106d7decbe4a1a47f Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 20 Oct 2025 14:54:15 -0400 Subject: [PATCH 03/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). Push `TranscriptionFrame`s upstream, to be handled by the user context aggregator. This will require at least a couple of other changes: - Updating examples to put transcript processors upstream from `OpenAIRealtimeLLMService` - Maybe figuring out a way to preserve backward compatibility with existing pipelines that put transcript processors downstream from `OpenAIRealtimeLLMService` - Updating `OpenAIRealtimeLLMService` to ignore new received context frames, since the upstream user context aggregator will generate those after each newly-added user message; hopefully nobody was reliant on the old behavior of resetting the conversation upon receiving a new context! --- examples/foundational/19-openai-realtime.py | 2 +- src/pipecat/services/openai/realtime/llm.py | 22 ++++++++++----------- 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/examples/foundational/19-openai-realtime.py b/examples/foundational/19-openai-realtime.py index 685070d5e..3bf12f916 100644 --- a/examples/foundational/19-openai-realtime.py +++ b/examples/foundational/19-openai-realtime.py @@ -176,8 +176,8 @@ Remember, your responses should be short. Just one or two sentences, usually. Re [ transport.input(), # Transport user input context_aggregator.user(), + transcript.user(), # LLM pushes TranscriptionFrames upstream llm, # LLM - transcript.user(), # Placed after the LLM, as LLM pushes TranscriptionFrames downstream transport.output(), # Transport bot output transcript.assistant(), # After the transcript output, to time with the audio output context_aggregator.assistant(), diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index ace8be655..7e65cc2f7 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -607,11 +607,11 @@ class OpenAIRealtimeLLMService(LLMService): # For now, no additional logic needed beyond the event handler call async def _handle_evt_input_audio_transcription_delta(self, evt): - if self._send_transcription_frames: - await self.push_frame( - # no way to get a language code? - InterimTranscriptionFrame(evt.delta, "", time_now_iso8601(), result=evt) - ) + await self.push_frame( + # no way to get a language code? + InterimTranscriptionFrame(evt.delta, "", time_now_iso8601(), result=evt), + direction=FrameDirection.UPSTREAM, + ) @traced_stt async def _handle_user_transcription( @@ -628,12 +628,12 @@ class OpenAIRealtimeLLMService(LLMService): """ await self._call_event_handler("on_conversation_item_updated", evt.item_id, None) - if self._send_transcription_frames: - await self.push_frame( - # no way to get a language code? - TranscriptionFrame(evt.transcript, "", time_now_iso8601(), result=evt) - ) - await self._handle_user_transcription(evt.transcript, True, Language.EN) + await self.push_frame( + # no way to get a language code? + TranscriptionFrame(evt.transcript, "", time_now_iso8601(), result=evt), + FrameDirection.UPSTREAM, + ) + await self._handle_user_transcription(evt.transcript, True, Language.EN) pair = self._user_and_response_message_tuple if pair: user, assistant = pair From 8a151235c36fbf4a6d8e3f2be86cd64520962e5b Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 20 Oct 2025 15:47:46 -0400 Subject: [PATCH 04/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deprecate `send_transcription_frames`—transcription frames are now always sent. --- src/pipecat/services/openai/realtime/llm.py | 22 ++++++++++++++++++--- 1 file changed, 19 insertions(+), 3 deletions(-) diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index 7e65cc2f7..f5f54f8dd 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -110,7 +110,7 @@ class OpenAIRealtimeLLMService(LLMService): base_url: str = "wss://api.openai.com/v1/realtime", session_properties: Optional[events.SessionProperties] = None, start_audio_paused: bool = False, - send_transcription_frames: bool = True, + send_transcription_frames: Optional[bool] = None, **kwargs, ): """Initialize the OpenAI Realtime LLM service. @@ -123,9 +123,26 @@ class OpenAIRealtimeLLMService(LLMService): session_properties: Configuration properties for the realtime session. If None, uses default SessionProperties. start_audio_paused: Whether to start with audio input paused. Defaults to False. - send_transcription_frames: Whether to emit transcription frames. Defaults to True. + send_transcription_frames: Whether to emit transcription frames. + + .. deprecated:: 0.0.92 + This parameter is deprecated and will be removed in a future version. + Transcription frames are always sent. + **kwargs: Additional arguments passed to parent LLMService. """ + if send_transcription_frames is not None: + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "`send_transcription_frames` is deprecated and will be removed in a future version. " + "Transcription frames are always sent.", + DeprecationWarning, + stacklevel=2, + ) + full_url = f"{base_url}?model={model}" super().__init__(base_url=full_url, **kwargs) @@ -137,7 +154,6 @@ class OpenAIRealtimeLLMService(LLMService): session_properties or events.SessionProperties() ) self._audio_input_paused = start_audio_paused - self._send_transcription_frames = send_transcription_frames self._websocket = None self._receive_task = None # "Last received context" is only needed while we still support From 5fa56df01409ae47ff3a34d5b63ad0d4da0dc305 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 20 Oct 2025 15:51:00 -0400 Subject: [PATCH 05/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). Update 19b example with new pattern. --- examples/foundational/19b-openai-realtime-text.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/examples/foundational/19b-openai-realtime-text.py b/examples/foundational/19b-openai-realtime-text.py index bb63a4814..fab029bb6 100644 --- a/examples/foundational/19b-openai-realtime-text.py +++ b/examples/foundational/19b-openai-realtime-text.py @@ -18,6 +18,8 @@ from pipecat.frames.frames import LLMRunFrame, TranscriptionMessage 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 @@ -169,20 +171,20 @@ 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( [ transport.input(), # Transport user input context_aggregator.user(), + transcript.user(), # LLM pushes TranscriptionFrames upstream llm, # LLM tts, # TTS - transcript.user(), # Placed after the LLM, as LLM pushes TranscriptionFrames downstream transport.output(), # Transport bot output transcript.assistant(), # After the transcript output, to time with the audio output context_aggregator.assistant(), From 47756319beda42cb4ae107ff45916ddba769cc19 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 20 Oct 2025 16:04:09 -0400 Subject: [PATCH 06/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Receiving a new context (via a context frame) no longer serves as a signal to reset the conversation. That’s because we’re now receiving new contexts from the user aggregator every time new messages are added, and from the assistant aggregator when function call results come in. The code pattern we're heading towards, of “diffing” each new context with the previous on, sets us up for doing more sophisticated things in the future, like sending specific messages to OpenAI to edit its internally-tracked context. Also, remove code that was directly modifying context. --- src/pipecat/services/openai/realtime/llm.py | 54 ++++----------------- 1 file changed, 9 insertions(+), 45 deletions(-) diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index f5f54f8dd..a9d890bae 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -156,15 +156,7 @@ class OpenAIRealtimeLLMService(LLMService): self._audio_input_paused = start_audio_paused self._websocket = None self._receive_task = 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 @@ -176,7 +168,6 @@ class OpenAIRealtimeLLMService(LLMService): self._current_audio_response = None self._messages_added_manually = {} - self._user_and_response_message_tuple = None self._pending_function_calls = {} # Track function calls by call_id self._register_event_handler("on_conversation_item_created") @@ -382,15 +373,15 @@ class OpenAIRealtimeLLMService(LLMService): else LLMContext.from_openai_context(frame.context) ) if not self._context: - self._last_received_context = frame.context + # We got our initial context + # Run the LLM at next opportunity self._context = 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() + await self._create_response() + else: + # We got an updated context + # Send results for any newly-completed function calls + # TODO: to implement + pass elif isinstance(frame, InputAudioRawFrame): if not self._audio_input_paused: await self._send_user_audio(frame) @@ -607,12 +598,7 @@ class OpenAIRealtimeLLMService(LLMService): del self._messages_added_manually[evt.item.id] return - if evt.item.role == "user": - # We need to wait for completion of both user message and response message. Then we'll - # add both to the context. User message is complete when we have a "transcript" field - # that is not None. Response message is complete when we get a "response.done" event. - self._user_and_response_message_tuple = (evt.item, {"done": False, "output": []}) - elif evt.item.role == "assistant": + if evt.item.role == "assistant": self._current_assistant_response = evt.item await self.push_frame(LLMFullResponseStartFrame()) @@ -650,16 +636,6 @@ class OpenAIRealtimeLLMService(LLMService): FrameDirection.UPSTREAM, ) await self._handle_user_transcription(evt.transcript, True, Language.EN) - pair = self._user_and_response_message_tuple - if pair: - user, assistant = pair - user.content[0].transcript = evt.transcript - if assistant["done"]: - self._user_and_response_message_tuple = None - self._context.add_user_content_item_as_message(user) - else: - # User message without preceding conversation.item.created. Bug? - logger.warning(f"Transcript for unknown user message: {evt}") async def _handle_conversation_item_retrieved(self, evt: events.ConversationItemRetrieved): futures = self._retrieve_conversation_item_futures.pop(evt.item.id, None) @@ -689,18 +665,6 @@ class OpenAIRealtimeLLMService(LLMService): # response content for item in evt.response.output: await self._call_event_handler("on_conversation_item_updated", item.id, item) - pair = self._user_and_response_message_tuple - if pair: - user, assistant = pair - assistant["done"] = True - assistant["output"] = evt.response.output - if user.content[0].transcript is not None: - self._user_and_response_message_tuple = None - self._context.add_user_content_item_as_message(user) - else: - # Response message without preceding user message (standalone response) - # Function calls in this response were already processed immediately when arguments were complete - logger.debug(f"Handling standalone response: {evt.response.id}") async def _handle_evt_text_delta(self, evt): # We receive text deltas (as opposed to audio transcript deltas) when From 61944d22ef148b16b5864c79bece7c2a98451910 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 20 Oct 2025 16:35:04 -0400 Subject: [PATCH 07/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). Implement sending tool call results to the OpenAI server based on reading context updates. This lets us use the normal assistant context aggregator and not a special OpenAI Realtime subclass that pushes up a special frame for function call results. --- .../services/open_ai_realtime_adapter.py | 2 +- src/pipecat/services/openai/realtime/llm.py | 67 ++++++++++++------- 2 files changed, 45 insertions(+), 24 deletions(-) diff --git a/src/pipecat/adapters/services/open_ai_realtime_adapter.py b/src/pipecat/adapters/services/open_ai_realtime_adapter.py index 67bdfb6ae..58cf284b1 100644 --- a/src/pipecat/adapters/services/open_ai_realtime_adapter.py +++ b/src/pipecat/adapters/services/open_ai_realtime_adapter.py @@ -129,7 +129,7 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): or answer the question. If the last message is an assistant response, simple say that you are ready to continue the conversation.""" - self.ConvertedMessages( + return self.ConvertedMessages( messages=[ { "role": "user", diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index a9d890bae..a3421cb75 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -169,6 +169,7 @@ class OpenAIRealtimeLLMService(LLMService): self._messages_added_manually = {} self._pending_function_calls = {} # Track function calls by call_id + self._completed_tool_calls = set() self._register_event_handler("on_conversation_item_created") self._register_event_handler("on_conversation_item_updated") @@ -372,16 +373,7 @@ class OpenAIRealtimeLLMService(LLMService): if isinstance(frame, LLMContextFrame) else LLMContext.from_openai_context(frame.context) ) - if not self._context: - # We got our initial context - # Run the LLM at next opportunity - self._context = context - await self._create_response() - else: - # We got an updated context - # Send results for any newly-completed function calls - # TODO: to implement - pass + await self._handle_context(context) elif isinstance(frame, InputAudioRawFrame): if not self._audio_input_paused: await self._send_user_audio(frame) @@ -395,30 +387,32 @@ class OpenAIRealtimeLLMService(LLMService): await self._handle_bot_stopped_speaking() 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) await self._update_settings() elif isinstance(frame, LLMSetToolsFrame): await self._update_settings() - elif isinstance(frame, RealtimeFunctionCallResultFrame): - await self._handle_function_call_result(frame.result_frame) await self.push_frame(frame, direction) + async def _handle_context(self, context: LLMContext): + if not self._context: + # We got our initial context + self._context = context + # Initialize our bookkeeping of already-completed tool calls in + # the context + await self._process_completed_function_calls(send_new_results=False) + # Run the LLM at next opportunity + await self._create_response() + else: + # We got an updated context + self._context = context + # Send results for any newly-completed function calls + await self._process_completed_function_calls(send_new_results=True) + async def _handle_messages_append(self, frame): logger.error("!!! NEED TO IMPLEMENT MESSAGES APPEND") - async def _handle_function_call_result(self, frame): - item = events.ConversationItem( - type="function_call_output", - call_id=frame.tool_call_id, - output=json.dumps(frame.result), - ) - await self.send_client_event(events.ConversationItemCreateEvent(item=item)) - # # websocket communication # @@ -459,6 +453,7 @@ class OpenAIRealtimeLLMService(LLMService): if self._receive_task: await self.cancel_task(self._receive_task, timeout=1.0) self._receive_task = None + self._completed_tool_calls = set() self._disconnecting = False except Exception as e: logger.error(f"{self} error disconnecting: {e}") @@ -801,10 +796,36 @@ class OpenAIRealtimeLLMService(LLMService): ) ) + async def _process_completed_function_calls(self, send_new_results: bool): + # Check for set of completed function calls in the context + sent_new_result = False + for message in self._context.get_messages(): + if message.get("role") and message.get("content") != "IN_PROGRESS": + tool_call_id = message.get("tool_call_id") + if tool_call_id and tool_call_id not in self._completed_tool_calls: + # Found a newly-completed function call - send the result to the service + if send_new_results: + sent_new_result = True + await self._send_tool_result(tool_call_id, message.get("content")) + self._completed_tool_calls.add(tool_call_id) + + # If we sent any new tool call results to the service, trigger another + # response + if sent_new_result: + await self._create_response() + async def _send_user_audio(self, frame): payload = base64.b64encode(frame.audio).decode("utf-8") await self.send_client_event(events.InputAudioBufferAppendEvent(audio=payload)) + async def _send_tool_result(self, tool_call_id: str, result: str): + item = events.ConversationItem( + type="function_call_output", + call_id=tool_call_id, + output=json.dumps(result), + ) + await self.send_client_event(events.ConversationItemCreateEvent(item=item)) + def create_context_aggregator( self, context: OpenAILLMContext, From bab0aaf585a4eb4ad089f36d7f9529661aacc95c Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 20 Oct 2025 16:58:44 -0400 Subject: [PATCH 08/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). Update `create_context_aggregator()` (which we're keeping around for backward compatibility) to create a `LLMContextAggregatorPair` rather than OpenAI-Realtime-specific aggregators. --- .../services/openai/realtime/context.py | 243 +++++++++++++++++- .../services/openai/realtime/frames.py | 23 +- src/pipecat/services/openai/realtime/llm.py | 25 +- 3 files changed, 275 insertions(+), 16 deletions(-) diff --git a/src/pipecat/services/openai/realtime/context.py b/src/pipecat/services/openai/realtime/context.py index da08f78ad..57979406c 100644 --- a/src/pipecat/services/openai/realtime/context.py +++ b/src/pipecat/services/openai/realtime/context.py @@ -4,7 +4,94 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""OpenAI Realtime LLM context and aggregator implementations.""" +"""OpenAI Realtime LLM context and aggregator implementations. + +.. deprecated:: 0.0.92 + OpenAI Realtime no longer uses types from this module under the hood. + It now uses `LLMContext` and `LLMContextAggregatorPair`. + Using the new patterns should allow you to not need types from this module. + + BEFORE: + ``` + # Setup + context = OpenAILLMContext(messages, tools) + context_aggregator = llm.create_context_aggregator(context) + + # Context aggregator type + context_aggregator: OpenAIContextAggregatorPair + + # Context frame type + frame: OpenAILLMContextFrame + + # Context type + context: OpenAIRealtimeLLMContext + # or + context: OpenAILLMContext + + # Reading messages from context + messages = context.messages + ``` + + AFTER: + ``` + # Setup + context = LLMContext(messages, tools) + context_aggregator = LLMContextAggregatorPair(context) + + # Context aggregator type + context_aggregator: LLMContextAggregatorPair + + # Context frame type + frame: LLMContextFrame + + # Context type + context: LLMContext + + # Reading messages from context + messages = context.get_messages() + ``` +""" + +import warnings + +with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "Types in pipecat.services.openai.realtime.llm are deprecated. \n" + "OpenAI Realtime no longer uses types from this module under the hood. \n" + "It now uses `LLMContext` and `LLMContextAggregatorPair`. \n" + "Using the new patterns should allow you to not need types from this module.\n\n" + "BEFORE:\n" + "```\n" + "# Setup\n" + "context = OpenAILLMContext(messages, tools)\n" + "context_aggregator = llm.create_context_aggregator(context)\n\n" + "# Context aggregator type\n" + "context_aggregator: OpenAIContextAggregatorPair\n\n" + "# Context frame type\n" + "frame: OpenAILLMContextFrame\n\n" + "# Context type\n" + "context: OpenAIRealtimeLLMContext\n" + "# or\n" + "context: OpenAILLMContext\n\n" + "# Reading messages from context\n" + "messages = context.messages\n" + "```\n\n" + "AFTER:\n" + "```\n" + "# Setup\n" + "context = LLMContext(messages, tools)\n" + "context_aggregator = LLMContextAggregatorPair(context)\n\n" + "# Context aggregator type\n" + "context_aggregator: LLMContextAggregatorPair\n\n" + "# Context frame type\n" + "frame: LLMContextFrame\n\n" + "# Context type\n" + "context: LLMContext\n\n" + "# Reading messages from context\n" + "messages = context.get_messages()\n" + "```\n", + ) import copy import json @@ -31,6 +118,160 @@ 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/frames.py b/src/pipecat/services/openai/realtime/frames.py index 8617c6efd..39cfd9757 100644 --- a/src/pipecat/services/openai/realtime/frames.py +++ b/src/pipecat/services/openai/realtime/frames.py @@ -4,7 +4,28 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Custom frame types for OpenAI Realtime API integration.""" +"""Custom frame types for OpenAI Realtime API integration. + +.. deprecated:: 0.0.92 + OpenAI Realtime no longer uses types from this module under the hood. + + It now works more like most LLM services in Pipecat, relying on updates to + its context, pushed by context aggregators, to update its internal state. + + Listen for `LLMContextFrame`s for context updates. +""" + +import warnings + +with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "Types in pipecat.services.openai.realtime.frames are deprecated. \n" + "OpenAI Realtime no longer uses types from this module under the hood. \n\n" + "It now works more like other LLM services in Pipecat, relying on updates to \n" + "its context, pushed by context aggregators, to update its internal state.\n\n" + "Listen for `LLMContextFrame`s for context updates.\n" + ) from dataclasses import dataclass from typing import TYPE_CHECKING diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index a3421cb75..eb2ba5ef4 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -48,6 +48,7 @@ from pipecat.processors.aggregators.llm_response import ( LLMAssistantAggregatorParams, LLMUserAggregatorParams, ) +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContext, OpenAILLMContextFrame, @@ -60,11 +61,6 @@ from pipecat.utils.time import time_now_iso8601 from pipecat.utils.tracing.service_decorators import traced_openai_realtime, traced_stt from . import events -from .context import ( - OpenAIRealtimeAssistantContextAggregator, - OpenAIRealtimeUserContextAggregator, -) -from .frames import RealtimeFunctionCallResultFrame, RealtimeMessagesUpdateFrame try: from websockets.asyncio.client import connect as websocket_connect @@ -832,9 +828,14 @@ class OpenAIRealtimeLLMService(LLMService): *, user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), - ) -> OpenAIContextAggregatorPair: + ) -> LLMContextAggregatorPair: """Create an instance of OpenAIContextAggregatorPair from an OpenAILLMContext. + NOTE: this method exists only for backward compatibility. New code + should instead do: + context = LLMContext(...) + context_aggregator = LLMContextAggregatorPair(context) + Constructor keyword arguments for both the user and assistant aggregators can be provided. Args: @@ -847,11 +848,7 @@ class OpenAIRealtimeLLMService(LLMService): the user and one for the assistant, encapsulated in an OpenAIContextAggregatorPair. """ - context.set_llm_adapter(self.get_llm_adapter()) - - OpenAIRealtimeLLMContext.upgrade_to_realtime(context) - user = OpenAIRealtimeUserContextAggregator(context, params=user_params) - - assistant_params.expect_stripped_words = False - assistant = OpenAIRealtimeAssistantContextAggregator(context, params=assistant_params) - return OpenAIContextAggregatorPair(_user=user, _assistant=assistant) + context = LLMContext.from_openai_context(context) + return LLMContextAggregatorPair( + context, user_params=user_params, assistant_params=assistant_params + ) From b34461bf937030be3bcccf148d884ac60eb14314 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Mon, 20 Oct 2025 17:36:30 -0400 Subject: [PATCH 09/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). --- CHANGELOG.md | 94 ++++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 91 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 359b2a2f4..a152022f1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,85 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Expanded support for univeral `LLMContext` to `OpenAIRealtimeLLMService`. + As a reminder, the context-setup pattern when using `LLMContext` is: + + ```python + context = LLMContext(messages, tools) + context_aggregator = LLMContextAggregatorPair(context) + ``` + + (Note that even though `OpenAIRealtimeLLMService` now supports the universal + `LLMContext`, it is not meant to be swapped out for another LLM service at + runtime.) + + Note: `TranscriptionFrame`s now go upstream from `OpenAIRealtimeLLMService`, + so if you're using `TranscriptProcessor`, say, you'll want to adjust + accordingly: + + ```python + pipeline = Pipeline( + [ + transport.input(), + context_aggregator.user(), + + # BEFORE + llm, + transcript.user(), + + # AFTER + transcript.user(), + llm, + + transport.output(), + transcript.assistant(), + context_aggregator.assistant(), + ] + ) + ``` + + Also worth noting: whether or not you use the new context-setup pattern with + `OpenAIRealtimeLLMService`, some types have changed under the hood: + + ```python + ## BEFORE: + + # Context aggregator type + context_aggregator: OpenAIContextAggregatorPair + + # Context frame type + frame: OpenAILLMContextFrame + + # Context type + context: OpenAIRealtimeLLMContext + # or + context: OpenAILLMContext + + # Reading messages from context + messages = context.messages + + ## AFTER: + + # Context aggregator type + context_aggregator: LLMContextAggregatorPair + + # Context frame type + frame: LLMContextFrame + + # Context type + context: LLMContext + + # Reading messages from context + messages = context.get_messages() + ``` + + Also note that `RealtimeMessagesUpdateFrame` and + `RealtimeFunctionCallResultFrame` have been deprecated, since they're no + longer used by `OpenAIRealtimeLLMService`. OpenAI Realtime now works more + like other LLM services in Pipecat, relying on updates to its context, pushed + by context aggregators, to update its internal state. Listen for + `LLMContextFrame`s for context updates. + - Expanded support for universal `LLMContext` to `GeminiLiveLLMService`. As a reminder, the context-setup pattern when using `LLMContext` is: @@ -79,6 +158,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Deprecated +- The `send_transcription_frames` argument to `OpenAIRealtimeLLMService` is + deprecated. Transcription frames are now always sent. They go upstream, to be + handled by the user context aggregator. See "Added" section for details. + +- Types in `pipecat.services.openai.realtime.context` and + `pipecat.services.openai.realtime.frames` are deprecated, as they're no + longer used by `OpenAIRealtimeLLMService`. See "Added" section for details. + - `SimliVideoService` `simli_config` parameter is deprecated. Use `api_key` and `face_id` parameters instead. @@ -200,8 +287,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 deprecated. Transcription frames are now always sent. They go upstream, to be handled by the user context aggregator. See "Added" section for details. -- Types in `pipecat.services.aws.nova_sonic.context` have been deprecated due - to changes to support `LLMContext`. See "Changed" section for details. +- Types in `pipecat.services.aws.nova_sonic.context` are deprecated, as they're + no longer used by `AWSNovaSonicLLMService`. See "Added" section for + details. ### Fixed @@ -5375,4 +5463,4 @@ a bit. ## [0.0.2] - 2024-03-12 -Initial public release. \ No newline at end of file +Initial public release. From 19770b76b499a8e2c2783b22f4819845912e7950 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 21 Oct 2025 10:15:11 -0400 Subject: [PATCH 10/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). Add back file that was removed, when it should've just been deprecated. Also, fix version numbers in deprecation messages to match the next expected release. --- .../services/openai/realtime/context.py | 5 +++-- src/pipecat/services/openai/realtime/frames.py | 2 +- src/pipecat/services/openai/realtime/llm.py | 2 +- .../services/openai_realtime/context.py | 18 ++++++++++++++++++ 4 files changed, 23 insertions(+), 4 deletions(-) create mode 100644 src/pipecat/services/openai_realtime/context.py diff --git a/src/pipecat/services/openai/realtime/context.py b/src/pipecat/services/openai/realtime/context.py index 57979406c..96a714565 100644 --- a/src/pipecat/services/openai/realtime/context.py +++ b/src/pipecat/services/openai/realtime/context.py @@ -6,7 +6,7 @@ """OpenAI Realtime LLM context and aggregator implementations. -.. deprecated:: 0.0.92 +.. deprecated:: 0.0.91 OpenAI Realtime no longer uses types from this module under the hood. It now uses `LLMContext` and `LLMContextAggregatorPair`. Using the new patterns should allow you to not need types from this module. @@ -57,7 +57,8 @@ import warnings with warnings.catch_warnings(): warnings.simplefilter("always") warnings.warn( - "Types in pipecat.services.openai.realtime.llm are deprecated. \n" + "Types in pipecat.services.openai.realtime.llm (or " + "pipecat.services.openai_realtime.llm) are deprecated. \n" "OpenAI Realtime no longer uses types from this module under the hood. \n" "It now uses `LLMContext` and `LLMContextAggregatorPair`. \n" "Using the new patterns should allow you to not need types from this module.\n\n" diff --git a/src/pipecat/services/openai/realtime/frames.py b/src/pipecat/services/openai/realtime/frames.py index 39cfd9757..1f800af89 100644 --- a/src/pipecat/services/openai/realtime/frames.py +++ b/src/pipecat/services/openai/realtime/frames.py @@ -6,7 +6,7 @@ """Custom frame types for OpenAI Realtime API integration. -.. deprecated:: 0.0.92 +.. deprecated:: 0.0.91 OpenAI Realtime no longer uses types from this module under the hood. It now works more like most LLM services in Pipecat, relying on updates to diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index eb2ba5ef4..5e042cff3 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -121,7 +121,7 @@ class OpenAIRealtimeLLMService(LLMService): start_audio_paused: Whether to start with audio input paused. Defaults to False. send_transcription_frames: Whether to emit transcription frames. - .. deprecated:: 0.0.92 + .. deprecated:: 0.0.91 This parameter is deprecated and will be removed in a future version. Transcription frames are always sent. diff --git a/src/pipecat/services/openai_realtime/context.py b/src/pipecat/services/openai_realtime/context.py new file mode 100644 index 000000000..79a01b980 --- /dev/null +++ b/src/pipecat/services/openai_realtime/context.py @@ -0,0 +1,18 @@ +# +# Copyright (c) 2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""OpenAI Realtime LLM context and aggregator implementations. + +.. deprecated:: 0.0.91 + OpenAI Realtime no longer uses types from this module under the hood. + It now uses `LLMContext` and `LLMContextAggregatorPair`. + Using the new patterns should allow you to not need types from this module. + + See deprecation warning in pipecat.services.openai.realtime.context for + more details. +""" + +from pipecat.services.openai.realtime.context import * From 46e97c57c26c0ecb17f5d94d6c8175abacf59616 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 21 Oct 2025 11:14:40 -0400 Subject: [PATCH 11/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). Update 20b example to use new `LLMContext` pattern. --- .../20b-persistent-context-openai-realtime.py | 78 ++++++++----------- .../services/open_ai_realtime_adapter.py | 2 +- 2 files changed, 35 insertions(+), 45 deletions(-) diff --git a/examples/foundational/20b-persistent-context-openai-realtime.py b/examples/foundational/20b-persistent-context-openai-realtime.py index 629a17c67..33328ac24 100644 --- a/examples/foundational/20b-persistent-context-openai-realtime.py +++ b/examples/foundational/20b-persistent-context-openai-realtime.py @@ -13,11 +13,15 @@ from datetime import datetime from dotenv import load_dotenv from loguru import logger +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import LLMRunFrame 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, ) @@ -97,14 +101,12 @@ async def load_conversation(params: FunctionCallParams): asyncio.create_task(_reset()) -tools = [ - { - "type": "function", - "name": "get_current_weather", - "description": "Get the current weather", - "parameters": { - "type": "object", - "properties": { +tools = ToolsSchema( + standard_tools=[ + FunctionSchema( + name="get_current_weather", + description="Get the current weather", + properties={ "location": { "type": "string", "description": "The city and state, e.g. San Francisco, CA", @@ -115,45 +117,33 @@ tools = [ "description": "The temperature unit to use. Infer this from the users location.", }, }, - "required": ["location", "format"], - }, - }, - { - "type": "function", - "name": "save_conversation", - "description": "Save the current conversatione. Use this function to persist the current conversation to external storage.", - "parameters": { - "type": "object", - "properties": {}, - "required": [], - }, - }, - { - "type": "function", - "name": "get_saved_conversation_filenames", - "description": "Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.", - "parameters": { - "type": "object", - "properties": {}, - "required": [], - }, - }, - { - "type": "function", - "name": "load_conversation", - "description": "Load a conversation history. Use this function to load a conversation history into the current session.", - "parameters": { - "type": "object", - "properties": { + required=["location", "format"], + ), + FunctionSchema( + name="save_conversation", + description="Save the current conversatione. Use this function to persist the current conversation to external storage.", + properties={}, + required=[], + ), + FunctionSchema( + name="get_saved_conversation_filenames", + description="Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.", + properties={}, + required=[], + ), + FunctionSchema( + name="load_conversation", + description="Load a conversation history. Use this function to load a conversation history into the current session.", + properties={ "filename": { "type": "string", "description": "The filename of the conversation history to load.", } }, - "required": ["filename"], - }, - }, -] + required=["filename"], + ), + ] +) # We store functions so objects (e.g. SileroVADAnalyzer) don't get @@ -224,8 +214,8 @@ Remember, your responses should be short. Just one or two sentences, usually.""" llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames) llm.register_function("load_conversation", load_conversation) - context = OpenAILLMContext([], tools) - context_aggregator = llm.create_context_aggregator(context) + context = LLMContext([], tools) + 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 58cf284b1..7a6fc4d02 100644 --- a/src/pipecat/adapters/services/open_ai_realtime_adapter.py +++ b/src/pipecat/adapters/services/open_ai_realtime_adapter.py @@ -93,7 +93,7 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): # message as a single input. if not universal_context_messages: - return self.ConvertedMessages() + return self.ConvertedMessages(messages=[]) messages = copy.deepcopy(universal_context_messages) system_instruction = None From 376180414664aba011d1f3a36783e474f3033a22 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 21 Oct 2025 11:41:02 -0400 Subject: [PATCH 12/32] Make `OpenAIRealtimeLLMService`'s websocket send method more resilient. Previously, it was possible for a websocket send attempt to occur during a disconnect. --- src/pipecat/services/openai/realtime/llm.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index 5e042cff3..3d64f6ccb 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -456,10 +456,14 @@ class OpenAIRealtimeLLMService(LLMService): async def _ws_send(self, realtime_message): try: - if self._websocket: + if not self._disconnecting and self._websocket: await self._websocket.send(json.dumps(realtime_message)) except Exception as e: - if self._disconnecting: + if self._disconnecting or not self._websocket: + # We're in the process of disconnecting. + # (If not self._websocket, that could indicate that we + # somehow *started* the websocket send attempt while we still + # had a connection) return logger.error(f"Error sending message to websocket: {e}") # In server-to-server contexts, a WebSocket error should be quite rare. Given how hard From 42d0a097c5d7f80a8ffeea3c88f64f00946c0cd4 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 21 Oct 2025 16:29:13 -0400 Subject: [PATCH 13/32] Tweaks to 20b example --- .../foundational/20b-persistent-context-openai-realtime.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/examples/foundational/20b-persistent-context-openai-realtime.py b/examples/foundational/20b-persistent-context-openai-realtime.py index 33328ac24..c0aa9bc95 100644 --- a/examples/foundational/20b-persistent-context-openai-realtime.py +++ b/examples/foundational/20b-persistent-context-openai-realtime.py @@ -73,7 +73,7 @@ async def save_conversation(params: FunctionCallParams): timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S") filename = f"{BASE_FILENAME}{timestamp}.json" logger.debug( - f"writing conversation to {filename}\n{json.dumps(params.context.messages, indent=4)}" + f"writing conversation to {filename}\n{json.dumps(params.context.get_messages(), indent=4)}" ) try: with open(filename, "w") as file: @@ -214,7 +214,7 @@ Remember, your responses should be short. Just one or two sentences, usually.""" llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames) llm.register_function("load_conversation", load_conversation) - context = LLMContext([], tools) + context = LLMContext([{"role": "user", "content": "Say hello!"}], tools) context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline( From b6a1886daef5d444760d7b2774a2a0c1edf5209c Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 21 Oct 2025 17:00:40 -0400 Subject: [PATCH 14/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). --- .../20b-persistent-context-openai-realtime.py | 6 +++++- src/pipecat/services/openai/realtime/llm.py | 19 ++++++++++++++----- 2 files changed, 19 insertions(+), 6 deletions(-) diff --git a/examples/foundational/20b-persistent-context-openai-realtime.py b/examples/foundational/20b-persistent-context-openai-realtime.py index c0aa9bc95..346a5b4bd 100644 --- a/examples/foundational/20b-persistent-context-openai-realtime.py +++ b/examples/foundational/20b-persistent-context-openai-realtime.py @@ -77,7 +77,7 @@ async def save_conversation(params: FunctionCallParams): ) try: with open(filename, "w") as file: - messages = params.context.get_messages_for_persistent_storage() + messages = params.context.get_messages() # remove the last message, which is the instruction we just gave to save the conversation messages.pop() json.dump(messages, file, indent=2) @@ -94,6 +94,10 @@ async def load_conversation(params: FunctionCallParams): with open(filename, "r") as file: params.context.set_messages(json.load(file)) await params.llm.reset_conversation() + # NOTE: we manually create a response here rather than relying + # on the function callback to trigger one since we've reset the + # conversation so the remote service doesn't know about the + # in-progress tool call. await params.llm._create_response() except Exception as e: await params.result_callback({"success": False, "error": str(e)}) diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index 3d64f6ccb..1abbce01f 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -401,9 +401,10 @@ class OpenAIRealtimeLLMService(LLMService): # Run the LLM at next opportunity await self._create_response() else: - # We got an updated context + # We got an updated context. + # This may contain a new user message or tool call result. self._context = context - # Send results for any newly-completed function calls + # Send results for newly-completed function calls, if any. await self._process_completed_function_calls(send_new_results=True) async def _handle_messages_append(self, frame): @@ -758,7 +759,11 @@ class OpenAIRealtimeLLMService(LLMService): """ logger.debug("Resetting conversation") await self._disconnect() + + # Prepare to setup server-side conversation from local context again self._llm_needs_conversation_setup = True + await self._process_completed_function_calls(send_new_results=False) + await self._connect() @traced_openai_realtime(operation="llm_request") @@ -771,6 +776,10 @@ class OpenAIRealtimeLLMService(LLMService): # Configure the LLM for this session if needed if self._llm_needs_conversation_setup: + logger.debug( + f"Setting up conversation on OpenAI Realtime LLM service with initial messages: {adapter.get_messages_for_logging(self._context)}" + ) + # Send initial messages llm_invocation_params = adapter.get_llm_invocation_params(self._context) messages = llm_invocation_params["messages"] @@ -785,7 +794,7 @@ class OpenAIRealtimeLLMService(LLMService): # 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)}") + logger.debug(f"Creating response") await self.push_frame(LLMFullResponseStartFrame()) await self.start_processing_metrics() @@ -809,8 +818,8 @@ class OpenAIRealtimeLLMService(LLMService): await self._send_tool_result(tool_call_id, message.get("content")) self._completed_tool_calls.add(tool_call_id) - # If we sent any new tool call results to the service, trigger another - # response + # If we reported any new tool call results to the service, trigger + # another response if sent_new_result: await self._create_response() From 6140fdb2c98b3b65dc526ed004a3d4d58072e013 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 21 Oct 2025 17:40:59 -0400 Subject: [PATCH 15/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). In anticipation of `messages` property being added to `LLMContext` (in another PR), remove warnings about the need to use `get_messages()` instead. --- src/pipecat/services/openai/realtime/context.py | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/src/pipecat/services/openai/realtime/context.py b/src/pipecat/services/openai/realtime/context.py index 96a714565..b5d68b8b4 100644 --- a/src/pipecat/services/openai/realtime/context.py +++ b/src/pipecat/services/openai/realtime/context.py @@ -27,9 +27,6 @@ context: OpenAIRealtimeLLMContext # or context: OpenAILLMContext - - # Reading messages from context - messages = context.messages ``` AFTER: @@ -46,9 +43,6 @@ # Context type context: LLMContext - - # Reading messages from context - messages = context.get_messages() ``` """ @@ -75,8 +69,6 @@ with warnings.catch_warnings(): "context: OpenAIRealtimeLLMContext\n" "# or\n" "context: OpenAILLMContext\n\n" - "# Reading messages from context\n" - "messages = context.messages\n" "```\n\n" "AFTER:\n" "```\n" @@ -89,8 +81,6 @@ with warnings.catch_warnings(): "frame: LLMContextFrame\n\n" "# Context type\n" "context: LLMContext\n\n" - "# Reading messages from context\n" - "messages = context.get_messages()\n" "```\n", ) From 9bc02afd0d82bbabc73e069742d66e27a91ed8f5 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 23 Oct 2025 09:34:06 -0400 Subject: [PATCH 16/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). CHANGELOG tweak. --- CHANGELOG.md | 6 ------ 1 file changed, 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a152022f1..2449276ab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -63,9 +63,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 # or context: OpenAILLMContext - # Reading messages from context - messages = context.messages - ## AFTER: # Context aggregator type @@ -76,9 +73,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 # Context type context: LLMContext - - # Reading messages from context - messages = context.get_messages() ``` Also note that `RealtimeMessagesUpdateFrame` and From 0495de52b60ec7c8e22e711a21b8e9a5e251b120 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 23 Oct 2025 09:57:44 -0400 Subject: [PATCH 17/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). Log warning about transcription frame direction change. --- CHANGELOG.md | 6 +++--- src/pipecat/services/openai/realtime/llm.py | 22 +++++++++++++++++++++ 2 files changed, 25 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2449276ab..4a85fc19d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,9 +21,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `LLMContext`, it is not meant to be swapped out for another LLM service at runtime.) - Note: `TranscriptionFrame`s now go upstream from `OpenAIRealtimeLLMService`, - so if you're using `TranscriptProcessor`, say, you'll want to adjust - accordingly: + Note: `TranscriptionFrame`s and `InterimTranscriptionFrame`s now go upstream + from `OpenAIRealtimeLLMService`, so if you're using `TranscriptProcessor`, + say, you'll want to adjust accordingly: ```python pipeline = Pipeline( diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index 1abbce01f..36ec7930c 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -139,6 +139,28 @@ class OpenAIRealtimeLLMService(LLMService): stacklevel=2, ) + # Log warning about transcription frame direction change in 0.0.92 + logger.warning( + "As of version 0.0.92, TranscriptionFrames and InterimTranscriptionFrames " + "now go upstream from OpenAIRealtimeLLMService, so if you're using " + "TranscriptProcessor, say, you'll want to adjust accordingly:\n\n" + "pipeline = Pipeline(\n" + " [\n" + " transport.input(),\n" + " context_aggregator.user(),\n\n" + " # BEFORE\n" + " llm,\n" + " transcript.user(),\n\n" + " # AFTER\n" + " transcript.user(),\n" + " llm,\n\n" + " transport.output(),\n" + " transcript.assistant(),\n" + " context_aggregator.assistant(),\n" + " ]\n" + ")" + ) + full_url = f"{base_url}?model={model}" super().__init__(base_url=full_url, **kwargs) From e42cf78e79760d9ac6c48ce5cceb16e097ea6e6e Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 23 Oct 2025 10:02:45 -0400 Subject: [PATCH 18/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). Update deprecation versions. --- src/pipecat/services/openai/realtime/context.py | 2 +- src/pipecat/services/openai/realtime/frames.py | 2 +- src/pipecat/services/openai/realtime/llm.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/pipecat/services/openai/realtime/context.py b/src/pipecat/services/openai/realtime/context.py index b5d68b8b4..91c6e74d5 100644 --- a/src/pipecat/services/openai/realtime/context.py +++ b/src/pipecat/services/openai/realtime/context.py @@ -6,7 +6,7 @@ """OpenAI Realtime LLM context and aggregator implementations. -.. deprecated:: 0.0.91 +.. deprecated:: 0.0.92 OpenAI Realtime no longer uses types from this module under the hood. It now uses `LLMContext` and `LLMContextAggregatorPair`. Using the new patterns should allow you to not need types from this module. diff --git a/src/pipecat/services/openai/realtime/frames.py b/src/pipecat/services/openai/realtime/frames.py index 1f800af89..39cfd9757 100644 --- a/src/pipecat/services/openai/realtime/frames.py +++ b/src/pipecat/services/openai/realtime/frames.py @@ -6,7 +6,7 @@ """Custom frame types for OpenAI Realtime API integration. -.. deprecated:: 0.0.91 +.. deprecated:: 0.0.92 OpenAI Realtime no longer uses types from this module under the hood. It now works more like most LLM services in Pipecat, relying on updates to diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index 36ec7930c..bf5fe7679 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -121,7 +121,7 @@ class OpenAIRealtimeLLMService(LLMService): start_audio_paused: Whether to start with audio input paused. Defaults to False. send_transcription_frames: Whether to emit transcription frames. - .. deprecated:: 0.0.91 + .. deprecated:: 0.0.92 This parameter is deprecated and will be removed in a future version. Transcription frames are always sent. From df19011080e2e072cbe6d78d0ddc5236734f23ba Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 23 Oct 2025 11:17:00 -0400 Subject: [PATCH 19/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). Improve warning about transcription frame direction change. --- src/pipecat/services/openai/realtime/llm.py | 52 ++++++++++++--------- 1 file changed, 30 insertions(+), 22 deletions(-) diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index bf5fe7679..284d19a90 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -139,28 +139,6 @@ class OpenAIRealtimeLLMService(LLMService): stacklevel=2, ) - # Log warning about transcription frame direction change in 0.0.92 - logger.warning( - "As of version 0.0.92, TranscriptionFrames and InterimTranscriptionFrames " - "now go upstream from OpenAIRealtimeLLMService, so if you're using " - "TranscriptProcessor, say, you'll want to adjust accordingly:\n\n" - "pipeline = Pipeline(\n" - " [\n" - " transport.input(),\n" - " context_aggregator.user(),\n\n" - " # BEFORE\n" - " llm,\n" - " transcript.user(),\n\n" - " # AFTER\n" - " transcript.user(),\n" - " llm,\n\n" - " transport.output(),\n" - " transcript.assistant(),\n" - " context_aggregator.assistant(),\n" - " ]\n" - ")" - ) - full_url = f"{base_url}?model={model}" super().__init__(base_url=full_url, **kwargs) @@ -883,6 +861,36 @@ class OpenAIRealtimeLLMService(LLMService): the user and one for the assistant, encapsulated in an OpenAIContextAggregatorPair. """ + # Log warning about transcription frame direction change in 0.0.92. + # We're putting this warning here rather than in the constructor so + # that it shows up for folks who haven't updated their code at all + # since 0.0.92, gives them a way to acknowledge and dismiss the + # warning, and encourages adoption of a new preferred pattern. + logger.warning( + "As of version 0.0.92, TranscriptionFrames and InterimTranscriptionFrames " + "now go upstream from OpenAIRealtimeLLMService, so if you're using " + "TranscriptProcessor, say, you'll want to adjust accordingly:\n\n" + "pipeline = Pipeline(\n" + " [\n" + " transport.input(),\n" + " context_aggregator.user(),\n\n" + " # BEFORE\n" + " llm,\n" + " transcript.user(),\n\n" + " # AFTER\n" + " transcript.user(),\n" + " llm,\n\n" + " transport.output(),\n" + " transcript.assistant(),\n" + " context_aggregator.assistant(),\n" + " ]\n" + ")\n\n" + "Once you've done that (if needed), you can dismiss this warning " + "by updating to the new context-setup pattern:\n\n" + " context = LLMContext(messages, tools)\n" + " context_aggregator = LLMContextAggregatorPair(context)\n" + ) + context = LLMContext.from_openai_context(context) return LLMContextAggregatorPair( context, user_params=user_params, assistant_params=assistant_params From 95be1510ac47cec294882a5f191b96ebd1664ce4 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 23 Oct 2025 11:34:33 -0400 Subject: [PATCH 20/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). Improve `OpenAIRealtimeLLMAdapter.get_messages_for_logging()`. --- .../services/open_ai_realtime_adapter.py | 20 ++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/src/pipecat/adapters/services/open_ai_realtime_adapter.py b/src/pipecat/adapters/services/open_ai_realtime_adapter.py index 7a6fc4d02..3cdfefc86 100644 --- a/src/pipecat/adapters/services/open_ai_realtime_adapter.py +++ b/src/pipecat/adapters/services/open_ai_realtime_adapter.py @@ -75,7 +75,25 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): Returns: List of messages in a format ready for logging about OpenAI Realtime. """ - return self._from_universal_context_messages(self.get_messages(context)).messages + # NOTE: this is the same as in OpenAIAdapter, as that's what it was + # prior to a refactor. Worth noting that for OpenAI Realtime + # specifically, not everything handled here is necessarily supported + # (or supported yet). + msgs = [] + for message in self.get_messages(context): + 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 item["type"] == "input_audio": + item["input_audio"]["data"] = "..." + if "mime_type" in msg and msg["mime_type"].startswith("image/"): + msg["data"] = "..." + msgs.append(msg) + return msgs @dataclass class ConvertedMessages: From 75b3ea9c96dfd3233136dd5846bd2527eb845e21 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 23 Oct 2025 12:01:58 -0400 Subject: [PATCH 21/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). Fix tracing. --- src/pipecat/utils/tracing/service_decorators.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index cf1ba912c..3935a4afc 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -905,7 +905,9 @@ def traced_openai_realtime(operation: str) -> Callable: # Capture context messages being sent if hasattr(self, "_context") and self._context: try: - messages = self._context.get_messages_for_logging() + messages = self.get_llm_adapter().get_messages_for_logging( + self._context + ) if messages: operation_attrs["context_messages"] = json.dumps(messages) except Exception as e: From 8ac421f8fd45127e76350f394196e3e0d1011de8 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 23 Oct 2025 12:03:45 -0400 Subject: [PATCH 22/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). Remove unused imports. --- examples/foundational/19-openai-realtime.py | 1 - examples/foundational/19b-openai-realtime-text.py | 1 - .../foundational/20b-persistent-context-openai-realtime.py | 3 --- 3 files changed, 5 deletions(-) diff --git a/examples/foundational/19-openai-realtime.py b/examples/foundational/19-openai-realtime.py index 3bf12f916..e883322ef 100644 --- a/examples/foundational/19-openai-realtime.py +++ b/examples/foundational/19-openai-realtime.py @@ -21,7 +21,6 @@ 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 from pipecat.runner.utils import create_transport diff --git a/examples/foundational/19b-openai-realtime-text.py b/examples/foundational/19b-openai-realtime-text.py index fab029bb6..c1f33b7bf 100644 --- a/examples/foundational/19b-openai-realtime-text.py +++ b/examples/foundational/19b-openai-realtime-text.py @@ -20,7 +20,6 @@ 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 from pipecat.runner.utils import create_transport diff --git a/examples/foundational/20b-persistent-context-openai-realtime.py b/examples/foundational/20b-persistent-context-openai-realtime.py index 346a5b4bd..e3f018c16 100644 --- a/examples/foundational/20b-persistent-context-openai-realtime.py +++ b/examples/foundational/20b-persistent-context-openai-realtime.py @@ -22,9 +22,6 @@ 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.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.deepgram.stt import DeepgramSTTService From 15aa76efbaabbe7e760242522772eab70836161b Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 23 Oct 2025 12:49:48 -0400 Subject: [PATCH 23/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). Maintain backward compatibility with functions specified in dict format. --- src/pipecat/adapters/schemas/tools_schema.py | 3 +++ .../services/aws_nova_sonic_adapter.py | 18 ++++++++++++++++-- .../services/open_ai_realtime_adapter.py | 18 ++++++++++++++++-- .../processors/aggregators/llm_context.py | 12 ++++++++++-- 4 files changed, 45 insertions(+), 6 deletions(-) diff --git a/src/pipecat/adapters/schemas/tools_schema.py b/src/pipecat/adapters/schemas/tools_schema.py index 05710616d..d0d798569 100644 --- a/src/pipecat/adapters/schemas/tools_schema.py +++ b/src/pipecat/adapters/schemas/tools_schema.py @@ -22,9 +22,12 @@ class AdapterType(Enum): Parameters: GEMINI: Google Gemini adapter - currently the only service supporting custom tools. + SHIM: Backward compatibility shim for creating ToolsSchemas from lists of tools in + any format, used by LLMContext.from_openai_context. """ GEMINI = "gemini" # that is the only service where we are able to add custom tools for now + SHIM = "shim" # for use as backward compatibility shim for creating ToolsSchemas from list of tools in any format class ToolsSchema: diff --git a/src/pipecat/adapters/services/aws_nova_sonic_adapter.py b/src/pipecat/adapters/services/aws_nova_sonic_adapter.py index 60f12798b..dcc42ba68 100644 --- a/src/pipecat/adapters/services/aws_nova_sonic_adapter.py +++ b/src/pipecat/adapters/services/aws_nova_sonic_adapter.py @@ -16,7 +16,7 @@ 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.adapters.schemas.tools_schema import AdapterType, ToolsSchema from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextMessage @@ -210,4 +210,18 @@ class AWSNovaSonicLLMAdapter(BaseLLMAdapter[AWSNovaSonicLLMInvocationParams]): List of dictionaries in AWS Nova Sonic function format. """ functions_schema = tools_schema.standard_tools - return [self._to_aws_nova_sonic_function_format(func) for func in functions_schema] + standard_tools = [ + self._to_aws_nova_sonic_function_format(func) for func in functions_schema + ] + + # For backward compatibility, AWS Nova Sonic can still be used with + # tools in dict format, even though it always uses `LLMContext` under + # the hood (via `LLMContext.from_openai_context()`). + # To support this behavior, we use "shimmed" custom tools here. + # (We maintain this backward compatibility because users aren't + # *knowingly* opting into the new `LLMContext`.) + shimmed_tools = [] + if tools_schema.custom_tools: + shimmed_tools = tools_schema.custom_tools.get(AdapterType.SHIM, []) + + return standard_tools + shimmed_tools diff --git a/src/pipecat/adapters/services/open_ai_realtime_adapter.py b/src/pipecat/adapters/services/open_ai_realtime_adapter.py index 3cdfefc86..3d3650633 100644 --- a/src/pipecat/adapters/services/open_ai_realtime_adapter.py +++ b/src/pipecat/adapters/services/open_ai_realtime_adapter.py @@ -15,7 +15,7 @@ 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.adapters.schemas.tools_schema import AdapterType, ToolsSchema from pipecat.processors.aggregators.llm_context import LLMContext, LLMContextMessage from pipecat.services.openai.realtime import events @@ -225,4 +225,18 @@ class OpenAIRealtimeLLMAdapter(BaseLLMAdapter): List of function definitions in OpenAI Realtime format. """ functions_schema = tools_schema.standard_tools - return [self._to_openai_realtime_function_format(func) for func in functions_schema] + standard_tools = [ + self._to_openai_realtime_function_format(func) for func in functions_schema + ] + + # For backward compatibility, OpenAI Realtime can still be used with + # tools in dict format, even though it always uses `LLMContext` under + # the hood (via `LLMContext.from_openai_context()`). + # To support this behavior, we use "shimmed" custom tools here. + # (We maintain this backward compatibility because users aren't + # *knowingly* opting into the new `LLMContext`.) + shimmed_tools = [] + if tools_schema.custom_tools: + shimmed_tools = tools_schema.custom_tools.get(AdapterType.SHIM, []) + + return standard_tools + shimmed_tools diff --git a/src/pipecat/processors/aggregators/llm_context.py b/src/pipecat/processors/aggregators/llm_context.py index 913566909..61241729e 100644 --- a/src/pipecat/processors/aggregators/llm_context.py +++ b/src/pipecat/processors/aggregators/llm_context.py @@ -29,7 +29,7 @@ from openai.types.chat import ( ) from PIL import Image -from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema from pipecat.frames.frames import AudioRawFrame if TYPE_CHECKING: @@ -83,9 +83,17 @@ class LLMContext: Returns: New LLMContext instance with converted messages and settings. """ + # Convert tools to ToolsSchema if needed. + # If the tools are already a ToolsSchema, this is a no-op. + # Otherwise, we wrap them in a shim ToolsSchema. + converted_tools = openai_context.tools + if isinstance(converted_tools, list): + converted_tools = ToolsSchema( + standard_tools=[], custom_tools={AdapterType.SHIM: converted_tools} + ) return LLMContext( messages=openai_context.get_messages(), - tools=openai_context.tools, + tools=converted_tools, tool_choice=openai_context.tool_choice, ) From 8c03df1463a537e00454ea48321c5f9124dfef3e Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 23 Oct 2025 14:32:32 -0400 Subject: [PATCH 24/32] Update some docstring arg descriptions to be a bit more current or accurate --- src/pipecat/services/azure/realtime/llm.py | 2 +- src/pipecat/services/openai/realtime/llm.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/pipecat/services/azure/realtime/llm.py b/src/pipecat/services/azure/realtime/llm.py index 1193b82d4..66ba95eea 100644 --- a/src/pipecat/services/azure/realtime/llm.py +++ b/src/pipecat/services/azure/realtime/llm.py @@ -38,7 +38,7 @@ class AzureRealtimeLLMService(OpenAIRealtimeLLMService): Args: api_key: The API key for the Azure OpenAI service. base_url: The full Azure WebSocket endpoint URL including api-version and deployment. - Example: "wss://my-project.openai.azure.com/openai/realtime?api-version=2024-10-01-preview&deployment=my-realtime-deployment" + Example: "wss://my-project.openai.azure.com/openai/realtime?api-version=2025-04-01-preview&deployment=my-realtime-deployment" **kwargs: Additional arguments passed to parent OpenAIRealtimeLLMService. """ super().__init__(base_url=base_url, api_key=api_key, **kwargs) diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index 284d19a90..213c27df7 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -113,7 +113,7 @@ class OpenAIRealtimeLLMService(LLMService): Args: api_key: OpenAI API key for authentication. - model: OpenAI model name. Defaults to "gpt-4o-realtime-preview-2025-06-03". + model: OpenAI model name. Defaults to "gpt-realtime". base_url: WebSocket base URL for the realtime API. Defaults to "wss://api.openai.com/v1/realtime". session_properties: Configuration properties for the realtime session. From 917ea273525a229ce29ac25de340a405b2a9fb55 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 23 Oct 2025 14:36:22 -0400 Subject: [PATCH 25/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). Update `AzureRealtimeLLMService` example (19a) to use new `LLMContext` pattern. --- examples/foundational/19a-azure-realtime.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/examples/foundational/19a-azure-realtime.py b/examples/foundational/19a-azure-realtime.py index c4b0fc02a..7b07985be 100644 --- a/examples/foundational/19a-azure-realtime.py +++ b/examples/foundational/19a-azure-realtime.py @@ -18,7 +18,8 @@ from pipecat.frames.frames import LLMRunFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.azure.realtime.llm import AzureRealtimeLLMService @@ -155,10 +156,10 @@ Remember, your responses should be short. Just one or two sentences, usually. Re llm.register_function("get_current_weather", fetch_weather_from_api) llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) - # Create a standard OpenAI LLM context object using the normal messages format. The + # Create a standard LLM context object using the normal messages format. The # OpenAIRealtimeBetaLLMService will convert this internally to messages that the # openai WebSocket API can understand. - context = OpenAILLMContext( + context = LLMContext( [{"role": "user", "content": "Say hello!"}], # [{"role": "user", "content": [{"type": "text", "text": "Say hello!"}]}], # [ @@ -173,7 +174,7 @@ Remember, your responses should be short. Just one or two sentences, usually. Re tools, ) - context_aggregator = llm.create_context_aggregator(context) + context_aggregator = LLMContextAggregatorPair(context) pipeline = Pipeline( [ From 0282033208c5099dbe79442b8fd8a07f3ed61773 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 23 Oct 2025 15:08:58 -0400 Subject: [PATCH 26/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). Add `LLMContext.get_messages_for_persistent_storage()` for compatibility with `OpenAILLMContext`, to avoid tripping up users who we're unknowingly migrating to using `LLMContext`. --- .../processors/aggregators/llm_context.py | 28 ++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/src/pipecat/processors/aggregators/llm_context.py b/src/pipecat/processors/aggregators/llm_context.py index 61241729e..8dc79fb50 100644 --- a/src/pipecat/processors/aggregators/llm_context.py +++ b/src/pipecat/processors/aggregators/llm_context.py @@ -15,7 +15,6 @@ service-specific adapter. """ import base64 -import copy import io from dataclasses import dataclass from typing import TYPE_CHECKING, Any, List, Optional, TypeAlias, Union @@ -127,6 +126,33 @@ class LLMContext: """ return self.get_messages() + def get_messages_for_persistent_storage(self) -> List[LLMContextMessage]: + """Get messages suitable for persistent storage. + + NOTE: the only reason this method exists is because we're "silently" + switching from OpenAILLMContext to LLMContext under the hood in some + services and don't want to trip up users who may have been relying on + this method, which is part of the public API of OpenAILLMContext but + doesn't need to be for LLMContext. + + .. deprecated:: + Use `get_messages()` instead. + + Returns: + List of conversation messages. + """ + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "get_messages_for_persistent_storage() is deprecated, use get_messages() instead.", + DeprecationWarning, + stacklevel=2, + ) + + return self.get_messages() + def get_messages(self, llm_specific_filter: Optional[str] = None) -> List[LLMContextMessage]: """Get the current messages list. From 1f96cdf9707f1d690f83e1fe39e5a5f84144e080 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 23 Oct 2025 16:05:41 -0400 Subject: [PATCH 27/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Make `LLMUserAggregator` push the `LLMSetToolsFrame`s, in case a speech-to-speech service that needs to handle the frame itself—like `OpenAIRealtimeLLMService`—is downstream. As far as I can tell, pushing `LLMSetToolsFrame` should otherwise have no unwanted side effects. --- examples/foundational/19-openai-realtime.py | 37 ++++++++++++++++--- .../aggregators/llm_response_universal.py | 6 +++ 2 files changed, 38 insertions(+), 5 deletions(-) diff --git a/examples/foundational/19-openai-realtime.py b/examples/foundational/19-openai-realtime.py index e883322ef..5f215a07b 100644 --- a/examples/foundational/19-openai-realtime.py +++ b/examples/foundational/19-openai-realtime.py @@ -5,6 +5,7 @@ # +import asyncio import os from datetime import datetime @@ -14,7 +15,7 @@ from loguru import logger from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import LLMRunFrame, TranscriptionMessage +from pipecat.frames.frames import LLMRunFrame, LLMSetToolsFrame, TranscriptionMessage from pipecat.observers.loggers.transcription_log_observer import TranscriptionLogObserver from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner @@ -53,6 +54,18 @@ async def fetch_weather_from_api(params: FunctionCallParams): ) +async def get_news(params: FunctionCallParams): + await params.result_callback( + { + "news": [ + "Massive UFO currently hovering above New York City", + "Stock markets reach all-time highs", + "Living dinosaur species discovered in the Amazon rainforest", + ], + } + ) + + async def fetch_restaurant_recommendation(params: FunctionCallParams): await params.result_callback({"name": "The Golden Dragon"}) @@ -74,6 +87,13 @@ weather_function = FunctionSchema( required=["location", "format"], ) +get_news_function = FunctionSchema( + name="get_news", + description="Get the current news.", + properties={}, + required=[], +) + restaurant_function = FunctionSchema( name="get_restaurant_recommendation", description="Get a restaurant recommendation", @@ -141,10 +161,6 @@ even if you're asked about them. You are participating in a voice conversation. Keep your responses concise, short, and to the point unless specifically asked to elaborate on a topic. -You have access to the following tools: -- get_current_weather: Get the current weather for a given location. -- get_restaurant_recommendation: Get a restaurant recommendation for a given location. - Remember, your responses should be short. Just one or two sentences, usually. Respond in English.""", ) @@ -158,6 +174,7 @@ Remember, your responses should be short. Just one or two sentences, usually. Re # llm.register_function(None, fetch_weather_from_api) llm.register_function("get_current_weather", fetch_weather_from_api) llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) + llm.register_function("get_news", get_news) transcript = TranscriptProcessor() @@ -199,6 +216,16 @@ Remember, your responses should be short. Just one or two sentences, usually. Re # Kick off the conversation. await task.queue_frames([LLMRunFrame()]) + async def set_tools_after_delay(): + await asyncio.sleep(15) + new_tools = ToolsSchema( + standard_tools=[weather_function, restaurant_function, get_news_function] + ) + logger.info("Registering new tool with LLMSetToolsFrame") + await task.queue_frames([LLMSetToolsFrame(tools=new_tools)]) + + asyncio.create_task(set_tools_after_delay()) + @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client): logger.info(f"Client disconnected") diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 69a8dd280..9f1e04fe0 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -290,6 +290,12 @@ class LLMUserAggregator(LLMContextAggregator): await self._handle_llm_messages_update(frame) elif isinstance(frame, LLMSetToolsFrame): self.set_tools(frame.tools) + # Push the LLMSetToolsFrame as well, since speech-to-speech LLM + # services (like OpenAI Realtime) may need to know about tool + # changes; unlike text-based LLM services they won't just "pick up + # the change" on the next LLM run, as the LLM is continuously + # running. + await self.push_frame(frame, direction) elif isinstance(frame, LLMSetToolChoiceFrame): self.set_tool_choice(frame.tool_choice) elif isinstance(frame, SpeechControlParamsFrame): From 8894db429026742a077fd4cc1ec7f0237c142090 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 23 Oct 2025 16:39:13 -0400 Subject: [PATCH 28/32] Update `OpenAIRealtimeLLMService` to work with `LLMContext` and `LLMContextAggregatorPair` (cont'd). Add warning about no longer pushing `TTSTextFrame`s. --- CHANGELOG.md | 4 ++++ src/pipecat/services/openai/realtime/llm.py | 7 +++++-- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4a85fc19d..f208bf1d9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -82,6 +82,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 by context aggregators, to update its internal state. Listen for `LLMContextFrame`s for context updates. + Finally, `LLMTextFrame`s are no longer pushed from `OpenAIRealtimeLLMService` + when it's configured with `output_modalities=['audio']`. If you need + to process its output, listen for `TTSTextFrame`s instead. + - Expanded support for universal `LLMContext` to `GeminiLiveLLMService`. As a reminder, the context-setup pattern when using `LLMContext` is: diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index 213c27df7..c46f3435d 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -885,8 +885,11 @@ class OpenAIRealtimeLLMService(LLMService): " context_aggregator.assistant(),\n" " ]\n" ")\n\n" - "Once you've done that (if needed), you can dismiss this warning " - "by updating to the new context-setup pattern:\n\n" + "Also, LLMTextFrames are no longer pushed from " + "OpenAIRealtimeLLMService when it's configured with " + "output_modalities=['audio']. Listen for TTSTextFrames instead.\n\n" + "Once you've made the appropriate changes (if needed), you can" + "dismiss this warning by updating to the new context-setup pattern:\n\n" " context = LLMContext(messages, tools)\n" " context_aggregator = LLMContextAggregatorPair(context)\n" ) From d0f52feba3a36f7ecf9c85dd200bc00594230ed1 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 29 Oct 2025 16:15:16 -0400 Subject: [PATCH 29/32] OpenAI Realtime needs the assistant context aggregator to have `expect_stripped_words=False` --- CHANGELOG.md | 8 +++++++- examples/foundational/19-openai-realtime.py | 8 +++++++- examples/foundational/19a-azure-realtime.py | 8 +++++++- src/pipecat/services/openai/realtime/llm.py | 1 + 4 files changed, 22 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f208bf1d9..6ff37d47c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,7 +14,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ```python context = LLMContext(messages, tools) - context_aggregator = LLMContextAggregatorPair(context) + context_aggregator = LLMContextAggregatorPair( + context, + # This part is `OpenAIRealtimeLLMService`-specific. + # `expect_stripped_words=False` needed when OpenAI Realtime used with + # "audio" modality (the default). + assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), + ) ``` (Note that even though `OpenAIRealtimeLLMService` now supports the universal diff --git a/examples/foundational/19-openai-realtime.py b/examples/foundational/19-openai-realtime.py index 5f215a07b..3451c9da8 100644 --- a/examples/foundational/19-openai-realtime.py +++ b/examples/foundational/19-openai-realtime.py @@ -21,6 +21,7 @@ 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 import LLMAssistantAggregatorParams from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.transcript_processor import TranscriptProcessor from pipecat.runner.types import RunnerArguments @@ -186,7 +187,12 @@ Remember, your responses should be short. Just one or two sentences, usually. Re tools, ) - context_aggregator = LLMContextAggregatorPair(context) + context_aggregator = LLMContextAggregatorPair( + context, + # `expect_stripped_words=False` needed when OpenAI Realtime used with + # "audio" modality (the default) + assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), + ) pipeline = Pipeline( [ diff --git a/examples/foundational/19a-azure-realtime.py b/examples/foundational/19a-azure-realtime.py index 7b07985be..7d9cf1b4b 100644 --- a/examples/foundational/19a-azure-realtime.py +++ b/examples/foundational/19a-azure-realtime.py @@ -19,6 +19,7 @@ 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 import LLMAssistantAggregatorParams from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport @@ -174,7 +175,12 @@ Remember, your responses should be short. Just one or two sentences, usually. Re tools, ) - context_aggregator = LLMContextAggregatorPair(context) + context_aggregator = LLMContextAggregatorPair( + context, + # `expect_stripped_words=False` needed when OpenAI Realtime used with + # "audio" modality (the default) + assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False), + ) pipeline = Pipeline( [ diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index c46f3435d..f9e8b1a3f 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -895,6 +895,7 @@ class OpenAIRealtimeLLMService(LLMService): ) context = LLMContext.from_openai_context(context) + assistant_params.expect_stripped_words = False return LLMContextAggregatorPair( context, user_params=user_params, assistant_params=assistant_params ) From ddac24e6c9a44c8c658783fa2805415ea0e8e2d1 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 29 Oct 2025 16:17:05 -0400 Subject: [PATCH 30/32] Fix a missing space in a warning message --- src/pipecat/services/openai/realtime/llm.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index f9e8b1a3f..012604eb8 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -888,7 +888,7 @@ class OpenAIRealtimeLLMService(LLMService): "Also, LLMTextFrames are no longer pushed from " "OpenAIRealtimeLLMService when it's configured with " "output_modalities=['audio']. Listen for TTSTextFrames instead.\n\n" - "Once you've made the appropriate changes (if needed), you can" + "Once you've made the appropriate changes (if needed), you can " "dismiss this warning by updating to the new context-setup pattern:\n\n" " context = LLMContext(messages, tools)\n" " context_aggregator = LLMContextAggregatorPair(context)\n" From 89e9acf0e1b5673e00764b88dbd28535fd9876bb Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 29 Oct 2025 16:21:04 -0400 Subject: [PATCH 31/32] CHANGELOG and code comment tweaks --- CHANGELOG.md | 6 +++--- examples/foundational/19-openai-realtime.py | 1 + 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6ff37d47c..5ab7c7121 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,7 +25,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 (Note that even though `OpenAIRealtimeLLMService` now supports the universal `LLMContext`, it is not meant to be swapped out for another LLM service at - runtime.) + runtime with `LLMSwitcher`.) Note: `TranscriptionFrame`s and `InterimTranscriptionFrame`s now go upstream from `OpenAIRealtimeLLMService`, so if you're using `TranscriptProcessor`, @@ -108,7 +108,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 (Note that even though `GeminiLiveLLMService` now supports the universal `LLMContext`, it is not meant to be swapped out for another LLM service at - runtime.) + runtime with `LLMSwitcher`.) Worth noting: whether or not you use the new context-setup pattern with `GeminiLiveLLMService`, some types have changed under the hood: @@ -212,7 +212,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 (Note that even though `AWSNovaSonicLLMService` now supports the universal `LLMContext`, it is not meant to be swapped out for another LLM service at - runtime.) + runtime with `LLMSwitcher`.) Worth noting: whether or not you use the new context-setup pattern with `AWSNovaSonicLLMService`, some types have changed under the hood: diff --git a/examples/foundational/19-openai-realtime.py b/examples/foundational/19-openai-realtime.py index 3451c9da8..4604fbd02 100644 --- a/examples/foundational/19-openai-realtime.py +++ b/examples/foundational/19-openai-realtime.py @@ -222,6 +222,7 @@ Remember, your responses should be short. Just one or two sentences, usually. Re # Kick off the conversation. await task.queue_frames([LLMRunFrame()]) + # Add a new tool at runtime after a delay. async def set_tools_after_delay(): await asyncio.sleep(15) new_tools = ToolsSchema( From 8f15980c674f69d1349936b085b365b8f0a815b6 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 29 Oct 2025 16:23:50 -0400 Subject: [PATCH 32/32] Get rid of unnecessary new task in example file --- examples/foundational/19-openai-realtime.py | 14 +++++--------- 1 file changed, 5 insertions(+), 9 deletions(-) diff --git a/examples/foundational/19-openai-realtime.py b/examples/foundational/19-openai-realtime.py index 4604fbd02..6907ec196 100644 --- a/examples/foundational/19-openai-realtime.py +++ b/examples/foundational/19-openai-realtime.py @@ -223,15 +223,11 @@ Remember, your responses should be short. Just one or two sentences, usually. Re await task.queue_frames([LLMRunFrame()]) # Add a new tool at runtime after a delay. - async def set_tools_after_delay(): - await asyncio.sleep(15) - new_tools = ToolsSchema( - standard_tools=[weather_function, restaurant_function, get_news_function] - ) - logger.info("Registering new tool with LLMSetToolsFrame") - await task.queue_frames([LLMSetToolsFrame(tools=new_tools)]) - - asyncio.create_task(set_tools_after_delay()) + await asyncio.sleep(15) + new_tools = ToolsSchema( + standard_tools=[weather_function, restaurant_function, get_news_function] + ) + await task.queue_frames([LLMSetToolsFrame(tools=new_tools)]) @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client):