Module livekit.agents.inference.responses

Responses API LLM served by the LiveKit inference gateway.

Classes

class LLM (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: ResponsesOptions | dict[str, Any] | None = None,
prompt_cache_breakpoints: "bool | Literal['auto']" = 'auto',
use_websocket: bool = False,
reasoning: NotGivenOr[ReasoningSetting] = NOT_GIVEN)

Expand source code

class LLM(llm.LLM):
    """Responses API LLM routed through the LiveKit inference gateway.

    Credentials and gateway URL resolve the same way as ``inference.LLM``. REST is the
    default. With ``use_websocket=True`` turns stream over ``/v1/responses`` on pooled
    sockets, so ``inference_class`` and the other upgrade headers are fixed per connection.

    Conversation state is client-side. The gateway strips ``previous_response_id`` and
    forces ``store=false``, so every turn sends the full context.

    Web search is the only hosted tool the gateway accepts. Any other hosted tool is
    rejected with HTTP 400, or a ``gateway.error`` over WebSocket.

    ``prompt_cache_options`` (including ``ttl``) is forwarded as given.

    ``reasoning`` takes an effort such as ``"low"``, or the whole Responses object such as
    ``{"effort": "low", "summary": "auto"}``. It overrides ``extra_kwargs["reasoning"]``,
    and a ``chat()`` call's own ``extra_kwargs["reasoning"]`` overrides it for that turn.
    Left unset, with no ``extra_kwargs["reasoning"]`` either, it defaults to the lowest effort
    the model supports (:func:`min_reasoning_effort`), so a voice turn doesn't wait on
    reasoning; models without an effort control get none. ``None`` sends no reasoning, for the
    provider's default.

    Each read waits ``conn_options.timeout`` for the next event. REST streams get gateway
    keepalives while a model reasons silently; over WebSocket they are rolling out behind a
    gateway flag, so until then raise ``conn_options.timeout`` for models that reason silently
    for longer than it.
    """

    _provider_tool_type: type[llm.ProviderTool] = llm.ProviderTool

    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: ResponsesOptions | dict[str, Any] | None = None,
        prompt_cache_breakpoints: bool | Literal["auto"] = "auto",
        use_websocket: bool = False,
        reasoning: NotGivenOr[ReasoningSetting] = NOT_GIVEN,
    ) -> None:
        super().__init__()

        lk_api_key, lk_api_secret = resolve_credentials(api_key, api_secret)
        self._opts = _LLMOptions(
            model=model,
            provider=provider,
            base_url=base_url if base_url else get_default_inference_url(),
            api_key=lk_api_key,
            api_secret=lk_api_secret,
            inference_class=inference_class,
            extra_kwargs=_copy_kwargs(extra_kwargs or {}),
            prompt_cache_breakpoints=prompt_cache_breakpoints,
            use_websocket=use_websocket,
            reasoning=_reasoning_fields(reasoning) if is_given(reasoning) else NOT_GIVEN,
        )
        self._check_reasoning()
        self._ws: ResponsesWebsocket | None = None
        self._retired_ws: set[_GatewayResponsesWebsocket] = set()
        self._ws_close_tasks: set[asyncio.Task[None]] = set()
        self._closed = False
        # the gateway bounds upstream silence, so reads only need to outlast its keepalives
        self._stream_idle_timeout: float | None = None
        self._active_streams: int = 0
        self._parallel_generation: bool = False
        self._prev_resp_id = ""
        self._prev_resp_items: list[dict[str, Any]] | None = None
        self._client: openai.AsyncClient | None = None
        if use_websocket:
            self._ws = self._make_ws()
        else:
            self._client = self._make_client()

    def _make_ws(self) -> _GatewayResponsesWebsocket:
        return _GatewayResponsesWebsocket(
            base_url=self._opts.base_url,
            model=self._opts.model,
            api_key=self._opts.api_key,
            api_secret=self._opts.api_secret,
            provider=self._opts.provider,
            inference_class=self._opts.inference_class,
            extra_headers=lambda: dict(self._opts.extra_kwargs.get("extra_headers") or {}),
            timeout=None,
        )

    def _make_client(self) -> openai.AsyncClient:
        return 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:
        if self._ws is not None:
            self._ws._pool.prewarm()
        elif self._client is not None:
            await self._client.models.list()

    async def aclose(self) -> None:
        self._closed = True
        await super().aclose()
        if self._ws is not None:
            await self._ws.aclose()
        # clear before awaiting: a turn finishing mid-close then finds its transport
        # already gone instead of mutating the set or closing it twice
        retired = list(self._retired_ws)
        self._retired_ws.clear()
        await asyncio.gather(*(ws.aclose() for ws in retired), return_exceptions=True)
        if self._ws_close_tasks:
            await asyncio.gather(*self._ws_close_tasks, return_exceptions=True)
        if self._client is not None:
            await self._client.close()

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

        Each option is read on the next ``chat()`` call. ``extra_kwargs`` *replaces* the
        persistent kwargs dict rather than merging; pass ``{}`` to clear it. ``reasoning``
        changes only reasoning and leaves ``extra_kwargs`` as is; ``None`` clears it. Over
        WebSocket a new ``model`` gets a new pool and new ``extra_headers`` invalidate it, since
        both are bound to the socket at the upgrade. A turn created before a model swap keeps
        the old model's pool, so its retries dial the model it sends.
        """
        # read first, so a rejected setting changes nothing
        reasoning_fields = _reasoning_fields(reasoning) if is_given(reasoning) else NOT_GIVEN
        if is_given(model) and model != self._opts.model:
            self._opts.model = model
            if isinstance(self._ws, _GatewayResponsesWebsocket):
                self._retire_ws(self._ws)
                self._ws = self._make_ws()
                self._prewarm_swapped_ws(self._ws)
        if is_given(extra_kwargs):
            old_headers = self._opts.extra_kwargs.get("extra_headers")
            self._opts.extra_kwargs = _copy_kwargs(extra_kwargs)
            # headers ride on the upgrade, so pooled sockets carry the old ones
            if self._ws is not None and extra_kwargs.get("extra_headers") != old_headers:
                self._ws._pool.invalidate()
        # 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
        if is_given(reasoning_fields):
            self._opts.reasoning = reasoning_fields
        if is_given(model) or is_given(extra_kwargs) or is_given(reasoning):
            self._check_reasoning()

    def _check_reasoning(self) -> None:
        """Warn about a reasoning setting that is unlikely to do what the caller meant.

        Logs when it overrides a different ``extra_kwargs["reasoning"]``, or its effort is
        below the model's lowest. It only warns; nothing it checks is changed.
        """
        reasoning = self._opts.reasoning
        if not is_given(reasoning) or not reasoning:
            return
        configured = self._opts.extra_kwargs.get("reasoning")
        if configured is not None and _reasoning_fields(configured) != reasoning:
            logger.warning(
                "reasoning overrides extra_kwargs['reasoning']",
                extra={"reasoning": reasoning, "configured": _reasoning_fields(configured)},
            )
        _warn_if_below_min_effort(self._opts.model, reasoning.get("effort"))

    def _prewarm_swapped_ws(self, ws: _GatewayResponsesWebsocket) -> None:
        # prewarm dials a socket, so only replace a warm pool with a warm one, and never
        # after aclose(), which nothing would follow to close it
        if self._closed or self._prewarm_task is None:
            return
        try:
            asyncio.get_running_loop()
        except RuntimeError:
            return
        ws._pool.prewarm()

    def _retire_ws(self, ws: _GatewayResponsesWebsocket) -> None:
        ws._pool.invalidate()
        self._retired_ws.add(ws)
        self._close_retired_ws(ws)

    def _release_ws(self, ws: _GatewayResponsesWebsocket) -> None:
        ws._holders -= 1
        self._close_retired_ws(ws)

    def _close_retired_ws(self, ws: _GatewayResponsesWebsocket) -> None:
        if ws not in self._retired_ws or ws._holders > 0:
            return
        try:
            task = asyncio.get_running_loop().create_task(ws.aclose())
        except RuntimeError:
            return  # no loop to close on; aclose() closes it
        self._retired_ws.discard(ws)
        self._ws_close_tasks.add(task)
        task.add_done_callback(self._ws_close_tasks.discard)

    @property
    def model(self) -> str:
        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[type[llm_utils.ResponseFormatT] | dict[str, Any]] = NOT_GIVEN,
        inference_class: NotGivenOr[InferenceClass] = NOT_GIVEN,
        extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
    ) -> ResponsesLLMStream:
        extra: dict[str, Any] = dict(self._opts.extra_kwargs)
        if (reasoning := self._resolve_reasoning()) is not None:
            extra["reasoning"] = reasoning
        # after reasoning: a turn's own extra_kwargs override the LLM's settings
        if is_given(extra_kwargs):
            extra.update(extra_kwargs)

        if is_given(response_format):
            text = dict(extra.get("text") or {})
            text["format"] = _to_text_format(response_format)
            extra["text"] = text

        if is_given(parallel_tool_calls):
            extra["parallel_tool_calls"] = parallel_tool_calls

        merged_tool_choice = extra.pop("tool_choice", None)
        oai_tool_choice = _to_responses_tool_choice(
            tool_choice if is_given(tool_choice) else merged_tool_choice
        )
        # the API rejects a tool_choice with no tools
        if oai_tool_choice is not None and tools:
            extra["tool_choice"] = oai_tool_choice

        if self._client is not None:
            effective_inference_class = (
                inference_class if is_given(inference_class) else self._opts.inference_class
            )
            headers = {
                **extra.get("extra_headers", {}),
                **get_inference_headers(inference_class=effective_inference_class),
            }
            if self._opts.provider:
                headers[HEADER_INFERENCE_PROVIDER] = self._opts.provider
            extra["extra_headers"] = headers
            self._client.api_key = create_access_token(self._opts.api_key, self._opts.api_secret)
        else:
            # headers ride on the upgrade, so only the LLM's own extra_headers reach the
            # gateway; the payload must not carry them
            extra.pop("extra_headers", None)
            if is_given(extra_kwargs) and "extra_headers" in extra_kwargs:
                logger.warning(
                    "extra_headers cannot change per request over WebSocket; "
                    "using the LLM's extra_kwargs headers"
                )
            if is_given(inference_class) and inference_class != self._opts.inference_class:
                logger.warning(
                    "inference_class cannot change per request over WebSocket; "
                    "using the LLM's value",
                    extra={
                        "requested": inference_class,
                        "configured": self._opts.inference_class,
                    },
                )

        stream = _GatewayResponsesLLMStream(
            self,
            model=self._opts.model,
            strict_tool_schema=True,
            client=self._client,
            chat_ctx=chat_ctx,
            tools=tools or [],
            conn_options=conn_options,
            extra_kwargs=extra,
            prev_resp=None,
            prompt_cache_breakpoints=self._resolve_prompt_cache_breakpoints(),
        )
        if isinstance(self._ws, _GatewayResponsesWebsocket):
            ws = self._ws
            ws._holders += 1
            stream._task.add_done_callback(lambda _: self._release_ws(ws))
        return stream

    def _resolve_reasoning(self) -> dict[str, Any] | None:
        """Return the ``reasoning`` fields this LLM sends, or None to send none.

        ``reasoning=`` wins; an ``extra_kwargs["reasoning"]`` is sent as part of the
        kwargs instead; otherwise the model's lowest effort is the default. Called per
        ``chat()`` so ``update_options(model=...)`` re-evaluates the default.
        """
        reasoning = self._opts.reasoning
        if is_given(reasoning):
            return dict(reasoning) if reasoning else None
        if "reasoning" in self._opts.extra_kwargs:
            return None
        effort = min_reasoning_effort(self._opts.model)
        return {"effort": effort} if effort is not None else None

    def _resolve_prompt_cache_breakpoints(self) -> bool:
        setting = self._opts.prompt_cache_breakpoints
        if isinstance(setting, bool):
            return setting
        return supports_prompt_cache_breakpoints(self._opts.model)

Responses API LLM routed through the LiveKit inference gateway.

Credentials and gateway URL resolve the same way as inference.LLM. REST is the default. With use_websocket=True turns stream over /v1/responses on pooled sockets, so inference_class and the other upgrade headers are fixed per connection.

Conversation state is client-side. The gateway strips previous_response_id and forces store=false, so every turn sends the full context.

Web search is the only hosted tool the gateway accepts. Any other hosted tool is rejected with HTTP 400, or a gateway.error over WebSocket.

prompt_cache_options (including ttl) is forwarded as given.

reasoning takes an effort such as "low", or the whole Responses object such as {"effort": "low", "summary": "auto"}. It overrides extra_kwargs["reasoning"], and a chat() call's own extra_kwargs["reasoning"] overrides it for that turn. Left unset, with no extra_kwargs["reasoning"] either, it defaults to the lowest effort the model supports (:func:min_reasoning_effort), so a voice turn doesn't wait on reasoning; models without an effort control get none. None sends no reasoning, for the provider's default.

Each read waits conn_options.timeout for the next event. REST streams get gateway keepalives while a model reasons silently; over WebSocket they are rolling out behind a gateway flag, so until then raise conn_options.timeout for models that reason silently for longer than it.

Ancestors

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

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 "livekit"

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:
    self._closed = True
    await super().aclose()
    if self._ws is not None:
        await self._ws.aclose()
    # clear before awaiting: a turn finishing mid-close then finds its transport
    # already gone instead of mutating the set or closing it twice
    retired = list(self._retired_ws)
    self._retired_ws.clear()
    await asyncio.gather(*(ws.aclose() for ws in retired), return_exceptions=True)
    if self._ws_close_tasks:
        await asyncio.gather(*self._ws_close_tasks, return_exceptions=True)
    if self._client is not None:
        await self._client.close()

def chat(self,
*,
chat_ctx: ChatContext,
tools: list[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[type[llm_utils.ResponseFormatT] | dict[str, Any]] = NOT_GIVEN,
inference_class: NotGivenOr[InferenceClass] = NOT_GIVEN,
extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN) ‑> livekit.agents.llm._responses_engine.ResponsesLLMStream

Expand source code

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[type[llm_utils.ResponseFormatT] | dict[str, Any]] = NOT_GIVEN,
    inference_class: NotGivenOr[InferenceClass] = NOT_GIVEN,
    extra_kwargs: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
) -> ResponsesLLMStream:
    extra: dict[str, Any] = dict(self._opts.extra_kwargs)
    if (reasoning := self._resolve_reasoning()) is not None:
        extra["reasoning"] = reasoning
    # after reasoning: a turn's own extra_kwargs override the LLM's settings
    if is_given(extra_kwargs):
        extra.update(extra_kwargs)

    if is_given(response_format):
        text = dict(extra.get("text") or {})
        text["format"] = _to_text_format(response_format)
        extra["text"] = text

    if is_given(parallel_tool_calls):
        extra["parallel_tool_calls"] = parallel_tool_calls

    merged_tool_choice = extra.pop("tool_choice", None)
    oai_tool_choice = _to_responses_tool_choice(
        tool_choice if is_given(tool_choice) else merged_tool_choice
    )
    # the API rejects a tool_choice with no tools
    if oai_tool_choice is not None and tools:
        extra["tool_choice"] = oai_tool_choice

    if self._client is not None:
        effective_inference_class = (
            inference_class if is_given(inference_class) else self._opts.inference_class
        )
        headers = {
            **extra.get("extra_headers", {}),
            **get_inference_headers(inference_class=effective_inference_class),
        }
        if self._opts.provider:
            headers[HEADER_INFERENCE_PROVIDER] = self._opts.provider
        extra["extra_headers"] = headers
        self._client.api_key = create_access_token(self._opts.api_key, self._opts.api_secret)
    else:
        # headers ride on the upgrade, so only the LLM's own extra_headers reach the
        # gateway; the payload must not carry them
        extra.pop("extra_headers", None)
        if is_given(extra_kwargs) and "extra_headers" in extra_kwargs:
            logger.warning(
                "extra_headers cannot change per request over WebSocket; "
                "using the LLM's extra_kwargs headers"
            )
        if is_given(inference_class) and inference_class != self._opts.inference_class:
            logger.warning(
                "inference_class cannot change per request over WebSocket; "
                "using the LLM's value",
                extra={
                    "requested": inference_class,
                    "configured": self._opts.inference_class,
                },
            )

    stream = _GatewayResponsesLLMStream(
        self,
        model=self._opts.model,
        strict_tool_schema=True,
        client=self._client,
        chat_ctx=chat_ctx,
        tools=tools or [],
        conn_options=conn_options,
        extra_kwargs=extra,
        prev_resp=None,
        prompt_cache_breakpoints=self._resolve_prompt_cache_breakpoints(),
    )
    if isinstance(self._ws, _GatewayResponsesWebsocket):
        ws = self._ws
        ws._holders += 1
        stream._task.add_done_callback(lambda _: self._release_ws(ws))
    return stream

def update_options(self,
*,
model: NotGivenOr[LLMModels | str] = NOT_GIVEN,
extra_kwargs: NotGivenOr[ResponsesOptions | dict[str, Any]] = NOT_GIVEN,
prompt_cache_breakpoints: "NotGivenOr[bool | Literal['auto']]" = NOT_GIVEN,
reasoning: NotGivenOr[ReasoningSetting] = NOT_GIVEN) ‑> None

Expand source code

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

    Each option is read on the next ``chat()`` call. ``extra_kwargs`` *replaces* the
    persistent kwargs dict rather than merging; pass ``{}`` to clear it. ``reasoning``
    changes only reasoning and leaves ``extra_kwargs`` as is; ``None`` clears it. Over
    WebSocket a new ``model`` gets a new pool and new ``extra_headers`` invalidate it, since
    both are bound to the socket at the upgrade. A turn created before a model swap keeps
    the old model's pool, so its retries dial the model it sends.
    """
    # read first, so a rejected setting changes nothing
    reasoning_fields = _reasoning_fields(reasoning) if is_given(reasoning) else NOT_GIVEN
    if is_given(model) and model != self._opts.model:
        self._opts.model = model
        if isinstance(self._ws, _GatewayResponsesWebsocket):
            self._retire_ws(self._ws)
            self._ws = self._make_ws()
            self._prewarm_swapped_ws(self._ws)
    if is_given(extra_kwargs):
        old_headers = self._opts.extra_kwargs.get("extra_headers")
        self._opts.extra_kwargs = _copy_kwargs(extra_kwargs)
        # headers ride on the upgrade, so pooled sockets carry the old ones
        if self._ws is not None and extra_kwargs.get("extra_headers") != old_headers:
            self._ws._pool.invalidate()
    # 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
    if is_given(reasoning_fields):
        self._opts.reasoning = reasoning_fields
    if is_given(model) or is_given(extra_kwargs) or is_given(reasoning):
        self._check_reasoning()

