Merge pull request #2889 from pipecat-ai/pk/openai-realtime-universal-llmcontext-2

Support new `LLMContext` pattern with `OpenAIRealtimeLLMService`

CHANGELOG.md

@@ -9,6 +9,89 @@ 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,
+    # 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
+  `LLMContext`, it is not meant to be swapped out for another LLM service at
+  runtime with `LLMSwitcher`.)
+
+  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(
+    [
+      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
+
+  ## AFTER:
+
+  # Context aggregator type
+  context_aggregator: LLMContextAggregatorPair
+
+  # Context frame type
+  frame: LLMContextFrame
+
+  # Context type
+  context: LLMContext
+  ```
+
+  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.
+
+  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:
 
@@ -25,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:
@@ -79,6 +162,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.
 
@@ -121,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:
@@ -200,8 +291,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
+- Types in `pipecat.services.aws.nova_sonic.context` are deprecated, as they're
-  to changes to support `LLMContext`. See "Changed" section for details.
+  no longer used by `AWSNovaSonicLLMService`. See "Added" section for
+  details.
 
 ### Fixed
 
@@ -5375,4 +5467,4 @@ a bit.
 
 ## [0.0.2] - 2024-03-12
 
-Initial public release.
+Initial public release.

examples/foundational/19-openai-realtime.py

@@ -5,6 +5,7 @@
 #
 
 
+import asyncio
 import os
 from datetime import datetime
 
@@ -14,12 +15,14 @@ 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
 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 import LLMAssistantAggregatorParams
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.transcript_processor import TranscriptProcessor
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
@@ -52,6 +55,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"})
 
@@ -73,6 +88,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",
@@ -140,10 +162,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.""",
     )
 
@@ -157,25 +175,31 @@ 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()
 
     # 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,
+        # `expect_stripped_words=False` needed when OpenAI Realtime used with
+        # "audio" modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     pipeline = Pipeline(
         [
             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(),
@@ -198,6 +222,13 @@ 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.
+        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):
         logger.info(f"Client disconnected")

examples/foundational/19a-azure-realtime.py

@@ -18,7 +18,9 @@ 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 import LLMAssistantAggregatorParams
+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 +157,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 +175,12 @@ 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,
+        # `expect_stripped_words=False` needed when OpenAI Realtime used with
+        # "audio" modality (the default)
+        assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=False),
+    )
 
     pipeline = Pipeline(
         [

examples/foundational/19b-openai-realtime-text.py

@@ -18,7 +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.openai_llm_context import OpenAILLMContext
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.transcript_processor import TranscriptProcessor
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
@@ -169,20 +170,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(),

examples/foundational/20b-persistent-context-openai-realtime.py

@@ -13,14 +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.openai_llm_context import (
+from pipecat.processors.aggregators.llm_context import LLMContext
-    OpenAILLMContext,
+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.deepgram.stt import DeepgramSTTService
@@ -69,11 +70,11 @@ 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:
-            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)
@@ -90,6 +91,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)})
@@ -97,14 +102,12 @@ async def load_conversation(params: FunctionCallParams):
     asyncio.create_task(_reset())
 
 
-tools = [
+tools = ToolsSchema(
-    {
+    standard_tools=[
-        "type": "function",
+        FunctionSchema(
-        "name": "get_current_weather",
+            name="get_current_weather",
-        "description": "Get the current weather",
+            description="Get the current weather",
-        "parameters": {
+            properties={
-            "type": "object",
-            "properties": {
                 "location": {
                     "type": "string",
                     "description": "The city and state, e.g. San Francisco, CA",
@@ -115,45 +118,33 @@ tools = [
                     "description": "The temperature unit to use. Infer this from the users location.",
                 },
             },
-            "required": ["location", "format"],
+            required=["location", "format"],
-        },
+        ),
-    },
+        FunctionSchema(
-    {
+            name="save_conversation",
-        "type": "function",
+            description="Save the current conversatione. Use this function to persist the current conversation to external storage.",
-        "name": "save_conversation",
+            properties={},
-        "description": "Save the current conversatione. Use this function to persist the current conversation to external storage.",
+            required=[],
-        "parameters": {
+        ),
-            "type": "object",
+        FunctionSchema(
-            "properties": {},
+            name="get_saved_conversation_filenames",
-            "required": [],
+            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=[],
-    {
+        ),
-        "type": "function",
+        FunctionSchema(
-        "name": "get_saved_conversation_filenames",
+            name="load_conversation",
-        "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.",
+            description="Load a conversation history. Use this function to load a conversation history into the current session.",
-        "parameters": {
+            properties={
-            "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": {
                 "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 +215,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 = LLMContext([{"role": "user", "content": "Say hello!"}], tools)
-    context_aggregator = llm.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(context)
 
     pipeline = Pipeline(
         [

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:

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

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.adapters.schemas.tools_schema import AdapterType, 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,124 @@ 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.")
+        # 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:
+        """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=[])
+
+        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."""
+
+        return 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]:
@@ -94,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

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
@@ -29,7 +28,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 +82,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,
         )
 
@@ -119,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.
 

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):

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)

src/pipecat/services/openai/realtime/context.py

@@ -4,7 +4,85 @@
 # 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
+    ```
+
+    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
+    ```
+"""
+
+import warnings
+
+with warnings.catch_warnings():
+    warnings.simplefilter("always")
+    warnings.warn(
+        "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"
+        "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"
+        "```\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"
+        "```\n",
+    )
 
 import copy
 import json

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

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,10 +43,12 @@ 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,
 )
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
 from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContext,
     OpenAILLMContextFrame,
