Module livekit.plugins.openai

OpenAI plugin for LiveKit Agents

Support for OpenAI Realtime API, Decisions API, LLM, TTS, and STT APIs.

Also includes support for a large number of OpenAI-compatible APIs including Azure OpenAI, Cerebras, Fireworks, Perplexity, Telnyx, xAI, Ollama, DeepSeek, OpenRouter, and OVHcloud AI Endpoints.

See https://docs.livekit.io/agents/integrations/openai/ and https://docs.livekit.io/agents/integrations/llm/ for more information.

Sub-modules

livekit.plugins.openai.realtime

livekit.plugins.openai.responses

livekit.plugins.openai.tools

Functions

async def create_embeddings(*,
input: list[str],
model: models.EmbeddingModels = 'text-embedding-3-small',
dimensions: int | None = None,
api_key: str | None = None,
http_session: aiohttp.ClientSession | None = None) ‑> list[livekit.plugins.openai.embeddings.EmbeddingData]

Expand source code

async def create_embeddings(
    *,
    input: list[str],
    model: models.EmbeddingModels = "text-embedding-3-small",
    dimensions: int | None = None,
    api_key: str | None = None,
    http_session: aiohttp.ClientSession | None = None,
) -> list[EmbeddingData]:
    http_session = http_session or utils.http_context.http_session()

    api_key = api_key or os.environ.get("OPENAI_API_KEY")
    if not api_key:
        raise ValueError("OPENAI_API_KEY must be set")

    async with http_session.post(
        "https://api.openai.com/v1/embeddings",
        headers={"Authorization": f"Bearer {api_key}"},
        json={
            "model": model,
            "input": input,
            "encoding_format": "base64",
            "dimensions": dimensions,
        },
    ) as resp:
        json = await resp.json()
        data = json["data"]
        list_data = []
        for d in data:
            bytes = base64.b64decode(d["embedding"])
            num_floats = len(bytes) // 4
            floats = list(struct.unpack("f" * num_floats, bytes))
            list_data.append(EmbeddingData(index=d["index"], embedding=floats))

        return list_data

Classes

class DecisionModel (*,
model: DecisionModels | str = 'gpt-6-luna-decisions',
api_key: NotGivenOr[str] = NOT_GIVEN,
base_url: NotGivenOr[str] = NOT_GIVEN,
client: openai.AsyncClient | None = None,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
max_retries: NotGivenOr[int] = NOT_GIVEN)

Expand source code

class DecisionModel(llm.DecisionModel):
    """OpenAI's Decisions API (``POST /v1/decisions``).

    One request answers a batch of probability, choice, and score questions about the
    same input.

    Example::

        model = openai.DecisionModel()
        response = await model.evaluate(
            input=chat_ctx,
            decisions={"handoff": llm.Probability("The caller requests a human.")},
        )

    Args:
        model: OpenAI decision model identifier.
        api_key: Defaults to OPENAI_API_KEY.
        base_url: Defaults to the OpenAI API.
        client: Optional caller-owned client, used instead of the arguments above.
        safety_identifier: Opaque end-user identifier passed to OpenAI.
        max_retries: Retries inside the OpenAI client. Decisions are retried by
            ``evaluate`` through ``conn_options``, so this defaults to none.
    """

    def __init__(
        self,
        *,
        model: DecisionModels | str = "gpt-6-luna-decisions",
        api_key: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        client: openai.AsyncClient | None = None,
        safety_identifier: NotGivenOr[str] = NOT_GIVEN,
        max_retries: NotGivenOr[int] = NOT_GIVEN,
    ) -> None:
        super().__init__(
            capabilities=DecisionCapabilities(decision_kinds=["probability", "choice", "score"])
        )
        if is_given(api_key) and not api_key:
            raise ValueError(
                "OpenAI API key is required, either as argument or set"
                " OPENAI_API_KEY environment variable"
            )

        self._model = model
        self._safety_identifier = safety_identifier
        self._owns_client = client is None
        self._client = client or openai.AsyncClient(
            api_key=api_key if is_given(api_key) else None,
            base_url=base_url if is_given(base_url) else None,
            max_retries=max_retries if is_given(max_retries) else 0,
        )

    @property
    def model(self) -> str:
        return self._model

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

    async def _evaluate_impl(
        self,
        *,
        input: str,
        decisions: Mapping[str, Decision],
        conn_options: APIConnectOptions,
    ) -> DecisionResponse:
        names = list(decisions)
        try:
            raw = await self._client.decisions.with_raw_response.create(
                model=self._model,
                input=input,
                questions=[_encode_question(name, decisions[name]) for name in names],
                safety_identifier=self._safety_identifier
                if is_given(self._safety_identifier)
                else openai.omit,
                timeout=conn_options.timeout,
            )
        except openai.APITimeoutError:
            raise APITimeoutError("OpenAI decision request timed out") from None
        except openai.APIStatusError as e:
            raise APIStatusError(
                e.message, status_code=e.status_code, request_id=e.request_id, body=e.body
            ) from None
        except openai.APIError as e:
            raise APIConnectionError("OpenAI decision request failed") from e

        try:
            decision = raw.parse()
        except (ValidationError, ValueError) as e:
            raise APIError("invalid OpenAI decision response", retryable=False) from e

        results, errors = _decode_answers(decision.answers, names)
        return DecisionResponse(
            results=results,
            errors=errors,
            model=decision.model,
            provider=self.provider,
            request_id=raw.request_id or "",
            input_tokens=decision.usage.input_tokens,
            output_tokens=decision.usage.output_tokens,
        )

    async def aclose(self) -> None:
        if self._owns_client:
            await self._client.close()

OpenAI's Decisions API (POST /v1/decisions).

One request answers a batch of probability, choice, and score questions about the same input.

Example::

model = openai.DecisionModel()
response = await model.evaluate(
    input=chat_ctx,
    decisions={"handoff": llm.Probability("The caller requests a human.")},
)

Args

model

OpenAI decision model identifier.

api_key

Defaults to OPENAI_API_KEY.

base_url

Defaults to the OpenAI API.

client

Optional caller-owned client, used instead of the arguments above.

safety_identifier

Opaque end-user identifier passed to OpenAI.

max_retries

Retries inside the OpenAI client. Decisions are retried by evaluate through conn_options, so this defaults to none.

Ancestors

• livekit.agents.llm.decisions.DecisionModel
• abc.ABC
• EventEmitter
• typing.Generic

Instance variables

prop model : str

Expand source code

@property
def model(self) -> str:
    return self._model

prop provider : str

Expand source code

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

Methods

async def aclose(self) ‑> None

Expand source code

async def aclose(self) -> None:
    if self._owns_client:
        await self._client.close()

Release resources owned by this model, if any.

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once

class EmbeddingData (index: int, embedding: list[float])

Expand source code

@dataclass
class EmbeddingData:
    index: int
    embedding: list[float]

EmbeddingData(index: 'int', embedding: 'list[float]')

Instance variables

var embedding : list[float]

var index : int

class LLM (*,
model: str | ChatModels = 'gpt-4.1',
api_key: NotGivenOr[str] = NOT_GIVEN,
base_url: NotGivenOr[str] = NOT_GIVEN,
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
store: NotGivenOr[bool] = NOT_GIVEN,
metadata: NotGivenOr[dict[str, str]] = NOT_GIVEN,
max_completion_tokens: NotGivenOr[int] = NOT_GIVEN,
timeout: httpx.Timeout | None = None,
max_retries: NotGivenOr[int] = NOT_GIVEN,
service_tier: NotGivenOr[str] = NOT_GIVEN,
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
verbosity: NotGivenOr[Verbosity] = NOT_GIVEN,
prompt_cache_retention: NotGivenOr[PromptCacheRetention] = NOT_GIVEN,
prompt_cache_options: NotGivenOr[PromptCacheOptions] = NOT_GIVEN,
prompt_cache_breakpoints: "bool | Literal['auto']" = 'auto',
extra_body: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
extra_headers: NotGivenOr[dict[str, str]] = NOT_GIVEN,
extra_query: NotGivenOr[dict[str, str]] = NOT_GIVEN)

Expand source code

