Module livekit.agents.inference.realtime.openai

Classes

class RealtimeModel (model: str,
*,
provider: str | None = None,
base_url: str | None = None,
api_key: str | None = None,
api_secret: str | None = None,
inference_class: InferenceClass | None = None,
voice: NotGivenOr[str] = NOT_GIVEN,
modalities: "NotGivenOr[list[Literal['text', 'audio']]]" = NOT_GIVEN,
input_audio_transcription: NotGivenOr[AudioTranscription | None] = NOT_GIVEN,
input_audio_noise_reduction: NotGivenOr[NoiseReductionType | NoiseReduction | None] = NOT_GIVEN,
turn_detection: NotGivenOr[RealtimeAudioInputTurnDetection | None] = NOT_GIVEN,
tool_choice: NotGivenOr[llm.ToolChoice | None] = NOT_GIVEN,
speed: NotGivenOr[float] = NOT_GIVEN,
tracing: NotGivenOr[Tracing | None] = NOT_GIVEN,
truncation: NotGivenOr[RealtimeTruncation | None] = NOT_GIVEN,
reasoning: NotGivenOr[RealtimeReasoning | None] = NOT_GIVEN,
http_session: aiohttp.ClientSession | None = None,
max_session_duration: NotGivenOr[float | None] = NOT_GIVEN,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0))

Expand source code