@@ -57,12 +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,
-    OpenAIRealtimeLLMContext,
-    OpenAIRealtimeUserContextAggregator,
-)
-from .frames import RealtimeFunctionCallResultFrame, RealtimeMessagesUpdateFrame
 
 try:
     from websockets.asyncio.client import connect as websocket_connect
@@ -108,22 +106,39 @@ 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.
 
         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.
                 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)
 
@@ -135,10 +150,11 @@ 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
-        self._context = None
+        self._context: LLMContext = None
+
+        self._llm_needs_conversation_setup = True
 
         self._disconnecting = False
         self._api_session_ready = False
@@ -148,8 +164,8 @@ 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._completed_tool_calls = set()
 
         self._register_event_handler("on_conversation_item_created")
         self._register_event_handler("on_conversation_item_updated")
@@ -347,22 +363,13 @@ class OpenAIRealtimeLLMService(LLMService):
 
         if isinstance(frame, TranscriptionFrame):
             pass
-        elif isinstance(frame, OpenAILLMContextFrame):
+        elif isinstance(frame, (LLMContextFrame, OpenAILLMContextFrame)):
-            context: OpenAIRealtimeLLMContext = OpenAIRealtimeLLMContext.upgrade_to_realtime(
+            context = (
                 frame.context
+                if isinstance(frame, LLMContextFrame)
+                else LLMContext.from_openai_context(frame.context)
             )
-            if not self._context:
+            await self._handle_context(context)
-                self._context = context
-            elif frame.context is not self._context:
-                # If the context has changed, reset the conversation
-                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)
@@ -376,29 +383,33 @@ class OpenAIRealtimeLLMService(LLMService):
             await self._handle_bot_stopped_speaking()
         elif isinstance(frame, LLMMessagesAppendFrame):
             await self._handle_messages_append(frame)
-        elif isinstance(frame, RealtimeMessagesUpdateFrame):
-            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.
+            # This may contain a new user message or tool call result.
+            self._context = context
+            # 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):
         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
     #
@@ -439,16 +450,21 @@ 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}")
 
     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
