OpenAIRealtimeClient

async def add_event(self, event: Optional[RealtimeEvent]):
    await self._eventQueue.put(event)

async def get_event(self) -> Optional[RealtimeEvent]:
    return await self._eventQueue.get()

async def queue_input_audio_buffer_delta(self, audio: str) -> None:
    """queue InputAudioBufferDelta.

    Args:
        audio (str): The audio.
    """
    await self.add_event(InputAudioBufferDelta(delta=audio, item_id=None, raw_message=dict()))

async def send_function_result(self, call_id: str, result: str) -> None:
    """Send the result of a function call to the OpenAI Realtime API.

    Args:
        call_id (str): The ID of the function call.
        result (str): The result of the function call.
    """
    await self.connection.conversation.item.create(
        item={
            "type": "function_call_output",
            "call_id": call_id,
            "output": result,
        },
    )

    await self.connection.response.create()

async def send_text(self, *, role: Role, text: str) -> None:
    """Send a text message to the OpenAI Realtime API.

    Args:
        role (str): The role of the message.
        text (str): The text of the message.
    """
    await self.connection.response.cancel()
    await self.connection.conversation.item.create(
        item={"type": "message", "role": role, "content": [{"type": "input_text", "text": text}]}
    )
    await self.connection.response.create()

async def send_audio(self, audio: str) -> None:
    """Send audio to the OpenAI Realtime API.

    Args:
        audio (str): The audio to send.
    """
    await self.queue_input_audio_buffer_delta(audio)
    await self.connection.input_audio_buffer.append(audio=audio)

async def truncate_audio(self, audio_end_ms: int, content_index: int, item_id: str) -> None:
    """Truncate audio in the OpenAI Realtime API.

    Args:
        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.
    """
    await self.connection.conversation.item.truncate(
        audio_end_ms=audio_end_ms, content_index=content_index, item_id=item_id
    )

async def session_update(self, session_options: dict[str, Any]) -> None:
    """Send a session update to the OpenAI Realtime API.

    Args:
        session_options (dict[str, Any]): The session options to update.
    """
    logger = self.logger
    logger.info(f"Sending session update: {session_options}")
    await self.connection.session.update(session=session_options)  # type: ignore[arg-type]
    logger.info("Sending session update finished")

@asynccontextmanager
async def connect(self) -> AsyncGenerator[None, None]:
    """Connect to the OpenAI Realtime API."""
    try:
        if not self._client:
            self._client = AsyncOpenAI(
                api_key=self.config.get("api_key", None),
                organization=self.config.get("organization", None),
                project=self.config.get("project", None),
                base_url=self.config.get("base_url", None),
                websocket_base_url=self.config.get("websocket_base_url", None),
                timeout=self.config.get("timeout", NOT_GIVEN),
                max_retries=self.config.get("max_retries", DEFAULT_MAX_RETRIES),
                default_headers=self.config.get("default_headers", None),
                default_query=self.config.get("default_query", None),
            )
        async with self._client.beta.realtime.connect(
            model=self._model,
        ) as self._connection:
            await self._initialize_session()
            yield
    finally:
        self._connection = None

async def read_events(self) -> AsyncGenerator[RealtimeEvent, None]:
    """Read messages from the OpenAI Realtime API."""
    if self._connection is None:
        raise RuntimeError("Client is not connected, call connect() first.")

    try:
        async for event in self._read_events():
            yield event

    finally:
        self._connection = None

@classmethod
def get_factory(
    cls, llm_config: dict[str, Any], logger: Logger, **kwargs: Any
) -> Optional[Callable[[], "RealtimeClientProtocol"]]:
    """Create a Realtime API client.

    Args:
        llm_config: The config for the client.
        logger: The logger to use for logging events.
        kwargs: Additional arguments.

    Returns:
        RealtimeClientProtocol: The Realtime API client is returned if the model matches the pattern
    """
    if llm_config["config_list"][0].get("api_type", "openai") == "openai" and list(kwargs.keys()) == []:
        return lambda: OpenAIRealtimeClient(llm_config=llm_config, logger=logger, **kwargs)
    return None