OpenAIRealtimeClient

OpenAIRealtimeClient(
    *,
    llm_config: LLMConfig | dict[str, Any],
    logger: Logger | None = None
)
(Experimental) Client for OpenAI Realtime API.
(Experimental) Client for OpenAI Realtime API.
Parameters:
NameDescription
llm_configType: LLMConfig | dict[str, typing.Any]
loggerType: logging.Logger | None

Default: None

Static Methods

get_factory

get_factory(
    llm_config: LLMConfig | dict[str, Any],
    logger: Logger,
    **kwargs: Any
) -> 
Create a Realtime API client.
Parameters:
NameDescription
llm_configThe config for the client.

Type: LLMConfig | dict[str, typing.Any]
loggerThe logger to use for logging events.

Type: logging.Logger
**kwargsType: Any

Instance Attributes

connection


Get the OpenAI WebSocket connection.

logger


Get the logger for the OpenAI Realtime API.

Instance Methods

add_event

add_event(self, event: RealtimeEvent | None) -> 
Parameters:
NameDescription
eventType: RealtimeEvent | None

connect

connect(self) -> AsyncGenerator[NoneNone]
Connect to the OpenAI Realtime API.

get_event

get_event(self) -> RealtimeEvent | None

queue_input_audio_buffer_delta

queue_input_audio_buffer_delta(self, audio: str) -> None
queue InputAudioBufferDelta.
Parameters:
NameDescription
audioThe audio.

Type: str

read_events

read_events(self) -> AsyncGenerator[RealtimeEvent, None]
Read messages from the OpenAI Realtime API.

send_audio

send_audio(self, audio: str) -> None
Send audio to the OpenAI Realtime API.
Parameters:
NameDescription
audioThe audio to send.

Type: str

send_function_result

send_function_result(
    self,
    call_id: str,
    result: str
) -> None
Send the result of a function call to the OpenAI Realtime API.
Parameters:
NameDescription
call_idThe ID of the function call.

Type: str
resultThe result of the function call.

Type: str

send_text

send_text(
    self,
    *,
    role: Literal['user''assistant''system'],
    text: str
) -> None
Send a text message to the OpenAI Realtime API.
Parameters:
NameDescription
roleThe role of the message.

Type: Literal[‘user’, ‘assistant’, ‘system’]
textThe text of the message.

Type: str

session_update

session_update(self, session_options: dict[str, Any]) -> None
Send a session update to the OpenAI Realtime API.
Parameters:
NameDescription
session_optionsThe session options to update.

Type: dict[str, typing.Any]

truncate_audio

truncate_audio(
    self,
    audio_end_ms: int,
    content_index: int,
    item_id: str
) -> None
Truncate audio in the OpenAI Realtime API.
Parameters:
NameDescription
audio_end_msThe end of the audio to truncate.

Type: int
content_indexThe index of the content to truncate.

Type: int
item_idThe ID of the item to truncate.

Type: str