beginning of realtime impl

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

@@ -0,0 +1,94 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import aiohttp
+import os
+import sys
+
+
+from pipecat.frames.frames import LLMMessagesFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.logger import FrameLogger
+from pipecat.services.openai import OpenAILLMServiceRealtimeBeta
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.vad.silero import SileroVADAnalyzer
+
+from runner import configure
+
+from loguru import logger
+
+from dotenv import load_dotenv
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        (room_url, token) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            token,
+            "Respond bot",
+            DailyParams(
+                audio_out_enabled=True,
+                transcription_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+            ),
+        )
+
+        fl1 = FrameLogger("fl-1")
+        llm = OpenAILLMServiceRealtimeBeta(api_key=os.getenv("OPENAI_API_KEY"))
+        fl2 = FrameLogger("fl-2")
+
+        messages = [
+            {
+                "role": "system",
+                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+            },
+        ]
+
+        pipeline = Pipeline(
+            [
+                transport.input(),  # Transport user input
+                fl1,
+                llm,  # LLM
+                fl2,
+                transport.output(),  # Transport bot output
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            PipelineParams(
+                allow_interruptions=True,
+                enable_metrics=True,
+                enable_usage_metrics=True,
+                report_only_initial_ttfb=True,
+            ),
+        )
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            transport.capture_participant_transcription(participant["id"])
+            # Kick off the conversation.
+            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+            await task.queue_frames([LLMMessagesFrame(messages)])
+
+        runner = PipelineRunner()
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

pyproject.toml

@@ -52,11 +52,11 @@ 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.37.2" ]
+openai = [ "openai~=1.50.2" ]
 openpipe = [ "openpipe~=4.24.0" ]
 playht = [ "pyht~=0.0.28" ]
 silero = [ "onnxruntime>=1.16.1" ]
-together = [ "together~=1.2.7" ]
+together = [ "openai~=1.50.2" ]
 websocket = [ "websockets~=12.0", "fastapi~=0.115.0" ]
 whisper = [ "faster-whisper~=1.0.3" ]
 xtts = [ "resampy~=0.4.3" ]

src/pipecat/services/openai.py

@@ -4,6 +4,7 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+import asyncio
 import base64
 import io
 import json
@@ -17,6 +18,8 @@ from PIL import Image
 from pydantic import BaseModel, Field
 
 from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
     ErrorFrame,
     Frame,
     FunctionCallInProgressFrame,
@@ -25,6 +28,7 @@ from pipecat.frames.frames import (
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
     LLMUpdateSettingsFrame,
+    StartFrame,
     StartInterruptionFrame,
     TextFrame,
     TTSAudioRawFrame,
@@ -56,6 +60,7 @@ try:
         DefaultAsyncHttpxClient,
     )
     from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam
+    import websockets
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error(
@@ -63,6 +68,15 @@ except ModuleNotFoundError as e:
     )
     raise Exception(f"Missing module: {e}")
 
+# websocket logger
+import logging
+
+logging.basicConfig(
+    format="%(message)s",
+    level=logging.DEBUG,
+)
+
+
 ValidVoice = Literal["alloy", "echo", "fable", "onyx", "nova", "shimmer"]
 
 VALID_VOICES: Dict[str, ValidVoice] = {
@@ -573,3 +587,91 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
 
         except Exception as e:
             logger.error(f"Error processing frame: {e}")
+
+
+class OpenAILLMServiceRealtimeBeta(LLMService):
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        base_url="wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01",
+        **kwargs,
+    ):
+        super().__init__(base_url=base_url, **kwargs)
+        self.api_key = api_key
+        self.base_url = base_url
+        self._websocket = None
+        self._receive_task = None
+
+    async def start(self, frame: StartFrame):
+        await super().start(frame)
+        await self._connect()
+
+    async def stop(self, frame: EndFrame):
+        await super().stop(frame)
+        await self._disconnect()
+
+    async def cancel(self, frame: CancelFrame):
+        await super().cancel(frame)
+        await self._disconnect()
+
+    async def _connect(self):
+        try:
+            logger.debug(f"connecting to {self.base_url} with api_key {self.api_key}")
+            self._websocket = await websockets.connect(
+                uri=self.base_url,
+                extra_headers={
+                    "Authorization": f"Bearer {self.api_key}",
+                    "OpenAI-Beta": "realtime=v1",
+                },
+            )
+            self._receive_task = self.get_event_loop().create_task(self._receive_task_handler())
+        except Exception as e:
+            logger.error(f"{self} initialization error: {e}")
+            self._websocket = None
+
+    async def _disconnect(self):
+        pass
+
+    async def _receive_task_handler(self):
+        try:
+            async for message in self._get_websocket():
+                msg = json.loads(message)
+                logger.debug(f"Received message: {msg}")
+        except asyncio.CancelledError:
+            pass
+        except Exception as e:
+            logger.error(f"{self} exception: {e}")
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        await self.push_frame(frame, direction)
+
+        # if isinstance(frame, TranscriptionFrame):
+        #     self._websocket.send(
+        #         json.dumps(
+        #             {
+        #                 {
+        #                     "type": "response.create",
+        #                     "response": {
+        #                         "modalities": ["text"],
+        #                         "instructions": frame.text,
+        #                     },
+        #                 }
+        #             }
+        #         )
+        #     )
+
+    # async def get_chat_completions(
+    #     self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
+    # ) -> AsyncStream[ChatCompletionChunk]:
+    #     async def _empty_async_generator() -> AsyncGenerator[str, None]:
+    #         try:
+    #             if False:
+    #                 yield ""
+    #         except asyncio.CancelledError:
+    #             return
+    #         except Exception as e:
+    #             logger.error(f"{self} exception: {e}")
+
+    #     return _empty_async_generator()