class RealtimeModel(_RealtimeModel):
    """OpenAI-compatible realtime model authenticated through LiveKit Inference."""

    def __init__(
        self,
        model: str,
        *,
        provider: str | None = None,
        base_url: str | None = None,
        api_key: str | None = None,
        api_secret: str | None = None,
        inference_class: InferenceClass | None = None,
        voice: NotGivenOr[str] = NOT_GIVEN,
        modalities: NotGivenOr[list[Literal["text", "audio"]]] = NOT_GIVEN,
        input_audio_transcription: NotGivenOr[AudioTranscription | None] = NOT_GIVEN,
        input_audio_noise_reduction: NotGivenOr[
            NoiseReductionType | NoiseReduction | None
        ] = NOT_GIVEN,
        turn_detection: NotGivenOr[RealtimeAudioInputTurnDetection | None] = NOT_GIVEN,
        tool_choice: NotGivenOr[llm.ToolChoice | None] = NOT_GIVEN,
        speed: NotGivenOr[float] = NOT_GIVEN,
        tracing: NotGivenOr[Tracing | None] = NOT_GIVEN,
        truncation: NotGivenOr[RealtimeTruncation | None] = NOT_GIVEN,
        reasoning: NotGivenOr[RealtimeReasoning | None] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        max_session_duration: NotGivenOr[float | None] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> None:
        if "/" not in model:
            raise ValueError("model must be provider-prefixed, for example 'openai/gpt-realtime'")

        resolved_api_key, resolved_api_secret = resolve_credentials(api_key, api_secret)

        is_xai = model.startswith("xai/")
        resolved_voice = voice if is_given(voice) else "eve" if is_xai else DEFAULT_VOICE
        resolved_transcription = input_audio_transcription
        resolved_turn_detection = turn_detection
        # Preserve whether xAI's server VAD is a default the framework may disable.
        can_disable_turn_detection = not is_given(turn_detection)
        if is_xai:
            if not is_given(resolved_transcription):
                resolved_transcription = _XAI_DEFAULT_INPUT_AUDIO_TRANSCRIPTION
            if not is_given(resolved_turn_detection):
                resolved_turn_detection = _XAI_DEFAULT_TURN_DETECTION

        super().__init__(
            model=model,
            voice=resolved_voice,
            modalities=modalities,
            input_audio_transcription=resolved_transcription,
            input_audio_noise_reduction=input_audio_noise_reduction,
            turn_detection=resolved_turn_detection,
            tool_choice=tool_choice,
            speed=speed,
            tracing=tracing,
            truncation=truncation,
            reasoning=reasoning,
            api_key="livekit-inference",
            base_url=base_url or get_default_inference_url(),
            http_session=http_session,
            max_session_duration=max_session_duration,
            conn_options=conn_options,
        )
        # LiveKit Inference always uses the OpenAI-compatible protocol; ambient Azure
        # settings must not change its URL or session wire format.
        self._opts.is_azure = False
        self._opts.api_version = None
        if is_xai:
            self._capabilities.can_disable_turn_detection = can_disable_turn_detection
        self._inference_opts = _InferenceOptions(
            provider=provider,
            api_key=resolved_api_key,
            api_secret=resolved_api_secret,
            inference_class=inference_class,
        )
        self._provider_label = "LiveKit Inference Realtime"

    @classmethod
    def from_model_string(cls, model: str) -> RealtimeModel:
        """Create a RealtimeModel instance from a model string"""
        return cls(model)

    @property
    def provider(self) -> str:
        return "livekit"

    def session(self, *, turn_detection_disabled: bool = False) -> RealtimeSession:
        sess = RealtimeSession(self, turn_detection_disabled=turn_detection_disabled)
        self._sessions.add(sess)
        return sess

OpenAI-compatible realtime model authenticated through LiveKit Inference.

Initialize a Realtime model client for OpenAI or Azure OpenAI.

Args

model : str

Realtime model name, e.g., "gpt-realtime".

voice : str

Voice used for audio responses. Defaults to "marin".

modalities (list[Literal["text", "audio"]] | NotGiven): Modalities to enable. Defaults to ["text", "audio"] if not provided.

tool_choice : llm.ToolChoice | None | NotGiven

Tool selection policy for responses.

base_url : str | NotGiven

HTTP base URL of the OpenAI/Azure API. If not provided, uses OPENAI_BASE_URL for OpenAI; for Azure, constructed from AZURE_OPENAI_ENDPOINT.

input_audio_transcription : AudioTranscription | None | NotGiven

Options for transcribing input audio.

input_audio_noise_reduction : NoiseReductionType | NoiseReduction | InputAudioNoiseReduction | None | NotGiven

Input audio noise reduction settings.

turn_detection : RealtimeAudioInputTurnDetection | None | NotGiven

Server-side turn-detection options.

speed : float | NotGiven

Audio playback speed multiplier.

tracing : Tracing | None | NotGiven

Tracing configuration for OpenAI Realtime.

truncation : RealtimeTruncation | None | NotGiven

Truncation configuration for OpenAI Realtime.

reasoning : RealtimeReasoning | None | NotGiven

Reasoning config for reasoning-capable models (e.g. gpt-realtime-2), e.g. RealtimeReasoning(effort="low").

api_key : str | None

OpenAI API key. If None and not using Azure, read from OPENAI_API_KEY.

http_session : aiohttp.ClientSession | None

Optional shared HTTP session.

azure_deployment : str | None

Azure deployment name. Presence of any Azure-specific option enables Azure mode.

entra_token : str | None

Azure Entra token auth (alternative to api_key).

max_session_duration : float | None | NotGiven

Seconds before recycling the connection.

conn_options : APIConnectOptions

Retry/backoff and connection settings.

temperature : float | NotGiven

Deprecated; ignored by Realtime v1.

Raises

ValueError

If OPENAI_API_KEY is missing in non-Azure mode, or if Azure endpoint cannot be determined when in Azure mode.

Examples

Basic OpenAI usage:

from livekit.plugins.openai.realtime import RealtimeModel
from openai.types import realtime

model = RealtimeModel(
    voice="marin",
    modalities=["audio"],
    input_audio_transcription=realtime.AudioTranscription(
        model="gpt-4o-transcribe",
    ),
    input_audio_noise_reduction="near_field",
    turn_detection=realtime.realtime_audio_input_turn_detection.SemanticVad(
        type="semantic_vad",
        create_response=True,
        eagerness="auto",
        interrupt_response=True,
    ),
)
session = AgentSession(llm=model)

Ancestors

• livekit.agents.llm._realtime.openai.RealtimeModel
• livekit.agents.llm.realtime.RealtimeModel

Static methods

def from_model_string(model: str) ‑> RealtimeModel

Create a RealtimeModel instance from a model string

Instance variables

prop provider : str

Expand source code

@property
def provider(self) -> str:
    return "livekit"

Methods

def session(self, *, turn_detection_disabled: bool = False) ‑> RealtimeSession

Expand source code

def session(self, *, turn_detection_disabled: bool = False) -> RealtimeSession:
    sess = RealtimeSession(self, turn_detection_disabled=turn_detection_disabled)
    self._sessions.add(sess)
    return sess

Create a new session, optionally with server-side turn detection disabled.

turn_detection_disabled is honored only by plugins reporting can_disable_turn_detection; the model itself is left unchanged and reusable.

class RealtimeSession (realtime_model: RealtimeModel,
*,
turn_detection_disabled: bool = False)

Expand source code

class RealtimeSession(_RealtimeSession):
    def __init__(
        self,
        realtime_model: RealtimeModel,
        *,
        turn_detection_disabled: bool = False,
    ) -> None:
        self._inference_model = realtime_model
        super().__init__(realtime_model, turn_detection_disabled=turn_detection_disabled)

    def _create_ws_url_and_headers(self) -> tuple[str, dict[str, str]]:
        opts = self._inference_model._inference_opts
        url, _ = super()._create_ws_url_and_headers()
        headers = get_inference_headers(inference_class=opts.inference_class)
        headers["Authorization"] = f"Bearer {create_access_token(opts.api_key, opts.api_secret)}"
        if opts.provider:
            headers[HEADER_INFERENCE_PROVIDER] = opts.provider
        return url, headers

    def _wrap_session_update(self, event_id: str, session: Any) -> dict[str, Any]:
        event = super()._wrap_session_update(event_id, session)
        if hasattr(event, "model_dump"):
            event = event.model_dump(by_alias=True, exclude_unset=True, exclude_defaults=False)
        else:
            event = dict(event)

        session_payload = event.get("session")
        if isinstance(session_payload, dict):
            session_payload.pop("model", None)
        return event

    def _is_fatal_error(self, error: object | None) -> bool:
        code = getattr(error, "code", None) or getattr(error, "type", None)
        return (
            isinstance(code, str)
            and code
            in {
                "unsupported_transcription_model",
                "unsupported_audio_transport",
                "unsupported_audio_format",
            }
        ) or super()._is_fatal_error(error)

A session for the OpenAI Realtime API.

This class is used to interact with the OpenAI Realtime API. It is responsible for sending events to the OpenAI Realtime API and receiving events from it.

It exposes two more events: - openai_server_event_received: expose the raw server events from the OpenAI Realtime API - openai_client_event_queued: expose the raw client events sent to the OpenAI Realtime API

Ancestors

• livekit.agents.llm._realtime.openai.RealtimeSession
• livekit.agents.llm.realtime.RealtimeSession
• abc.ABC
• EventEmitter
• typing.Generic

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once