agentchat.realtime_agent.oai_realtime_client

OpenAIRealtimeClient

class OpenAIRealtimeClient()

(Experimental) Client for OpenAI Realtime API.

__init__

def __init__(*,
             llm_config: dict[str, Any],
             voice: str,
             system_message: str,
             logger: Optional[Logger] = None) -> None

(Experimental) Client for OpenAI Realtime API.

Arguments:

• llm_config dict[str, Any] - The config for the client.

logger

@property
def logger() -> Logger

Get the logger for the OpenAI Realtime API.

connection

@property
def connection() -> AsyncRealtimeConnection

Get the OpenAI WebSocket connection.

send_function_result

async def send_function_result(call_id: str, result: str) -> None

Send the result of a function call to the OpenAI Realtime API.

Arguments:

• call_id str - The ID of the function call.
• result str - The result of the function call.

send_text

async def send_text(*, role: Role, text: str) -> None

Send a text message to the OpenAI Realtime API.

Arguments:

• role str - The role of the message.
• text str - The text of the message.

send_audio

async def send_audio(audio: str) -> None

Send audio to the OpenAI Realtime API.

Arguments:

• audio str - The audio to send.

truncate_audio

async def truncate_audio(audio_end_ms: int, content_index: int,
                         item_id: str) -> None

Truncate audio in the OpenAI Realtime API.

Arguments:

• audio_end_ms int - The end of the audio to truncate.
• content_index int - The index of the content to truncate.
• item_id str - The ID of the item to truncate.

session_update

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

Send a session update to the OpenAI Realtime API.

Arguments:

• session_options dict[str, Any] - The session options to update.

connect

@asynccontextmanager
async def connect() -> AsyncGenerator[None, None]

Connect to the OpenAI Realtime API.

read_events

async def read_events() -> AsyncGenerator[dict[str, Any], None]

Read messages from the OpenAI Realtime API.

OpenAIRealtimeWebRTCClient

class OpenAIRealtimeWebRTCClient()

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

__init__

def __init__(*,
             llm_config: dict[str, Any],
             voice: str,
             system_message: str,
             websocket: "WebSocket",
             logger: Optional[Logger] = None) -> None

(Experimental) Client for OpenAI Realtime API.

Arguments:

• llm_config dict[str, Any] - The config for the client.

logger

@property
def logger() -> Logger

Get the logger for the OpenAI Realtime API.

send_function_result

async def send_function_result(call_id: str, result: str) -> None

Send the result of a function call to the OpenAI Realtime API.

Arguments:

• call_id str - The ID of the function call.
• result str - The result of the function call.

send_text

async def send_text(*, role: Role, text: str) -> None

Send a text message to the OpenAI Realtime API.

Arguments:

• role str - The role of the message.
• text str - The text of the message.

send_audio

async def send_audio(audio: str) -> None

Send audio to the OpenAI Realtime API.

Arguments:

• audio str - The audio to send.

truncate_audio

async def truncate_audio(audio_end_ms: int, content_index: int,
                         item_id: str) -> None

Truncate audio in the OpenAI Realtime API.

Arguments:

• audio_end_ms int - The end of the audio to truncate.
• content_index int - The index of the content to truncate.
• item_id str - The ID of the item to truncate.

session_update

async def session_update(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

Arguments:

• session_options dict[str, Any] - The session options to update.

connect

@asynccontextmanager
async def connect() -> 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

async def read_events() -> AsyncGenerator[dict[str, Any], 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.