lots of debugging statements. multiple function calls broken

examples/foundational/19-openai-realtime-beta.py

@@ -9,11 +9,13 @@ import aiohttp
 import os
 import sys
 
-from pipecat.frames.frames import TranscriptionFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.services.openai import OpenAILLMContext
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
+    OpenAILLMContextFrame,
+)
 from pipecat.services.openai_realtime_beta import (
     OpenAILLMServiceRealtimeBeta,
     OpenAITurnDetection,
@@ -100,8 +102,6 @@ You are participating in a voice conversation. Keep your responses concise, shor
 unless specifically asked to elaborate on a topic.
 
 Remember, your responses should be short. Just one or two sentences, usually.
-
-Start by suggesting that you have a conversation about space exploration.
 """,
         )
 
@@ -109,7 +109,11 @@ Start by suggesting that you have a conversation about space exploration.
             api_key=os.getenv("OPENAI_API_KEY"), session_properties=session_properties
         )
         llm.register_function(None, fetch_weather_from_api)
-        context = OpenAILLMContext([], tools)
+        context = OpenAILLMContext(
+            # [{"role": "user", "content": "What's the weather right now in San Francisco?"}], tools
+            [{"role": "user", "content": "Say 'hello'."}],
+            tools,
+        )
         context_aggregator = llm.create_context_aggregator(context)
 
         pipeline = Pipeline(
@@ -136,15 +140,7 @@ Start by suggesting that you have a conversation about space exploration.
         async def on_first_participant_joined(transport, participant):
             transport.capture_participant_transcription(participant["id"])
             # Kick off the conversation.
-            await task.queue_frames(
+            await task.queue_frames([OpenAILLMContextFrame(context=context)])
-                [
-                    TranscriptionFrame(
-                        user_id="foo",
-                        timestamp=0,
-                        text="What's the weather like in San Francisco right now?",
-                    )
-                ]
-            )
 
         runner = PipelineRunner()
 

pyproject.toml

@@ -53,7 +53,7 @@ livekit = [ "livekit~=0.13.1", "tenacity~=9.0.0" ]
 lmnt = [ "lmnt~=1.1.4" ]
 local = [ "pyaudio~=0.2.14" ]
 moondream = [ "einops~=0.8.0", "timm~=1.0.8", "transformers~=4.44.0" ]
-openai = [ "openai~=1.50.2" ]
+openai = [ "openai~=1.50.2", "websockets~=12.0", "python-deepcompare~=1.0.1" ]
 openpipe = [ "openpipe~=4.24.0" ]
 playht = [ "pyht~=0.0.28" ]
 silero = [ "onnxruntime>=1.16.1" ]

src/pipecat/processors/aggregators/openai_llm_context.py

@@ -168,6 +168,7 @@ class OpenAILLMContext:
         llm: FrameProcessor,
         run_llm: bool = True,
     ) -> None:
+        logger.debug(f"Calling function {function_name} with arguments {arguments}")
         # Push a SystemFrame downstream. This frame will let our assistant context aggregator
         # know that we are in the middle of a function call. Some contexts/aggregators may
         # not need this. But some definitely do (Anthropic, for example).

src/pipecat/services/openai.py

@@ -497,8 +497,10 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
             self._function_calls_in_progress.clear()
             self._function_call_finished = None
         elif isinstance(frame, FunctionCallInProgressFrame):
+            logger.debug(f"FunctionCallInProgressFrame: {frame}")
             self._function_calls_in_progress[frame.tool_call_id] = frame
         elif isinstance(frame, FunctionCallResultFrame):
+            logger.debug(f"FunctionCallResultFrame: {frame}")
             if frame.tool_call_id in self._function_calls_in_progress:
                 del self._function_calls_in_progress[frame.tool_call_id]
                 self._function_call_result = frame
@@ -514,6 +516,7 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
             await self._push_aggregation()
 
     async def _push_aggregation(self):
+        logger.debug("!!! Pushing aggregation")
         if not (
             self._aggregation or self._function_call_result or self._pending_image_frame_message
         ):

src/pipecat/services/openai_realtime_beta.py

@@ -1,11 +1,14 @@
 import asyncio
 import base64
+import random
 import json
 import websockets
 
+from copy import deepcopy
 from typing import List, Optional
 from pydantic import BaseModel, Field
 
+
 from pipecat.frames.frames import (
     CancelFrame,
     LLMFullResponseStartFrame,
@@ -21,7 +24,15 @@ from pipecat.frames.frames import (
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_services import LLMService
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.services.openai import (
+    OpenAIAssistantContextAggregator,
+    OpenAIUserContextAggregator,
+    OpenAIContextAggregatorPair,
+)
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
+    OpenAILLMContextFrame,
+)
 
 from loguru import logger
 
@@ -33,6 +44,52 @@ from loguru import logger
 # )
 
 
+class OpenAIUnhandledFunctionException(Exception):
+    pass
+
+
+class OpenAIRealtimeLLMContext(OpenAILLMContext):
+    @staticmethod
+    def upgrade_to_realtime(obj: OpenAILLMContext) -> "OpenAIRealtimeLLMContext":
+        if isinstance(obj, OpenAILLMContext) and not isinstance(obj, OpenAIRealtimeLLMContext):
+            obj.__class__ = OpenAIRealtimeLLMContext
+            obj._unsent_messages = deepcopy(obj._messages)
+            obj._marker = random.randint(1, 1000)
+        return obj
+
+    def add_message(self, message):
+        super().add_message(message)
+        if message.get("role") == "tool":
+            self._unsent_messages.append(message)
+
+    def set_messages(self, messages):
+        super().set_messages(messages)
+        self._unsent_messages = deepcopy(self._messages)
+
+    def get_unsent_messages(self):
+        return self._unsent_messages
+
+    def update_all_messages_sent(self):
+        logger.debug("!!! Updating all messages sent")
+        self._unsent_messages = []
+
+
+class OpenAIRealtimeUserContextAggregator(OpenAIUserContextAggregator):
+    async def _push_aggregation(self):
+        pass
+        # idx = len(self._context.messages)
+        # logger.debug(f"!!! 1 {idx}")
+
+        # await super()._push_aggregation()
+        # self._context._unsent_messages.extend(self._context.messages[idx:])
+        # logger.debug(f"!!! 2 {self._context._unsent_messages}")
+
+
+class OpenAIRealtimeAssistantContextAggregator(OpenAIAssistantContextAggregator):
+    async def _push_aggregation(self):
+        await super()._push_aggregation()
+
+
 class OpenAIInputTranscription(BaseModel):
     # enabled: bool = Field(description="Whether to enable input audio transcription.", default=True)
     model: str = Field(
@@ -67,7 +124,7 @@ class RealtimeSessionProperties(BaseModel):
         default=OpenAIInputTranscription()
     )
     turn_detection: Optional[OpenAITurnDetection] = Field(default=None)
-    tools: List[str] = Field(default=[])
+    tools: List[dict] = Field(default=[])
     tool_choice: str = Field(default="auto")
     temperature: float = Field(default=0.8)
     max_response_output_tokens: int = Field(default=4096)
@@ -89,7 +146,7 @@ class OpenAILLMServiceRealtimeBeta(LLMService):
         self._receive_task = None
 
         self._session_properties = session_properties
-        self._responses_in_flight = {}
+        self._context = None
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
@@ -103,15 +160,22 @@ class OpenAILLMServiceRealtimeBeta(LLMService):
         await super().cancel(frame)
         await self._disconnect()
 
+    async def _ws_send(self, realtime_message):
+        try:
+            if realtime_message.get("type") != "input_audio_buffer.append":
+                logger.debug(f"!!! Sending message to websocket: {realtime_message}")
+
+            await self._websocket.send(json.dumps(realtime_message))
+        except Exception as e:
+            logger.error(f"Error sending message to websocket: {e}")
+
     async def update_session_properties(self):
         logger.debug(f"Updating session properties: {self._session_properties.dict()}")
-        await self._websocket.send(
+        await self._ws_send(
-            json.dumps(
+            {
-                {
+                "type": "session.update",
-                    "type": "session.update",
+                "session": self._session_properties.dict(),
-                    "session": self._session_properties.dict(),
+            }
-                }
-            )
         )
 
     async def _connect(self):
@@ -158,14 +222,39 @@ class OpenAILLMServiceRealtimeBeta(LLMService):
                 if not msg:
                     continue
                 if msg["type"] == "session.created":
-                    logger.debug(f"Received session.created: {msg}")
                     await self.update_session_properties()
                 elif msg["type"] == "session.updated":
                     logger.debug(f"Received session configuration: {msg}")
                     self._session_properties = msg["session"]
-                elif msg["type"] == "response.created":
+                elif msg["type"] == "input_audio_buffer.speech_started":
+                    # user started speaking
                     pass
+                elif msg["type"] == "input_audio_buffer.speech_stopped":
+                    # user stopped speaking
+                    pass
+                elif msg["type"] == "conversation.item.created":
+                    # for input, this will get sent from the server whether the
+                    # conversation item is created by audio transcription or by
+                    # sending a client conversation.item.create message.
+                    # for function calls
+                    logger.debug(f"Received {msg}")
+                    pass
+                elif msg["type"] == "response.created":
+                    # could use for processing metrics
+                    pass
+                elif msg["type"] == "conversation.item.input_audio_transcription.completed":
+                    # or here maybe (possible send upstream to user context aggregator)
+                    logger.debug(f"Received {msg}")
+                    if msg.get("transcript"):
+                        self._context.add_message({"role": "user", "content": msg["transcript"]})
                 elif msg["type"] == "response.output_item.added":
+                    # maybe ignore for now but could be useful for UI updates
+                    pass
+                elif msg["type"] == "response.content_part.added":
+                    # same thing, ignore for now until we think more about UI updates
+                    pass
+                elif msg["type"] == "response.audio_transcript.delta":
+                    # openai playground app uses this, not "text"
                     pass
                 elif msg["type"] == "response.audio.delta":
                     frame = TTSAudioRawFrame(
@@ -174,17 +263,36 @@ class OpenAILLMServiceRealtimeBeta(LLMService):
                         num_channels=1,
                     )
                     await self.push_frame(frame)
-                elif msg["type"] == "response.text.delta":
+                elif msg["type"] == "response.audio.done":
-                    logger.debug(f"!!! {msg['delta']}")
+                    # bot stopped speaking - or do that at the end of the response?
+                    pass
+                elif msg["type"] == "response.audio_transcript.done":
+                    # probably ignore for now
+                    pass
+                elif msg["type"] == "response.content_part.done":
                     pass
                 elif msg["type"] == "response.output_item.done":
-                    if msg["item"]["type"] == "message":
+                    logger.debug(f"Received response item done: {msg}")
-                        for item in msg["item"]["content"]:
+                    item = msg["item"]
-                            if item["type"] == "text":
+                    if item["type"] == "message" and item["status"] == "completed":
-                                await self.push_frame(TextFrame(item["text"]))
+                        for item in item["content"]:
+                            # output text
+                            if item["type"] == "audio" and item["transcript"] is not None:
+                                logger.debug(f"!!! >{item['transcript']}")
+                                await self.push_frame(TextFrame(item["transcript"]))
                 elif msg["type"] == "response.done":
+                    logger.debug(f"Received response done: {msg}")
                     await self.stop_processing_metrics()
+                    # send usage metrics from here
+                    # ...
+                    # function calls - do all calls here to support parallel function calls
+                    items = msg["response"]["output"]
+                    function_calls = [item for item in items if item.get("type") == "function_call"]
+                    if function_calls:
+                        await self._handle_function_call_items(function_calls)
                     await self.push_frame(LLMFullResponseEndFrame())
+                elif msg["type"] == "rate_limits.updated":
+                    pass
                 elif msg["type"] == "error":
                     raise Exception(f"Error: {msg}")
 
@@ -193,74 +301,121 @@ class OpenAILLMServiceRealtimeBeta(LLMService):
         except Exception as e:
             logger.error(f"{self} exception: {e}")
 
-    async def _create_response(self, context: OpenAILLMContext, messages: list):
+    async def _handle_function_call_items(self, items):
+        logger.debug(f"Handling function call items: {items}")
+        total_items = len(items)
+        logger.debug("!!!!!")
+        for index, item in enumerate(items):
+            logger.debug(f"!!! function call item: {item}")
+            function_name = item["name"]
+            tool_id = item["call_id"]
+            arguments = json.loads(item["arguments"])
+            if self.has_function(function_name):
+                run_llm = index == total_items - 1
+                if function_name in self._callbacks.keys():
+                    f = self._callbacks[function_name]
+                elif None in self._callbacks.keys():
+                    await self.call_function(
+                        context=self._context,
+                        tool_call_id=tool_id,
+                        function_name=function_name,
+                        arguments=arguments,
+                        run_llm=run_llm,
+                    )
+            else:
+                raise OpenAIUnhandledFunctionException(
+                    f"The LLM tried to call a function named '{function_name}', but there isn't a callback registered for that function."
+                )
+
+    async def _reset_conversation(self):
+        # need to think about how to implement this, and how to think about interop with messages lists
+        # used with the HTTP API
+        pass
+
+    async def _create_response(self, context: OpenAIRealtimeLLMContext):
         try:
+            messages = context.get_unsent_messages()
+            context.update_all_messages_sent()
+            logger.debug(
+                f"Creating response: {context._marker} {context.get_messages_for_logging()}"
+            )
+
+            items = []
+            for m in messages:
+                if m and m.get("role") == "user":
+                    content = m.get("content")
+                    if isinstance(content, str):
+                        items.append(
+                            {
+                                "type": "message",
+                                "status": "completed",
+                                "role": "user",
+                                "content": [{"type": "input_text", "text": content}],
+                            }
+                        )
+                    else:
+                        raise Exception(f"Invalid message content {m}")
+                elif m and m.get("role") == "tool":
+                    items.append(
+                        {
+                            "type": "function_call_output",
+                            "call_id": m.get("tool_call_id"),
+                            "output": m["content"],
+                        }
+                    )
+
             await self.push_frame(LLMFullResponseStartFrame())
             await self.start_processing_metrics()
-            await self._websocket.send(
+            for item in items:
-                json.dumps(
+                logger.debug(f"||| {item}")
-                    {
+                await self._ws_send({"type": "conversation.item.create", "item": item})
-                        "type": "conversation.item.create",
+            logger.debug("||| RESPONSE.CREATE")
-                        "item": {
+            await self._ws_send(
-                            "type": "message",
+                {
-                            "status": "completed",
+                    "type": "response.create",
-                            "role": "user",
+                    "response": {
-                            "content": [{"type": "input_text", "text": messages[0]["content"]}],
+                        "modalities": ["audio", "text"],
-                        },
-                    }
-                )
-            )
-            await self._websocket.send(
-                json.dumps(
-                    {
-                        "type": "response.create",
-                        "response": {
-                            "modalities": ["audio", "text"],
-                        },
                     },
-                )
+                },
             )
         except Exception as e:
             logger.error(f"{self} exception: {e}")
 
     async def _send_user_audio(self, frame):
         payload = base64.b64encode(frame.audio).decode("utf-8")
-        await self._websocket.send(
+        await self._ws_send(
-            json.dumps(
+            {
-                {
+                "type": "input_audio_buffer.append",
-                    "type": "input_audio_buffer.append",
+                "audio": payload,
-                    "audio": payload,
+            },
-                },
-            )
         )
-        # await self._websocket.send(json.dumps(({"type": "input_audio_buffer.commit"})))
 
     async def _handle_interruption(self, frame):
         logger.debug(f"Handling interruption: {frame}")
         await self.stop_all_metrics()
         await self.push_frame(LLMFullResponseEndFrame())
-        await self._websocket.send(
+        # await self._ws_send(
-            json.dumps(
+        #     {
-                {
+        #         "type": "response.cancel",
-                    "type": "response.cancel",
+        #     },
-                },
+        # )
-            )
+        # await self._ws_send(
-        )
+        #     {
-        await self._websocket.send(
+        #         "type": "input_audio_buffer.clear",
-            json.dumps(
+        #     },
-                {
+        # )
-                    "type": "input_audio_buffer.clear",
-                },
-            )
-        )
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
         if isinstance(frame, TranscriptionFrame):
-            messages = [{"role": "user", "content": frame.text}]
+            pass
-            context = OpenAILLMContext(messages)
+        elif isinstance(frame, OpenAILLMContextFrame):
-            # await self._create_response(context, messages)
+            context: OpenAIRealtimeLLMContext = OpenAIRealtimeLLMContext.upgrade_to_realtime(
+                frame.context
+            )
+            self._context = context
+            await self._create_response(context)
         elif isinstance(frame, InputAudioRawFrame):
             await self._send_user_audio(frame)
         elif isinstance(frame, StartInterruptionFrame):
@@ -268,16 +423,12 @@ class OpenAILLMServiceRealtimeBeta(LLMService):
 
         await self.push_frame(frame, direction)
 
-    # async def get_chat_completions(
+    def create_context_aggregator(
-    #     self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
+        self, context: OpenAILLMContext, *, assistant_expect_stripped_words: bool = False
-    # ) -> AsyncStream[ChatCompletionChunk]:
+    ) -> OpenAIContextAggregatorPair:
-    #     async def _empty_async_generator() -> AsyncGenerator[str, None]:
+        OpenAIRealtimeLLMContext.upgrade_to_realtime(context)
-    #         try:
+        user = OpenAIRealtimeUserContextAggregator(context)
-    #             if False:
+        assistant = OpenAIRealtimeAssistantContextAggregator(
-    #                 yield ""
+            user, expect_stripped_words=assistant_expect_stripped_words
-    #         except asyncio.CancelledError:
+        )
-    #             return
+        return OpenAIContextAggregatorPair(_user=user, _assistant=assistant)
-    #         except Exception as e:
-    #             logger.error(f"{self} exception: {e}")
-
-    #     return _empty_async_generator()