Module `livekit.plugins.xai.realtime.realtime_model`

Classes

class FileSearch (vector_store_ids: list[str] = <factory>, max_num_results: int | None = None)

Expand source code

@dataclass(slots=True)
class FileSearch(XAITool):
    """Enable file search tool for searching uploaded document collections."""

    vector_store_ids: list[str] = field(default_factory=list)
    max_num_results: int | None = None

    def to_dict(self) -> dict[str, Any]:
        result: dict[str, Any] = {
            "type": "file_search",
            "vector_store_ids": self.vector_store_ids,
        }
        if self.max_num_results is not None:
            result["max_num_results"] = self.max_num_results

        return result

Enable file search tool for searching uploaded document collections.

Ancestors

XAITool
livekit.agents.llm.tool_context.ProviderTool
livekit.agents.llm.tool_context.Tool
abc.ABC

Instance variables

var max_num_results : int | None

Expand source code

@dataclass(slots=True)
class FileSearch(XAITool):
    """Enable file search tool for searching uploaded document collections."""

    vector_store_ids: list[str] = field(default_factory=list)
    max_num_results: int | None = None

    def to_dict(self) -> dict[str, Any]:
        result: dict[str, Any] = {
            "type": "file_search",
            "vector_store_ids": self.vector_store_ids,
        }
        if self.max_num_results is not None:
            result["max_num_results"] = self.max_num_results

        return result

var vector_store_ids : list[str]

Expand source code

@dataclass(slots=True)
class FileSearch(XAITool):
    """Enable file search tool for searching uploaded document collections."""

    vector_store_ids: list[str] = field(default_factory=list)
    max_num_results: int | None = None

    def to_dict(self) -> dict[str, Any]:
        result: dict[str, Any] = {
            "type": "file_search",
            "vector_store_ids": self.vector_store_ids,
        }
        if self.max_num_results is not None:
            result["max_num_results"] = self.max_num_results

        return result

Methods

def to_dict(self) ‑> dict[str, typing.Any]

Expand source code

def to_dict(self) -> dict[str, Any]:
    result: dict[str, Any] = {
        "type": "file_search",
        "vector_store_ids": self.vector_store_ids,
    }
    if self.max_num_results is not None:
        result["max_num_results"] = self.max_num_results

    return result

class RealtimeModel (*, voice: Literal['Ara', 'Eve', 'Leo', 'Rex', 'Sal'] | str | livekit.agents.types.NotGiven | None = 'Ara', api_key: str | None = None, base_url: str | livekit.agents.types.NotGiven = NOT_GIVEN, turn_detection: openai.types.beta.realtime.session.TurnDetection | livekit.agents.types.NotGiven | None = NOT_GIVEN, http_session: aiohttp.client.ClientSession | None = None, max_session_duration: float | livekit.agents.types.NotGiven | None = NOT_GIVEN, conn_options: livekit.agents.types.APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0))

Expand source code

