Module livekit.agents.inference.llm

def drop_unsupported_params(
    model: str, params: dict[str, Any], tools: list[Any] | None = None
) -> dict[str, Any]:
    """Remove parameters that are not supported by the given model.

    Strips any provider prefix (e.g. ``openai/o3-pro`` -> ``o3-pro``) before
    matching against known model prefixes.
    """
    model_name = model.split("/")[-1] if "/" in model else model
    if tools and any(
        model_name.startswith(p) for p in _REASONING_EFFORT_TOOL_INCOMPATIBLE_PREFIXES
    ):
        params = {k: v for k, v in params.items() if k != "reasoning_effort"}
    for prefix, unsupported in _UNSUPPORTED_PARAMS.items():
        if model_name.startswith(prefix):
            kept_at_none = _KEPT_AT_EFFORT_NONE.get(prefix)
            if (
                kept_at_none
                and min_reasoning_effort(model_name) == "none"
                and _reasoning_effort_is_none(params)
            ):
                # These models accept their sampling params again at effort
                # "none" (their lowest supported effort). Verified against the
                # API on gpt-5.6-luna (2026-10): "none" + temperature -> 200,
                # "low" + temperature -> 400, on both the chat completions and
                # responses APIs; re-verified for gpt-6 on gpt-6.1-sol,
                # gpt-6-sol and gpt-6-luna (2026-10-20). The reasoning guide no
                # longer documents this restriction. Applies only to models in
                # _MIN_REASONING_EFFORT whose floor is "none" — new gpt-5.1+
                # and gpt-6 models must be added there or the params stay
                # stripped.
                unsupported = unsupported - kept_at_none
            params = {k: v for k, v in params.items() if k not in unsupported}
            break
    return params

def llm_from_model_string(model: str) -> llm.LLM | llm.RealtimeModel | llm.DuplexModel:
    """Create the inference model a ``llm=`` string names.

    Duplex model strings resolve to :class:`livekit.agents.inference.DuplexModel`,
    realtime (speech-to-speech) model strings to
    :class:`livekit.agents.inference.RealtimeModel`, every other string to
    :class:`livekit.agents.inference.LLM`.
    """
    if is_duplex_model(model):
        # imported lazily: the realtime package imports this module
        from .realtime import DuplexModel

        return DuplexModel.from_model_string(model)

    if is_realtime_model(model):
        # imported lazily: the realtime package imports this module
        from .realtime import RealtimeModel

        return RealtimeModel.from_model_string(model)

    return LLM.from_model_string(model)

def min_reasoning_effort(model: str) -> ReasoningEffort | None:
    """Lowest reasoning effort the model supports, or None if the model has no
    reasoning-effort control.

    Strips any provider prefix (e.g. ``openai/gpt-5`` -> ``gpt-5``) before matching.
    """
    return _MIN_REASONING_EFFORT.get(model.split("/")[-1])

def supports_prompt_cache_breakpoints(model: str) -> bool:
    """Whether an OpenAI model accepts ``prompt_cache_breakpoint`` content parts.

    OpenAI supports breakpoints on GPT-5.6 and later, so any ``gpt-<major>.<minor>``
    at or above 5.6 qualifies, future families such as gpt-6 included. Every other
    name is out: gpt-oss, the o-series, the chat-latest aliases, and models from
    other providers (``google/...``). Older models answer HTTP 400 to the field.
    """
    provider, _, name = model.lower().rpartition("/")
    if provider not in ("", "openai"):
        return False
    version = _GPT_VERSION.match(name)
    if version is None:
        return False
    return (int(version.group(1)), int(version.group(2) or 0)) >= (5, 6)

class ChatCompletionOptions(TypedDict, total=False):
    frequency_penalty: float | None
    logit_bias: dict[str, int] | None
    logprobs: bool | None
    max_completion_tokens: int | None
    max_tokens: int | None
    metadata: Metadata | None
    modalities: list[Literal["text", "audio"]] | None
    n: int | None
    parallel_tool_calls: bool
    prediction: ChatCompletionPredictionContentParam | None
    presence_penalty: float | None
    prompt_cache_key: str
    prompt_cache_options: PromptCacheOptions
    prompt_cache_retention: Literal["in_memory", "24h"] | None
    reasoning_effort: ReasoningEffort | None
    safety_identifier: str
    seed: int | None
    service_tier: Literal["auto", "default", "flex", "scale", "priority"] | None
    stop: str | None | list[str] | None
    store: bool | None
    temperature: float | None
    top_logprobs: int | None
    top_p: float | None
    user: str
    verbosity: Literal["low", "medium", "high"] | None
    web_search_options: completion_create_params.WebSearchOptions

    # livekit-typed arguments
    tool_choice: ToolChoice
    # TODO(theomonnomn): support repsonse format
    # response_format: completion_create_params.ResponseFormat