class LLM(llm.LLM):
    def __init__(
        self,
        *,
        model: str | ChatModels = "gpt-4.1",
        api_key: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        client: openai.AsyncClient | None = None,
        user: NotGivenOr[str] = NOT_GIVEN,
        safety_identifier: NotGivenOr[str] = NOT_GIVEN,
        prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
        store: NotGivenOr[bool] = NOT_GIVEN,
        metadata: NotGivenOr[dict[str, str]] = NOT_GIVEN,
        max_completion_tokens: NotGivenOr[int] = NOT_GIVEN,
        timeout: httpx.Timeout | None = None,
        max_retries: NotGivenOr[int] = NOT_GIVEN,
        service_tier: NotGivenOr[str] = NOT_GIVEN,
        reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
        verbosity: NotGivenOr[Verbosity] = NOT_GIVEN,
        prompt_cache_retention: NotGivenOr[PromptCacheRetention] = NOT_GIVEN,
        prompt_cache_options: NotGivenOr[PromptCacheOptions] = NOT_GIVEN,
        prompt_cache_breakpoints: bool | Literal["auto"] = "auto",
        extra_body: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
        extra_headers: NotGivenOr[dict[str, str]] = NOT_GIVEN,
        extra_query: NotGivenOr[dict[str, str]] = NOT_GIVEN,
        _provider_fmt: NotGivenOr[str] = NOT_GIVEN,
        _strict_tool_schema: bool = True,
    ) -> None:
        """
        Create a new instance of OpenAI LLM.

        ``api_key`` must be set to your OpenAI API key, either using the argument or by setting the
        ``OPENAI_API_KEY`` environmental variable.

        ``prompt_cache_breakpoints`` sends each :class:`livekit.agents.llm.CacheBreakpoint` in the
        chat context as an OpenAI ``prompt_cache_breakpoint``. ``"auto"`` does so against
        ``api.openai.com`` for models that accept the field (GPT-5.6 and later); Azure and
        OpenAI-compatible endpoints opt in with ``True``.
        """
        super().__init__()

        if not is_given(reasoning_effort) and _supports_reasoning_effort(model):
            if model in [
                "gpt-5.1",
                "gpt-5.2",
                "gpt-5.4",
                "gpt-5.4-mini",
                "gpt-5.5",
                "gpt-5.6-luna",
                "gpt-5.6-sol",
                "gpt-5.6-terra",
            ]:
                reasoning_effort = "none"
            else:
                reasoning_effort = "minimal"

        self._opts = _LLMOptions(
            model=model,
            user=user,
            temperature=temperature,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            store=store,
            metadata=metadata,
            max_completion_tokens=max_completion_tokens,
            service_tier=service_tier,
            reasoning_effort=reasoning_effort,
            safety_identifier=safety_identifier,
            prompt_cache_key=prompt_cache_key,
            top_p=top_p,
            verbosity=verbosity,
            prompt_cache_retention=prompt_cache_retention,
            prompt_cache_options=prompt_cache_options,
            prompt_cache_breakpoints=prompt_cache_breakpoints,
            extra_body=extra_body,
            extra_headers=extra_headers,
            extra_query=extra_query,
        )
        if is_given(api_key) and not api_key:
            raise ValueError(
                "OpenAI API key is required, either as argument or set"
                " OPENAI_API_KEY environment variable"
            )

        self._provider_fmt = _provider_fmt or "openai"
        self._strict_tool_schema = _strict_tool_schema
        self._owns_client = client is None
        self._client = client or openai.AsyncClient(
            api_key=api_key if is_given(api_key) else None,
            base_url=base_url if is_given(base_url) else None,
            max_retries=max_retries if is_given(max_retries) else 0,
            http_client=httpx.AsyncClient(
                timeout=timeout
                if timeout
                else httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0),
                follow_redirects=True,
                limits=httpx.Limits(
                    max_connections=50,
                    max_keepalive_connections=50,
                    keepalive_expiry=120,
                ),
            ),
        )

    async def _prewarm_impl(self) -> None:
        # token-free request supported by openai and openai-compatible servers
        await self._client.models.list()

    async def aclose(self) -> None:
        await super().aclose()

        if self._owns_client:
            await self._client.close()

    @property
    def model(self) -> str:
        return self._opts.model

    @property
    def provider(self) -> str:
        return self._client._base_url.netloc.decode("utf-8")

    @staticmethod
    def with_azure(
        *,
        model: str | ChatModels = "gpt-4o",
        azure_endpoint: str | None = None,
        azure_deployment: str | None = None,
        api_version: str | None = None,
        api_key: str | None = None,
        azure_ad_token: str | None = None,
        azure_ad_token_provider: AsyncAzureADTokenProvider | None = None,
        organization: str | None = None,
        project: str | None = None,
        base_url: str | None = None,
        user: NotGivenOr[str] = NOT_GIVEN,
        safety_identifier: NotGivenOr[str] = NOT_GIVEN,
        prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
        timeout: httpx.Timeout | None = None,
        reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
        verbosity: NotGivenOr[Verbosity] = NOT_GIVEN,
        max_completion_tokens: NotGivenOr[int] = NOT_GIVEN,
        prompt_cache_breakpoints: bool | Literal["auto"] = "auto",
    ) -> LLM:
        """
        This automatically infers the following arguments from their corresponding environment variables if they are not provided:
        - `api_key` from `AZURE_OPENAI_API_KEY`
        - `organization` from `OPENAI_ORG_ID`
        - `project` from `OPENAI_PROJECT_ID`
        - `azure_ad_token` from `AZURE_OPENAI_AD_TOKEN`
        - `api_version` from `OPENAI_API_VERSION`
        - `azure_endpoint` from `AZURE_OPENAI_ENDPOINT`
        """  # noqa: E501

        azure_client = openai.AsyncAzureOpenAI(
            max_retries=0,
            azure_endpoint=azure_endpoint,
            azure_deployment=azure_deployment,
            api_version=api_version,
            api_key=api_key,
            azure_ad_token=azure_ad_token,
            azure_ad_token_provider=azure_ad_token_provider,
            organization=organization,
            project=project,
            base_url=base_url,
            timeout=timeout
            if timeout
            else httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0),
        )  # type: ignore

        llm = LLM(
            model=model,
            client=azure_client,
            user=user,
            temperature=temperature,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            reasoning_effort=reasoning_effort,
            safety_identifier=safety_identifier,
            prompt_cache_key=prompt_cache_key,
            top_p=top_p,
            verbosity=verbosity,
            max_completion_tokens=max_completion_tokens,
            prompt_cache_breakpoints=prompt_cache_breakpoints,
        )
        llm._owns_client = True
        return llm

    @staticmethod
    def with_cerebras(
        *,
        model: str | CerebrasChatModels = "gpt-oss-120b",
        api_key: str | None = None,
        base_url: str = "https://api.cerebras.ai/v1",
        client: openai.AsyncClient | None = None,
        user: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
        reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
        safety_identifier: NotGivenOr[str] = NOT_GIVEN,
        prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
    ) -> LLM:
        """
        Create a new instance of Cerebras LLM.

        ``api_key`` must be set to your Cerebras API key, either using the argument or by setting
        the ``CEREBRAS_API_KEY`` environment variable.
        """

        api_key = api_key or os.environ.get("CEREBRAS_API_KEY")
        if api_key is None:
            raise ValueError(
                "Cerebras API key is required, either as argument or set CEREBRAS_API_KEY environment variable"  # noqa: E501
            )

        return LLM(
            model=model,
            api_key=api_key,
            base_url=base_url,
            client=client,
            user=user,
            temperature=temperature,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            reasoning_effort=reasoning_effort,
            safety_identifier=safety_identifier,
            prompt_cache_key=prompt_cache_key,
            top_p=top_p,
            _strict_tool_schema=False,
        )

    @staticmethod
    def with_sambanova(
        *,
        model: str | SambaNovaChatModels = "DeepSeek-R1-0528",
        api_key: str | None = None,
        base_url: str = "https://api.sambanova.ai/v1",
        client: openai.AsyncClient | None = None,
        user: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: ToolChoice = "auto",
        reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
        safety_identifier: NotGivenOr[str] = NOT_GIVEN,
        prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
    ) -> LLM:
        """
        Create a new instance of SambaNova LLM (OpenAI-compatible).

        ``api_key`` must be set to your SambaNova API key, either using the argument or by setting
        the ``SAMBANOVA_API_KEY`` environment variable.
        """

        api_key = api_key or os.environ.get("SAMBANOVA_API_KEY")
        if api_key is None:
            raise ValueError(
                "SambaNova API key is required, either as argument or set SAMBANOVA_API_KEY environment variable"
            )

        return LLM(
            model=model,
            api_key=api_key,
            base_url=base_url,
            client=client,
            user=user,
            temperature=temperature,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            reasoning_effort=reasoning_effort,
            safety_identifier=safety_identifier,
            prompt_cache_key=prompt_cache_key,
            top_p=top_p,
            _strict_tool_schema=False,
        )

    @staticmethod
    def with_fireworks(
        *,
        model: str = "accounts/fireworks/models/llama-v3p3-70b-instruct",
        api_key: str | None = None,
        base_url: str = "https://api.fireworks.ai/inference/v1",
        client: openai.AsyncClient | None = None,
        user: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: ToolChoice = "auto",
        reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
        safety_identifier: NotGivenOr[str] = NOT_GIVEN,
        prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
    ) -> LLM:
        """
        Create a new instance of Fireworks LLM.

        ``api_key`` must be set to your Fireworks API key, either using the argument or by setting
        the ``FIREWORKS_API_KEY`` environmental variable.
        """

        api_key = api_key or os.environ.get("FIREWORKS_API_KEY")
        if api_key is None:
            raise ValueError(
                "Fireworks API key is required, either as argument or set FIREWORKS_API_KEY environmental variable"  # noqa: E501
            )

        return LLM(
            model=model,
            api_key=api_key,
            base_url=base_url,
            client=client,
            user=user,
            temperature=temperature,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            reasoning_effort=reasoning_effort,
            safety_identifier=safety_identifier,
            prompt_cache_key=prompt_cache_key,
            top_p=top_p,
        )

    @staticmethod
    def with_x_ai(
        *,
        model: str | XAIChatModels = "grok-3-fast",
        api_key: str | None = None,
        base_url: str = "https://api.x.ai/v1",
        client: openai.AsyncClient | None = None,
        user: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: ToolChoice = "auto",
        reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
        safety_identifier: NotGivenOr[str] = NOT_GIVEN,
        prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
    ) -> LLM:
        """
        Create a new instance of XAI LLM.

        ``api_key`` must be set to your XAI API key, either using the argument or by setting
        the ``XAI_API_KEY`` environmental variable.
        """
        api_key = api_key or os.environ.get("XAI_API_KEY")
        if api_key is None:
            raise ValueError(
                "XAI API key is required, either as argument or set XAI_API_KEY environmental variable"  # noqa: E501
            )

        return LLM(
            model=model,
            api_key=api_key,
            base_url=base_url,
            client=client,
            user=user,
            temperature=temperature,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            # TODO(long): add provider fmt for grok
            reasoning_effort=reasoning_effort,
            safety_identifier=safety_identifier,
            prompt_cache_key=prompt_cache_key,
            top_p=top_p,
        )

    @staticmethod
    def with_openrouter(
        *,
        model: str = "auto",
        api_key: str | None = None,
        base_url: str = "https://openrouter.ai/api/v1",
        client: openai.AsyncClient | None = None,
        site_url: str | None = None,
        app_name: str | None = None,
        fallback_models: list[str] | None = None,
        provider: OpenRouterProviderPreferences | None = None,
        plugins: list[OpenRouterWebPlugin] | None = None,
        user: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: ToolChoice = "auto",
        reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
        safety_identifier: NotGivenOr[str] = NOT_GIVEN,
        prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
        timeout: httpx.Timeout | None = None,
    ) -> LLM:
        """
        Create a new instance of OpenRouter LLM.

        ``api_key`` must be set to your OpenRouter API key, either using the argument or by setting
        the ``OPENROUTER_API_KEY`` environment variable.
        """

        api_key = api_key or os.environ.get("OPENROUTER_API_KEY")
        if api_key is None:
            raise ValueError(
                "OpenRouter API key is required, either as argument or set OPENROUTER_API_KEY environment variable"
            )

        # Set up analytics headers for OpenRouter
        default_headers: dict[str, str] = {}
        if site_url:
            default_headers["HTTP-Referer"] = site_url
        if app_name:
            default_headers["X-Title"] = app_name

        # Build OpenRouter-specific request body
        or_body: dict[str, Any] = {}
        if provider:
            or_body["provider"] = provider
        if fallback_models:
            # Set fallback models for routing
            or_body["models"] = [model, *fallback_models]
        if plugins:
            or_body["plugins"] = [
                {k: v for k, v in asdict(p).items() if v is not None} for p in plugins
            ]

        return LLM(
            model=model,
            api_key=api_key,
            client=client,
            base_url=base_url,
            user=user,
            temperature=temperature,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            reasoning_effort=reasoning_effort,
            safety_identifier=safety_identifier,
            prompt_cache_key=prompt_cache_key,
            top_p=top_p,
            extra_body=or_body,
            extra_headers=default_headers,
            timeout=timeout,
        )

    @staticmethod
    def with_deepseek(
        *,
        model: str | DeepSeekChatModels = "deepseek-chat",
        api_key: str | None = None,
        base_url: str = "https://api.deepseek.com/v1",
        client: openai.AsyncClient | None = None,
        user: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: ToolChoice = "auto",
        reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
        safety_identifier: NotGivenOr[str] = NOT_GIVEN,
        prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
    ) -> LLM:
        """
        Create a new instance of DeepSeek LLM.

        ``api_key`` must be set to your DeepSeek API key, either using the argument or by setting
        the ``DEEPSEEK_API_KEY`` environmental variable.
        """

        api_key = api_key or os.environ.get("DEEPSEEK_API_KEY")
        if api_key is None:
            raise ValueError(
                "DeepSeek API key is required, either as argument or set DEEPSEEK_API_KEY environmental variable"  # noqa: E501
            )

        return LLM(
            model=model,
            api_key=api_key,
            base_url=base_url,
            client=client,
            user=user,
            temperature=temperature,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            reasoning_effort=reasoning_effort,
            safety_identifier=safety_identifier,
            prompt_cache_key=prompt_cache_key,
            top_p=top_p,
        )

    @staticmethod
    def with_cometapi(
        *,
        model: str | CometAPIChatModels = "gpt-5-chat-latest",
        api_key: str | None = None,
        base_url: str = "https://api.cometapi.com/v1/",
        client: openai.AsyncClient | None = None,
        user: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: ToolChoice = "auto",
        reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
        safety_identifier: NotGivenOr[str] = NOT_GIVEN,
        prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
    ) -> LLM:
        """
        Create a new instance of CometAPI LLM.

        ``api_key`` must be set to your CometAPI API key, either using the argument or by setting
        the ``COMETAPI_API_KEY`` environmental variable.

        CometAPI provides access to 500+ AI models from multiple providers including OpenAI,
        Anthropic, Google, xAI, DeepSeek, and Qwen through a unified API.

        Get your API key at: https://api.cometapi.com/console/token
        Learn more: https://www.cometapi.com/?utm_source=livekit&utm_campaign=integration&utm_medium=integration&utm_content=integration
        """

        api_key = api_key or os.environ.get("COMETAPI_API_KEY")
        if api_key is None:
            raise ValueError(
                "CometAPI API key is required, either as argument or set COMETAPI_API_KEY environmental variable"  # noqa: E501
            )

        return LLM(
            model=model,
            api_key=api_key,
            base_url=base_url,
            client=client,
            user=user,
            temperature=temperature,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            reasoning_effort=reasoning_effort,
            safety_identifier=safety_identifier,
            prompt_cache_key=prompt_cache_key,
            top_p=top_p,
        )

    @staticmethod
    def with_octo(
        *,
        model: str | OctoChatModels = "llama-2-13b-chat",
        api_key: str | None = None,
        base_url: str = "https://text.octoai.run/v1",
        client: openai.AsyncClient | None = None,
        user: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: ToolChoice = "auto",
        reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
        safety_identifier: NotGivenOr[str] = NOT_GIVEN,
        prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
    ) -> LLM:
        """
        Create a new instance of OctoAI LLM.

        ``api_key`` must be set to your OctoAI API key, either using the argument or by setting
        the ``OCTOAI_TOKEN`` environmental variable.
        """

        api_key = api_key or os.environ.get("OCTOAI_TOKEN")
        if api_key is None:
            raise ValueError(
                "OctoAI API key is required, either as argument or set OCTOAI_TOKEN environmental variable"  # noqa: E501
            )

        return LLM(
            model=model,
            api_key=api_key,
            base_url=base_url,
            client=client,
            user=user,
            temperature=temperature,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            reasoning_effort=reasoning_effort,
            safety_identifier=safety_identifier,
            prompt_cache_key=prompt_cache_key,
            top_p=top_p,
        )

    @staticmethod
    def with_ollama(
        *,
        model: str = "llama3.1",
        base_url: str = "http://localhost:11434/v1",
        client: openai.AsyncClient | None = None,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: ToolChoice = "auto",
        reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
        safety_identifier: NotGivenOr[str] = NOT_GIVEN,
        prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
    ) -> LLM:
        """
        Create a new instance of Ollama LLM.
        """

        return LLM(
            model=model,
            api_key="ollama",
            base_url=base_url,
            client=client,
            temperature=temperature,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            reasoning_effort=reasoning_effort,
            safety_identifier=safety_identifier,
            prompt_cache_key=prompt_cache_key,
            top_p=top_p,
        )

    @staticmethod
    def with_ovhcloud(
        *,
        model: str = "gpt-oss-120b",
        api_key: str | None = None,
        base_url: str = "https://oai.endpoints.kepler.ai.cloud.ovh.net/v1",
        client: openai.AsyncClient | None = None,
        user: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: ToolChoice = "auto",
        reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
        safety_identifier: NotGivenOr[str] = NOT_GIVEN,
        prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
    ) -> LLM:
        """
        Create a new instance of OVHcloud AI Endpoints LLM.

        ``api_key`` must be set to your OVHcloud AI Endpoints API key, either using the argument or by setting
        the ``OVHCLOUD_API_KEY`` environmental variable.
        """

        api_key = api_key or os.environ.get("OVHCLOUD_API_KEY")
        if api_key is None:
            raise ValueError(
                "OVHcloud AI Endpoints API key is required, either as argument or set OVHCLOUD_API_KEY environmental variable"  # noqa: E501
            )

        return LLM(
            model=model,
            api_key=api_key,
            base_url=base_url,
            client=client,
            user=user,
            temperature=temperature,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            reasoning_effort=reasoning_effort,
            safety_identifier=safety_identifier,
            prompt_cache_key=prompt_cache_key,
            top_p=top_p,
        )

    @staticmethod
    def with_perplexity(
        *,
        model: str | PerplexityChatModels = "llama-3.1-sonar-small-128k-chat",
        api_key: str | None = None,
        base_url: str = "https://api.perplexity.ai",
        client: openai.AsyncClient | None = None,
        user: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: ToolChoice = "auto",
        reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
        safety_identifier: NotGivenOr[str] = NOT_GIVEN,
        prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
    ) -> LLM:
        """
        Create a new instance of PerplexityAI LLM.

        ``api_key`` must be set to your TogetherAI API key, either using the argument or by setting
        the ``PERPLEXITY_API_KEY`` environmental variable.
        """

        api_key = api_key or os.environ.get("PERPLEXITY_API_KEY")
        if api_key is None:
            raise ValueError(
                "Perplexity AI API key is required, either as argument or set PERPLEXITY_API_KEY environmental variable"  # noqa: E501
            )

        return LLM(
            model=model,
            api_key=api_key,
            base_url=base_url,
            client=client,
            user=user,
            temperature=temperature,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            reasoning_effort=reasoning_effort,
            safety_identifier=safety_identifier,
            prompt_cache_key=prompt_cache_key,
            top_p=top_p,
        )

    @staticmethod
    def with_together(
        *,
        model: str | TogetherChatModels = "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo",
        api_key: str | None = None,
        base_url: str = "https://api.together.xyz/v1",
        client: openai.AsyncClient | None = None,
        user: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: ToolChoice = "auto",
        reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
        safety_identifier: NotGivenOr[str] = NOT_GIVEN,
        prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
    ) -> LLM:
        """
        Create a new instance of TogetherAI LLM.

        ``api_key`` must be set to your TogetherAI API key, either using the argument or by setting
        the ``TOGETHER_API_KEY`` environmental variable.
        """

        api_key = api_key or os.environ.get("TOGETHER_API_KEY")
        if api_key is None:
            raise ValueError(
                "Together AI API key is required, either as argument or set TOGETHER_API_KEY environmental variable"  # noqa: E501
            )

        return LLM(
            model=model,
            api_key=api_key,
            base_url=base_url,
            client=client,
            user=user,
            temperature=temperature,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            reasoning_effort=reasoning_effort,
            safety_identifier=safety_identifier,
            prompt_cache_key=prompt_cache_key,
            top_p=top_p,
        )

    @staticmethod
    def with_telnyx(
        *,
        model: str | TelnyxChatModels = "meta-llama/Meta-Llama-3.1-70B-Instruct",
        api_key: str | None = None,
        base_url: str = "https://api.telnyx.com/v2/ai",
        client: openai.AsyncClient | None = None,
        user: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: ToolChoice = "auto",
        reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
        safety_identifier: NotGivenOr[str] = NOT_GIVEN,
        prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
    ) -> LLM:
        """
        Create a new instance of Telnyx LLM.

        ``api_key`` must be set to your Telnyx API key, either using the argument or by setting
        the ``TELNYX_API_KEY`` environmental variable.
        """

        api_key = api_key or os.environ.get("TELNYX_API_KEY")
        if api_key is None:
            raise ValueError(
                "Telnyx AI API key is required, either as argument or set TELNYX_API_KEY environmental variable"  # noqa: E501
            )

        return LLM(
            model=model,
            api_key=api_key,
            base_url=base_url,
            client=client,
            user=user,
            temperature=temperature,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            reasoning_effort=reasoning_effort,
            safety_identifier=safety_identifier,
            prompt_cache_key=prompt_cache_key,
            top_p=top_p,
        )

    @staticmethod
    def with_nebius(
        *,
        model: str | NebiusChatModels = "meta-llama/Meta-Llama-3.1-70B-Instruct",
        api_key: str | None = None,
        base_url: str = "https://api.studio.nebius.com/v1/",
        client: openai.AsyncClient | None = None,
        user: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: ToolChoice = "auto",
        reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
        safety_identifier: NotGivenOr[str] = NOT_GIVEN,
        prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
        top_p: NotGivenOr[float] = NOT_GIVEN,
    ) -> LLM:
        """
        Create a new instance of Nebius LLM.

        ``api_key`` must be set to your Nebius API key, either using the argument or by setting
        the ``NEBIUS_API_KEY`` environmental variable.
        """

        api_key = api_key or os.environ.get("NEBIUS_API_KEY")
        if api_key is None:
            raise ValueError(
                "Nebius API key is required, either as argument or set NEBIUS_API_KEY environmental variable"  # noqa: E501
            )

        return LLM(
            model=model,
            api_key=api_key,
            base_url=base_url,
            client=client,
            user=user,
            temperature=temperature,
            parallel_tool_calls=parallel_tool_calls,
            tool_choice=tool_choice,
            reasoning_effort=reasoning_effort,
            safety_identifier=safety_identifier,
            prompt_cache_key=prompt_cache_key,
            top_p=top_p,
        )

    @staticmethod
    def with_letta(
        *,
        agent_id: str,
        base_url: str = "https://api.letta.com/v1/chat/completions",
        api_key: str | None = None,
    ) -> LLM:
        """
        Create a new Letta-backed LLM instance connected to the specified Letta agent.

        Args:
            agent_id (str): The Letta agent ID (must be prefixed with 'agent-').
            base_url (str): The URL of the Letta server (e.g., http://localhost:8283/v1/chat/completions for local or https://api.letta.com/v1/chat/completions for cloud).
            api_key (str | None, optional): Optional API key for authentication, required if
                                            the Letta server enforces auth.

        Returns:
            LLM: A configured LLM instance for interacting with the given Letta agent.
        """

        parsed = urlparse(base_url)
        if parsed.scheme not in {"http", "https"}:
            raise ValueError(f"Invalid URL scheme: '{parsed.scheme}'. Must be 'http' or 'https'.")
        if not parsed.netloc:
            raise ValueError(f"URL '{base_url}' is missing a network location (e.g., domain name).")

        api_key = api_key or os.environ.get("LETTA_API_KEY")
        if api_key is None:
            raise ValueError(
                "Letta API key is required, either as argument or set LETTA_API_KEY environmental variable"  # noqa: E501
            )

        return LLM(
            model=agent_id,
            api_key=api_key,
            base_url=base_url,
            client=None,
            user=NOT_GIVEN,
            temperature=NOT_GIVEN,
            parallel_tool_calls=NOT_GIVEN,
            tool_choice=NOT_GIVEN,
        )

    def chat(
        self,
        *,
        chat_ctx: ChatContext,
        tools: list[llm.Tool] | None = None,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
        parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
        tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
        response_format: NotGivenOr[
            completion_create_params.ResponseFormat | type[llm_utils.ResponseFormatT]
        ] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
    ) -> LLMStream:
        extra = {}
        if is_given(extra_kwargs):
            extra.update(extra_kwargs)

        if is_given(self._opts.extra_body):
            extra["extra_body"] = self._opts.extra_body

        if is_given(self._opts.extra_headers):
            extra["extra_headers"] = self._opts.extra_headers

        if is_given(self._opts.extra_query):
            extra["extra_query"] = self._opts.extra_query

        if is_given(self._opts.metadata):
            extra["metadata"] = self._opts.metadata

        if is_given(self._opts.user):
            extra["user"] = self._opts.user

        if is_given(self._opts.max_completion_tokens):
            extra["max_completion_tokens"] = self._opts.max_completion_tokens

        if is_given(self._opts.temperature):
            extra["temperature"] = self._opts.temperature

        if is_given(self._opts.service_tier):
            extra["service_tier"] = self._opts.service_tier

        if is_given(self._opts.reasoning_effort):
            extra["reasoning_effort"] = self._opts.reasoning_effort

        if is_given(self._opts.safety_identifier):
            extra["safety_identifier"] = self._opts.safety_identifier

        if is_given(self._opts.prompt_cache_key):
            extra["prompt_cache_key"] = self._opts.prompt_cache_key

        if is_given(self._opts.top_p):
            extra["top_p"] = self._opts.top_p

        if is_given(self._opts.verbosity):
            extra["verbosity"] = self._opts.verbosity

        if is_given(self._opts.prompt_cache_retention):
            extra["prompt_cache_retention"] = self._opts.prompt_cache_retention

        if is_given(self._opts.prompt_cache_options):
            extra["prompt_cache_options"] = self._opts.prompt_cache_options

        parallel_tool_calls = (
            parallel_tool_calls if is_given(parallel_tool_calls) else self._opts.parallel_tool_calls
        )
        if is_given(parallel_tool_calls):
            extra["parallel_tool_calls"] = parallel_tool_calls

        tool_choice = tool_choice if is_given(tool_choice) else self._opts.tool_choice
        if is_given(tool_choice):
            oai_tool_choice: ChatCompletionToolChoiceOptionParam
            if isinstance(tool_choice, dict):
                oai_tool_choice = {
                    "type": "function",
                    "function": {"name": tool_choice["function"]["name"]},
                }
                extra["tool_choice"] = oai_tool_choice
            elif tool_choice in ("auto", "required", "none"):
                oai_tool_choice = tool_choice
                extra["tool_choice"] = oai_tool_choice

        if is_given(response_format):
            extra["response_format"] = llm_utils.to_openai_response_format(response_format)  # type: ignore

        return LLMStream(
            self,
            model=self._opts.model,
            provider_fmt=self._provider_fmt,
            strict_tool_schema=self._strict_tool_schema,
            client=self._client,
            chat_ctx=chat_ctx,
            tools=tools or [],
            conn_options=conn_options,
            extra_kwargs=extra,
            prompt_cache_breakpoints=self._resolve_prompt_cache_breakpoints(),
        )

    def _resolve_prompt_cache_breakpoints(self) -> bool:
        setting = self._opts.prompt_cache_breakpoints
        if isinstance(setting, bool):
            return setting
        # Azure deployment names hide the model version and OpenAI-compatible vendors have
        # not been verified to accept the field, so only api.openai.com resolves on its own
        return (
            self._client._base_url.host == "api.openai.com"
            and supports_prompt_cache_breakpoints(self._opts.model)
        )

