autogen.agentchat.realtime.experimental.clients.oai.OpenAIRealtimeWebRTCClient

OpenAIRealtimeWebRTCClient

OpenAIRealtimeWebRTCClient(
    *,
    llm_config: LLMConfig | dict[str, Any],
    websocket: WebSocket,
    logger: Logger | None = None
)

(Experimental) Client for OpenAI Realtime API that uses WebRTC protocol.
(Experimental) Client for OpenAI Realtime API.

Parameters:

Name	Description
llm_config	Type: LLMConfig | dict[str, typing.Any]
websocket	Type: WebSocket
logger	Type: logging.Logger | None Default: None

Static Methods

get_factory

get_factory(
    llm_config: LLMConfig | dict[str, Any],
    logger: Logger,
    **kwargs: Any
) -> Callable[[], RealtimeClientProtocol] | None

Create a Realtime API client.

Parameters:

Name	Description
llm_config	The config for the client. Type: LLMConfig | dict[str, typing.Any]
logger	The logger to use for logging events. Type: logging.Logger
**kwargs	Additional arguments. Type: Any

Returns:

Type	Description
Callable[[], RealtimeClientProtocol] | None	RealtimeClientProtocol: The Realtime API client is returned if the model matches the pattern

Instance Attributes

logger

Get the logger for the OpenAI Realtime API.

Instance Methods

connect

connect(self) -> AsyncGenerator[None, None]

Connect to the OpenAI Realtime API.
In the case of WebRTC, we pass connection information over the websocket, so that javascript on the other end of websocket open actual connection to OpenAI

read_events

read_events(self) -> AsyncGenerator[RealtimeEvent, None]

Read events from the OpenAI Realtime API.

send_audio

send_audio(self, audio: str) -> None

Send audio to the OpenAI Realtime API.
in case of WebRTC, audio is already sent by js client, so we just queue it in order to be logged.

Parameters:

Name	Description
audio	The 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:

Name	Description
call_id	The ID of the function call. Type: str
result	The 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:

Name	Description
role	The role of the message. Type: Literal[‘user’, ‘assistant’, ‘system’]
text	The text of the message. Type: str

session_init_data

session_init_data(self) -> list[dict[str, Any]]

Control initial session with OpenAI.

session_update

session_update(self, session_options: dict[str, Any]) -> None

Send a session update to the OpenAI Realtime API.
In the case of WebRTC we can not send it directly, but we can send it to the javascript over the websocket, and rely on it to send session update to OpenAI

Parameters:

Name	Description
session_options	The 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:

Name	Description
audio_end_ms	The end of the audio to truncate. Type: int
content_index	The index of the content to truncate. Type: int
item_id	The ID of the item to truncate. Type: str