Merge pull request #1710 from pipecat-ai/mb/openai-context-aggregation

fix: OpenAIRealtimeBetaLLMService writes two assistant messages to th…

CHANGELOG.md

@@ -87,6 +87,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an issue where `OpenAIRealtimeBetaLLMService` would add two assistant
+  messages to the context.
+
 - Fixed an issue with `GeminiMultimodalLiveLLMService` where the context
   contained tokens instead of words.
 

src/pipecat/services/openai_realtime_beta/context.py

@@ -12,9 +12,11 @@ from loguru import logger
 from pipecat.frames.frames import (
     Frame,
     FunctionCallResultFrame,
+    InterimTranscriptionFrame,
     LLMMessagesUpdateFrame,
     LLMSetToolsFrame,
     LLMTextFrame,
+    TranscriptionFrame,
 )
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.frame_processor import FrameDirection
@@ -137,15 +139,6 @@ class OpenAIRealtimeLLMContext(OpenAILLMContext):
         }
         self.add_message(message)
 
-    def add_assistant_content_item_as_message(self, item):
-        message = {"role": "assistant", "content": []}
-        for content in item.content:
-            if content.type == "audio":
-                message["content"].append({"type": "text", "text": content.transcript})
-            else:
-                logger.error(f"Unhandled content type in assistant item: {content.type} - {item}")
-        self.add_message(message)
-
 
 class OpenAIRealtimeUserContextAggregator(OpenAIUserContextAggregator):
     async def process_frame(
@@ -175,8 +168,10 @@ class OpenAIRealtimeAssistantContextAggregator(OpenAIAssistantContextAggregator)
     # but the OpenAIRealtimeLLMService pushes LLMTextFrames and TTSTextFrames. We
     # need to override this proces_frame for LLMTextFrame, so that only the TTSTextFrames
     # are process. This ensures that the context gets only one set of messages.
+    # OpenAIRealtimeLLMService also pushes TranscriptionFrames and InterimTranscriptionFrames,
+    # so we need to ignore pushing those as well, as they're also TextFrames.
     async def process_frame(self, frame: Frame, direction: FrameDirection):
-        if not isinstance(frame, LLMTextFrame):
+        if not isinstance(frame, (LLMTextFrame, TranscriptionFrame, InterimTranscriptionFrame)):
             await super().process_frame(frame, direction)
 
     async def handle_function_call_result(self, frame: FunctionCallResultFrame):

src/pipecat/services/openai_realtime_beta/openai.py

@@ -562,13 +562,11 @@ class OpenAIRealtimeBetaLLMService(LLMService):
         await self.push_error(ErrorFrame(error=f"Error: {evt}", fatal=True))
 
     async def _handle_assistant_output(self, output):
-        # logger.debug(f"!!! HANDLE Assistant output: {output}")
         # We haven't seen intermixed audio and function_call items in the same response. But let's
         # try to write logic that handles that, if it does happen.
-        messages = [item for item in output if item.type == "message"]
+        # Also, the assistant output is pushed as LLMTextFrame and TTSTextFrame to be handled by
+        # the assistant context aggregator.
         function_calls = [item for item in output if item.type == "function_call"]
-        for item in messages:
-            self._context.add_assistant_content_item_as_message(item)
         await self._handle_function_call_items(function_calls)
 
     async def _handle_function_call_items(self, items):