Helper class that provides a standard way to create an ABC using inheritance.

Create a new instance of OpenAI LLM.

api_key must be set to your OpenAI API key, either using the argument or by setting the OPENAI_API_KEY environmental variable.

prompt_cache_breakpoints sends each :class:CacheBreakpoint in the chat context as an OpenAI prompt_cache_breakpoint. "auto" does so against api.openai.com for models that accept the field (GPT-5.6 and later); Azure and OpenAI-compatible endpoints opt in with True.

Ancestors

• livekit.agents.llm.llm.LLM
• abc.ABC
• EventEmitter
• typing.Generic

Subclasses

• livekit.plugins.baseten.llm.LLM
• livekit.plugins.cerebras.llm.LLM
• livekit.plugins.groq.services.LLM
• livekit.plugins.perplexity.llm.LLM
• livekit.plugins.sarvam.llm.client.LLM

Static methods

def with_azure(*,
model: str | ChatModels = 'gpt-4o',
azure_endpoint: str | None = None,
azure_deployment: str | None = None,
api_version: str | None = None,
api_key: str | None = None,
azure_ad_token: str | None = None,
azure_ad_token_provider: AsyncAzureADTokenProvider | None = None,
organization: str | None = None,
project: str | None = None,
base_url: str | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
timeout: httpx.Timeout | None = None,
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN,
verbosity: NotGivenOr[Verbosity] = NOT_GIVEN,
max_completion_tokens: NotGivenOr[int] = NOT_GIVEN,
prompt_cache_breakpoints: "bool | Literal['auto']" = 'auto') ‑> livekit.plugins.openai.llm.LLM

Expand source code

@staticmethod
def with_azure(
    *,
    model: str | ChatModels = "gpt-4o",
    azure_endpoint: str | None = None,
    azure_deployment: str | None = None,
    api_version: str | None = None,
    api_key: str | None = None,
    azure_ad_token: str | None = None,
    azure_ad_token_provider: AsyncAzureADTokenProvider | None = None,
    organization: str | None = None,
    project: str | None = None,
    base_url: str | None = None,
    user: NotGivenOr[str] = NOT_GIVEN,
    safety_identifier: NotGivenOr[str] = NOT_GIVEN,
    prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
    temperature: NotGivenOr[float] = NOT_GIVEN,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
    timeout: httpx.Timeout | None = None,
    reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
    top_p: NotGivenOr[float] = NOT_GIVEN,
    verbosity: NotGivenOr[Verbosity] = NOT_GIVEN,
    max_completion_tokens: NotGivenOr[int] = NOT_GIVEN,
    prompt_cache_breakpoints: bool | Literal["auto"] = "auto",
) -> LLM:
    """
    This automatically infers the following arguments from their corresponding environment variables if they are not provided:
    - `api_key` from `AZURE_OPENAI_API_KEY`
    - `organization` from `OPENAI_ORG_ID`
    - `project` from `OPENAI_PROJECT_ID`
    - `azure_ad_token` from `AZURE_OPENAI_AD_TOKEN`
    - `api_version` from `OPENAI_API_VERSION`
    - `azure_endpoint` from `AZURE_OPENAI_ENDPOINT`
    """  # noqa: E501

    azure_client = openai.AsyncAzureOpenAI(
        max_retries=0,
        azure_endpoint=azure_endpoint,
        azure_deployment=azure_deployment,
        api_version=api_version,
        api_key=api_key,
        azure_ad_token=azure_ad_token,
        azure_ad_token_provider=azure_ad_token_provider,
        organization=organization,
        project=project,
        base_url=base_url,
        timeout=timeout
        if timeout
        else httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0),
    )  # type: ignore

    llm = LLM(
        model=model,
        client=azure_client,
        user=user,
        temperature=temperature,
        parallel_tool_calls=parallel_tool_calls,
        tool_choice=tool_choice,
        reasoning_effort=reasoning_effort,
        safety_identifier=safety_identifier,
        prompt_cache_key=prompt_cache_key,
        top_p=top_p,
        verbosity=verbosity,
        max_completion_tokens=max_completion_tokens,
        prompt_cache_breakpoints=prompt_cache_breakpoints,
    )
    llm._owns_client = True
    return llm

This automatically infers the following arguments from their corresponding environment variables if they are not provided: - api_key from AZURE_OPENAI_API_KEY - organization from OPENAI_ORG_ID - project from OPENAI_PROJECT_ID - azure_ad_token from AZURE_OPENAI_AD_TOKEN - api_version from OPENAI_API_VERSION - azure_endpoint from AZURE_OPENAI_ENDPOINT