Update LLM configuration options.

Each option is read on the next chat() call. extra_kwargs replaces the persistent kwargs dict rather than merging; pass {} to clear it. reasoning changes only reasoning and leaves extra_kwargs as is; None clears it. Over WebSocket a new model gets a new pool and new extra_headers invalidate it, since both are bound to the socket at the upgrade. A turn created before a model swap keeps the old model's pool, so its retries dial the model it sends.

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once

class ResponsesOptions (*args, **kwargs)

Expand source code

class ResponsesOptions(TypedDict, total=False):
    include: list[ResponseIncludable] | None
    instructions: str | None
    max_output_tokens: int | None
    max_tool_calls: int | None
    metadata: Metadata | None
    parallel_tool_calls: bool | None
    prompt_cache_options: PromptCacheOptions
    reasoning: Reasoning | dict[str, Any] | None
    temperature: float | None
    text: ResponseTextConfigParam
    top_p: float | None
    truncation: Literal["auto", "disabled"] | None
    user: str

    # livekit-typed argument
    tool_choice: ToolChoice

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

Ancestors

• builtins.dict

Class variables

var include : list[typing.Literal['file_search_call.results', 'web_search_call.results', 'web_search_call.action.sources', 'message.input_image.image_url', 'computer_call_output.output.image_url', 'code_interpreter_call.outputs', 'reasoning.encrypted_content', 'message.output_text.logprobs']] | None

var instructions : str | None

var max_output_tokens : int | None

var max_tool_calls : int | None

var metadata : Dict[str, str] | None

var parallel_tool_calls : bool | None

var prompt_cache_options : openai.types.responses.response_create_params.PromptCacheOptions

var reasoning : openai.types.shared.reasoning.Reasoning | dict[str, typing.Any] | None

var temperature : float | None

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

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

var top_p : float | None

var truncation : Literal['auto', 'disabled'] | None

var user : str