diff --git a/examples/foundational/19-openai-realtime-beta.py b/examples/foundational/19-openai-realtime-beta.py index a8d706b03..910391a9b 100644 --- a/examples/foundational/19-openai-realtime-beta.py +++ b/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( - [ - TranscriptionFrame( - user_id="foo", - timestamp=0, - text="What's the weather like in San Francisco right now?", - ) - ] - ) + await task.queue_frames([OpenAILLMContextFrame(context=context)]) runner = PipelineRunner() diff --git a/pyproject.toml b/pyproject.toml index 30e1fef76..87dffce27 100644 --- a/pyproject.toml +++ b/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" ] diff --git a/src/pipecat/processors/aggregators/openai_llm_context.py b/src/pipecat/processors/aggregators/openai_llm_context.py index f099c372d..69edc9df0 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/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). diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 2d24f938e..ac9087165 100644 --- a/src/pipecat/services/openai.py +++ b/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 ): diff --git a/src/pipecat/services/openai_realtime_beta.py b/src/pipecat/services/openai_realtime_beta.py index d106880fd..35985c510 100644 --- a/src/pipecat/services/openai_realtime_beta.py +++ b/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( - json.dumps( - { - "type": "session.update", - "session": self._session_properties.dict(), - } - ) + await self._ws_send( + { + "type": "session.update", + "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": - logger.debug(f"!!! {msg['delta']}") + elif msg["type"] == "response.audio.done": + # 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": - for item in msg["item"]["content"]: - if item["type"] == "text": - await self.push_frame(TextFrame(item["text"])) + logger.debug(f"Received response item done: {msg}") + item = msg["item"] + if item["type"] == "message" and item["status"] == "completed": + 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( - json.dumps( - { - "type": "conversation.item.create", - "item": { - "type": "message", - "status": "completed", - "role": "user", - "content": [{"type": "input_text", "text": messages[0]["content"]}], - }, - } - ) - ) - await self._websocket.send( - json.dumps( - { - "type": "response.create", - "response": { - "modalities": ["audio", "text"], - }, + for item in items: + logger.debug(f"||| {item}") + await self._ws_send({"type": "conversation.item.create", "item": item}) + logger.debug("||| RESPONSE.CREATE") + await self._ws_send( + { + "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( - json.dumps( - { - "type": "input_audio_buffer.append", - "audio": payload, - }, - ) + await self._ws_send( + { + "type": "input_audio_buffer.append", + "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( - json.dumps( - { - "type": "response.cancel", - }, - ) - ) - await self._websocket.send( - json.dumps( - { - "type": "input_audio_buffer.clear", - }, - ) - ) + # await self._ws_send( + # { + # "type": "response.cancel", + # }, + # ) + # await self._ws_send( + # { + # "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}] - context = OpenAILLMContext(messages) - # await self._create_response(context, messages) + pass + elif isinstance(frame, OpenAILLMContextFrame): + 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( - # 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() + def create_context_aggregator( + self, context: OpenAILLMContext, *, assistant_expect_stripped_words: bool = False + ) -> OpenAIContextAggregatorPair: + OpenAIRealtimeLLMContext.upgrade_to_realtime(context) + user = OpenAIRealtimeUserContextAggregator(context) + assistant = OpenAIRealtimeAssistantContextAggregator( + user, expect_stripped_words=assistant_expect_stripped_words + ) + return OpenAIContextAggregatorPair(_user=user, _assistant=assistant)