def with_cerebras(*,
model: str | CerebrasChatModels = 'gpt-oss-120b',
api_key: str | None = None,
base_url: str = 'https://api.cerebras.ai/v1',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM

Expand source code

@staticmethod
def with_cerebras(
    *,
    model: str | CerebrasChatModels = "gpt-oss-120b",
    api_key: str | None = None,
    base_url: str = "https://api.cerebras.ai/v1",
    client: openai.AsyncClient | None = None,
    user: NotGivenOr[str] = NOT_GIVEN,
    temperature: NotGivenOr[float] = NOT_GIVEN,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
    reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
    safety_identifier: NotGivenOr[str] = NOT_GIVEN,
    prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
    top_p: NotGivenOr[float] = NOT_GIVEN,
) -> LLM:
    """
    Create a new instance of Cerebras LLM.

    ``api_key`` must be set to your Cerebras API key, either using the argument or by setting
    the ``CEREBRAS_API_KEY`` environment variable.
    """

    api_key = api_key or os.environ.get("CEREBRAS_API_KEY")
    if api_key is None:
        raise ValueError(
            "Cerebras API key is required, either as argument or set CEREBRAS_API_KEY environment variable"  # noqa: E501
        )

    return LLM(
        model=model,
        api_key=api_key,
        base_url=base_url,
        client=client,
        user=user,
        temperature=temperature,
        parallel_tool_calls=parallel_tool_calls,
        tool_choice=tool_choice,
        reasoning_effort=reasoning_effort,
        safety_identifier=safety_identifier,
        prompt_cache_key=prompt_cache_key,
        top_p=top_p,
        _strict_tool_schema=False,
    )

Create a new instance of Cerebras LLM.

api_key must be set to your Cerebras API key, either using the argument or by setting the CEREBRAS_API_KEY environment variable.

def with_cometapi(*,
model: str | CometAPIChatModels = 'gpt-5-chat-latest',
api_key: str | None = None,
base_url: str = 'https://api.cometapi.com/v1/',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM

Expand source code

@staticmethod
def with_cometapi(
    *,
    model: str | CometAPIChatModels = "gpt-5-chat-latest",
    api_key: str | None = None,
    base_url: str = "https://api.cometapi.com/v1/",
    client: openai.AsyncClient | None = None,
    user: NotGivenOr[str] = NOT_GIVEN,
    temperature: NotGivenOr[float] = NOT_GIVEN,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: ToolChoice = "auto",
    reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
    safety_identifier: NotGivenOr[str] = NOT_GIVEN,
    prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
    top_p: NotGivenOr[float] = NOT_GIVEN,
) -> LLM:
    """
    Create a new instance of CometAPI LLM.

    ``api_key`` must be set to your CometAPI API key, either using the argument or by setting
    the ``COMETAPI_API_KEY`` environmental variable.

    CometAPI provides access to 500+ AI models from multiple providers including OpenAI,
    Anthropic, Google, xAI, DeepSeek, and Qwen through a unified API.

    Get your API key at: https://api.cometapi.com/console/token
    Learn more: https://www.cometapi.com/?utm_source=livekit&utm_campaign=integration&utm_medium=integration&utm_content=integration
    """

    api_key = api_key or os.environ.get("COMETAPI_API_KEY")
    if api_key is None:
        raise ValueError(
            "CometAPI API key is required, either as argument or set COMETAPI_API_KEY environmental variable"  # noqa: E501
        )

    return LLM(
        model=model,
        api_key=api_key,
        base_url=base_url,
        client=client,
        user=user,
        temperature=temperature,
        parallel_tool_calls=parallel_tool_calls,
        tool_choice=tool_choice,
        reasoning_effort=reasoning_effort,
        safety_identifier=safety_identifier,
        prompt_cache_key=prompt_cache_key,
        top_p=top_p,
    )

Create a new instance of CometAPI LLM.

api_key must be set to your CometAPI API key, either using the argument or by setting the COMETAPI_API_KEY environmental variable.

CometAPI provides access to 500+ AI models from multiple providers including OpenAI, Anthropic, Google, xAI, DeepSeek, and Qwen through a unified API.

Get your API key at: https://api.cometapi.com/console/token Learn more: https://www.cometapi.com/?utm_source=livekit&utm_campaign=integration&utm_medium=integration&utm_content=integration

def with_deepseek(*,
model: str | DeepSeekChatModels = 'deepseek-chat',
api_key: str | None = None,
base_url: str = 'https://api.deepseek.com/v1',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM

Expand source code

@staticmethod
def with_deepseek(
    *,
    model: str | DeepSeekChatModels = "deepseek-chat",
    api_key: str | None = None,
    base_url: str = "https://api.deepseek.com/v1",
    client: openai.AsyncClient | None = None,
    user: NotGivenOr[str] = NOT_GIVEN,
    temperature: NotGivenOr[float] = NOT_GIVEN,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: ToolChoice = "auto",
    reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
    safety_identifier: NotGivenOr[str] = NOT_GIVEN,
    prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
    top_p: NotGivenOr[float] = NOT_GIVEN,
) -> LLM:
    """
    Create a new instance of DeepSeek LLM.

    ``api_key`` must be set to your DeepSeek API key, either using the argument or by setting
    the ``DEEPSEEK_API_KEY`` environmental variable.
    """

    api_key = api_key or os.environ.get("DEEPSEEK_API_KEY")
    if api_key is None:
        raise ValueError(
            "DeepSeek API key is required, either as argument or set DEEPSEEK_API_KEY environmental variable"  # noqa: E501
        )

    return LLM(
        model=model,
        api_key=api_key,
        base_url=base_url,
        client=client,
        user=user,
        temperature=temperature,
        parallel_tool_calls=parallel_tool_calls,
        tool_choice=tool_choice,
        reasoning_effort=reasoning_effort,
        safety_identifier=safety_identifier,
        prompt_cache_key=prompt_cache_key,
        top_p=top_p,
    )

Create a new instance of DeepSeek LLM.

api_key must be set to your DeepSeek API key, either using the argument or by setting the DEEPSEEK_API_KEY environmental variable.

def with_fireworks(*,
model: str = 'accounts/fireworks/models/llama-v3p3-70b-instruct',
api_key: str | None = None,
base_url: str = 'https://api.fireworks.ai/inference/v1',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM

Expand source code

@staticmethod
def with_fireworks(
    *,
    model: str = "accounts/fireworks/models/llama-v3p3-70b-instruct",
    api_key: str | None = None,
    base_url: str = "https://api.fireworks.ai/inference/v1",
    client: openai.AsyncClient | None = None,
    user: NotGivenOr[str] = NOT_GIVEN,
    temperature: NotGivenOr[float] = NOT_GIVEN,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: ToolChoice = "auto",
    reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
    safety_identifier: NotGivenOr[str] = NOT_GIVEN,
    prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
    top_p: NotGivenOr[float] = NOT_GIVEN,
) -> LLM:
    """
    Create a new instance of Fireworks LLM.

    ``api_key`` must be set to your Fireworks API key, either using the argument or by setting
    the ``FIREWORKS_API_KEY`` environmental variable.
    """

    api_key = api_key or os.environ.get("FIREWORKS_API_KEY")
    if api_key is None:
        raise ValueError(
            "Fireworks API key is required, either as argument or set FIREWORKS_API_KEY environmental variable"  # noqa: E501
        )

    return LLM(
        model=model,
        api_key=api_key,
        base_url=base_url,
        client=client,
        user=user,
        temperature=temperature,
        parallel_tool_calls=parallel_tool_calls,
        tool_choice=tool_choice,
        reasoning_effort=reasoning_effort,
        safety_identifier=safety_identifier,
        prompt_cache_key=prompt_cache_key,
        top_p=top_p,
    )

Create a new instance of Fireworks LLM.

api_key must be set to your Fireworks API key, either using the argument or by setting the FIREWORKS_API_KEY environmental variable.

def with_letta(*,
agent_id: str,
base_url: str = 'https://api.letta.com/v1/chat/completions',
api_key: str | None = None) ‑> livekit.plugins.openai.llm.LLM

Expand source code

@staticmethod
def with_letta(
    *,
    agent_id: str,
    base_url: str = "https://api.letta.com/v1/chat/completions",
    api_key: str | None = None,
) -> LLM:
    """
    Create a new Letta-backed LLM instance connected to the specified Letta agent.

    Args:
        agent_id (str): The Letta agent ID (must be prefixed with 'agent-').
        base_url (str): The URL of the Letta server (e.g., http://localhost:8283/v1/chat/completions for local or https://api.letta.com/v1/chat/completions for cloud).
        api_key (str | None, optional): Optional API key for authentication, required if
                                        the Letta server enforces auth.

    Returns:
        LLM: A configured LLM instance for interacting with the given Letta agent.
    """

    parsed = urlparse(base_url)
    if parsed.scheme not in {"http", "https"}:
        raise ValueError(f"Invalid URL scheme: '{parsed.scheme}'. Must be 'http' or 'https'.")
    if not parsed.netloc:
        raise ValueError(f"URL '{base_url}' is missing a network location (e.g., domain name).")

    api_key = api_key or os.environ.get("LETTA_API_KEY")
    if api_key is None:
        raise ValueError(
            "Letta API key is required, either as argument or set LETTA_API_KEY environmental variable"  # noqa: E501
        )

    return LLM(
        model=agent_id,
        api_key=api_key,
        base_url=base_url,
        client=None,
        user=NOT_GIVEN,
        temperature=NOT_GIVEN,
        parallel_tool_calls=NOT_GIVEN,
        tool_choice=NOT_GIVEN,
    )

Create a new Letta-backed LLM instance connected to the specified Letta agent.

Args

agent_id : str

The Letta agent ID (must be prefixed with 'agent-').

base_url : str

The URL of the Letta server (e.g., http://localhost:8283/v1/chat/completions for local or https://api.letta.com/v1/chat/completions for cloud).

api_key : str | None, optional

Optional API key for authentication, required if the Letta server enforces auth.

Returns

LLM

A configured LLM instance for interacting with the given Letta agent.

def with_nebius(*,
model: str | NebiusChatModels = 'meta-llama/Meta-Llama-3.1-70B-Instruct',
api_key: str | None = None,
base_url: str = 'https://api.studio.nebius.com/v1/',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM

Expand source code

@staticmethod
def with_nebius(
    *,
    model: str | NebiusChatModels = "meta-llama/Meta-Llama-3.1-70B-Instruct",
    api_key: str | None = None,
    base_url: str = "https://api.studio.nebius.com/v1/",
    client: openai.AsyncClient | None = None,
    user: NotGivenOr[str] = NOT_GIVEN,
    temperature: NotGivenOr[float] = NOT_GIVEN,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: ToolChoice = "auto",
    reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
    safety_identifier: NotGivenOr[str] = NOT_GIVEN,
    prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
    top_p: NotGivenOr[float] = NOT_GIVEN,
) -> LLM:
    """
    Create a new instance of Nebius LLM.

    ``api_key`` must be set to your Nebius API key, either using the argument or by setting
    the ``NEBIUS_API_KEY`` environmental variable.
    """

    api_key = api_key or os.environ.get("NEBIUS_API_KEY")
    if api_key is None:
        raise ValueError(
            "Nebius API key is required, either as argument or set NEBIUS_API_KEY environmental variable"  # noqa: E501
        )

    return LLM(
        model=model,
        api_key=api_key,
        base_url=base_url,
        client=client,
        user=user,
        temperature=temperature,
        parallel_tool_calls=parallel_tool_calls,
        tool_choice=tool_choice,
        reasoning_effort=reasoning_effort,
        safety_identifier=safety_identifier,
        prompt_cache_key=prompt_cache_key,
        top_p=top_p,
    )

Create a new instance of Nebius LLM.

api_key must be set to your Nebius API key, either using the argument or by setting the NEBIUS_API_KEY environmental variable.

def with_octo(*,
model: str | OctoChatModels = 'llama-2-13b-chat',
api_key: str | None = None,
base_url: str = 'https://text.octoai.run/v1',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM

Expand source code

@staticmethod
def with_octo(
    *,
    model: str | OctoChatModels = "llama-2-13b-chat",
    api_key: str | None = None,
    base_url: str = "https://text.octoai.run/v1",
    client: openai.AsyncClient | None = None,
    user: NotGivenOr[str] = NOT_GIVEN,
    temperature: NotGivenOr[float] = NOT_GIVEN,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: ToolChoice = "auto",
    reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
    safety_identifier: NotGivenOr[str] = NOT_GIVEN,
    prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
    top_p: NotGivenOr[float] = NOT_GIVEN,
) -> LLM:
    """
    Create a new instance of OctoAI LLM.

    ``api_key`` must be set to your OctoAI API key, either using the argument or by setting
    the ``OCTOAI_TOKEN`` environmental variable.
    """

    api_key = api_key or os.environ.get("OCTOAI_TOKEN")
    if api_key is None:
        raise ValueError(
            "OctoAI API key is required, either as argument or set OCTOAI_TOKEN environmental variable"  # noqa: E501
        )

    return LLM(
        model=model,
        api_key=api_key,
        base_url=base_url,
        client=client,
        user=user,
        temperature=temperature,
        parallel_tool_calls=parallel_tool_calls,
        tool_choice=tool_choice,
        reasoning_effort=reasoning_effort,
        safety_identifier=safety_identifier,
        prompt_cache_key=prompt_cache_key,
        top_p=top_p,
    )

Create a new instance of OctoAI LLM.

api_key must be set to your OctoAI API key, either using the argument or by setting the OCTOAI_TOKEN environmental variable.

def with_ollama(*,
model: str = 'llama3.1',
base_url: str = 'http://localhost:11434/v1',
client: openai.AsyncClient | None = None,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM

Expand source code

@staticmethod
def with_ollama(
    *,
    model: str = "llama3.1",
    base_url: str = "http://localhost:11434/v1",
    client: openai.AsyncClient | None = None,
    temperature: NotGivenOr[float] = NOT_GIVEN,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: ToolChoice = "auto",
    reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
    safety_identifier: NotGivenOr[str] = NOT_GIVEN,
    prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
    top_p: NotGivenOr[float] = NOT_GIVEN,
) -> LLM:
    """
    Create a new instance of Ollama LLM.
    """

    return LLM(
        model=model,
        api_key="ollama",
        base_url=base_url,
        client=client,
        temperature=temperature,
        parallel_tool_calls=parallel_tool_calls,
        tool_choice=tool_choice,
        reasoning_effort=reasoning_effort,
        safety_identifier=safety_identifier,
        prompt_cache_key=prompt_cache_key,
        top_p=top_p,
    )

Create a new instance of Ollama LLM.

def with_openrouter(*,
model: str = 'auto',
api_key: str | None = None,
base_url: str = 'https://openrouter.ai/api/v1',
client: openai.AsyncClient | None = None,
site_url: str | None = None,
app_name: str | None = None,
fallback_models: list[str] | None = None,
provider: OpenRouterProviderPreferences | None = None,
plugins: list[OpenRouterWebPlugin] | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN,
timeout: httpx.Timeout | None = None) ‑> livekit.plugins.openai.llm.LLM

Expand source code

@staticmethod
def with_openrouter(
    *,
    model: str = "auto",
    api_key: str | None = None,
    base_url: str = "https://openrouter.ai/api/v1",
    client: openai.AsyncClient | None = None,
    site_url: str | None = None,
    app_name: str | None = None,
    fallback_models: list[str] | None = None,
    provider: OpenRouterProviderPreferences | None = None,
    plugins: list[OpenRouterWebPlugin] | None = None,
    user: NotGivenOr[str] = NOT_GIVEN,
    temperature: NotGivenOr[float] = NOT_GIVEN,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: ToolChoice = "auto",
    reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
    safety_identifier: NotGivenOr[str] = NOT_GIVEN,
    prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
    top_p: NotGivenOr[float] = NOT_GIVEN,
    timeout: httpx.Timeout | None = None,
) -> LLM:
    """
    Create a new instance of OpenRouter LLM.

    ``api_key`` must be set to your OpenRouter API key, either using the argument or by setting
    the ``OPENROUTER_API_KEY`` environment variable.
    """

    api_key = api_key or os.environ.get("OPENROUTER_API_KEY")
    if api_key is None:
        raise ValueError(
            "OpenRouter API key is required, either as argument or set OPENROUTER_API_KEY environment variable"
        )

    # Set up analytics headers for OpenRouter
    default_headers: dict[str, str] = {}
    if site_url:
        default_headers["HTTP-Referer"] = site_url
    if app_name:
        default_headers["X-Title"] = app_name

    # Build OpenRouter-specific request body
    or_body: dict[str, Any] = {}
    if provider:
        or_body["provider"] = provider
    if fallback_models:
        # Set fallback models for routing
        or_body["models"] = [model, *fallback_models]
    if plugins:
        or_body["plugins"] = [
            {k: v for k, v in asdict(p).items() if v is not None} for p in plugins
        ]

    return LLM(
        model=model,
        api_key=api_key,
        client=client,
        base_url=base_url,
        user=user,
        temperature=temperature,
        parallel_tool_calls=parallel_tool_calls,
        tool_choice=tool_choice,
        reasoning_effort=reasoning_effort,
        safety_identifier=safety_identifier,
        prompt_cache_key=prompt_cache_key,
        top_p=top_p,
        extra_body=or_body,
        extra_headers=default_headers,
        timeout=timeout,
    )

Create a new instance of OpenRouter LLM.

api_key must be set to your OpenRouter API key, either using the argument or by setting the OPENROUTER_API_KEY environment variable.

def with_ovhcloud(*,
model: str = 'gpt-oss-120b',
api_key: str | None = None,
base_url: str = 'https://oai.endpoints.kepler.ai.cloud.ovh.net/v1',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM

Expand source code

@staticmethod
def with_ovhcloud(
    *,
    model: str = "gpt-oss-120b",
    api_key: str | None = None,
    base_url: str = "https://oai.endpoints.kepler.ai.cloud.ovh.net/v1",
    client: openai.AsyncClient | None = None,
    user: NotGivenOr[str] = NOT_GIVEN,
    temperature: NotGivenOr[float] = NOT_GIVEN,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: ToolChoice = "auto",
    reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
    safety_identifier: NotGivenOr[str] = NOT_GIVEN,
    prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
    top_p: NotGivenOr[float] = NOT_GIVEN,
) -> LLM:
    """
    Create a new instance of OVHcloud AI Endpoints LLM.

    ``api_key`` must be set to your OVHcloud AI Endpoints API key, either using the argument or by setting
    the ``OVHCLOUD_API_KEY`` environmental variable.
    """

    api_key = api_key or os.environ.get("OVHCLOUD_API_KEY")
    if api_key is None:
        raise ValueError(
            "OVHcloud AI Endpoints API key is required, either as argument or set OVHCLOUD_API_KEY environmental variable"  # noqa: E501
        )

    return LLM(
        model=model,
        api_key=api_key,
        base_url=base_url,
        client=client,
        user=user,
        temperature=temperature,
        parallel_tool_calls=parallel_tool_calls,
        tool_choice=tool_choice,
        reasoning_effort=reasoning_effort,
        safety_identifier=safety_identifier,
        prompt_cache_key=prompt_cache_key,
        top_p=top_p,
    )

Create a new instance of OVHcloud AI Endpoints LLM.

api_key must be set to your OVHcloud AI Endpoints API key, either using the argument or by setting the OVHCLOUD_API_KEY environmental variable.

def with_perplexity(*,
model: str | PerplexityChatModels = 'llama-3.1-sonar-small-128k-chat',
api_key: str | None = None,
base_url: str = 'https://api.perplexity.ai',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM

Expand source code

@staticmethod
def with_perplexity(
    *,
    model: str | PerplexityChatModels = "llama-3.1-sonar-small-128k-chat",
    api_key: str | None = None,
    base_url: str = "https://api.perplexity.ai",
    client: openai.AsyncClient | None = None,
    user: NotGivenOr[str] = NOT_GIVEN,
    temperature: NotGivenOr[float] = NOT_GIVEN,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: ToolChoice = "auto",
    reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
    safety_identifier: NotGivenOr[str] = NOT_GIVEN,
    prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
    top_p: NotGivenOr[float] = NOT_GIVEN,
) -> LLM:
    """
    Create a new instance of PerplexityAI LLM.

    ``api_key`` must be set to your TogetherAI API key, either using the argument or by setting
    the ``PERPLEXITY_API_KEY`` environmental variable.
    """

    api_key = api_key or os.environ.get("PERPLEXITY_API_KEY")
    if api_key is None:
        raise ValueError(
            "Perplexity AI API key is required, either as argument or set PERPLEXITY_API_KEY environmental variable"  # noqa: E501
        )

    return LLM(
        model=model,
        api_key=api_key,
        base_url=base_url,
        client=client,
        user=user,
        temperature=temperature,
        parallel_tool_calls=parallel_tool_calls,
        tool_choice=tool_choice,
        reasoning_effort=reasoning_effort,
        safety_identifier=safety_identifier,
        prompt_cache_key=prompt_cache_key,
        top_p=top_p,
    )

Create a new instance of PerplexityAI LLM.

api_key must be set to your TogetherAI API key, either using the argument or by setting the PERPLEXITY_API_KEY environmental variable.

def with_sambanova(*,
model: str | SambaNovaChatModels = 'DeepSeek-R1-0528',
api_key: str | None = None,
base_url: str = 'https://api.sambanova.ai/v1',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM

Expand source code

@staticmethod
def with_sambanova(
    *,
    model: str | SambaNovaChatModels = "DeepSeek-R1-0528",
    api_key: str | None = None,
    base_url: str = "https://api.sambanova.ai/v1",
    client: openai.AsyncClient | None = None,
    user: NotGivenOr[str] = NOT_GIVEN,
    temperature: NotGivenOr[float] = NOT_GIVEN,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: ToolChoice = "auto",
    reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
    safety_identifier: NotGivenOr[str] = NOT_GIVEN,
    prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
    top_p: NotGivenOr[float] = NOT_GIVEN,
) -> LLM:
    """
    Create a new instance of SambaNova LLM (OpenAI-compatible).

    ``api_key`` must be set to your SambaNova API key, either using the argument or by setting
    the ``SAMBANOVA_API_KEY`` environment variable.
    """

    api_key = api_key or os.environ.get("SAMBANOVA_API_KEY")
    if api_key is None:
        raise ValueError(
            "SambaNova API key is required, either as argument or set SAMBANOVA_API_KEY environment variable"
        )

    return LLM(
        model=model,
        api_key=api_key,
        base_url=base_url,
        client=client,
        user=user,
        temperature=temperature,
        parallel_tool_calls=parallel_tool_calls,
        tool_choice=tool_choice,
        reasoning_effort=reasoning_effort,
        safety_identifier=safety_identifier,
        prompt_cache_key=prompt_cache_key,
        top_p=top_p,
        _strict_tool_schema=False,
    )

Create a new instance of SambaNova LLM (OpenAI-compatible).

api_key must be set to your SambaNova API key, either using the argument or by setting the SAMBANOVA_API_KEY environment variable.

def with_telnyx(*,
model: str | TelnyxChatModels = 'meta-llama/Meta-Llama-3.1-70B-Instruct',
api_key: str | None = None,
base_url: str = 'https://api.telnyx.com/v2/ai',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM

Expand source code

@staticmethod
def with_telnyx(
    *,
    model: str | TelnyxChatModels = "meta-llama/Meta-Llama-3.1-70B-Instruct",
    api_key: str | None = None,
    base_url: str = "https://api.telnyx.com/v2/ai",
    client: openai.AsyncClient | None = None,
    user: NotGivenOr[str] = NOT_GIVEN,
    temperature: NotGivenOr[float] = NOT_GIVEN,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: ToolChoice = "auto",
    reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
    safety_identifier: NotGivenOr[str] = NOT_GIVEN,
    prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
    top_p: NotGivenOr[float] = NOT_GIVEN,
) -> LLM:
    """
    Create a new instance of Telnyx LLM.

    ``api_key`` must be set to your Telnyx API key, either using the argument or by setting
    the ``TELNYX_API_KEY`` environmental variable.
    """

    api_key = api_key or os.environ.get("TELNYX_API_KEY")
    if api_key is None:
        raise ValueError(
            "Telnyx AI API key is required, either as argument or set TELNYX_API_KEY environmental variable"  # noqa: E501
        )

    return LLM(
        model=model,
        api_key=api_key,
        base_url=base_url,
        client=client,
        user=user,
        temperature=temperature,
        parallel_tool_calls=parallel_tool_calls,
        tool_choice=tool_choice,
        reasoning_effort=reasoning_effort,
        safety_identifier=safety_identifier,
        prompt_cache_key=prompt_cache_key,
        top_p=top_p,
    )

Create a new instance of Telnyx LLM.

api_key must be set to your Telnyx API key, either using the argument or by setting the TELNYX_API_KEY environmental variable.

def with_together(*,
model: str | TogetherChatModels = 'meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo',
api_key: str | None = None,
base_url: str = 'https://api.together.xyz/v1',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM

Expand source code

@staticmethod
def with_together(
    *,
    model: str | TogetherChatModels = "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo",
    api_key: str | None = None,
    base_url: str = "https://api.together.xyz/v1",
    client: openai.AsyncClient | None = None,
    user: NotGivenOr[str] = NOT_GIVEN,
    temperature: NotGivenOr[float] = NOT_GIVEN,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: ToolChoice = "auto",
    reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
    safety_identifier: NotGivenOr[str] = NOT_GIVEN,
    prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
    top_p: NotGivenOr[float] = NOT_GIVEN,
) -> LLM:
    """
    Create a new instance of TogetherAI LLM.

    ``api_key`` must be set to your TogetherAI API key, either using the argument or by setting
    the ``TOGETHER_API_KEY`` environmental variable.
    """

    api_key = api_key or os.environ.get("TOGETHER_API_KEY")
    if api_key is None:
        raise ValueError(
            "Together AI API key is required, either as argument or set TOGETHER_API_KEY environmental variable"  # noqa: E501
        )

    return LLM(
        model=model,
        api_key=api_key,
        base_url=base_url,
        client=client,
        user=user,
        temperature=temperature,
        parallel_tool_calls=parallel_tool_calls,
        tool_choice=tool_choice,
        reasoning_effort=reasoning_effort,
        safety_identifier=safety_identifier,
        prompt_cache_key=prompt_cache_key,
        top_p=top_p,
    )

Create a new instance of TogetherAI LLM.

api_key must be set to your TogetherAI API key, either using the argument or by setting the TOGETHER_API_KEY environmental variable.

def with_x_ai(*,
model: str | XAIChatModels = 'grok-3-fast',
api_key: str | None = None,
base_url: str = 'https://api.x.ai/v1',
client: openai.AsyncClient | None = None,
user: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: ToolChoice = 'auto',
reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
safety_identifier: NotGivenOr[str] = NOT_GIVEN,
prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
top_p: NotGivenOr[float] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLM

Expand source code

@staticmethod
def with_x_ai(
    *,
    model: str | XAIChatModels = "grok-3-fast",
    api_key: str | None = None,
    base_url: str = "https://api.x.ai/v1",
    client: openai.AsyncClient | None = None,
    user: NotGivenOr[str] = NOT_GIVEN,
    temperature: NotGivenOr[float] = NOT_GIVEN,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: ToolChoice = "auto",
    reasoning_effort: NotGivenOr[ReasoningEffort] = NOT_GIVEN,
    safety_identifier: NotGivenOr[str] = NOT_GIVEN,
    prompt_cache_key: NotGivenOr[str] = NOT_GIVEN,
    top_p: NotGivenOr[float] = NOT_GIVEN,
) -> LLM:
    """
    Create a new instance of XAI LLM.

    ``api_key`` must be set to your XAI API key, either using the argument or by setting
    the ``XAI_API_KEY`` environmental variable.
    """
    api_key = api_key or os.environ.get("XAI_API_KEY")
    if api_key is None:
        raise ValueError(
            "XAI API key is required, either as argument or set XAI_API_KEY environmental variable"  # noqa: E501
        )

    return LLM(
        model=model,
        api_key=api_key,
        base_url=base_url,
        client=client,
        user=user,
        temperature=temperature,
        parallel_tool_calls=parallel_tool_calls,
        tool_choice=tool_choice,
        # TODO(long): add provider fmt for grok
        reasoning_effort=reasoning_effort,
        safety_identifier=safety_identifier,
        prompt_cache_key=prompt_cache_key,
        top_p=top_p,
    )

Create a new instance of XAI LLM.

api_key must be set to your XAI API key, either using the argument or by setting the XAI_API_KEY environmental variable.

Instance variables

prop model : str

Expand source code

@property
def model(self) -> str:
    return self._opts.model

Get the model name/identifier for this LLM instance.

Returns

The model name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their model information.

prop provider : str

Expand source code

@property
def provider(self) -> str:
    return self._client._base_url.netloc.decode("utf-8")

Get the provider name/identifier for this LLM instance.

Returns

The provider name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their provider information.

Methods

async def aclose(self) ‑> None

Expand source code

async def aclose(self) -> None:
    await super().aclose()

    if self._owns_client:
        await self._client.close()

def chat(self,
*,
chat_ctx: ChatContext,
tools: list[llm.Tool] | None = None,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0, liveness_timeout=6.0),
parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
response_format: NotGivenOr[completion_create_params.ResponseFormat | type[llm_utils.ResponseFormatT]] = NOT_GIVEN,
extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN) ‑> livekit.plugins.openai.llm.LLMStream

Expand source code

def chat(
    self,
    *,
    chat_ctx: ChatContext,
    tools: list[llm.Tool] | None = None,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    parallel_tool_calls: NotGivenOr[bool] = NOT_GIVEN,
    tool_choice: NotGivenOr[ToolChoice] = NOT_GIVEN,
    response_format: NotGivenOr[
        completion_create_params.ResponseFormat | type[llm_utils.ResponseFormatT]
    ] = NOT_GIVEN,
    extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
) -> LLMStream:
    extra = {}
    if is_given(extra_kwargs):
        extra.update(extra_kwargs)

    if is_given(self._opts.extra_body):
        extra["extra_body"] = self._opts.extra_body

    if is_given(self._opts.extra_headers):
        extra["extra_headers"] = self._opts.extra_headers

    if is_given(self._opts.extra_query):
        extra["extra_query"] = self._opts.extra_query

    if is_given(self._opts.metadata):
        extra["metadata"] = self._opts.metadata

    if is_given(self._opts.user):
        extra["user"] = self._opts.user

    if is_given(self._opts.max_completion_tokens):
        extra["max_completion_tokens"] = self._opts.max_completion_tokens

    if is_given(self._opts.temperature):
        extra["temperature"] = self._opts.temperature

    if is_given(self._opts.service_tier):
        extra["service_tier"] = self._opts.service_tier

    if is_given(self._opts.reasoning_effort):
        extra["reasoning_effort"] = self._opts.reasoning_effort

    if is_given(self._opts.safety_identifier):
        extra["safety_identifier"] = self._opts.safety_identifier

    if is_given(self._opts.prompt_cache_key):
        extra["prompt_cache_key"] = self._opts.prompt_cache_key

    if is_given(self._opts.top_p):
        extra["top_p"] = self._opts.top_p

    if is_given(self._opts.verbosity):
        extra["verbosity"] = self._opts.verbosity

    if is_given(self._opts.prompt_cache_retention):
        extra["prompt_cache_retention"] = self._opts.prompt_cache_retention

    if is_given(self._opts.prompt_cache_options):
        extra["prompt_cache_options"] = self._opts.prompt_cache_options

    parallel_tool_calls = (
        parallel_tool_calls if is_given(parallel_tool_calls) else self._opts.parallel_tool_calls
    )
    if is_given(parallel_tool_calls):
        extra["parallel_tool_calls"] = parallel_tool_calls

    tool_choice = tool_choice if is_given(tool_choice) else self._opts.tool_choice
    if is_given(tool_choice):
        oai_tool_choice: ChatCompletionToolChoiceOptionParam
        if isinstance(tool_choice, dict):
            oai_tool_choice = {
                "type": "function",
                "function": {"name": tool_choice["function"]["name"]},
            }
            extra["tool_choice"] = oai_tool_choice
        elif tool_choice in ("auto", "required", "none"):
            oai_tool_choice = tool_choice
            extra["tool_choice"] = oai_tool_choice

    if is_given(response_format):
        extra["response_format"] = llm_utils.to_openai_response_format(response_format)  # type: ignore

    return LLMStream(
        self,
        model=self._opts.model,
        provider_fmt=self._provider_fmt,
        strict_tool_schema=self._strict_tool_schema,
        client=self._client,
        chat_ctx=chat_ctx,
        tools=tools or [],
        conn_options=conn_options,
        extra_kwargs=extra,
        prompt_cache_breakpoints=self._resolve_prompt_cache_breakpoints(),
    )

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once

class LLMStream (llm: LLM,
*,
model: str | ChatModels,
provider_fmt: str,
strict_tool_schema: bool,
client: openai.AsyncClient,
chat_ctx: llm.ChatContext,
tools: list[llm.Tool],
conn_options: APIConnectOptions,
extra_kwargs: dict[str, Any],
prompt_cache_breakpoints: bool = False)

Expand source code

class LLMStream(_LLMStream):
    def __init__(
        self,
        llm: LLM,
        *,
        model: str | ChatModels,
        provider_fmt: str,
        strict_tool_schema: bool,
        client: openai.AsyncClient,
        chat_ctx: llm.ChatContext,
        tools: list[llm.Tool],
        conn_options: APIConnectOptions,
        extra_kwargs: dict[str, Any],
        prompt_cache_breakpoints: bool = False,
    ) -> None:
        super().__init__(
            llm,
            model=model,
            provider_fmt=provider_fmt,
            strict_tool_schema=strict_tool_schema,
            client=client,
            chat_ctx=chat_ctx,
            tools=tools,
            conn_options=conn_options,
            extra_kwargs=extra_kwargs,
            prompt_cache_breakpoints=prompt_cache_breakpoints,
        )

Helper class that provides a standard way to create an ABC using inheritance.

Ancestors

• LLMStream
• livekit.agents.llm.llm.LLMStream
• abc.ABC

class OpenRouterProviderPreferences (*args, **kwargs)

Expand source code

class OpenRouterProviderPreferences(TypedDict, total=False):
    """OpenRouter provider routing preferences."""

    order: list[str]
    allow_fallbacks: bool
    require_parameters: bool
    data_collection: Literal["allow", "deny"]
    only: list[str]
    ignore: list[str]
    quantizations: list[str]
    sort: Literal["price", "throughput", "latency"]
    max_price: dict[str, float]

OpenRouter provider routing preferences.

Ancestors

• builtins.dict

Class variables

var allow_fallbacks : bool

var data_collection : Literal['allow', 'deny']

var ignore : list[str]

var max_price : dict[str, float]

var only : list[str]

var order : list[str]

var quantizations : list[str]

var require_parameters : bool

var sort : Literal['price', 'throughput', 'latency']

class OpenRouterWebPlugin (max_results: int = 5, search_prompt: str | None = None, id: str = 'web')

Expand source code

@dataclass
class OpenRouterWebPlugin:
    """OpenRouter web search plugin configuration"""

    max_results: int = 5
    search_prompt: str | None = None
    id: str = "web"

OpenRouter web search plugin configuration

Instance variables

var id : str

var max_results : int

var search_prompt : str | None

class STT (*,
language: str | list[str] = 'en',
detect_language: bool = False,
model: STTModels | str = 'gpt-4o-mini-transcribe',
prompt: NotGivenOr[str] = NOT_GIVEN,
keywords: NotGivenOr[list[str]] = NOT_GIVEN,
turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN,
noise_reduction_type: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
base_url: NotGivenOr[str] = NOT_GIVEN,
api_key: NotGivenOr[str] = NOT_GIVEN,
client: openai.AsyncClient | None = None,
use_realtime: NotGivenOr[bool] = NOT_GIVEN,
vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN)

Expand source code

class STT(stt.STT):
    def __init__(
        self,
        *,
        language: str | list[str] = "en",
        detect_language: bool = False,
        model: STTModels | str = "gpt-4o-mini-transcribe",
        prompt: NotGivenOr[str] = NOT_GIVEN,
        keywords: NotGivenOr[list[str]] = NOT_GIVEN,
        turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN,
        noise_reduction_type: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        client: openai.AsyncClient | None = None,
        use_realtime: NotGivenOr[bool] = NOT_GIVEN,
        vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
    ):
        """
        Create a new instance of OpenAI STT.

        Args:
            language: The language code to use for transcription (e.g., "en" for English).
                gpt-transcribe and gpt-live-transcribe accept a list for code-switched audio.
            detect_language: Whether to automatically detect the language.
            model: The OpenAI model to use for transcription.
            prompt: Optional free-form description of the audio, such as its topic or setting.
            keywords: Literal terms to expect, such as product names or acronyms. Only for
                gpt-transcribe and gpt-live-transcribe, and only a hint.
            turn_detection: When using realtime transcription, this controls how model detects the user is done speaking.
                Final transcripts are generated only after the turn is over. See: https://platform.openai.com/docs/guides/realtime-vad
                Ignored for `gpt-realtime-whisper` and `gpt-live-transcribe`, which do not
                support server-side turn detection.
            noise_reduction_type: Type of noise reduction to apply. "near_field" or "far_field"
                This isn't needed when using LiveKit's noise cancellation.
            temperature: Sampling temperature between 0 and 1. Lower values make the
                transcription more deterministic. Not supported for realtime transcription.
            base_url: Custom base URL for OpenAI API.
            api_key: Your OpenAI API key. If not provided, will use the OPENAI_API_KEY environment variable.
            client: Optional pre-configured OpenAI AsyncClient instance.
            use_realtime: Whether to use the realtime transcription API. Defaults to True for
                `gpt-realtime-whisper` and `gpt-live-transcribe`, which are served only there,
                and to False otherwise.
            vad: Optional Voice Activity Detector used to commit the audio buffer when the model
                does not support server-side turn detection (`gpt-realtime-whisper`,
                `gpt-live-transcribe`).
                When not provided and the model requires it, the bundled Silero VAD is used with
                default settings. Pass `vad=None` to opt out and drive
                `input_audio_buffer.commit` yourself.
        """  # noqa: E501

        if not is_given(use_realtime):
            use_realtime = _is_realtime_only(model)

        if use_realtime and is_given(temperature):
            logger.warning(
                "temperature is not supported for realtime transcription; "
                "ignoring the provided value"
            )
            temperature = NOT_GIVEN

        if use_realtime and _is_realtime_only(model):
            if is_given(turn_detection):
                logger.warning(
                    "turn_detection is not supported for %s; ignoring the provided value", model
                )
                turn_detection = NOT_GIVEN
            if not is_given(vad):
                vad = inference.VAD(model="silero")

        super().__init__(
            capabilities=stt.STTCapabilities(
                streaming=use_realtime,
                interim_results=use_realtime,
                aligned_transcript=False,
                keyterms=_supports_context_hints(model),
            )
        )
        # the last language asked for, kept while detection is on so it can be restored
        self._specified_languages = _as_languages(language)
        languages = [] if detect_language else self._specified_languages
        resolved_keywords = list(keywords) if is_given(keywords) else []
        _validate_context(model, languages, resolved_keywords)

        if not is_given(turn_detection):
            turn_detection = {
                "type": "server_vad",
                "threshold": 0.5,
                "prefix_padding_ms": 600,
                "silence_duration_ms": 350,
            }

        self._opts = _STTOptions(
            languages=languages,
            detect_language=detect_language,
            model=model,
            prompt=prompt,
            keywords=resolved_keywords,
            turn_detection=_as_turn_detection(turn_detection),
            temperature=temperature,
        )
        if is_given(noise_reduction_type):
            self._opts.noise_reduction_type = noise_reduction_type

        # user keywords; _opts.keywords holds the effective set (user + session)
        self._user_keywords: list[str] = list(self._opts.keywords)
        self._session_keyterms: list[str] = []

        self._vad = vad if is_given(vad) else None
        # an explicit `vad=None` means the caller commits the audio buffer itself
        self._vad_opted_out = vad is None

        if is_given(api_key) and not api_key:
            raise ValueError(
                "OpenAI API key is required, either as argument or set"
                " OPENAI_API_KEY environment variable"
            )

        self._client = client or openai.AsyncClient(
            max_retries=0,
            api_key=api_key if is_given(api_key) else None,
            base_url=base_url if is_given(base_url) else None,
            http_client=httpx.AsyncClient(
                timeout=httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0),
                follow_redirects=True,
                limits=httpx.Limits(
                    max_connections=50,
                    max_keepalive_connections=50,
                    keepalive_expiry=120,
                ),
            ),
        )

        self._streams = weakref.WeakSet[SpeechStream]()
        self._session: aiohttp.ClientSession | None = None
        self._pool = utils.ConnectionPool[aiohttp.ClientWebSocketResponse](
            max_session_duration=_max_session_duration,
            connect_cb=self._connect_ws,
            close_cb=self._close_ws,
        )

    @property
    def model(self) -> str:
        return self._opts.model

    @property
    def provider(self) -> str:
        return self._client._base_url.netloc.decode("utf-8")

    @staticmethod
    def with_azure(
        *,
        language: str | list[str] = "en",
        detect_language: bool = False,
        model: STTModels | str = "gpt-4o-mini-transcribe",
        prompt: NotGivenOr[str] = NOT_GIVEN,
        keywords: NotGivenOr[list[str]] = NOT_GIVEN,
        turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN,
        noise_reduction_type: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
        azure_endpoint: str | None = None,
        azure_deployment: str | None = None,
        api_version: str | None = None,
        api_key: str | None = None,
        azure_ad_token: str | None = None,
        azure_ad_token_provider: AsyncAzureADTokenProvider | None = None,
        organization: str | None = None,
        project: str | None = None,
        base_url: str | None = None,
        use_realtime: NotGivenOr[bool] = NOT_GIVEN,
        timeout: httpx.Timeout | None = None,
        vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
    ) -> STT:
        """
        Create a new instance of Azure OpenAI STT.

        This automatically infers the following arguments from their corresponding environment variables if they are not provided:
        - `api_key` from `AZURE_OPENAI_API_KEY`
        - `organization` from `OPENAI_ORG_ID`
        - `project` from `OPENAI_PROJECT_ID`
        - `azure_ad_token` from `AZURE_OPENAI_AD_TOKEN`
        - `api_version` from `OPENAI_API_VERSION`
        - `azure_endpoint` from `AZURE_OPENAI_ENDPOINT`
        """  # noqa: E501

        azure_client = openai.AsyncAzureOpenAI(
            max_retries=0,
            azure_endpoint=azure_endpoint,
            azure_deployment=azure_deployment,
            api_version=api_version,
            api_key=api_key,
            azure_ad_token=azure_ad_token,
            azure_ad_token_provider=azure_ad_token_provider,
            organization=organization,
            project=project,
            base_url=base_url,
            timeout=timeout
            if timeout
            else httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0),
        )  # type: ignore

        return STT(
            language=language,
            detect_language=detect_language,
            model=model,
            prompt=prompt,
            keywords=keywords,
            turn_detection=turn_detection,
            noise_reduction_type=noise_reduction_type,
            temperature=temperature,
            client=azure_client,
            use_realtime=use_realtime,
            vad=vad,
        )

    @staticmethod
    def with_ovhcloud(
        *,
        model: str = "whisper-large-v3-turbo",
        api_key: NotGivenOr[str] = NOT_GIVEN,
        base_url: str = "https://oai.endpoints.kepler.ai.cloud.ovh.net/v1",
        client: openai.AsyncClient | None = None,
        language: str | list[str] = "en",
        detect_language: bool = False,
        prompt: NotGivenOr[str] = NOT_GIVEN,
    ) -> STT:
        """
        Create a new instance of OVHcloud AI Endpoints STT.

        ``api_key`` must be set to your OVHcloud AI Endpoints API key, either using the argument or by setting
        the ``OVHCLOUD_API_KEY`` environmental variable.
        """
        ovhcloud_api_key = api_key if is_given(api_key) else os.environ.get("OVHCLOUD_API_KEY")
        if not ovhcloud_api_key:
            raise ValueError("OVHcloud AI Endpoints API key is required")

        return STT(
            model=model,
            api_key=ovhcloud_api_key,
            base_url=base_url,
            client=client,
            language=language,
            detect_language=detect_language,
            prompt=prompt,
            use_realtime=False,
        )

    def stream(
        self,
        *,
        language: NotGivenOr[str | list[str]] = NOT_GIVEN,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> SpeechStream:
        opts = dataclasses.replace(self._opts)
        if is_given(language):
            opts.languages = _as_languages(language)
            _validate_context(opts.model, opts.languages, opts.keywords)
        stream = SpeechStream(
            stt=self,
            pool=self._pool,
            conn_options=conn_options,
            opts=opts,
            vad_instance=self._vad,
        )
        self._streams.add(stream)
        return stream

    def update_options(
        self,
        *,
        model: NotGivenOr[STTModels | str] = NOT_GIVEN,
        language: NotGivenOr[str | list[str]] = NOT_GIVEN,
        detect_language: NotGivenOr[bool] = NOT_GIVEN,
        prompt: NotGivenOr[str] = NOT_GIVEN,
        keywords: NotGivenOr[list[str]] = NOT_GIVEN,
        turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN,
        noise_reduction_type: NotGivenOr[str] = NOT_GIVEN,
        temperature: NotGivenOr[float] = NOT_GIVEN,
    ) -> None:
        """
        Update the options for the speech stream. Open streams apply the change in place;
        only a new `model` reconnects them.

        Args:
            language: The language to transcribe in, or a list for gpt-transcribe and
                gpt-live-transcribe. An empty list detects the language.
            detect_language: Whether to detect the language. Turning it off falls back to the
                last language asked for.
            model: The model to use for transcription.
            prompt: Optional free-form description of the audio.
            keywords: Literal terms to expect. Only for gpt-transcribe and gpt-live-transcribe.
            turn_detection: When using realtime, this controls how model detects the user is done speaking.
            noise_reduction_type: Type of noise reduction to apply. "near_field" or "far_field"
            temperature: Sampling temperature between 0 and 1. Not supported for realtime transcription.
        """  # noqa: E501
        # resolve first: an unsupported combination must raise before anything is applied
        resolved_model = model if is_given(model) else self._opts.model
        if is_given(language):
            languages = _as_languages(language)
        elif detect_language:
            languages = []
        elif detect_language is False and not self._opts.languages:
            logger.warning(
                "detect_language=False names no language, falling back to %s; "
                "pass `language` to transcribe in another",
                self._specified_languages,
            )
            languages = self._specified_languages
        else:
            languages = self._opts.languages
        user_keywords = list(keywords) if is_given(keywords) else self._user_keywords
        _validate_context(resolved_model, languages, user_keywords)
        if not is_given(language):
            # a stream keeps its own language, which the new model may not take
            for stream in self._streams:
                _validate_context(resolved_model, stream._opts.languages, user_keywords)

        # the transport is fixed at construction: AgentSession wraps a non-streaming STT once
        if is_given(model) and _is_realtime_only(model):
            if not self.capabilities.streaming:
                raise ValueError(
                    f"{resolved_model} is served only over the realtime API, and this STT was "
                    "created for the transcriptions endpoint; pass `use_realtime=True` to the "
                    "constructor to reach it"
                )
            if self._vad is None and not self._vad_opted_out:
                raise ValueError(
                    f"{resolved_model} has no server-side endpointing, so it needs a `vad` to "
                    "commit the audio buffer; pass one to the constructor, or pass `vad=None` "
                    "to drive `input_audio_buffer.commit` yourself"
                )

        languages_changed = languages != self._opts.languages
        model_changed = resolved_model != self._opts.model
        # a stream keeps a language of its own unless this call names or moves the language
        languages_given = is_given(language) or languages_changed
        self._opts.model = resolved_model
        self._capabilities.keyterms = _supports_context_hints(resolved_model)
        self._opts.languages = languages
        if languages:
            self._specified_languages = languages
        self._user_keywords = user_keywords
        # detected keyterms must not survive a switch to a model that rejects keywords
        self._opts.keywords = (
            list(dict.fromkeys([*user_keywords, *self._session_keyterms]))
            if self.capabilities.keyterms
            else []
        )
        if is_given(detect_language):
            self._opts.detect_language = detect_language
        if is_given(prompt):
            self._opts.prompt = prompt
        if is_given(turn_detection):
            self._opts.turn_detection = _as_turn_detection(turn_detection)
        if is_given(noise_reduction_type):
            self._opts.noise_reduction_type = noise_reduction_type
        if is_given(temperature):
            if self.capabilities.streaming:
                logger.warning(
                    "temperature is not supported for realtime transcription; "
                    "ignoring the provided value"
                )
            else:
                self._opts.temperature = temperature

        for stream in self._streams:
            stream.update_options(language=languages if languages_given else NOT_GIVEN)

        if model_changed:
            # every stream has dropped its own socket by now, so this reaches the idle ones the
            # pool holds between two speech sessions
            self._pool.invalidate()

    def _update_session_keyterms(self, keyterms: list[str]) -> None:
        if not self.capabilities.keyterms:
            super()._update_session_keyterms(keyterms)
            return
        if keyterms == self._session_keyterms:
            return
        self._session_keyterms = list(keyterms)
        self._opts.keywords = list(dict.fromkeys([*self._user_keywords, *keyterms]))
        for stream in self._streams:
            stream.update_options()

    async def _connect_ws(self, timeout: float) -> aiohttp.ClientWebSocketResponse:
        query_params: dict[str, str] = {
            "intent": "transcription",
        }
        # OpenAI's native realtime endpoint treats ?model= as selecting a
        # conversation session and rejects the transcription-mode
        # session.update with invalid_model — the model is conveyed via
        # audio.input.transcription.model instead. Gateways need the model
        # on the upgrade URL to route the connection before the first frame.
        if urlparse(str(self._client.base_url)).hostname != "api.openai.com":
            query_params["model"] = self._opts.model
        headers = {
            "User-Agent": "LiveKit Agents",
            "Authorization": f"Bearer {self._client.api_key}",
        }
        url = f"{str(self._client.base_url).rstrip('/')}/realtime?{urlencode(query_params)}"
        if url.startswith("http"):
            url = url.replace("http", "ws", 1)

        session = self._ensure_session()
        # the config is sent once the stream acquires the connection, since the pool also
        # hands back sockets it opened earlier
        return await asyncio.wait_for(
            session.ws_connect(url, headers=headers, autoping=False), timeout
        )

    async def _close_ws(self, ws: aiohttp.ClientWebSocketResponse) -> None:
        await ws.close()

    async def aclose(self) -> None:
        """Close the websocket connection pool this STT owns.

        Without this the pool's ``close_cb`` never runs for pooled sockets, so
        they stay open after the STT is closed.
        """
        await self._pool.aclose()

    def _ensure_session(self) -> aiohttp.ClientSession:
        if not self._session:
            self._session = utils.http_context.http_session()

        return self._session

    async def _recognize_impl(
        self,
        buffer: AudioBuffer,
        *,
        language: NotGivenOr[str | list[str]] = NOT_GIVEN,
        conn_options: APIConnectOptions,
    ) -> stt.SpeechEvent:
        opts = self._opts
        if is_given(language):
            languages = _as_languages(language)
            _validate_context(opts.model, languages, opts.keywords)
            opts = dataclasses.replace(opts, languages=languages)

        try:
            data = rtc.combine_audio_frames(buffer).to_wav_bytes()

            format = "json"
            if opts.model == "whisper-1":
                # verbose_json returns language and other details, only supported for whisper-1
                format = "verbose_json"

            transcription = _transcription(opts)
            resp = await self._client.audio.transcriptions.create(
                file=(
                    "file.wav",
                    data,
                    "audio/wav",
                ),
                model=opts.model,  # type: ignore
                language=transcription.language or openai.omit,
                languages=transcription.languages or openai.omit,
                keywords=transcription.keywords or openai.omit,
                prompt=transcription.prompt or openai.omit,
                response_format=format,
                temperature=opts.temperature if is_given(opts.temperature) else openai.omit,
                timeout=httpx.Timeout(30, connect=conn_options.timeout),
            )

            # the detected language beats the hint, dominant code first; an empty list keeps it
            sd = stt.SpeechData(text=resp.text, language=_transcript_language(opts.languages))
            if isinstance(resp, TranscriptionVerbose) and resp.language:
                sd.language = LanguageCode(resp.language)
            elif isinstance(resp, Transcription) and resp.languages:
                sd.language = LanguageCode(resp.languages[0].code)

            return stt.SpeechEvent(
                type=stt.SpeechEventType.FINAL_TRANSCRIPT,
                alternatives=[sd],
            )

        except openai.APITimeoutError:
            raise APITimeoutError() from None
        except openai.APIStatusError as e:
            raise APIStatusError(
                e.message, status_code=e.status_code, request_id=e.request_id, body=e.body
            ) from None
        except Exception as e:
            raise APIConnectionError() from e

Helper class that provides a standard way to create an ABC using inheritance.

Create a new instance of OpenAI STT.

Args

language

The language code to use for transcription (e.g., "en" for English). gpt-transcribe and gpt-live-transcribe accept a list for code-switched audio.

detect_language

Whether to automatically detect the language.

model

The OpenAI model to use for transcription.

prompt

Optional free-form description of the audio, such as its topic or setting.

keywords

Literal terms to expect, such as product names or acronyms. Only for gpt-transcribe and gpt-live-transcribe, and only a hint.

turn_detection

When using realtime transcription, this controls how model detects the user is done speaking. Final transcripts are generated only after the turn is over. See: https://platform.openai.com/docs/guides/realtime-vad Ignored for gpt-realtime-whisper and gpt-live-transcribe, which do not support server-side turn detection.

noise_reduction_type

Type of noise reduction to apply. "near_field" or "far_field" This isn't needed when using LiveKit's noise cancellation.

temperature

Sampling temperature between 0 and 1. Lower values make the transcription more deterministic. Not supported for realtime transcription.

base_url

Custom base URL for OpenAI API.

api_key

Your OpenAI API key. If not provided, will use the OPENAI_API_KEY environment variable.

client

Optional pre-configured OpenAI AsyncClient instance.

use_realtime

Whether to use the realtime transcription API. Defaults to True for gpt-realtime-whisper and gpt-live-transcribe, which are served only there, and to False otherwise.

vad

Optional Voice Activity Detector used to commit the audio buffer when the model does not support server-side turn detection (gpt-realtime-whisper, gpt-live-transcribe). When not provided and the model requires it, the bundled Silero VAD is used with default settings. Pass vad=None to opt out and drive input_audio_buffer.commit yourself.

Ancestors

• livekit.agents.stt.stt.STT
• abc.ABC
• EventEmitter
• typing.Generic

Subclasses

• livekit.plugins.groq.services.STT

Static methods

def with_azure(*,
language: str | list[str] = 'en',
detect_language: bool = False,
model: STTModels | str = 'gpt-4o-mini-transcribe',
prompt: NotGivenOr[str] = NOT_GIVEN,
keywords: NotGivenOr[list[str]] = NOT_GIVEN,
turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN,
noise_reduction_type: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN,
azure_endpoint: str | None = None,
azure_deployment: str | None = None,
api_version: str | None = None,
api_key: str | None = None,
azure_ad_token: str | None = None,
azure_ad_token_provider: AsyncAzureADTokenProvider | None = None,
organization: str | None = None,
project: str | None = None,
base_url: str | None = None,
use_realtime: NotGivenOr[bool] = NOT_GIVEN,
timeout: httpx.Timeout | None = None,
vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN) ‑> livekit.plugins.openai.stt.STT

Expand source code

@staticmethod
def with_azure(
    *,
    language: str | list[str] = "en",
    detect_language: bool = False,
    model: STTModels | str = "gpt-4o-mini-transcribe",
    prompt: NotGivenOr[str] = NOT_GIVEN,
    keywords: NotGivenOr[list[str]] = NOT_GIVEN,
    turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN,
    noise_reduction_type: NotGivenOr[str] = NOT_GIVEN,
    temperature: NotGivenOr[float] = NOT_GIVEN,
    azure_endpoint: str | None = None,
    azure_deployment: str | None = None,
    api_version: str | None = None,
    api_key: str | None = None,
    azure_ad_token: str | None = None,
    azure_ad_token_provider: AsyncAzureADTokenProvider | None = None,
    organization: str | None = None,
    project: str | None = None,
    base_url: str | None = None,
    use_realtime: NotGivenOr[bool] = NOT_GIVEN,
    timeout: httpx.Timeout | None = None,
    vad: NotGivenOr[vad.VAD | None] = NOT_GIVEN,
) -> STT:
    """
    Create a new instance of Azure OpenAI STT.

    This automatically infers the following arguments from their corresponding environment variables if they are not provided:
    - `api_key` from `AZURE_OPENAI_API_KEY`
    - `organization` from `OPENAI_ORG_ID`
    - `project` from `OPENAI_PROJECT_ID`
    - `azure_ad_token` from `AZURE_OPENAI_AD_TOKEN`
    - `api_version` from `OPENAI_API_VERSION`
    - `azure_endpoint` from `AZURE_OPENAI_ENDPOINT`
    """  # noqa: E501

    azure_client = openai.AsyncAzureOpenAI(
        max_retries=0,
        azure_endpoint=azure_endpoint,
        azure_deployment=azure_deployment,
        api_version=api_version,
        api_key=api_key,
        azure_ad_token=azure_ad_token,
        azure_ad_token_provider=azure_ad_token_provider,
        organization=organization,
        project=project,
        base_url=base_url,
        timeout=timeout
        if timeout
        else httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0),
    )  # type: ignore

    return STT(
        language=language,
        detect_language=detect_language,
        model=model,
        prompt=prompt,
        keywords=keywords,
        turn_detection=turn_detection,
        noise_reduction_type=noise_reduction_type,
        temperature=temperature,
        client=azure_client,
        use_realtime=use_realtime,
        vad=vad,
    )

Create a new instance of Azure OpenAI STT.

This automatically infers the following arguments from their corresponding environment variables if they are not provided: - api_key from AZURE_OPENAI_API_KEY - organization from OPENAI_ORG_ID - project from OPENAI_PROJECT_ID - azure_ad_token from AZURE_OPENAI_AD_TOKEN - api_version from OPENAI_API_VERSION - azure_endpoint from AZURE_OPENAI_ENDPOINT

def with_ovhcloud(*,
model: str = 'whisper-large-v3-turbo',
api_key: NotGivenOr[str] = NOT_GIVEN,
base_url: str = 'https://oai.endpoints.kepler.ai.cloud.ovh.net/v1',
client: openai.AsyncClient | None = None,
language: str | list[str] = 'en',
detect_language: bool = False,
prompt: NotGivenOr[str] = NOT_GIVEN) ‑> livekit.plugins.openai.stt.STT

Expand source code

@staticmethod
def with_ovhcloud(
    *,
    model: str = "whisper-large-v3-turbo",
    api_key: NotGivenOr[str] = NOT_GIVEN,
    base_url: str = "https://oai.endpoints.kepler.ai.cloud.ovh.net/v1",
    client: openai.AsyncClient | None = None,
    language: str | list[str] = "en",
    detect_language: bool = False,
    prompt: NotGivenOr[str] = NOT_GIVEN,
) -> STT:
    """
    Create a new instance of OVHcloud AI Endpoints STT.

    ``api_key`` must be set to your OVHcloud AI Endpoints API key, either using the argument or by setting
    the ``OVHCLOUD_API_KEY`` environmental variable.
    """
    ovhcloud_api_key = api_key if is_given(api_key) else os.environ.get("OVHCLOUD_API_KEY")
    if not ovhcloud_api_key:
        raise ValueError("OVHcloud AI Endpoints API key is required")

    return STT(
        model=model,
        api_key=ovhcloud_api_key,
        base_url=base_url,
        client=client,
        language=language,
        detect_language=detect_language,
        prompt=prompt,
        use_realtime=False,
    )

Create a new instance of OVHcloud AI Endpoints STT.

api_key must be set to your OVHcloud AI Endpoints API key, either using the argument or by setting the OVHCLOUD_API_KEY environmental variable.

Instance variables

prop model : str

Expand source code

@property
def model(self) -> str:
    return self._opts.model

Get the model name/identifier for this STT instance.

Returns

The model name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their model information.

prop provider : str

Expand source code

@property
def provider(self) -> str:
    return self._client._base_url.netloc.decode("utf-8")

Get the provider name/identifier for this STT instance.

Returns

The provider name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their provider information.

Methods

async def aclose(self) ‑> None

Expand source code

async def aclose(self) -> None:
    """Close the websocket connection pool this STT owns.

    Without this the pool's ``close_cb`` never runs for pooled sockets, so
    they stay open after the STT is closed.
    """
    await self._pool.aclose()

Close the websocket connection pool this STT owns.

Without this the pool's close_cb never runs for pooled sockets, so they stay open after the STT is closed.

def stream(self,
*,
language: NotGivenOr[str | list[str]] = NOT_GIVEN,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0, liveness_timeout=6.0)) ‑> livekit.plugins.openai.stt.SpeechStream

Expand source code

def stream(
    self,
    *,
    language: NotGivenOr[str | list[str]] = NOT_GIVEN,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
) -> SpeechStream:
    opts = dataclasses.replace(self._opts)
    if is_given(language):
        opts.languages = _as_languages(language)
        _validate_context(opts.model, opts.languages, opts.keywords)
    stream = SpeechStream(
        stt=self,
        pool=self._pool,
        conn_options=conn_options,
        opts=opts,
        vad_instance=self._vad,
    )
    self._streams.add(stream)
    return stream

def update_options(self,
*,
model: NotGivenOr[STTModels | str] = NOT_GIVEN,
language: NotGivenOr[str | list[str]] = NOT_GIVEN,
detect_language: NotGivenOr[bool] = NOT_GIVEN,
prompt: NotGivenOr[str] = NOT_GIVEN,
keywords: NotGivenOr[list[str]] = NOT_GIVEN,
turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN,
noise_reduction_type: NotGivenOr[str] = NOT_GIVEN,
temperature: NotGivenOr[float] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    model: NotGivenOr[STTModels | str] = NOT_GIVEN,
    language: NotGivenOr[str | list[str]] = NOT_GIVEN,
    detect_language: NotGivenOr[bool] = NOT_GIVEN,
    prompt: NotGivenOr[str] = NOT_GIVEN,
    keywords: NotGivenOr[list[str]] = NOT_GIVEN,
    turn_detection: NotGivenOr[SessionTurnDetection] = NOT_GIVEN,
    noise_reduction_type: NotGivenOr[str] = NOT_GIVEN,
    temperature: NotGivenOr[float] = NOT_GIVEN,
) -> None:
    """
    Update the options for the speech stream. Open streams apply the change in place;
    only a new `model` reconnects them.

    Args:
        language: The language to transcribe in, or a list for gpt-transcribe and
            gpt-live-transcribe. An empty list detects the language.
        detect_language: Whether to detect the language. Turning it off falls back to the
            last language asked for.
        model: The model to use for transcription.
        prompt: Optional free-form description of the audio.
        keywords: Literal terms to expect. Only for gpt-transcribe and gpt-live-transcribe.
        turn_detection: When using realtime, this controls how model detects the user is done speaking.
        noise_reduction_type: Type of noise reduction to apply. "near_field" or "far_field"
        temperature: Sampling temperature between 0 and 1. Not supported for realtime transcription.
    """  # noqa: E501
    # resolve first: an unsupported combination must raise before anything is applied
    resolved_model = model if is_given(model) else self._opts.model
    if is_given(language):
        languages = _as_languages(language)
    elif detect_language:
        languages = []
    elif detect_language is False and not self._opts.languages:
        logger.warning(
            "detect_language=False names no language, falling back to %s; "
            "pass `language` to transcribe in another",
            self._specified_languages,
        )
        languages = self._specified_languages
    else:
        languages = self._opts.languages
    user_keywords = list(keywords) if is_given(keywords) else self._user_keywords
    _validate_context(resolved_model, languages, user_keywords)
    if not is_given(language):
        # a stream keeps its own language, which the new model may not take
        for stream in self._streams:
            _validate_context(resolved_model, stream._opts.languages, user_keywords)

    # the transport is fixed at construction: AgentSession wraps a non-streaming STT once
    if is_given(model) and _is_realtime_only(model):
        if not self.capabilities.streaming:
            raise ValueError(
                f"{resolved_model} is served only over the realtime API, and this STT was "
                "created for the transcriptions endpoint; pass `use_realtime=True` to the "
                "constructor to reach it"
            )
        if self._vad is None and not self._vad_opted_out:
            raise ValueError(
                f"{resolved_model} has no server-side endpointing, so it needs a `vad` to "
                "commit the audio buffer; pass one to the constructor, or pass `vad=None` "
                "to drive `input_audio_buffer.commit` yourself"
            )

    languages_changed = languages != self._opts.languages
    model_changed = resolved_model != self._opts.model
    # a stream keeps a language of its own unless this call names or moves the language
    languages_given = is_given(language) or languages_changed
    self._opts.model = resolved_model
    self._capabilities.keyterms = _supports_context_hints(resolved_model)
    self._opts.languages = languages
    if languages:
        self._specified_languages = languages
    self._user_keywords = user_keywords
    # detected keyterms must not survive a switch to a model that rejects keywords
    self._opts.keywords = (
        list(dict.fromkeys([*user_keywords, *self._session_keyterms]))
        if self.capabilities.keyterms
        else []
    )
    if is_given(detect_language):
        self._opts.detect_language = detect_language
    if is_given(prompt):
        self._opts.prompt = prompt
    if is_given(turn_detection):
        self._opts.turn_detection = _as_turn_detection(turn_detection)
    if is_given(noise_reduction_type):
        self._opts.noise_reduction_type = noise_reduction_type
    if is_given(temperature):
        if self.capabilities.streaming:
            logger.warning(
                "temperature is not supported for realtime transcription; "
                "ignoring the provided value"
            )
        else:
            self._opts.temperature = temperature

    for stream in self._streams:
        stream.update_options(language=languages if languages_given else NOT_GIVEN)

    if model_changed:
        # every stream has dropped its own socket by now, so this reaches the idle ones the
        # pool holds between two speech sessions
        self._pool.invalidate()

Update the options for the speech stream. Open streams apply the change in place; only a new model reconnects them.

Args

language

The language to transcribe in, or a list for gpt-transcribe and gpt-live-transcribe. An empty list detects the language.

detect_language

Whether to detect the language. Turning it off falls back to the last language asked for.

model

The model to use for transcription.

prompt

Optional free-form description of the audio.

keywords

Literal terms to expect. Only for gpt-transcribe and gpt-live-transcribe.

turn_detection

When using realtime, this controls how model detects the user is done speaking.

noise_reduction_type

Type of noise reduction to apply. "near_field" or "far_field"

temperature

Sampling temperature between 0 and 1. Not supported for realtime transcription.

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once

class TTS (*,
model: TTSModels | str = 'gpt-4o-mini-tts',
voice: TTSVoices | str = 'ash',
speed: float = 1.0,
sample_rate: int = 24000,
instructions: NotGivenOr[str] = NOT_GIVEN,
base_url: NotGivenOr[str] = NOT_GIVEN,
api_key: NotGivenOr[str] = NOT_GIVEN,
client: openai.AsyncClient | None = None,
response_format: NotGivenOr[RESPONSE_FORMATS] = NOT_GIVEN)

Expand source code

class TTS(tts.TTS):
    def __init__(
        self,
        *,
        model: TTSModels | str = DEFAULT_MODEL,
        voice: TTSVoices | str = DEFAULT_VOICE,
        speed: float = 1.0,
        sample_rate: int = SAMPLE_RATE,
        instructions: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        client: openai.AsyncClient | None = None,
        response_format: NotGivenOr[RESPONSE_FORMATS] = NOT_GIVEN,
    ) -> None:
        """
        Create a new instance of OpenAI TTS.

        ``api_key`` must be set to your OpenAI API key, either using the argument or by setting the
        ``OPENAI_API_KEY`` environmental variable.

        ``sample_rate`` is the output sample rate, in Hz. Encoded audio, including the default MP3
        format, is decoded and resampled to this rate. For raw PCM, it must match the provider's
        sample rate; raw PCM is not resampled.
        """
        if isinstance(sample_rate, bool) or not isinstance(sample_rate, int) or sample_rate < 10:
            raise ValueError(
                f"OpenAI TTS sample_rate must be an integer of at least 10 Hz, got {sample_rate}"
            )

        super().__init__(
            capabilities=tts.TTSCapabilities(streaming=False),
            sample_rate=sample_rate,
            num_channels=NUM_CHANNELS,
        )

        self._opts = _TTSOptions(
            model=model,
            voice=voice,
            speed=speed,
            instructions=instructions if is_given(instructions) else None,
            response_format=response_format if is_given(response_format) else "mp3",
        )

        if is_given(api_key) and not api_key:
            raise ValueError(
                "OpenAI API key is required, either as argument or set"
                " OPENAI_API_KEY environment variable"
            )
        self._owns_client = client is None
        self._client = client or openai.AsyncClient(
            max_retries=0,
            api_key=api_key if is_given(api_key) else None,
            base_url=base_url if is_given(base_url) else None,
            http_client=httpx.AsyncClient(
                timeout=httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0),
                follow_redirects=True,
                limits=httpx.Limits(
                    max_connections=50, max_keepalive_connections=50, keepalive_expiry=120
                ),
            ),
        )

        self._prewarm_task: asyncio.Task | None = None

    @property
    def model(self) -> str:
        return self._opts.model

    @property
    def provider(self) -> str:
        return self._client._base_url.netloc.decode("utf-8")

    def update_options(
        self,
        *,
        model: NotGivenOr[TTSModels | str] = NOT_GIVEN,
        voice: NotGivenOr[TTSVoices | str] = NOT_GIVEN,
        speed: NotGivenOr[float] = NOT_GIVEN,
        instructions: NotGivenOr[str] = NOT_GIVEN,
    ) -> None:
        if is_given(model):
            self._opts.model = model
        if is_given(voice):
            self._opts.voice = voice
        if is_given(speed):
            self._opts.speed = speed
        if is_given(instructions):
            self._opts.instructions = instructions

    @staticmethod
    def with_azure(
        *,
        model: TTSModels | str = DEFAULT_MODEL,
        voice: TTSVoices | str = DEFAULT_VOICE,
        speed: float = 1.0,
        sample_rate: int = SAMPLE_RATE,
        instructions: NotGivenOr[str] = NOT_GIVEN,
        azure_endpoint: str | None = None,
        azure_deployment: str | None = None,
        api_version: str | None = None,
        api_key: str | None = None,
        azure_ad_token: str | None = None,
        azure_ad_token_provider: AsyncAzureADTokenProvider | None = None,
        organization: str | None = None,
        project: str | None = None,
        base_url: str | None = None,
        response_format: NotGivenOr[RESPONSE_FORMATS] = NOT_GIVEN,
        timeout: httpx.Timeout | None = None,
    ) -> TTS:
        """
        Create a new instance of Azure OpenAI TTS.

        This automatically infers the following arguments from their corresponding environment
        variables if they are not provided:
        - `api_key` from `AZURE_OPENAI_API_KEY`
        - `organization` from `OPENAI_ORG_ID`
        - `project` from `OPENAI_PROJECT_ID`
        - `azure_ad_token` from `AZURE_OPENAI_AD_TOKEN`
        - `api_version` from `OPENAI_API_VERSION`
        - `azure_endpoint` from `AZURE_OPENAI_ENDPOINT`
        """

        azure_client = openai.AsyncAzureOpenAI(
            max_retries=0,
            azure_endpoint=azure_endpoint,
            azure_deployment=azure_deployment,
            api_version=api_version,
            api_key=api_key,
            azure_ad_token=azure_ad_token,
            azure_ad_token_provider=azure_ad_token_provider,
            organization=organization,
            project=project,
            base_url=base_url,
            timeout=timeout
            if timeout
            else httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0),
        )  # type: ignore

        tts = TTS(
            model=model,
            voice=voice,
            speed=speed,
            sample_rate=sample_rate,
            instructions=instructions,
            client=azure_client,
            response_format=response_format,
        )
        tts._owns_client = True
        return tts

    def synthesize(
        self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
    ) -> tts.ChunkedStream:
        return ChunkedStream(tts=self, input_text=text, conn_options=conn_options)

    def prewarm(self) -> None:
        async def _prewarm() -> None:
            try:
                await self._client.get("/", cast_to=str)
            except Exception:
                pass

        # Don't replace a prewarm still in flight: the old task would lose its only reference,
        # and aclose() cancels just the latest one, so it could still be using the client after
        # close. utils.ConnectionPool.prewarm guards the same way.
        if self._prewarm_task is None or self._prewarm_task.done():
            self._prewarm_task = asyncio.create_task(_prewarm())

    async def aclose(self) -> None:
        if self._prewarm_task:
            await aio.cancel_and_wait(self._prewarm_task)

        if self._owns_client:
            await self._client.close()

Helper class that provides a standard way to create an ABC using inheritance.

Create a new instance of OpenAI TTS.

api_key must be set to your OpenAI API key, either using the argument or by setting the OPENAI_API_KEY environmental variable.

sample_rate is the output sample rate, in Hz. Encoded audio, including the default MP3 format, is decoded and resampled to this rate. For raw PCM, it must match the provider's sample rate; raw PCM is not resampled.

Ancestors

• livekit.agents.tts.tts.TTS
• abc.ABC
• EventEmitter
• typing.Generic

Static methods

def with_azure(*,
model: TTSModels | str = 'gpt-4o-mini-tts',
voice: TTSVoices | str = 'ash',
speed: float = 1.0,
sample_rate: int = 24000,
instructions: NotGivenOr[str] = NOT_GIVEN,
azure_endpoint: str | None = None,
azure_deployment: str | None = None,
api_version: str | None = None,
api_key: str | None = None,
azure_ad_token: str | None = None,
azure_ad_token_provider: AsyncAzureADTokenProvider | None = None,
organization: str | None = None,
project: str | None = None,
base_url: str | None = None,
response_format: NotGivenOr[RESPONSE_FORMATS] = NOT_GIVEN,
timeout: httpx.Timeout | None = None) ‑> livekit.plugins.openai.tts.TTS

Expand source code

@staticmethod
def with_azure(
    *,
    model: TTSModels | str = DEFAULT_MODEL,
    voice: TTSVoices | str = DEFAULT_VOICE,
    speed: float = 1.0,
    sample_rate: int = SAMPLE_RATE,
    instructions: NotGivenOr[str] = NOT_GIVEN,
    azure_endpoint: str | None = None,
    azure_deployment: str | None = None,
    api_version: str | None = None,
    api_key: str | None = None,
    azure_ad_token: str | None = None,
    azure_ad_token_provider: AsyncAzureADTokenProvider | None = None,
    organization: str | None = None,
    project: str | None = None,
    base_url: str | None = None,
    response_format: NotGivenOr[RESPONSE_FORMATS] = NOT_GIVEN,
    timeout: httpx.Timeout | None = None,
) -> TTS:
    """
    Create a new instance of Azure OpenAI TTS.

    This automatically infers the following arguments from their corresponding environment
    variables if they are not provided:
    - `api_key` from `AZURE_OPENAI_API_KEY`
    - `organization` from `OPENAI_ORG_ID`
    - `project` from `OPENAI_PROJECT_ID`
    - `azure_ad_token` from `AZURE_OPENAI_AD_TOKEN`
    - `api_version` from `OPENAI_API_VERSION`
    - `azure_endpoint` from `AZURE_OPENAI_ENDPOINT`
    """

    azure_client = openai.AsyncAzureOpenAI(
        max_retries=0,
        azure_endpoint=azure_endpoint,
        azure_deployment=azure_deployment,
        api_version=api_version,
        api_key=api_key,
        azure_ad_token=azure_ad_token,
        azure_ad_token_provider=azure_ad_token_provider,
        organization=organization,
        project=project,
        base_url=base_url,
        timeout=timeout
        if timeout
        else httpx.Timeout(connect=15.0, read=5.0, write=5.0, pool=5.0),
    )  # type: ignore

    tts = TTS(
        model=model,
        voice=voice,
        speed=speed,
        sample_rate=sample_rate,
        instructions=instructions,
        client=azure_client,
        response_format=response_format,
    )
    tts._owns_client = True
    return tts

Create a new instance of Azure OpenAI TTS.

This automatically infers the following arguments from their corresponding environment variables if they are not provided: - api_key from AZURE_OPENAI_API_KEY - organization from OPENAI_ORG_ID - project from OPENAI_PROJECT_ID - azure_ad_token from AZURE_OPENAI_AD_TOKEN - api_version from OPENAI_API_VERSION - azure_endpoint from AZURE_OPENAI_ENDPOINT

Instance variables

prop model : str

Expand source code

@property
def model(self) -> str:
    return self._opts.model

Get the model name/identifier for this TTS instance.

Returns

The model name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their model information.

prop provider : str

Expand source code

@property
def provider(self) -> str:
    return self._client._base_url.netloc.decode("utf-8")

Get the provider name/identifier for this TTS instance.

Returns

The provider name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their provider information.

Methods

async def aclose(self) ‑> None

Expand source code

async def aclose(self) -> None:
    if self._prewarm_task:
        await aio.cancel_and_wait(self._prewarm_task)

    if self._owns_client:
        await self._client.close()

def prewarm(self) ‑> None

Expand source code

def prewarm(self) -> None:
    async def _prewarm() -> None:
        try:
            await self._client.get("/", cast_to=str)
        except Exception:
            pass

    # Don't replace a prewarm still in flight: the old task would lose its only reference,
    # and aclose() cancels just the latest one, so it could still be using the client after
    # close. utils.ConnectionPool.prewarm guards the same way.
    if self._prewarm_task is None or self._prewarm_task.done():
        self._prewarm_task = asyncio.create_task(_prewarm())

Pre-warm connection to the TTS service

def synthesize(self,
text: str,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0, liveness_timeout=6.0)) ‑> livekit.agents.tts.tts.ChunkedStream

Expand source code

def synthesize(
    self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
) -> tts.ChunkedStream:
    return ChunkedStream(tts=self, input_text=text, conn_options=conn_options)

def update_options(self,
*,
model: NotGivenOr[TTSModels | str] = NOT_GIVEN,
voice: NotGivenOr[TTSVoices | str] = NOT_GIVEN,
speed: NotGivenOr[float] = NOT_GIVEN,
instructions: NotGivenOr[str] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    model: NotGivenOr[TTSModels | str] = NOT_GIVEN,
    voice: NotGivenOr[TTSVoices | str] = NOT_GIVEN,
    speed: NotGivenOr[float] = NOT_GIVEN,
    instructions: NotGivenOr[str] = NOT_GIVEN,
) -> None:
    if is_given(model):
        self._opts.model = model
    if is_given(voice):
        self._opts.voice = voice
    if is_given(speed):
        self._opts.speed = speed
    if is_given(instructions):
        self._opts.instructions = instructions

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once