class RealtimeModel(openai.realtime.RealtimeModel):
    def __init__(
        self,
        *,
        voice: NotGivenOr[GrokVoices | str | None] = "Ara",
        api_key: str | None = None,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        turn_detection: NotGivenOr[TurnDetection | 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:
        api_key = api_key or os.environ.get("XAI_API_KEY")
        if api_key is None:
            raise ValueError(
                "The api_key client option must be set either by passing api_key "
                "to the client or by setting the XAI_API_KEY environment variable"
            )

        resolved_voice = voice if is_given(voice) else "Ara"
        super().__init__(
            base_url=base_url if is_given(base_url) else XAI_BASE_URL,
            model="grok-4-1-fast-non-reasoning",
            voice=resolved_voice,  # type: ignore[arg-type]
            api_key=api_key,
            modalities=["audio"],
            turn_detection=turn_detection
            if is_given(turn_detection)
            else XAI_DEFAULT_TURN_DETECTION,
            http_session=http_session if is_given(http_session) else None,
            max_session_duration=max_session_duration if is_given(max_session_duration) else None,
            conn_options=conn_options,
        )

    def session(self) -> "RealtimeSession":
        sess = RealtimeSession(self)
        self._sessions.add(sess)
        return sess

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.
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).
api_version : str | None: Azure OpenAI API version appended as query parameter.
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.plugins.openai.realtime.realtime_model.RealtimeModel
livekit.agents.llm.realtime.RealtimeModel

Methods

def session(self) ‑> RealtimeSession

Expand source code

def session(self) -> "RealtimeSession":
    sess = RealtimeSession(self)
    self._sessions.add(sess)
    return sess

class RealtimeSession (realtime_model: RealtimeModel)

Expand source code

class RealtimeSession(openai.realtime.RealtimeSession):
    """xAI Realtime Session that supports xAI built-in tools."""

    def __init__(self, realtime_model: RealtimeModel) -> None:
        super().__init__(realtime_model)
        self._xai_model: RealtimeModel = realtime_model

    def _create_tools_update_event(self, tools: list[llm.Tool]) -> dict[str, Any]:
        event = super()._create_tools_update_event(tools)

        # inject supported Toolset
        xai_tools: list[dict[str, Any]] = []
        for tool in tools:
            if isinstance(tool, XAITool):
                xai_tools.append(tool.to_dict())

        event["session"]["tools"] += xai_tools
        return event

xAI Realtime Session that supports xAI built-in tools.

Ancestors

livekit.plugins.openai.realtime.realtime_model.RealtimeSession
livekit.agents.llm.realtime.RealtimeSession
abc.ABC
EventEmitter
typing.Generic

Inherited members

EventEmitter:
- emit
- off
- on
- once

class WebSearch

Expand source code

@dataclass(slots=True)
class WebSearch(XAITool):
    """Enable web search tool for real-time internet searches."""

    def to_dict(self) -> dict[str, Any]:
        return {"type": "web_search"}

Enable web search tool for real-time internet searches.

Ancestors

XAITool
livekit.agents.llm.tool_context.ProviderTool
livekit.agents.llm.tool_context.Tool
abc.ABC

Methods

def to_dict(self) ‑> dict[str, typing.Any]

Expand source code

def to_dict(self) -> dict[str, Any]:
    return {"type": "web_search"}

class XAITool

Expand source code

class XAITool(ProviderTool):
    @abstractmethod
    def to_dict(self) -> dict[str, Any]: ...

Tool that provided by the LLM provider.

Ancestors

livekit.agents.llm.tool_context.ProviderTool
livekit.agents.llm.tool_context.Tool
abc.ABC

Methods

def to_dict(self) ‑> dict[str, typing.Any]

Expand source code

@abstractmethod
def to_dict(self) -> dict[str, Any]: ...

class XSearch (allowed_x_handles: list[str] | None = None)

Expand source code

@dataclass(slots=True)
class XSearch(XAITool):
    """Enable X (Twitter) search tool for searching posts."""

    allowed_x_handles: list[str] | None = None

    def to_dict(self) -> dict[str, Any]:
        result: dict[str, Any] = {"type": "x_search"}
        if self.allowed_x_handles:
            result["allowed_x_handles"] = self.allowed_x_handles
        return result

Enable X (Twitter) search tool for searching posts.

Ancestors

XAITool
livekit.agents.llm.tool_context.ProviderTool
livekit.agents.llm.tool_context.Tool
abc.ABC

Instance variables

var allowed_x_handles : list[str] | None

Expand source code

@dataclass(slots=True)
class XSearch(XAITool):
    """Enable X (Twitter) search tool for searching posts."""

    allowed_x_handles: list[str] | None = None

    def to_dict(self) -> dict[str, Any]:
        result: dict[str, Any] = {"type": "x_search"}
        if self.allowed_x_handles:
            result["allowed_x_handles"] = self.allowed_x_handles
        return result

Methods

def to_dict(self) ‑> dict[str, typing.Any]

Expand source code

def to_dict(self) -> dict[str, Any]:
    result: dict[str, Any] = {"type": "x_search"}
    if self.allowed_x_handles:
        result["allowed_x_handles"] = self.allowed_x_handles
    return result