@@ -459,13 +475,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:
+        if self._context:
-            settings.tools = self._context.tools
+            adapter: OpenAIRealtimeLLMAdapter = self.get_llm_adapter()
-        # instructions in the context come from an initial "system" message in the
+            llm_invocation_params = adapter.get_llm_invocation_params(self._context)
-        # messages list, and override instructions in the session properties
+
-        if self._context and self._context._session_instructions:
+            # tools given in the context override the tools in the session properties
-            settings.instructions = self._context._session_instructions
+            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))
 
     #
@@ -571,12 +594,7 @@ class OpenAIRealtimeLLMService(LLMService):
             del self._messages_added_manually[evt.item.id]
             return
 
-        if evt.item.role == "user":
+        if evt.item.role == "assistant":
-            # 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":
             self._current_assistant_response = evt.item
             await self.push_frame(LLMFullResponseStartFrame())
 
@@ -587,11 +605,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(
-            await self.push_frame(
+            # no way to get a language code?
-                # no way to get a language code?
+            InterimTranscriptionFrame(evt.delta, "", time_now_iso8601(), result=evt),
-                InterimTranscriptionFrame(evt.delta, "", time_now_iso8601(), result=evt)
+            direction=FrameDirection.UPSTREAM,
-            )
+        )
 
     @traced_stt
     async def _handle_user_transcription(
@@ -608,22 +626,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(
-            await self.push_frame(
+            # no way to get a language code?
-                # no way to get a language code?
+            TranscriptionFrame(evt.transcript, "", time_now_iso8601(), result=evt),
-                TranscriptionFrame(evt.transcript, "", time_now_iso8601(), result=evt)
+            FrameDirection.UPSTREAM,
-            )
+        )
-            await self._handle_user_transcription(evt.transcript, True, Language.EN)
+        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)
@@ -653,26 +661,17 @@ 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
+        # 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):
@@ -760,9 +759,11 @@ class OpenAIRealtimeLLMService(LLMService):
         """
         logger.debug("Resetting conversation")
         await self._disconnect()
-        if self._context:
+
-            self._context.llm_needs_settings_update = True
+        # Prepare to setup server-side conversation from local context again
-            self._context.llm_needs_initial_messages = True
+        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,19 +772,29 @@ class OpenAIRealtimeLLMService(LLMService):
             self._run_llm_when_api_session_ready = True
             return
 
-        if self._context.llm_needs_initial_messages:
+        adapter: OpenAIRealtimeLLMAdapter = self.get_llm_adapter()
-            messages = self._context.get_messages_for_initializing_history()
+
+        # 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"]
             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")
 
         await self.push_frame(LLMFullResponseStartFrame())
         await self.start_processing_metrics()
@@ -794,19 +805,50 @@ 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 reported 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,
         *,
         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:
@@ -819,11 +861,41 @@ class OpenAIRealtimeLLMService(LLMService):
             the user and one for the assistant, encapsulated in an
             OpenAIContextAggregatorPair.
         """
-        context.set_llm_adapter(self.get_llm_adapter())
+        # Log warning about transcription frame direction change in 0.0.92.
-
+        # We're putting this warning here rather than in the constructor so
-        OpenAIRealtimeLLMContext.upgrade_to_realtime(context)
+        # that it shows up for folks who haven't updated their code at all
-        user = OpenAIRealtimeUserContextAggregator(context, params=user_params)
+        # 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"
+            "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"
+        )
 
+        context = LLMContext.from_openai_context(context)
         assistant_params.expect_stripped_words = False
-        assistant = OpenAIRealtimeAssistantContextAggregator(context, params=assistant_params)
+        return LLMContextAggregatorPair(
-        return OpenAIContextAggregatorPair(_user=user, _assistant=assistant)
+            context, user_params=user_params, assistant_params=assistant_params
+        )

src/pipecat/services/openai_realtime/context.py

@@ -4,18 +4,15 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-"""OpenAI Realtime LLM context and aggregator implementations."""
+"""OpenAI Realtime LLM context and aggregator implementations.
 
-import warnings
+.. 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 *
-
-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,
-    )

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: