From 00badaf98e96d1ac035d0f949c98eb740162cfbf Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Sun, 6 Oct 2024 16:39:11 -0700 Subject: [PATCH] more pydantic cleanup --- .../openai_realtime_beta/client_events.py | 340 +++++++++++++++++- .../openai_realtime_beta/llm_and_context.py | 117 +++--- 2 files changed, 386 insertions(+), 71 deletions(-) diff --git a/src/pipecat/services/openai_realtime_beta/client_events.py b/src/pipecat/services/openai_realtime_beta/client_events.py index 578e56822..5bf05fc07 100644 --- a/src/pipecat/services/openai_realtime_beta/client_events.py +++ b/src/pipecat/services/openai_realtime_beta/client_events.py @@ -1,5 +1,12 @@ -from pydantic import BaseModel -from typing import Dict, List, Optional, Literal +import json +import uuid + +from pydantic import BaseModel, Field +from typing import Any, Dict, List, Literal, Optional, Union + +# +# session properties +# class InputAudioTranscription(BaseModel): @@ -24,10 +31,333 @@ class SessionProperties(BaseModel): tools: Optional[List[Dict]] = [] tool_choice: Optional[Literal["auto", "none", "required"]] = "auto" temperature: Optional[float] = 0.8 - max_response_output_tokens: Optional[int] = 4096 + max_response_output_tokens: Optional[Union[int, Literal["inf"]]] = Field(default=4096) -class SessionUpdateEvent(BaseModel): +# +# context +# + + +class ItemContent(BaseModel): + type: Literal["text", "audio", "input_text", "input_audio"] + text: Optional[str] = None + audio: Optional[str] = None # base64-encoded audio + transcript: Optional[str] = None + + +class ConversationItem(BaseModel): + id: str + object: Literal["realtime.item"] + type: Literal["message", "function_call", "function_call_output"] + status: Optional[Literal["completed", "in_progress", "incomplete"]] = None + # role and content are present for message items + role: Optional[Literal["user", "assistant", "system"]] = None + content: Optional[List[ItemContent]] = None + # these four fields are present for function_call items + call_id: Optional[str] = None + name: Optional[str] = None + arguments: Optional[str] = None + Output: Optional[str] = None + + +class RealtimeConversation(BaseModel): + id: str + object: Literal["realtime.conversation"] + + +# +# error class +# +class RealtimeError(BaseModel): + type: str + code: str + message: str + param: Optional[str] = None + + +# +# client events +# + + +class ClientEvent(BaseModel): + event_id: str = Field(default_factory=lambda: str(uuid.uuid4())) + + +class SessionUpdateEvent(ClientEvent): + session_properties: Optional[SessionProperties] = None + + +# +# server events +# + + +class ServerEvent(BaseModel): event_id: str - type: Literal["session.update"] + type: str + + class Config: + arbitrary_types_allowed = True + + +class SessionCreatedEvent(ServerEvent): + type: Literal["session.created"] session: SessionProperties + + +class SessionUpdatedEvent(ServerEvent): + type: Literal["session.updated"] + session: SessionProperties + + +class ConversationCreated(ServerEvent): + type: Literal["conversation.created"] + conversation: RealtimeConversation + + +class ConversationItemCreated(ServerEvent): + type: Literal["conversation.item.created"] + previous_item_id: Optional[str] = None + item: ConversationItem + + +class ConversationItemInputAudioTranscriptionCompleted(ServerEvent): + type: Literal["conversation.item.input_audio_transcription.completed"] + item_id: str + content_index: int + transcript: str + + +class ConversationItemInputAudioTranscriptionFailed(ServerEvent): + type: Literal["conversation.item.input_audio_transcription.failed"] + item_id: str + content_index: int + error: RealtimeError + + +class ConversationItemTruncated(ServerEvent): + type: Literal["conversation.item.truncated"] + item_id: str + content_index: int + audio_end_ms: int + + +class ConversationItemDeleted(ServerEvent): + type: Literal["conversation.item.deleted"] + item_id: str + + +class ResponseCreated(ServerEvent): + type: Literal["response.created"] + response: "Response" + + +class ResponseDone(ServerEvent): + type: Literal["response.done"] + response: "Response" + + +class ResponseOutputItemAdded(ServerEvent): + type: Literal["response.output_item.added"] + response_id: str + output_index: int + item: ConversationItem + + +class ResponseOutputItemDone(ServerEvent): + type: Literal["response.output_item.done"] + response_id: str + output_index: int + item: ConversationItem + + +class ResponseContentPartAdded(ServerEvent): + type: Literal["response.content_part.added"] + response_id: str + item_id: str + output_index: int + content_index: int + part: ItemContent + + +class ResponseContentPartDone(ServerEvent): + type: Literal["response.content_part.done"] + response_id: str + item_id: str + output_index: int + content_index: int + part: ItemContent + + +class ResponseTextDelta(ServerEvent): + type: Literal["response.text.delta"] + response_id: str + item_id: str + output_index: int + content_index: int + delta: str + + +class ResponseTextDone(ServerEvent): + type: Literal["response.text.done"] + response_id: str + item_id: str + output_index: int + content_index: int + text: str + + +class ResponseAudioTranscriptDelta(ServerEvent): + type: Literal["response.audio_transcript.delta"] + response_id: str + item_id: str + output_index: int + content_index: int + delta: str + + +class ResponseAudioTranscriptDone(ServerEvent): + type: Literal["response.audio_transcript.done"] + response_id: str + item_id: str + output_index: int + content_index: int + transcript: str + + +class ResponseAudioDelta(ServerEvent): + type: Literal["response.audio.delta"] + response_id: str + item_id: str + output_index: int + content_index: int + delta: str # base64-encoded audio + + +class ResponseAudioDone(ServerEvent): + type: Literal["response.audio.done"] + response_id: str + item_id: str + output_index: int + content_index: int + + +class ResponseFunctionCallArgumentsDelta(ServerEvent): + type: Literal["response.function_call_arguments.delta"] + response_id: str + item_id: str + output_index: int + call_id: str + delta: str + + +class ResponseFunctionCallArgumentsDone(ServerEvent): + type: Literal["response.function_call_arguments.done"] + response_id: str + item_id: str + output_index: int + call_id: str + arguments: str + + +class InputAudioBufferSpeechStarted(ServerEvent): + type: Literal["input_audio_buffer.speech_started"] + audio_start_ms: int + item_id: str + + +class InputAudioBufferSpeechStopped(ServerEvent): + type: Literal["input_audio_buffer.speech_stopped"] + audio_end_ms: int + item_id: str + + +class InputAudioBufferCommitted(ServerEvent): + type: Literal["input_audio_buffer.committed"] + previous_item_id: Optional[str] = None + item_id: str + + +class InputAudioBufferCleared(ServerEvent): + type: Literal["input_audio_buffer.cleared"] + + +class ErrorEvent(ServerEvent): + type: Literal["error"] + error: RealtimeError + + +class RateLimitsUpdated(ServerEvent): + type: Literal["rate_limits.updated"] + rate_limits: List[Dict[str, Any]] + + +class TokenDetails(BaseModel): + cached_tokens: Optional[int] = 0 + text_tokens: Optional[int] = 0 + audio_tokens: Optional[int] = 0 + + class Config: + extra = "allow" + + +class Usage(BaseModel): + total_tokens: int + input_tokens: int + output_tokens: int + input_token_details: TokenDetails + output_token_details: TokenDetails + + +class Response(BaseModel): + id: str + object: Literal["realtime.response"] + status: Literal["completed", "in_progress", "incomplete"] + status_details: Any + output: List[ConversationItem] + usage: Optional[Usage] = None + + +_server_event_types = { + "error": ErrorEvent, + "session.created": SessionCreatedEvent, + "session.updated": SessionUpdatedEvent, + "conversation.created": ConversationCreated, + "input_audio_buffer.committed": InputAudioBufferCommitted, + "input_audio_buffer.cleared": InputAudioBufferCleared, + "input_audio_buffer.speech_started": InputAudioBufferSpeechStarted, + "input_audio_buffer.speech_stopped": InputAudioBufferSpeechStopped, + "conversation.item.created": ConversationItemCreated, + "conversation.item.input_audio_transcription.completed": ConversationItemInputAudioTranscriptionCompleted, + "conversation.item.input_audio_transcription.failed": ConversationItemInputAudioTranscriptionFailed, + "conversation.item.truncated": ConversationItemTruncated, + "conversation.item.deleted": ConversationItemDeleted, + "response.created": ResponseCreated, + "response.done": ResponseDone, + "response.output_item.added": ResponseOutputItemAdded, + "response.output_item.done": ResponseOutputItemDone, + "response.content_part.added": ResponseContentPartAdded, + "response.content_part.done": ResponseContentPartDone, + "response.text.delta": ResponseTextDelta, + "response.text.done": ResponseTextDone, + "response.audio_transcript.delta": ResponseAudioTranscriptDelta, + "response.audio_transcript.done": ResponseAudioTranscriptDone, + "response.audio.delta": ResponseAudioDelta, + "response.audio.done": ResponseAudioDone, + "response.function_call_arguments.delta": ResponseFunctionCallArgumentsDelta, + "response.function_call_arguments.done": ResponseFunctionCallArgumentsDone, + "rate_limits.updated": RateLimitsUpdated, +} + + +def parse_server_event(str): + try: + event = json.loads(str) + event_type = event["type"] + if event_type not in _server_event_types: + raise Exception(f"Unimplemented server event type: {event_type}") + return _server_event_types[event_type].model_validate(event) + except Exception as e: + raise Exception(f"{e} \n\n{str}") diff --git a/src/pipecat/services/openai_realtime_beta/llm_and_context.py b/src/pipecat/services/openai_realtime_beta/llm_and_context.py index 5781ca00a..d26bf0d31 100644 --- a/src/pipecat/services/openai_realtime_beta/llm_and_context.py +++ b/src/pipecat/services/openai_realtime_beta/llm_and_context.py @@ -149,6 +149,9 @@ class OpenAILLMServiceRealtimeBeta(LLMService): await super().cancel(frame) await self._disconnect() + async def send_client_event(self, event: events.ClientEvent): + await self._ws_send(event.dict()) + async def _ws_send(self, realtime_message): try: # if realtime_message.get("type") != "input_audio_buffer.append": @@ -204,112 +207,94 @@ class OpenAILLMServiceRealtimeBeta(LLMService): async def _receive_task_handler(self): try: async for message in self._get_websocket(): - msg = json.loads(message) - # logger.debug(f"Received message: {msg}") - if not msg: - continue - if msg["type"] == "session.created": + evt = events.parse_server_event(message) + # logger.debug(f"Received event: {evt}") + if evt.type == "session.created": # session.created is received right after connecting. send a message # to configure the session properties. await self.update_session_properties() - elif msg["type"] == "session.updated": - self._session_properties = msg["session"] - elif msg["type"] == "input_audio_buffer.speech_started": + elif evt.type == "session.updated": + self._session_properties = evt.session + elif evt.type == "input_audio_buffer.speech_started": # user started speaking + # todo: send user started speaking if configured pass - elif msg["type"] == "input_audio_buffer.speech_stopped": + elif evt.type == "input_audio_buffer.speech_stopped": # user stopped speaking + # todo: send user stopped speaking if configured await self.start_processing_metrics() await self.start_ttfb_metrics() - elif msg["type"] == "conversation.item.created": + elif evt.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}") + # we could listen to this event and track conversation item IDs to + # help with context bookkeeping. pass - elif msg["type"] == "response.created": + elif evt.type == "response.created": # todo: 1. figure out TTS started/stopped frame semantics better # 2. do not push these frames in text-only mode - logger.debug(f"Received response created: {msg}") if not self._bot_speaking: self._bot_speaking = True await self.push_frame(TTSStartedFrame()) pass - elif msg["type"] == "conversation.item.input_audio_transcription.completed": + elif evt.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 + if evt.transcript: + self._context.add_message({"role": "user", "content": evt.transcript}) + elif evt.type == "response.output_item.added": + # todo: think about adding a frame for this (generally, in Pipecat/RTVI), as + # it could be useful for managing UI state pass - elif msg["type"] == "response.content_part.added": - # same thing, ignore for now until we think more about UI updates + elif evt.type == "response.content_part.added": + # todo: same thing — possibly a useful event for client-side UI pass - elif msg["type"] == "response.audio_transcript.delta": - # openai playground app uses this, not "text" - if msg["delta"]: - await self.push_frame(TextFrame(msg["delta"])) - pass - elif msg["type"] == "response.audio.delta": + elif evt.type == "response.audio_transcript.delta": + # note: the openai playground app uses this, not "response.text.delta" + if evt.delta: + await self.push_frame(TextFrame(evt.delta)) + elif evt.type == "response.audio.delta": await self.stop_ttfb_metrics() frame = TTSAudioRawFrame( - audio=base64.b64decode(msg["delta"]), + audio=base64.b64decode(evt.delta), sample_rate=24000, num_channels=1, ) await self.push_frame(frame) - elif msg["type"] == "response.audio.done": + elif evt.type == "response.audio.done": if self._bot_speaking: self._bot_speaking = False await self.push_frame(TTSStoppedFrame()) + elif evt.type == "response.audio_transcript.done": + # this doesn't map to any Pipecat frame types pass - elif msg["type"] == "response.audio_transcript.done": - # probably ignore for now + elif evt.type == "response.content_part.done": + # this doesn't map to any Pipecat frame types pass - elif msg["type"] == "response.content_part.done": + elif evt.type == "response.output_item.done": + # this doesn't map to any Pipecat frame types pass - elif msg["type"] == "response.output_item.done": - # 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: - # could send full transcript here instead of streaming chunks - # logger.debug(f"!!! >{item['transcript']}") - pass - elif msg["type"] == "response.done": - # logger.debug(f"Received response done: {msg}") + elif evt.type == "response.done": # usage metrics - # example. - # response.usage.total_tokens:592 - # response.usage.input_tokens:425 - # response.usage.output_tokens:167 - # response.usage.input_token_details.cached_tokens:0 - # response.usage.input_token_details.text_tokens:310 - # response.usage.input_token_details.audio_tokens:115 - # response.usage.output_token_details.text_tokens:32 - # response.usage.output_token_details.audio_tokens:135 tokens = LLMTokenUsage( - prompt_tokens=msg["response"]["usage"]["input_tokens"], - completion_tokens=msg["response"]["usage"]["output_tokens"], - total_tokens=msg["response"]["usage"]["total_tokens"], + prompt_tokens=evt.response.usage.input_tokens, + completion_tokens=evt.response.usage.output_tokens, + total_tokens=evt.response.usage.total_tokens, ) await self.start_llm_usage_metrics(tokens) - # question for mrkb: don't seem to be getting processing time on the console except the first inference await self.stop_processing_metrics() # function calls - items = msg["response"]["output"] - function_calls = [item for item in items if item.get("type") == "function_call"] + items = evt.response.output + function_calls = [item for item in items if item.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": + elif evt.type == "rate_limits.updated": + # todo: add a Pipecat frame for this. (maybe?) pass - elif msg["type"] == "error": - raise Exception(f"Error: {msg}") + elif evt.type == "error": + # These errors seem to be fatal to this connection. So, close and send an ErrorFrame. + raise Exception(f"Error: {evt}") except asyncio.CancelledError: pass @@ -319,9 +304,9 @@ class OpenAILLMServiceRealtimeBeta(LLMService): async def _handle_function_call_items(self, items): total_items = len(items) for index, item in enumerate(items): - function_name = item["name"] - tool_id = item["call_id"] - arguments = json.loads(item["arguments"]) + 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():