From d8ce108ccd3f42258e0d701fd8912fe87b145fbd Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 26 Jun 2025 12:06:47 -0400 Subject: [PATCH] Update OpenAIRealtimeBetaLLMService docstrings --- .../services/openai_realtime_beta/context.py | 85 +++ .../services/openai_realtime_beta/events.py | 498 +++++++++++++++++- .../services/openai_realtime_beta/frames.py | 15 + .../services/openai_realtime_beta/openai.py | 104 +++- 4 files changed, 688 insertions(+), 14 deletions(-) diff --git a/src/pipecat/services/openai_realtime_beta/context.py b/src/pipecat/services/openai_realtime_beta/context.py index 85d1a5457..7caee0ece 100644 --- a/src/pipecat/services/openai_realtime_beta/context.py +++ b/src/pipecat/services/openai_realtime_beta/context.py @@ -4,6 +4,8 @@ # SPDX-License-Identifier: BSD 2-Clause License # +"""OpenAI Realtime LLM context and aggregator implementations.""" + import copy import json @@ -30,6 +32,18 @@ from .frames import RealtimeFunctionCallResultFrame, RealtimeMessagesUpdateFrame class OpenAIRealtimeLLMContext(OpenAILLMContext): + """OpenAI Realtime LLM context with session management and message conversion. + + Extends the standard OpenAI LLM context to support real-time session properties, + instruction management, and conversion between standard message formats and + realtime conversation items. + + Args: + messages: Initial conversation messages. Defaults to None. + tools: Available function tools. Defaults to None. + **kwargs: Additional arguments passed to parent OpenAILLMContext. + """ + def __init__(self, messages=None, tools=None, **kwargs): super().__init__(messages=messages, tools=tools, **kwargs) self.__setup_local() @@ -43,6 +57,14 @@ class OpenAIRealtimeLLMContext(OpenAILLMContext): @staticmethod def upgrade_to_realtime(obj: OpenAILLMContext) -> "OpenAIRealtimeLLMContext": + """Upgrade a standard OpenAI LLM context to a realtime context. + + Args: + obj: The OpenAILLMContext instance to upgrade. + + Returns: + The upgraded OpenAIRealtimeLLMContext instance. + """ if isinstance(obj, OpenAILLMContext) and not isinstance(obj, OpenAIRealtimeLLMContext): obj.__class__ = OpenAIRealtimeLLMContext obj.__setup_local() @@ -52,6 +74,14 @@ class OpenAIRealtimeLLMContext(OpenAILLMContext): # - finish implementing all frames def from_standard_message(self, message): + """Convert a standard message format to a realtime conversation item. + + Args: + message: The standard message dictionary to convert. + + Returns: + A ConversationItem instance for the realtime API. + """ if message.get("role") == "user": content = message.get("content") if isinstance(message.get("content"), list): @@ -79,6 +109,14 @@ class OpenAIRealtimeLLMContext(OpenAILLMContext): logger.error(f"Unhandled message type in from_standard_message: {message}") def get_messages_for_initializing_history(self): + """Get conversation items for initializing the realtime session history. + + Converts the context's messages to a format suitable for the realtime API, + handling system instructions and conversation history packaging. + + Returns: + List of conversation items for session initialization. + """ # We can't load a long conversation history into the openai realtime api yet. (The API/model # forgets that it can do audio, if you do a series of `conversation.item.create` calls.) So # our general strategy until this is fixed is just to put everything into a first "user" @@ -133,6 +171,11 @@ class OpenAIRealtimeLLMContext(OpenAILLMContext): ] def add_user_content_item_as_message(self, item): + """Add a user content item as a standard message to the context. + + Args: + item: The conversation item to add as a user message. + """ message = { "role": "user", "content": [{"type": "text", "text": item.content[0].transcript}], @@ -141,9 +184,25 @@ class OpenAIRealtimeLLMContext(OpenAILLMContext): class OpenAIRealtimeUserContextAggregator(OpenAIUserContextAggregator): + """User context aggregator for OpenAI Realtime API. + + Handles user input frames and generates appropriate context updates + for the realtime conversation, including message updates and tool settings. + + Args: + context: The OpenAI realtime LLM context. + **kwargs: Additional arguments passed to parent aggregator. + """ + async def process_frame( self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM ): + """Process incoming frames and handle realtime-specific frame types. + + Args: + frame: The frame to process. + direction: The direction of frame flow in the pipeline. + """ await super().process_frame(frame, direction) # Parent does not push LLMMessagesUpdateFrame. This ensures that in a typical pipeline, # messages are only processed by the user context aggregator, which is generally what we want. But @@ -157,6 +216,11 @@ class OpenAIRealtimeUserContextAggregator(OpenAIUserContextAggregator): await self.push_frame(frame, direction) async def push_aggregation(self): + """Push user input aggregation. + + Currently ignores all user input coming into the pipeline as realtime + audio input is handled directly by the service. + """ # for the moment, ignore all user input coming into the pipeline. # todo: think about whether/how to fix this to allow for text input from # upstream (transport/transcription, or other sources) @@ -164,6 +228,16 @@ class OpenAIRealtimeUserContextAggregator(OpenAIUserContextAggregator): class OpenAIRealtimeAssistantContextAggregator(OpenAIAssistantContextAggregator): + """Assistant context aggregator for OpenAI Realtime API. + + Handles assistant output frames from the realtime service, filtering + out duplicate text frames and managing function call results. + + Args: + context: The OpenAI realtime LLM context. + **kwargs: Additional arguments passed to parent aggregator. + """ + # The LLMAssistantContextAggregator uses TextFrames to aggregate the LLM output, # but the OpenAIRealtimeLLMService pushes LLMTextFrames and TTSTextFrames. We # need to override this proces_frame for LLMTextFrame, so that only the TTSTextFrames @@ -171,10 +245,21 @@ class OpenAIRealtimeAssistantContextAggregator(OpenAIAssistantContextAggregator) # OpenAIRealtimeLLMService also pushes TranscriptionFrames and InterimTranscriptionFrames, # so we need to ignore pushing those as well, as they're also TextFrames. async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process assistant frames, filtering out duplicate text content. + + Args: + frame: The frame to process. + direction: The direction of frame flow in the pipeline. + """ if not isinstance(frame, (LLMTextFrame, TranscriptionFrame, InterimTranscriptionFrame)): await super().process_frame(frame, direction) async def handle_function_call_result(self, frame: FunctionCallResultFrame): + """Handle function call result and notify the realtime service. + + Args: + frame: The function call result frame to handle. + """ await super().handle_function_call_result(frame) # The standard function callback code path pushes the FunctionCallResultFrame from the llm itself, diff --git a/src/pipecat/services/openai_realtime_beta/events.py b/src/pipecat/services/openai_realtime_beta/events.py index ef62248af..695cd3015 100644 --- a/src/pipecat/services/openai_realtime_beta/events.py +++ b/src/pipecat/services/openai_realtime_beta/events.py @@ -3,7 +3,8 @@ # # SPDX-License-Identifier: BSD 2-Clause License # -# + +"""Event models and data structures for OpenAI Realtime API communication.""" import json import uuid @@ -19,7 +20,7 @@ from pydantic import BaseModel, ConfigDict, Field class InputAudioTranscription(BaseModel): """Configuration for audio transcription settings. - Attributes: + Parameters: model: Transcription model to use (e.g., "gpt-4o-transcribe", "whisper-1"). language: Optional language code for transcription. prompt: Optional transcription hint text. @@ -39,6 +40,15 @@ class InputAudioTranscription(BaseModel): class TurnDetection(BaseModel): + """Server-side voice activity detection configuration. + + Parameters: + type: Detection type, must be "server_vad". + threshold: Voice activity detection threshold (0.0-1.0). Defaults to 0.5. + prefix_padding_ms: Padding before speech starts in milliseconds. Defaults to 300. + silence_duration_ms: Silence duration to detect speech end in milliseconds. Defaults to 800. + """ + type: Optional[Literal["server_vad"]] = "server_vad" threshold: Optional[float] = 0.5 prefix_padding_ms: Optional[int] = 300 @@ -46,6 +56,15 @@ class TurnDetection(BaseModel): class SemanticTurnDetection(BaseModel): + """Semantic-based turn detection configuration. + + Parameters: + type: Detection type, must be "semantic_vad". + eagerness: Turn detection eagerness level. Can be "low", "medium", "high", or "auto". + create_response: Whether to automatically create responses on turn detection. + interrupt_response: Whether to interrupt ongoing responses on turn detection. + """ + type: Optional[Literal["semantic_vad"]] = "semantic_vad" eagerness: Optional[Literal["low", "medium", "high", "auto"]] = None create_response: Optional[bool] = None @@ -53,10 +72,33 @@ class SemanticTurnDetection(BaseModel): class InputAudioNoiseReduction(BaseModel): + """Input audio noise reduction configuration. + + Parameters: + type: Noise reduction type for different microphone scenarios. + """ + type: Optional[Literal["near_field", "far_field"]] class SessionProperties(BaseModel): + """Configuration properties for an OpenAI Realtime session. + + Parameters: + modalities: Communication modalities to enable (text, audio, or both). + instructions: System instructions for the assistant. + voice: Voice ID for text-to-speech output. + input_audio_format: Format for input audio data. + output_audio_format: Format for output audio data. + input_audio_transcription: Configuration for input audio transcription. + input_audio_noise_reduction: Configuration for input audio noise reduction. + turn_detection: Turn detection configuration or False to disable. + tools: Available function tools for the assistant. + tool_choice: Tool usage strategy ("auto", "none", or "required"). + temperature: Sampling temperature for response generation. + max_response_output_tokens: Maximum tokens in response or "inf" for unlimited. + """ + modalities: Optional[List[Literal["text", "audio"]]] = None instructions: Optional[str] = None voice: Optional[str] = None @@ -80,6 +122,15 @@ class SessionProperties(BaseModel): class ItemContent(BaseModel): + """Content within a conversation item. + + Parameters: + type: Content type (text, audio, input_text, or input_audio). + text: Text content for text-based items. + audio: Base64-encoded audio data for audio items. + transcript: Transcribed text for audio items. + """ + type: Literal["text", "audio", "input_text", "input_audio"] text: Optional[str] = None audio: Optional[str] = None # base64-encoded audio @@ -87,6 +138,21 @@ class ItemContent(BaseModel): class ConversationItem(BaseModel): + """A conversation item in the realtime session. + + Parameters: + id: Unique identifier for the item, auto-generated if not provided. + object: Object type identifier for the realtime API. + type: Item type (message, function_call, or function_call_output). + status: Current status of the item. + role: Speaker role for message items (user, assistant, or system). + content: Content list for message items. + call_id: Function call identifier for function_call items. + name: Function name for function_call items. + arguments: Function arguments as JSON string for function_call items. + output: Function output as JSON string for function_call_output items. + """ + id: str = Field(default_factory=lambda: str(uuid.uuid4().hex)) object: Optional[Literal["realtime.item"]] = None type: Literal["message", "function_call", "function_call_output"] @@ -102,11 +168,31 @@ class ConversationItem(BaseModel): class RealtimeConversation(BaseModel): + """A realtime conversation session. + + Parameters: + id: Unique identifier for the conversation. + object: Object type identifier, always "realtime.conversation". + """ + id: str object: Literal["realtime.conversation"] class ResponseProperties(BaseModel): + """Properties for configuring assistant responses. + + Parameters: + modalities: Output modalities for the response. Defaults to ["audio", "text"]. + instructions: Specific instructions for this response. + voice: Voice ID for text-to-speech in this response. + output_audio_format: Audio format for this response. + tools: Available tools for this response. + tool_choice: Tool usage strategy for this response. + temperature: Sampling temperature for this response. + max_response_output_tokens: Maximum tokens for this response. + """ + modalities: Optional[List[Literal["text", "audio"]]] = ["audio", "text"] instructions: Optional[str] = None voice: Optional[str] = None @@ -121,6 +207,16 @@ class ResponseProperties(BaseModel): # error class # class RealtimeError(BaseModel): + """Error information from the realtime API. + + Parameters: + type: Error type identifier. + code: Specific error code. + message: Human-readable error message. + param: Parameter name that caused the error, if applicable. + event_id: Event ID associated with the error, if applicable. + """ + type: str code: Optional[str] = "" message: str @@ -134,14 +230,38 @@ class RealtimeError(BaseModel): class ClientEvent(BaseModel): + """Base class for client events sent to the realtime API. + + Parameters: + event_id: Unique identifier for the event, auto-generated if not provided. + """ + event_id: str = Field(default_factory=lambda: str(uuid.uuid4())) class SessionUpdateEvent(ClientEvent): + """Event to update session properties. + + Parameters: + type: Event type, always "session.update". + session: Updated session properties. + """ + type: Literal["session.update"] = "session.update" session: SessionProperties def model_dump(self, *args, **kwargs) -> Dict[str, Any]: + """Serialize the event to a dictionary. + + Handles special serialization for turn_detection where False becomes null. + + Args: + *args: Positional arguments passed to parent model_dump. + **kwargs: Keyword arguments passed to parent model_dump. + + Returns: + Dictionary representation of the event. + """ dump = super().model_dump(*args, **kwargs) # Handle turn_detection so that False is serialized as null @@ -153,25 +273,61 @@ class SessionUpdateEvent(ClientEvent): class InputAudioBufferAppendEvent(ClientEvent): + """Event to append audio data to the input buffer. + + Parameters: + type: Event type, always "input_audio_buffer.append". + audio: Base64-encoded audio data to append. + """ + type: Literal["input_audio_buffer.append"] = "input_audio_buffer.append" audio: str # base64-encoded audio class InputAudioBufferCommitEvent(ClientEvent): + """Event to commit the current input audio buffer. + + Parameters: + type: Event type, always "input_audio_buffer.commit". + """ + type: Literal["input_audio_buffer.commit"] = "input_audio_buffer.commit" class InputAudioBufferClearEvent(ClientEvent): + """Event to clear the input audio buffer. + + Parameters: + type: Event type, always "input_audio_buffer.clear". + """ + type: Literal["input_audio_buffer.clear"] = "input_audio_buffer.clear" class ConversationItemCreateEvent(ClientEvent): + """Event to create a new conversation item. + + Parameters: + type: Event type, always "conversation.item.create". + previous_item_id: ID of the item to insert after, if any. + item: The conversation item to create. + """ + type: Literal["conversation.item.create"] = "conversation.item.create" previous_item_id: Optional[str] = None item: ConversationItem class ConversationItemTruncateEvent(ClientEvent): + """Event to truncate a conversation item's audio content. + + Parameters: + type: Event type, always "conversation.item.truncate". + item_id: ID of the item to truncate. + content_index: Index of the content to truncate within the item. + audio_end_ms: End time in milliseconds for the truncated audio. + """ + type: Literal["conversation.item.truncate"] = "conversation.item.truncate" item_id: str content_index: int @@ -179,21 +335,48 @@ class ConversationItemTruncateEvent(ClientEvent): class ConversationItemDeleteEvent(ClientEvent): + """Event to delete a conversation item. + + Parameters: + type: Event type, always "conversation.item.delete". + item_id: ID of the item to delete. + """ + type: Literal["conversation.item.delete"] = "conversation.item.delete" item_id: str class ConversationItemRetrieveEvent(ClientEvent): + """Event to retrieve a conversation item by ID. + + Parameters: + type: Event type, always "conversation.item.retrieve". + item_id: ID of the item to retrieve. + """ + type: Literal["conversation.item.retrieve"] = "conversation.item.retrieve" item_id: str class ResponseCreateEvent(ClientEvent): + """Event to create a new assistant response. + + Parameters: + type: Event type, always "response.create". + response: Optional response configuration properties. + """ + type: Literal["response.create"] = "response.create" response: Optional[ResponseProperties] = None class ResponseCancelEvent(ClientEvent): + """Event to cancel the current assistant response. + + Parameters: + type: Event type, always "response.cancel". + """ + type: Literal["response.cancel"] = "response.cancel" @@ -203,6 +386,13 @@ class ResponseCancelEvent(ClientEvent): class ServerEvent(BaseModel): + """Base class for server events received from the realtime API. + + Parameters: + event_id: Unique identifier for the event. + type: Type of the server event. + """ + model_config = ConfigDict(arbitrary_types_allowed=True) event_id: str @@ -210,27 +400,65 @@ class ServerEvent(BaseModel): class SessionCreatedEvent(ServerEvent): + """Event indicating a session has been created. + + Parameters: + type: Event type, always "session.created". + session: The created session properties. + """ + type: Literal["session.created"] session: SessionProperties class SessionUpdatedEvent(ServerEvent): + """Event indicating a session has been updated. + + Parameters: + type: Event type, always "session.updated". + session: The updated session properties. + """ + type: Literal["session.updated"] session: SessionProperties class ConversationCreated(ServerEvent): + """Event indicating a conversation has been created. + + Parameters: + type: Event type, always "conversation.created". + conversation: The created conversation. + """ + type: Literal["conversation.created"] conversation: RealtimeConversation class ConversationItemCreated(ServerEvent): + """Event indicating a conversation item has been created. + + Parameters: + type: Event type, always "conversation.item.created". + previous_item_id: ID of the previous item, if any. + item: The created conversation item. + """ + type: Literal["conversation.item.created"] previous_item_id: Optional[str] = None item: ConversationItem class ConversationItemInputAudioTranscriptionDelta(ServerEvent): + """Event containing incremental input audio transcription. + + Parameters: + type: Event type, always "conversation.item.input_audio_transcription.delta". + item_id: ID of the conversation item being transcribed. + content_index: Index of the content within the item. + delta: Incremental transcription text. + """ + type: Literal["conversation.item.input_audio_transcription.delta"] item_id: str content_index: int @@ -238,6 +466,15 @@ class ConversationItemInputAudioTranscriptionDelta(ServerEvent): class ConversationItemInputAudioTranscriptionCompleted(ServerEvent): + """Event indicating input audio transcription is complete. + + Parameters: + type: Event type, always "conversation.item.input_audio_transcription.completed". + item_id: ID of the conversation item that was transcribed. + content_index: Index of the content within the item. + transcript: Complete transcription text. + """ + type: Literal["conversation.item.input_audio_transcription.completed"] item_id: str content_index: int @@ -245,6 +482,15 @@ class ConversationItemInputAudioTranscriptionCompleted(ServerEvent): class ConversationItemInputAudioTranscriptionFailed(ServerEvent): + """Event indicating input audio transcription failed. + + Parameters: + type: Event type, always "conversation.item.input_audio_transcription.failed". + item_id: ID of the conversation item that failed transcription. + content_index: Index of the content within the item. + error: Error details for the transcription failure. + """ + type: Literal["conversation.item.input_audio_transcription.failed"] item_id: str content_index: int @@ -252,6 +498,15 @@ class ConversationItemInputAudioTranscriptionFailed(ServerEvent): class ConversationItemTruncated(ServerEvent): + """Event indicating a conversation item has been truncated. + + Parameters: + type: Event type, always "conversation.item.truncated". + item_id: ID of the truncated conversation item. + content_index: Index of the content within the item. + audio_end_ms: End time in milliseconds for the truncated audio. + """ + type: Literal["conversation.item.truncated"] item_id: str content_index: int @@ -259,26 +514,63 @@ class ConversationItemTruncated(ServerEvent): class ConversationItemDeleted(ServerEvent): + """Event indicating a conversation item has been deleted. + + Parameters: + type: Event type, always "conversation.item.deleted". + item_id: ID of the deleted conversation item. + """ + type: Literal["conversation.item.deleted"] item_id: str class ConversationItemRetrieved(ServerEvent): + """Event containing a retrieved conversation item. + + Parameters: + type: Event type, always "conversation.item.retrieved". + item: The retrieved conversation item. + """ + type: Literal["conversation.item.retrieved"] item: ConversationItem class ResponseCreated(ServerEvent): + """Event indicating an assistant response has been created. + + Parameters: + type: Event type, always "response.created". + response: The created response object. + """ + type: Literal["response.created"] response: "Response" class ResponseDone(ServerEvent): + """Event indicating an assistant response is complete. + + Parameters: + type: Event type, always "response.done". + response: The completed response object. + """ + type: Literal["response.done"] response: "Response" class ResponseOutputItemAdded(ServerEvent): + """Event indicating an output item has been added to a response. + + Parameters: + type: Event type, always "response.output_item.added". + response_id: ID of the response. + output_index: Index of the output item. + item: The added conversation item. + """ + type: Literal["response.output_item.added"] response_id: str output_index: int @@ -286,6 +578,15 @@ class ResponseOutputItemAdded(ServerEvent): class ResponseOutputItemDone(ServerEvent): + """Event indicating an output item is complete. + + Parameters: + type: Event type, always "response.output_item.done". + response_id: ID of the response. + output_index: Index of the output item. + item: The completed conversation item. + """ + type: Literal["response.output_item.done"] response_id: str output_index: int @@ -293,6 +594,17 @@ class ResponseOutputItemDone(ServerEvent): class ResponseContentPartAdded(ServerEvent): + """Event indicating a content part has been added to a response. + + Parameters: + type: Event type, always "response.content_part.added". + response_id: ID of the response. + item_id: ID of the conversation item. + output_index: Index of the output item. + content_index: Index of the content part. + part: The added content part. + """ + type: Literal["response.content_part.added"] response_id: str item_id: str @@ -302,6 +614,17 @@ class ResponseContentPartAdded(ServerEvent): class ResponseContentPartDone(ServerEvent): + """Event indicating a content part is complete. + + Parameters: + type: Event type, always "response.content_part.done". + response_id: ID of the response. + item_id: ID of the conversation item. + output_index: Index of the output item. + content_index: Index of the content part. + part: The completed content part. + """ + type: Literal["response.content_part.done"] response_id: str item_id: str @@ -311,6 +634,17 @@ class ResponseContentPartDone(ServerEvent): class ResponseTextDelta(ServerEvent): + """Event containing incremental text from a response. + + Parameters: + type: Event type, always "response.text.delta". + response_id: ID of the response. + item_id: ID of the conversation item. + output_index: Index of the output item. + content_index: Index of the content part. + delta: Incremental text content. + """ + type: Literal["response.text.delta"] response_id: str item_id: str @@ -320,6 +654,17 @@ class ResponseTextDelta(ServerEvent): class ResponseTextDone(ServerEvent): + """Event indicating text content is complete. + + Parameters: + type: Event type, always "response.text.done". + response_id: ID of the response. + item_id: ID of the conversation item. + output_index: Index of the output item. + content_index: Index of the content part. + text: Complete text content. + """ + type: Literal["response.text.done"] response_id: str item_id: str @@ -329,6 +674,17 @@ class ResponseTextDone(ServerEvent): class ResponseAudioTranscriptDelta(ServerEvent): + """Event containing incremental audio transcript from a response. + + Parameters: + type: Event type, always "response.audio_transcript.delta". + response_id: ID of the response. + item_id: ID of the conversation item. + output_index: Index of the output item. + content_index: Index of the content part. + delta: Incremental transcript text. + """ + type: Literal["response.audio_transcript.delta"] response_id: str item_id: str @@ -338,6 +694,17 @@ class ResponseAudioTranscriptDelta(ServerEvent): class ResponseAudioTranscriptDone(ServerEvent): + """Event indicating audio transcript is complete. + + Parameters: + type: Event type, always "response.audio_transcript.done". + response_id: ID of the response. + item_id: ID of the conversation item. + output_index: Index of the output item. + content_index: Index of the content part. + transcript: Complete transcript text. + """ + type: Literal["response.audio_transcript.done"] response_id: str item_id: str @@ -347,6 +714,17 @@ class ResponseAudioTranscriptDone(ServerEvent): class ResponseAudioDelta(ServerEvent): + """Event containing incremental audio data from a response. + + Parameters: + type: Event type, always "response.audio.delta". + response_id: ID of the response. + item_id: ID of the conversation item. + output_index: Index of the output item. + content_index: Index of the content part. + delta: Base64-encoded incremental audio data. + """ + type: Literal["response.audio.delta"] response_id: str item_id: str @@ -356,6 +734,16 @@ class ResponseAudioDelta(ServerEvent): class ResponseAudioDone(ServerEvent): + """Event indicating audio content is complete. + + Parameters: + type: Event type, always "response.audio.done". + response_id: ID of the response. + item_id: ID of the conversation item. + output_index: Index of the output item. + content_index: Index of the content part. + """ + type: Literal["response.audio.done"] response_id: str item_id: str @@ -364,6 +752,17 @@ class ResponseAudioDone(ServerEvent): class ResponseFunctionCallArgumentsDelta(ServerEvent): + """Event containing incremental function call arguments. + + Parameters: + type: Event type, always "response.function_call_arguments.delta". + response_id: ID of the response. + item_id: ID of the conversation item. + output_index: Index of the output item. + call_id: ID of the function call. + delta: Incremental function arguments as JSON. + """ + type: Literal["response.function_call_arguments.delta"] response_id: str item_id: str @@ -373,6 +772,17 @@ class ResponseFunctionCallArgumentsDelta(ServerEvent): class ResponseFunctionCallArgumentsDone(ServerEvent): + """Event indicating function call arguments are complete. + + Parameters: + type: Event type, always "response.function_call_arguments.done". + response_id: ID of the response. + item_id: ID of the conversation item. + output_index: Index of the output item. + call_id: ID of the function call. + arguments: Complete function arguments as JSON string. + """ + type: Literal["response.function_call_arguments.done"] response_id: str item_id: str @@ -382,38 +792,90 @@ class ResponseFunctionCallArgumentsDone(ServerEvent): class InputAudioBufferSpeechStarted(ServerEvent): + """Event indicating speech has started in the input audio buffer. + + Parameters: + type: Event type, always "input_audio_buffer.speech_started". + audio_start_ms: Start time of speech in milliseconds. + item_id: ID of the associated conversation item. + """ + type: Literal["input_audio_buffer.speech_started"] audio_start_ms: int item_id: str class InputAudioBufferSpeechStopped(ServerEvent): + """Event indicating speech has stopped in the input audio buffer. + + Parameters: + type: Event type, always "input_audio_buffer.speech_stopped". + audio_end_ms: End time of speech in milliseconds. + item_id: ID of the associated conversation item. + """ + type: Literal["input_audio_buffer.speech_stopped"] audio_end_ms: int item_id: str class InputAudioBufferCommitted(ServerEvent): + """Event indicating the input audio buffer has been committed. + + Parameters: + type: Event type, always "input_audio_buffer.committed". + previous_item_id: ID of the previous item, if any. + item_id: ID of the committed conversation item. + """ + type: Literal["input_audio_buffer.committed"] previous_item_id: Optional[str] = None item_id: str class InputAudioBufferCleared(ServerEvent): + """Event indicating the input audio buffer has been cleared. + + Parameters: + type: Event type, always "input_audio_buffer.cleared". + """ + type: Literal["input_audio_buffer.cleared"] class ErrorEvent(ServerEvent): + """Event indicating an error occurred. + + Parameters: + type: Event type, always "error". + error: Error details. + """ + type: Literal["error"] error: RealtimeError class RateLimitsUpdated(ServerEvent): + """Event indicating rate limits have been updated. + + Parameters: + type: Event type, always "rate_limits.updated". + rate_limits: List of rate limit information. + """ + type: Literal["rate_limits.updated"] rate_limits: List[Dict[str, Any]] class TokenDetails(BaseModel): + """Detailed token usage information. + + Parameters: + cached_tokens: Number of cached tokens used. Defaults to 0. + text_tokens: Number of text tokens used. Defaults to 0. + audio_tokens: Number of audio tokens used. Defaults to 0. + """ + cached_tokens: Optional[int] = 0 text_tokens: Optional[int] = 0 audio_tokens: Optional[int] = 0 @@ -423,6 +885,16 @@ class TokenDetails(BaseModel): class Usage(BaseModel): + """Token usage statistics for a response. + + Parameters: + total_tokens: Total number of tokens used. + input_tokens: Number of input tokens used. + output_tokens: Number of output tokens used. + input_token_details: Detailed breakdown of input token usage. + output_token_details: Detailed breakdown of output token usage. + """ + total_tokens: int input_tokens: int output_tokens: int @@ -431,6 +903,17 @@ class Usage(BaseModel): class Response(BaseModel): + """A complete assistant response. + + Parameters: + id: Unique identifier for the response. + object: Object type, always "realtime.response". + status: Current status of the response. + status_details: Additional status information. + output: List of conversation items in the response. + usage: Token usage statistics for the response. + """ + id: str object: Literal["realtime.response"] status: Literal["completed", "in_progress", "incomplete", "cancelled", "failed"] @@ -474,6 +957,17 @@ _server_event_types = { def parse_server_event(str): + """Parse a server event from JSON string. + + Args: + str: JSON string containing the server event. + + Returns: + Parsed server event object of the appropriate type. + + Raises: + Exception: If the event type is unimplemented or parsing fails. + """ try: event = json.loads(str) event_type = event["type"] diff --git a/src/pipecat/services/openai_realtime_beta/frames.py b/src/pipecat/services/openai_realtime_beta/frames.py index 39de49b34..c28c9212f 100644 --- a/src/pipecat/services/openai_realtime_beta/frames.py +++ b/src/pipecat/services/openai_realtime_beta/frames.py @@ -4,16 +4,31 @@ # SPDX-License-Identifier: BSD 2-Clause License # +"""Custom frame types for OpenAI Realtime API integration.""" + from dataclasses import dataclass from pipecat.frames.frames import DataFrame, FunctionCallResultFrame +from pipecat.services.openai_realtime_beta.context import OpenAIRealtimeLLMContext @dataclass class RealtimeMessagesUpdateFrame(DataFrame): + """Frame indicating that the realtime context messages have been updated. + + Parameters: + context: The updated OpenAI realtime LLM context. + """ + context: "OpenAIRealtimeLLMContext" @dataclass class RealtimeFunctionCallResultFrame(DataFrame): + """Frame containing function call results for the realtime service. + + Parameters: + result_frame: The function call result frame to send to the realtime API. + """ + result_frame: FunctionCallResultFrame diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py index 8d5168c70..09761941b 100644 --- a/src/pipecat/services/openai_realtime_beta/openai.py +++ b/src/pipecat/services/openai_realtime_beta/openai.py @@ -4,6 +4,8 @@ # SPDX-License-Identifier: BSD 2-Clause License # +"""OpenAI Realtime Beta LLM service implementation with WebSocket support.""" + import base64 import json import time @@ -73,6 +75,15 @@ except ModuleNotFoundError as e: @dataclass class CurrentAudioResponse: + """Tracks the current audio response from the assistant. + + Parameters: + item_id: Unique identifier for the audio response item. + content_index: Index of the audio content within the item. + start_time_ms: Timestamp when the audio response started in milliseconds. + total_size: Total size of audio data received in bytes. Defaults to 0. + """ + item_id: str content_index: int start_time_ms: int @@ -80,6 +91,24 @@ class CurrentAudioResponse: class OpenAIRealtimeBetaLLMService(LLMService): + """OpenAI Realtime Beta LLM service providing real-time audio and text communication. + + Implements the OpenAI Realtime API Beta with WebSocket communication for low-latency + bidirectional audio and text interactions. Supports function calling, conversation + management, and real-time transcription. + + Args: + api_key: OpenAI API key for authentication. + model: OpenAI model name. Defaults to "gpt-4o-realtime-preview-2025-06-03". + base_url: WebSocket base URL for the realtime API. + Defaults to "wss://api.openai.com/v1/realtime". + session_properties: Configuration properties for the realtime session. + If None, uses default SessionProperties. + start_audio_paused: Whether to start with audio input paused. Defaults to False. + send_transcription_frames: Whether to emit transcription frames. Defaults to True. + **kwargs: Additional arguments passed to parent LLMService. + """ + # Overriding the default adapter to use the OpenAIRealtimeLLMAdapter one. adapter_class = OpenAIRealtimeLLMAdapter @@ -125,12 +154,30 @@ class OpenAIRealtimeBetaLLMService(LLMService): self._retrieve_conversation_item_futures = {} def can_generate_metrics(self) -> bool: + """Check if the service can generate usage metrics. + + Returns: + True if metrics generation is supported. + """ return True def set_audio_input_paused(self, paused: bool): + """Set whether audio input is paused. + + Args: + paused: True to pause audio input, False to resume. + """ self._audio_input_paused = paused async def retrieve_conversation_item(self, item_id: str): + """Retrieve a conversation item by ID from the server. + + Args: + item_id: The ID of the conversation item to retrieve. + + Returns: + The retrieved conversation item. + """ future = self.get_event_loop().create_future() retrieval_in_flight = False if not self._retrieve_conversation_item_futures.get(item_id): @@ -154,14 +201,29 @@ class OpenAIRealtimeBetaLLMService(LLMService): # async def start(self, frame: StartFrame): + """Start the service and establish WebSocket connection. + + Args: + frame: The start frame triggering service initialization. + """ await super().start(frame) await self._connect() async def stop(self, frame: EndFrame): + """Stop the service and close WebSocket connection. + + Args: + frame: The end frame triggering service shutdown. + """ await super().stop(frame) await self._disconnect() async def cancel(self, frame: CancelFrame): + """Cancel the service and close WebSocket connection. + + Args: + frame: The cancel frame triggering service cancellation. + """ await super().cancel(frame) await self._disconnect() @@ -247,6 +309,12 @@ class OpenAIRealtimeBetaLLMService(LLMService): # async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process incoming frames from the pipeline. + + Args: + frame: The frame to process. + direction: The direction of frame flow in the pipeline. + """ await super().process_frame(frame, direction) if isinstance(frame, TranscriptionFrame): @@ -304,6 +372,11 @@ class OpenAIRealtimeBetaLLMService(LLMService): # async def send_client_event(self, event: events.ClientEvent): + """Send a client event to the OpenAI Realtime API. + + Args: + event: The client event to send. + """ await self._ws_send(event.model_dump(exclude_none=True)) async def _connect(self): @@ -478,6 +551,11 @@ class OpenAIRealtimeBetaLLMService(LLMService): pass async def handle_evt_input_audio_transcription_completed(self, evt): + """Handle completion of input audio transcription. + + Args: + evt: The transcription completed event. + """ await self._call_event_handler("on_conversation_item_updated", evt.item_id, None) if self._send_transcription_frames: @@ -558,7 +636,9 @@ class OpenAIRealtimeBetaLLMService(LLMService): await self.push_frame(UserStoppedSpeakingFrame()) async def _maybe_handle_evt_retrieve_conversation_item_error(self, evt: events.ErrorEvent): - """If the given error event is an error retrieving a conversation item: + """Maybe handle an error event related to retrieving a conversation item. + + If the given error event is an error retrieving a conversation item: - set an exception on the future that retrieve_conversation_item() is waiting on - return true Otherwise: @@ -605,8 +685,11 @@ class OpenAIRealtimeBetaLLMService(LLMService): # async def reset_conversation(self): - # Disconnect/reconnect is the safest way to start a new conversation. - # Note that this will fail if called from the receive task. + """Reset the conversation by disconnecting and reconnecting. + + This is the safest way to start a new conversation. Note that this will + fail if called from the receive task. + """ logger.debug("Resetting conversation") await self._disconnect() if self._context: @@ -654,22 +737,19 @@ class OpenAIRealtimeBetaLLMService(LLMService): user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), ) -> OpenAIContextAggregatorPair: - """Create an instance of OpenAIContextAggregatorPair from an - OpenAILLMContext. Constructor keyword arguments for both the user and - assistant aggregators can be provided. + """Create an instance of OpenAIContextAggregatorPair from an OpenAILLMContext. + + Constructor keyword arguments for both the user and assistant aggregators can be provided. Args: - context (OpenAILLMContext): The LLM context. - user_params (LLMUserAggregatorParams, optional): User aggregator - parameters. - assistant_params (LLMAssistantAggregatorParams, optional): User - aggregator parameters. + context: The LLM context. + user_params: User aggregator parameters. + assistant_params: Assistant aggregator parameters. Returns: OpenAIContextAggregatorPair: A pair of context aggregators, one for the user and one for the assistant, encapsulated in an OpenAIContextAggregatorPair. - """ context.set_llm_adapter(self.get_llm_adapter())