class LLM(llm.LLM):
    def __init__(
        self,
        model: LLMModels | str,
        *,
        provider: str | None = None,
        base_url: str | None = None,
        api_key: str | None = None,
        api_secret: str | None = None,
        inference_class: InferenceClass | None = None,
        extra_kwargs: ChatCompletionOptions | dict[str, Any] | None = None,
        prompt_cache_breakpoints: bool | Literal["auto"] = "auto",
    ) -> None:
        """
        Args:
            prompt_cache_breakpoints: Send each :class:`llm.CacheBreakpoint` in the chat
                context as an OpenAI ``prompt_cache_breakpoint``. ``"auto"`` sends them
                when :func:`supports_prompt_cache_breakpoints` accepts the current model;
                ``True`` or ``False`` overrides that.
        """
        super().__init__()

        lk_base_url = base_url if base_url else get_default_inference_url()

        lk_api_key, lk_api_secret = resolve_credentials(api_key, api_secret)

        self._opts = _LLMOptions(
            model=model,
            provider=provider,
            base_url=lk_base_url,
            api_key=lk_api_key,
            api_secret=lk_api_secret,
            inference_class=inference_class,
            extra_kwargs=extra_kwargs or {},
            prompt_cache_breakpoints=prompt_cache_breakpoints,
        )
        self._client = openai.AsyncClient(
            api_key=create_access_token(self._opts.api_key, self._opts.api_secret),
            base_url=self._opts.base_url,
            max_retries=0,
            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
                ),
            ),
        )

    async def _prewarm_impl(self) -> None:
        await self._client.models.list()

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

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

    def update_options(
        self,
        *,
        model: NotGivenOr[LLMModels | str] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[ChatCompletionOptions | dict[str, Any]] = NOT_GIVEN,
        prompt_cache_breakpoints: NotGivenOr[bool | Literal["auto"]] = NOT_GIVEN,
    ) -> None:
        """Update LLM configuration options.

        Each option is read on the next ``chat()`` call, so a swap
        takes effect on the agent's next turn without recreating the
        LLM. ``extra_kwargs`` *replaces* the persistent kwargs dict
        rather than merging — pass ``{}`` to clear it.
        """
        if is_given(model):
            self._opts.model = model
        if is_given(extra_kwargs):
            self._opts.extra_kwargs = dict(extra_kwargs)
        # isinstance rather than is_given: mypy cannot narrow the bool | Literal union through it
        if not isinstance(prompt_cache_breakpoints, NotGiven):
            self._opts.prompt_cache_breakpoints = prompt_cache_breakpoints

    @property
    def model(self) -> str:
        """Get the model name for this LLM instance."""
        return self._opts.model

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

    def chat(
        self,
        *,
        chat_ctx: ChatContext,
        tools: list[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,
        inference_class: NotGivenOr[InferenceClass] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
    ) -> LLMStream:
        extra = {}
        if is_given(extra_kwargs):
            extra.update(extra_kwargs)

        parallel_tool_calls = (
            parallel_tool_calls
            if is_given(parallel_tool_calls)
            else self._opts.extra_kwargs.get("parallel_tool_calls", NOT_GIVEN)
        )
        if is_given(parallel_tool_calls):
            extra["parallel_tool_calls"] = parallel_tool_calls

        extra_tool_choice = self._opts.extra_kwargs.get("tool_choice", NOT_GIVEN)
        tool_choice = tool_choice if is_given(tool_choice) else extra_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

        extra.update(self._opts.extra_kwargs)

        effective_inference_class = (
            inference_class if is_given(inference_class) else self._opts.inference_class
        )

        self._client.api_key = create_access_token(self._opts.api_key, self._opts.api_secret)
        return LLMStream(
            self,
            model=self._opts.model,
            provider=self._opts.provider,
            inference_class=effective_inference_class,
            strict_tool_schema=True,
            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:
        # resolved per chat() so update_options(model=...) re-evaluates "auto"
        setting = self._opts.prompt_cache_breakpoints
        if isinstance(setting, bool):
            return setting
        return supports_prompt_cache_breakpoints(self._opts.model)

class LLMStream(llm.LLMStream):
    def __init__(
        self,
        llm_v: LLM | llm.LLM,
        *,
        model: LLMModels | str,
        provider: str | None = None,
        inference_class: InferenceClass | None = None,
        strict_tool_schema: bool,
        client: openai.AsyncClient,
        chat_ctx: llm.ChatContext,
        tools: list[Tool],
        conn_options: APIConnectOptions,
        extra_kwargs: dict[str, Any],
        provider_fmt: str = "openai",  # used internally for chat_ctx format
        prompt_cache_breakpoints: bool = False,
    ) -> None:
        super().__init__(llm_v, chat_ctx=chat_ctx, tools=tools, conn_options=conn_options)
        self._model = model
        self._provider = provider
        self._inference_class = inference_class
        self._provider_fmt = provider_fmt
        # the other formatters take no such argument and drop CacheBreakpoint themselves
        self._prompt_cache_breakpoints = prompt_cache_breakpoints and (
            provider_fmt in _BREAKPOINT_FORMATS
        )
        self._strict_tool_schema = strict_tool_schema
        self._client = client
        self._llm = llm_v
        self._extra_kwargs = drop_unsupported_params(model, extra_kwargs, tools=tools)
        self._tool_ctx = llm.ToolContext(tools)

    async def _run(self) -> None:
        # current function call that we're waiting for full completion (args are streamed)
        # (defined inside the _run method to make sure the state is reset for each run/attempt)
        self._oai_stream: openai.AsyncStream[ChatCompletionChunk] | None = None
        self._tool_call_id: str | None = None
        self._fnc_name: str | None = None
        self._fnc_raw_arguments: str | None = None
        self._tool_extra: dict[str, Any] | None = None
        self._tool_index: int | None = None
        retryable = True

        try:
            fmt_kwargs: dict[str, Any] = {}
            if self._prompt_cache_breakpoints:
                fmt_kwargs["prompt_cache_breakpoints"] = True
            chat_ctx, _ = self._chat_ctx.to_provider_format(format=self._provider_fmt, **fmt_kwargs)
            tool_schemas = cast(
                list[ChatCompletionToolParam],
                self._tool_ctx.parse_function_tools("openai", strict=self._strict_tool_schema),
            )
            if lk_oai_debug:
                tool_choice = self._extra_kwargs.get("tool_choice", NOT_GIVEN)
                logger.debug(
                    "chat.completions.create",
                    extra={
                        "fnc_ctx": tool_schemas,
                        "tool_choice": tool_choice,
                        "lk.pii.chat_ctx": chat_ctx,
                    },
                )
            if not self._tools:
                # remove tool_choice from extra_kwargs if no tools are provided
                self._extra_kwargs.pop("tool_choice", None)

            extra_headers = self._extra_kwargs.setdefault("extra_headers", {})
            extra_headers.update(get_inference_headers(inference_class=self._inference_class))
            if self._provider:
                extra_headers[HEADER_INFERENCE_PROVIDER] = self._provider

            self._oai_stream = stream = await self._client.chat.completions.create(
                messages=cast(list[ChatCompletionMessageParam], chat_ctx),
                tools=tool_schemas or openai.omit,
                model=self._model,
                stream_options={"include_usage": True},
                stream=True,
                timeout=httpx.Timeout(self._conn_options.timeout),
                **self._extra_kwargs,
            )

            thinking_filter = llm_utils.ThinkingTokenFilter(
                *_MODEL_THINK_TAGS.get(
                    self._model, (llm_utils.THINK_TAG_START, llm_utils.THINK_TAG_END)
                )
            )
            async with stream:
                async for chunk in stream:
                    for choice in chunk.choices:
                        chat_chunk = self._parse_choice(chunk.id, choice, thinking_filter)
                        if chat_chunk is not None:
                            if chat_chunk.has_response():
                                retryable = False
                            self._event_ch.send_nowait(chat_chunk)

                    if chunk.usage is not None:
                        tokens_details = chunk.usage.prompt_tokens_details
                        cached_tokens = tokens_details.cached_tokens if tokens_details else 0
                        cache_creation_tokens = (
                            tokens_details.cache_write_tokens if tokens_details else 0
                        )
                        completion_details = chunk.usage.completion_tokens_details
                        reasoning_tokens = (
                            completion_details.reasoning_tokens if completion_details else 0
                        )
                        usage_chunk = llm.ChatChunk(
                            id=chunk.id,
                            usage=llm.CompletionUsage(
                                completion_tokens=chunk.usage.completion_tokens or 0,
                                prompt_tokens=chunk.usage.prompt_tokens or 0,
                                prompt_cached_tokens=cached_tokens or 0,
                                cache_creation_tokens=cache_creation_tokens or 0,
                                reasoning_tokens=reasoning_tokens or 0,
                                total_tokens=chunk.usage.total_tokens or 0,
                                service_tier=getattr(chunk, "service_tier", None),
                            ),
                        )
                        self._event_ch.send_nowait(usage_chunk)

        except openai.APITimeoutError:
            raise APITimeoutError(retryable=retryable) from None
        except httpx.TimeoutException as e:
            # Only the request call runs inside the openai client's error mapping, so a
            # timeout waiting on the stream body arrives as the raw httpx exception.
            raise APITimeoutError(retryable=retryable) from e
        except openai.APIStatusError as e:
            if e.status_code == 429:
                self._log_rate_limited(e)
            raise APIStatusError(
                e.message,
                status_code=e.status_code,
                request_id=e.request_id,
                body=e.body,
                retryable=retryable,
            ) from None
        except Exception as e:
            raise APIConnectionError(retryable=retryable) from e

    def _log_rate_limited(self, e: openai.APIStatusError) -> None:
        """Log the gateway's quota snapshot when a request is rejected with 429.

        The gateway stamps X-LiveKit-Inference-{RPM,TPM,Credits}-{Limit,Used}
        on rejections so customers can see which limit they hit and by how much.
        """
        extra: dict[str, Any] = {"model": self._model, "request_id": e.request_id}
        extra.update(extract_quota_usage(e.response.headers))
        logger.warning("LLM request rate limited by inference gateway", extra=extra)

    def _parse_choice(
        self, id: str, choice: Choice, thinking_filter: llm_utils.ThinkingTokenFilter
    ) -> llm.ChatChunk | None:
        delta = choice.delta

        # https://github.com/livekit/agents/issues/688
        # the delta can be None when using Azure OpenAI (content filtering)
        if delta is None:
            return None

        delta.content = llm_utils.strip_thinking_tokens(
            delta.content, thinking_filter, final=choice.finish_reason is not None
        )

        # Calls completed by this delta. A provider can send several parallel calls
        # in one chunk (e.g. Ollama), so they are collected and emitted together.
        completed: list[llm.FunctionToolCall] = []

        if delta.tool_calls:
            for tool in delta.tool_calls:
                if not tool.function:
                    continue

                if self._tool_call_id and tool.id and tool.index != self._tool_index:
                    completed.append(
                        llm.FunctionToolCall(
                            arguments=self._fnc_raw_arguments or "",
                            name=self._fnc_name or "",
                            call_id=self._tool_call_id or "",
                            extra=self._tool_extra,
                        )
                    )
                    self._tool_call_id = self._fnc_name = self._fnc_raw_arguments = None
                    self._tool_extra = None

                if tool.function.name:
                    self._tool_index = tool.index
                    self._tool_call_id = tool.id
                    self._fnc_name = tool.function.name
                    self._fnc_raw_arguments = tool.function.arguments or ""
                    # Extract extra from tool call (e.g., Google thought signatures)
                    self._tool_extra = getattr(tool, "extra_content", None)
                elif tool.function.arguments:
                    self._fnc_raw_arguments += tool.function.arguments  # type: ignore

        if choice.finish_reason in ("tool_calls", "stop") and self._tool_call_id:
            completed.append(
                llm.FunctionToolCall(
                    arguments=self._fnc_raw_arguments or "",
                    name=self._fnc_name or "",
                    call_id=self._tool_call_id or "",
                    extra=self._tool_extra,
                )
            )
            self._tool_call_id = self._fnc_name = self._fnc_raw_arguments = None
            self._tool_extra = None

        if completed:
            return llm.ChatChunk(
                id=id,
                delta=llm.ChoiceDelta(
                    role="assistant",
                    content=delta.content,
                    extra=getattr(delta, "extra_content", None),
                    tool_calls=completed,
                ),
            )

        # Extract extra from delta (e.g., Google thought signatures on text parts)
        delta_extra = getattr(delta, "extra_content", None)

        if not delta.content and not delta_extra:
            return None

        return llm.ChatChunk(
            id=id,
            delta=llm.ChoiceDelta(
                content=delta.content,
                role="assistant",
                extra=delta_extra,
            ),
        )