OpenAIRealtimeWebRTCClient

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

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

(Experimental) Client for OpenAI Realtime API.

Parameters:
NameDescription
llm_configThe config for the client.

Type: dict[str, typing.Any]
websocketType: WebSocket
loggerType: logging.Logger | None

Default: None

Static Methods

get_factory

get_factory(
    llm_config: dict[str, typing.Any],
    logger: logging.Logger,
    **kwargs: Any
) -> Callable[[], autogen.agentchat.realtime.experimental.clients.realtime_client.RealtimeClientProtocol] | None

Create a Realtime API client.

Parameters:
NameDescription
llm_configType: dict[str, typing.Any]
loggerType: logging.Logger
**kwargsType: Any
Returns:
TypeDescription
Callable[[], autogen.agentchat.realtime.experimental.clients.realtime_client.RealtimeClientProtocol] | NoneRealtimeClientProtocol: 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[autogen.agentchat.realtime.experimental.realtime_events.RealtimeEvent, None]

Read messages from the OpenAI Realtime API. Again, in case of WebRTC, we do not read OpenAI messages directly since we do not hold connection to OpenAI. Instead we read messages from the websocket, and javascript client on the other side of the websocket that is connected to OpenAI is relaying events to us.


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_init_data

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

Control initial session with OpenAI.


session_update

session_update(self, session_options: dict[str, typing.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:
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