Module livekit.agents.inference.realtime.gpt_live

Classes

class DuplexModel (model: DuplexModels | str,
*,
provider: str | None = None,
base_url: str | None = None,
api_key: str | None = None,
api_secret: str | None = None,
voice: NotGivenOr[str] = NOT_GIVEN,
extra_kwargs: NotGivenOr[OpenAIDuplexOptions | dict[str, Any]] = 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 DuplexModel(_GPTLiveModel):
    """Full-duplex voice model, served through LiveKit Inference."""

    @overload
    def __init__(
        self,
        model: OpenAIDuplexModels,
        *,
        provider: str | None = None,
        base_url: str | None = None,
        api_key: str | None = None,
        api_secret: str | None = None,
        voice: NotGivenOr[str] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[OpenAIDuplexOptions] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        max_session_duration: NotGivenOr[float | None] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> None:
        pass

    @overload
    def __init__(
        self,
        model: str,
        *,
        provider: str | None = None,
        base_url: str | None = None,
        api_key: str | None = None,
        api_secret: str | None = None,
        voice: NotGivenOr[str] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        max_session_duration: NotGivenOr[float | None] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> None:
        pass

    def __init__(
        self,
        model: DuplexModels | str,
        *,
        provider: str | None = None,
        base_url: str | None = None,
        api_key: str | None = None,
        api_secret: str | None = None,
        voice: NotGivenOr[str] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[OpenAIDuplexOptions | dict[str, Any]] = NOT_GIVEN,
        http_session: aiohttp.ClientSession | None = None,
        max_session_duration: NotGivenOr[float | None] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> None:
        """
        Args:
            model: Provider-prefixed duplex model, for example ``openai/gpt-live-1``.
            provider: Host that must serve the model, e.g. ``"openai"``. The session is
                refused if no host in the region matches.
            voice: Output voice, ``marin`` by default for GPT-Live.
            extra_kwargs: Model-specific options; :class:`OpenAIDuplexOptions` for GPT-Live.
                Delegated work runs on the default service tier only.
            http_session: Optional shared HTTP session.
            max_session_duration: Seconds before the connection is recycled.
            conn_options: Retry/backoff and connection settings.
        """
        if "/" not in model:
            raise ValueError("model must be provider-prefixed, for example 'openai/gpt-live-1'")
        options: dict[str, Any] = dict(extra_kwargs) if is_given(extra_kwargs) else {}
        if unknown := options.keys() - OpenAIDuplexOptions.__optional_keys__:
            raise ValueError(f"unsupported extra_kwargs for {model}: {sorted(unknown)}")
        responses_options = options.get("responses_options")
        if responses_options and responses_options.get("service_tier") not in (None, "default"):
            raise ValueError("LiveKit Inference runs GPT-Live delegation on the default tier only")

        resolved_api_key, resolved_api_secret = resolve_credentials(api_key, api_secret)

        super().__init__(
            model=model,
            voice=voice if is_given(voice) else DEFAULT_VOICE,
            delegation=options.get("delegation", "responses"),
            responses_options=responses_options if responses_options is not None else NOT_GIVEN,
            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,
        )
        self._inference_opts = _InferenceOptions(
            provider=provider,
            api_key=resolved_api_key,
            api_secret=resolved_api_secret,
        )
        self._provider_label = "LiveKit Inference Duplex"

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

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

    def session(self) -> DuplexSession:
        return DuplexSession(self)

Full-duplex voice model, served through LiveKit Inference.

Args

model

Provider-prefixed duplex model, for example openai/gpt-live-1.

provider

Host that must serve the model, e.g. "openai". The session is refused if no host in the region matches.

voice

Output voice, marin by default for GPT-Live.

extra_kwargs

Model-specific options; :class:OpenAIDuplexOptions for GPT-Live. Delegated work runs on the default service tier only.

http_session

Optional shared HTTP session.

max_session_duration

Seconds before the connection is recycled.

conn_options

Retry/backoff and connection settings.

Ancestors

• livekit.agents.llm._realtime.gpt_live.GPTLiveModel
• livekit.agents.llm.duplex.DuplexModel
• abc.ABC

Static methods

def from_model_string(model: str) ‑> DuplexModel

Create a DuplexModel instance from a model string

Instance variables

prop provider : str

Expand source code

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

Methods

def session(self) ‑> DuplexSession

Expand source code

def session(self) -> DuplexSession:
    return DuplexSession(self)

Open a session; the adapter configures it with _update_session before use.

class DuplexSession (duplex_model: DuplexModel)

Expand source code

class DuplexSession(_GPTLiveSession):
    def __init__(self, duplex_model: DuplexModel) -> None:
        self._inference_model = duplex_model
        super().__init__(duplex_model)

    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()
        url = f"{url}?{urlencode({'model': self._opts.model})}"
        headers = get_inference_headers()
        # signed per connection, so a reconnect never presents an expired token
        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

    async def _update_tools(self, tools: list[llm.Tool]) -> None:
        hosted = sorted(
            tool.id
            for tool in llm.ToolContext(tools).flatten()
            if isinstance(tool, llm.ProviderTool)
        )
        if hosted:
            # hosted tools bill separately from the delegated tokens the gateway meters
            raise llm.RealtimeError(
                f"LiveKit Inference does not run hosted tools on GPT-Live, so the model can never "
                f"call {hosted}. Use function tools, or the OpenAI plugin's GPTLiveModel directly."
            )
        await super()._update_tools(tools)

    def _is_fatal_error(self, error: types.ErrorBody) -> bool:
        return (error.code or "") in _FATAL_ERROR_CODES or super()._is_fatal_error(error)

A session for the OpenAI GPT-Live API (WebSocket), reached with Agent.duplex_session.

Exposes three extra events: - openai_server_event_received: raw server events - openai_client_event_queued: raw client events sent to the server - delegation_created: a :class:GPTLiveDelegation, under client delegation

The append_* methods queue context and return without waiting. Their *.appended events arrive at the estimated context-injection end; they do not mean speech has finished.

Ancestors

• livekit.agents.llm._realtime.gpt_live.GPTLiveSession
• livekit.agents.llm.duplex.DuplexSession
• abc.ABC
• EventEmitter
• typing.Generic

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once

class OpenAIDuplexOptions (*args, **kwargs)

Expand source code

class OpenAIDuplexOptions(TypedDict, total=False):
    delegation: Literal["responses", "client"]
    """Where delegated work goes, fixed for the life of the session. ``responses`` (the default)
    runs it on a backend model, so ``@function_tool`` works as usual; ``client`` hands it to the
    application as a ``delegation_created`` event."""
    responses_options: OpenAIResponsesOptions

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Ancestors

• builtins.dict

Class variables

var delegation : Literal['responses', 'client']

Where delegated work goes, fixed for the life of the session. responses (the default) runs it on a backend model, so @function_tool works as usual; client hands it to the application as a delegation_created event.

var responses_options : OpenAIResponsesOptions

class OpenAIResponsesOptions (*args, **kwargs)

Expand source code

class OpenAIResponsesOptions(TypedDict, total=False):
    """The backend Responses model GPT-Live delegates work to, under ``delegation="responses"``.

    A key left unset is not sent, and the service's own default applies.
    """

    model: str
    """An OpenAI LLM LiveKit Inference serves; ``gpt-5.6-luna`` when unset."""
    instructions: str
    """Instructions for the backend model, distinct from the voice model's."""
    tool_choice: llm.ToolChoice | None
    parallel_tool_calls: bool
    reasoning: Reasoning
    """Responses reasoning settings, for example ``{"effort": "medium"}``."""
    text: ResponseTextConfigParam
    """Responses text settings, for example ``{"verbosity": "low"}``."""
    max_output_tokens: int
    """Upper bound on the tokens one backend response may generate; at least 16."""

The backend Responses model GPT-Live delegates work to, under delegation="responses".

A key left unset is not sent, and the service's own default applies.

Ancestors

• builtins.dict

Class variables

var instructions : str

Instructions for the backend model, distinct from the voice model's.

var max_output_tokens : int

Upper bound on the tokens one backend response may generate; at least 16.

var model : str

An OpenAI LLM LiveKit Inference serves; gpt-5.6-luna when unset.

var parallel_tool_calls : bool

var reasoning : openai.types.shared_params.reasoning.Reasoning

Responses reasoning settings, for example {"effort": "medium"}.

var text : openai.types.responses.response_text_config_param.ResponseTextConfigParam

Responses text settings, for example {"verbosity": "low"}.

var tool_choice : livekit.agents.llm.tool_context.NamedToolChoice | Literal['auto', 'required', 'none'] | None