more pydantic cleanup
This commit is contained in:
Kwindla Hultman Kramer
@@ -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}")
|
||||
|
||||
@@ -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():
|
||||
|
||||
Reference in New Issue
Block a user