diff --git a/examples/foundational/19-openai-realtime-beta.py b/examples/foundational/19-openai-realtime-beta.py new file mode 100644 index 000000000..9bf9faee9 --- /dev/null +++ b/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()) diff --git a/pyproject.toml b/pyproject.toml index 9d515c8f9..9006f23d9 100644 --- a/pyproject.toml +++ b/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" ] diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 2f98a7e10..46660bd48 100644 --- a/src/pipecat/services/openai.py +++ b/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()