more pydantic cleanup

src/pipecat/services/openai_realtime_beta/client_events.py

@@ -1,5 +1,12 @@
-from pydantic import BaseModel
+import json
-from typing import Dict, List, Optional, Literal
+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}")

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)
+                evt = events.parse_server_event(message)
-                # logger.debug(f"Received message: {msg}")
+                # logger.debug(f"Received event: {evt}")
-                if not msg:
+                if evt.type == "session.created":
-                    continue
-                if msg["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":
+                elif evt.type == "session.updated":
-                    self._session_properties = msg["session"]
+                    self._session_properties = evt.session
-                elif msg["type"] == "input_audio_buffer.speech_started":
+                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
+                    # we could listen to this event and track conversation item IDs to
-                    # logger.debug(f"Received {msg}")
+                    # 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 evt.transcript:
-                    if msg.get("transcript"):
+                        self._context.add_message({"role": "user", "content": evt.transcript})
-                        self._context.add_message({"role": "user", "content": msg["transcript"]})
+                elif evt.type == "response.output_item.added":
-                elif msg["type"] == "response.output_item.added":
+                    # todo: think about adding a frame for this (generally, in Pipecat/RTVI), as
-                    # maybe ignore for now but could be useful for UI updates
+                    # it could be useful for managing UI state
                     pass
-                elif msg["type"] == "response.content_part.added":
+                elif evt.type == "response.content_part.added":
-                    # same thing, ignore for now until we think more about UI updates
+                    # todo: same thing — possibly a useful event for client-side UI
                     pass
-                elif msg["type"] == "response.audio_transcript.delta":
+                elif evt.type == "response.audio_transcript.delta":
-                    # openai playground app uses this, not "text"
+                    # note: the openai playground app uses this, not "response.text.delta"
-                    if msg["delta"]:
+                    if evt.delta:
-                        await self.push_frame(TextFrame(msg["delta"]))
+                        await self.push_frame(TextFrame(evt.delta))
-                    pass
+                elif evt.type == "response.audio.delta":
-                elif msg["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":
+                elif evt.type == "response.content_part.done":
-                    # probably ignore for now
+                    # 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":
+                elif evt.type == "response.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}")
                     # 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"],
+                        prompt_tokens=evt.response.usage.input_tokens,
-                        completion_tokens=msg["response"]["usage"]["output_tokens"],
+                        completion_tokens=evt.response.usage.output_tokens,
-                        total_tokens=msg["response"]["usage"]["total_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"]
+                    items = evt.response.output
-                    function_calls = [item for item in items if item.get("type") == "function_call"]
+                    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":
+                elif evt.type == "error":
-                    raise Exception(f"Error: {msg}")
+                    # 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"]
+            function_name = item.name
-            tool_id = item["call_id"]
+            tool_id = item.call_id
-            arguments = json.loads(item["arguments"])
+            arguments = json.loads(item.arguments)
             if self.has_function(function_name):
                 run_llm = index == total_items - 1
                 if function_name in self._callbacks.keys():