Language Models API

class DeterministicStopError(ValueError):
    """An empty completion a retry cannot fix. Subclasses `ValueError` so
    existing handlers keep working."""

    def __init__(self, message, finish_reason):
        super().__init__(message)
        self.finish_reason = finish_reason

@synalinks_export(
    [
        "synalinks.LanguageModel",
        "synalinks.language_models.LanguageModel",
    ]
)
class LanguageModel(Module):
    """A language model API wrapper.

    A language model is a type of AI model designed to generate, and interpret human
    language. It is trained on large amounts of text data to learn patterns and
    structures in language. Language models can perform various tasks such as text
    generation, translation, summarization, and answering questions.

    We support providers that implement *constrained structured output*
    like Azure, Ollama or Mistral. In addition we support providers that otherwise
    allow to constrain the use of a specific tool like Groq or Anthropic.

    For the complete list of models, please refer to the providers documentation.
    The provider prefixes accepted in `model` (the part before the `/`) are
    returned by `LanguageModel.supported_providers()`.

    **Using OpenAI models**

    ```python
    import synalinks
    import os

    os.environ["OPENAI_API_KEY"] = "your-api-key"

    language_model = synalinks.LanguageModel(
        model="openai/gpt-4o-mini",
    )
    ```

    **Using Groq models**

    ```python
    import synalinks
    import os

    os.environ["GROQ_API_KEY"] = "your-api-key"

    language_model = synalinks.LanguageModel(
        model="groq/llama3-8b-8192",
    )
    ```

    **Using Anthropic models**

    ```python
    import synalinks
    import os

    os.environ["ANTHROPIC_API_KEY"] = "your-api-key"

    language_model = synalinks.LanguageModel(
        model="anthropic/claude-3-sonnet-20240229",
    )
    ```

    **Using Mistral models**

    ```python
    import synalinks
    import os

    os.environ["MISTRAL_API_KEY"] = "your-api-key"

    language_model = synalinks.LanguageModel(
        model="mistral/codestral-latest",
    )
    ```

    **Using Ollama models**

    ```python
    import synalinks
    import os

    language_model = synalinks.LanguageModel(
        model="ollama/mistral",
    )
    ```

    **Using Azure models**

    ```python
    import synalinks
    import os

    os.environ["AZURE_API_KEY"] = "your-api-key"
    os.environ["AZURE_API_BASE"] = "your-api-base"
    os.environ["AZURE_API_VERSION"] = "your-api-version"

    language_model = synalinks.LanguageModel(
        model="azure/<your_deployment_name>",
    )
    ```

    **Using Google Gemini models**

    ```python
    import synalinks
    import os

    os.environ["GEMINI_API_KEY"] = "your-api-key"

    language_model = synalinks.LanguageModel(
        model="gemini/gemini-3.1-flash-lite-preview",
    )
    ```

    **Using XAI models**

    ```python
    import synalinks
    import os

    os.environ["XAI_API_KEY"] = "your-api-key"

    language_model = synalinks.LanguageModel(
        model="xai/grok-code-fast-1",
    )
    ```

    **Using Cohere models**

    ```python
    import synalinks
    import os

    os.environ["COHERE_API_KEY"] = "your-api-key"

    language_model = synalinks.LanguageModel(
        model="cohere/command-r-plus",
    )
    ```

    **Using DeepSeek models**

    ```python
    import synalinks
    import os

    os.environ["DEEPSEEK_API_KEY"] = "your-api-key"

    language_model = synalinks.LanguageModel(
        model="deepseek/deepseek-chat",
    )
    ```

    **Using Together AI models**

    ```python
    import synalinks
    import os

    os.environ["TOGETHER_AI_API_KEY"] = "your-api-key"

    language_model = synalinks.LanguageModel(
        model="together_ai/meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo",
    )
    ```

    **Using OpenRouter models**

    ```python
    import synalinks
    import os

    os.environ["OPENROUTER_API_KEY"] = "your-api-key"

    language_model = synalinks.LanguageModel(
        model="openrouter/anthropic/claude-3-haiku",
    )
    ```

    **Using AWS Bedrock models**

    ```python
    import synalinks
    import os

    os.environ["AWS_ACCESS_KEY_ID"] = "your-access-key"
    os.environ["AWS_SECRET_ACCESS_KEY"] = "your-secret-key"
    os.environ["AWS_REGION_NAME"] = "us-east-1"

    language_model = synalinks.LanguageModel(
        model="bedrock/anthropic.claude-3-sonnet-20240229-v1:0",
    )
    ```

    **Using Doubleword models**

    Doubleword exposes an OpenAI-compatible API. The `doubleword/`
    prefix is rewritten to `openai/` internally and `api_base` is
    defaulted to `https://api.doubleword.ai/v1`, so structured outputs
    flow through the standard OpenAI path. Set `OPENAI_API_KEY` to your
    Doubleword API key (or pass `api_base` explicitly to override).

    ```python
    import synalinks
    import os

    os.environ["OPENAI_API_KEY"] = "your-doubleword-api-key"

    language_model = synalinks.LanguageModel(
        model="doubleword/qwen-qwen3-5-397b-a17b-fp8-dottxt",
    )
    ```

    **Using Mirai models**

    Mirai is an on-device inference engine for Apple silicon, similar to
    Ollama but faster (`brew install mirai`). Start the server first with the
    model registry id; it downloads the model if needed and serves an
    OpenAI-compatible API on `http://localhost:8000/v1`:

    ```bash
    mirai server --model trymirai/Qwen3.8-27B-M
    ```

    The server hosts that one model and ignores the `model` field of
    requests, so the name after `mirai/` is informational. The `mirai/`
    prefix is rewritten to `hosted_vllm/` internally (same wire format, no
    API key) and `api_base` is defaulted to the local server, so structured
    outputs flow through the JSON schema path. Pass `api_base` explicitly
    for another host or port (`--host`/`--port` on the server side).

    ```python
    import synalinks

    language_model = synalinks.LanguageModel(
        model="mirai/Qwen3.8-27B-M",
    )
    ```

    **Using SGLang models**

    SGLang serves an OpenAI-compatible API, on `http://localhost:30000/v1` by
    default:

    ```bash
    python -m sglang.launch_server --model-path Qwen/Qwen3-8B
    ```

    The `sglang/` prefix is rewritten to `hosted_vllm/` internally (same wire
    format) and `api_base` is defaulted to `SGLANG_API_BASE`, then to the local
    server, so structured outputs flow through the JSON schema path. For a
    server started with `--api-key`, set `HOSTED_VLLM_API_KEY`.

    ```python
    import synalinks

    language_model = synalinks.LanguageModel(
        model="sglang/Qwen/Qwen3-8B",
    )
    ```

    **Using a ChatGPT subscription (no API key)**

    The `codex/` prefix reaches OpenAI models through a ChatGPT subscription
    login instead of an API key. It does not go through LiteLLM: use
    `synalinks.OAuthLanguageModel`, which sends the requests through the
    locally signed-in Codex CLI (see its documentation for the setup and
    limitations).

    ```python
    import synalinks

    language_model = synalinks.OAuthLanguageModel(
        model="codex/gpt-6.1-sol",
    )
    ```

    To cascade models in case there is anything wrong with
    the model provider (hence making your pipelines more robust).
    Use the `fallback` argument like in this example:

    ```python
    import synalinks
    import os

    os.environ["GEMINI_API_KEY"] = "your-api-key"
    os.environ["ANTHROPIC_API_KEY"] = "your-api-key"

    language_model = synalinks.LanguageModel(
        model="anthropic/claude-3-sonnet-20240229",
        fallback=synalinks.LanguageModel(
            model="gemini/gemini-3.1-flash-lite-preview",
        )
    )
    ```

    **Note**: Obviously, use an `.env` file and `.gitignore` to avoid
    putting your API keys in the code or a config file that can lead to
    leackage when pushing it into repositories.

    **Controlling reasoning effort**

    Reasoning ("thinking") models can be steered with the `reasoning_effort`
    parameter, passed either as a default in the constructor or per call. It
    accepts:

    - `"none"` (the default): send nothing, leaving the model at its provider
      default reasoning behavior.
    - `"disable"`: actively turn native reasoning *off*. This only has an
      effect for providers that reason by default, i.e. ollama's thinking
      models (`qwen3`, `deepseek-r1`, ...), where it maps to ollama's
      `think=False` toggle (safely ignored by non-thinking ollama models), and
      the ChatGPT backend (`codex/`), where it sends the lowest effort every
      model accepts (`"low"`: most of them cannot turn reasoning off).
      Opt-in providers (OpenAI, Anthropic, Gemini, ...) reason only when
      explicitly enabled, so there is nothing to send and this value is a
      no-op for them.
    - any other value (e.g. `"low"`, `"medium"`, `"high"`): forwarded to the
      provider as the reasoning effort, but only when the model supports
      reasoning (otherwise it is silently dropped).

    ```python
    import synalinks

    language_model = synalinks.LanguageModel(
        model="openai/o3-mini",
        reasoning_effort="medium",
    )
    ```

    Args:
        model (str): The model to use.
        api_base (str): Optional. The endpoint to use.
        timeout (int): Optional. The timeout value in seconds (Default to 600).
        retry (int): Optional. The number of retry (default to 5).
        retry_max_wait (int): Optional. Max seconds to wait between retries when a
            rate-limit `Retry-After` header is honored (default to 60).
        fallback (LanguageModel): Optional. The language model to fallback
            if anything is wrong.
        caching (bool): Optional. Enable caching of LM calls (Default to False).
        cache_dir (str): Optional. Directory for a persistent on-disk response
            cache. When set, every successful response is saved as a JSON
            file keyed by the full request (model, messages, schema and
            generation parameters), and identical requests are answered from
            disk without calling the provider, including across runs and
            processes. Streamed responses are cached once the stream is fully
            consumed (an abandoned stream is not cached) and a cache hit on a
            streaming call is replayed as a stream. (Default to None,
            disabled).
        name (str): Optional. The name of the module.
        description (str): Optional. The description of the module.
        hooks (list): Optional. Hooks to attach to this module's calls.
        **default_kwargs: Optional. Default generation parameters (e.g.
            `temperature`, `top_p`, `top_k`, `max_tokens`, `reasoning_effort`)
            forwarded to every call. Per-call kwargs override these. See
            "Controlling reasoning effort" above for the `reasoning_effort`
            values and their semantics.
    """

    def __init__(
        self,
        model=None,
        api_base=None,
        timeout=600,
        retry=5,
        retry_max_wait=60,
        fallback=None,
        caching=False,
        cache_dir=None,
        name=None,
        description=None,
        hooks=None,
        **default_kwargs: object,
    ):
        super().__init__(
            trainable=False,
            name=name,
            description=description,
            hooks=hooks,
        )
        # `messages` may be passed as a Pydantic DataModel; the strict
        # JsonDataModel guard would otherwise reject it.
        self._allow_non_json_data_model_positional_args = True
        if model is None:
            raise ValueError("You need to set the `model` argument for any LanguageModel")
        model_provider = model.split("/")[0]
        if model_provider == "ollama":
            # Switch from `ollama` to `ollama_chat`
            # because it have better performance due to the chat prompts
            model = model.replace("ollama", "ollama_chat")
        if model_provider == "vllm":
            model = model.replace("vllm", "hosted_vllm")
        if model_provider == "doubleword":
            # Doubleword is OpenAI-compatible (strict JSON schema + same
            # request/response shape); route via litellm's `openai`
            # provider with the Doubleword endpoint as `api_base`.
            model = model.replace("doubleword", "openai", 1)
            if not api_base:
                api_base = "https://api.doubleword.ai/v1"
        if model_provider == "mirai":
            # Mirai serves a local OpenAI-compatible API without an API
            # key; `hosted_vllm` speaks that wire format and never asks
            # for one.
            model = model.replace("mirai", "hosted_vllm", 1)
            if not api_base:
                api_base = "http://localhost:8000/v1"
        if model_provider == "sglang":
            # SGLang serves the same OpenAI-compatible API as vLLM.
            model = model.replace("sglang", "hosted_vllm", 1)
            if not api_base:
                api_base = os.environ.get("SGLANG_API_BASE", "http://localhost:30000/v1")
        self.model = model
        if fallback is not None:
            # Lazy import: `get` lives in the package __init__ which imports
            # this file at load time.
            from synalinks.src.modules.language_models import get as _get_lm

            fallback = _get_lm(fallback)
        self.fallback = fallback
        if self.model.startswith("ollama") and not api_base:
            self.api_base = "http://localhost:11434"
        else:
            self.api_base = api_base
        if self.model.startswith("hosted_vllm") and not api_base:
            self.api_base = os.environ.get(
                "HOSTED_VLLM_API_BASE", "http://localhost:8000"
            )
        self.timeout = timeout
        self.retry = retry
        self.retry_max_wait = retry_max_wait
        self.caching = caching
        self.cache_dir = cache_dir
        self._file_cache = FileCache(cache_dir) if cache_dir else None
        self.default_kwargs = default_kwargs
        self.cumulated_cost = 0.0
        self.last_call_cost = 0.0
        # All-time counters across every LM call (training + inference).
        # Useful for raw debugging; operational metrics use the
        # inference-scoped counters below instead.
        self.cumulated_calls = 0
        self.cumulated_prompt_tokens = 0
        self.cumulated_completion_tokens = 0
        self.cumulated_tokens = 0
        self.cumulated_elapsed_s = 0.0
        self.cumulated_cached_tokens = 0
        self.cumulated_cache_creation_tokens = 0
        self.cumulated_reasoning_tokens = 0
        # Streaming latency counters: number of streamed calls and the summed
        # time-to-first-token / time-to-last-token (seconds, measured from the
        # provider request start). Averaged by the streaming metrics. Only
        # populated for `streaming=True` calls.
        self.cumulated_streaming_calls = 0
        self.cumulated_streaming_ttft_s = 0.0
        self.cumulated_streaming_ttlt_s = 0.0
        # Whole-trajectory time-to-first-token: for a streamed call made inside
        # an agent trajectory, the wall-clock from the (outermost) agent's start
        # to the first token of the final answer -- includes every tool-calling
        # round, unlike `streaming_ttft_s` which times the final LM call only.
        # Only populated when a `trajectory_scope` is active (see op_scope).
        self.cumulated_trajectory_calls = 0
        self.cumulated_trajectory_ttft_s = 0.0
        # Failure counters: a call that exhausts all retries (`failed_calls`)
        # and each time the `fallback` chain is invoked because of it
        # (`fallback_activations`). Successful calls bump `cumulated_calls`;
        # these are tracked separately so error rate is observable.
        self.cumulated_failed_calls = 0
        self.cumulated_fallback_activations = 0
        # Number of calls answered from the on-disk `cache_dir` cache
        # (no provider request, no token counters bumped).
        self.cumulated_cache_hits = 0
        self.cumulated_details = {}
        self.last_call_prompt_tokens = 0
        self.last_call_completion_tokens = 0
        self.last_call_tokens = 0
        self.last_call_elapsed_s = 0.0
        self.last_call_cached_tokens = 0
        self.last_call_cache_creation_tokens = 0
        self.last_call_reasoning_tokens = 0
        # Why the last completion stopped, from the response choice.
        self.last_call_finish_reason = None
        # Phase-scoped counters, populated based on `synalinks_op_scope` set
        # by the trainer: "inference" inside `predict_on_batch`, "reward"
        # inside `compute_reward`, "optimizer" inside `optimizer.optimize`.
        # Calls made outside any scope (e.g. standalone debugging) are
        # tracked only in the all-time `cumulated_*` set above.
        #
        # Tier 1 extras (first-class, drive dedicated KPI metrics):
        #   cached_tokens, cache_creation_tokens, reasoning_tokens.
        # Tier 2 long tail (multimodal split, tool use, LiteLLM overhead)
        # lives in `<phase>_cumulated_details`, a dict accumulated per call.
        for _phase in ("inference", "reward", "optimizer"):
            setattr(self, f"{_phase}_cumulated_calls", 0)
            setattr(self, f"{_phase}_cumulated_prompt_tokens", 0)
            setattr(self, f"{_phase}_cumulated_completion_tokens", 0)
            setattr(self, f"{_phase}_cumulated_tokens", 0)
            setattr(self, f"{_phase}_cumulated_elapsed_s", 0.0)
            setattr(self, f"{_phase}_cumulated_cost", 0.0)
            setattr(self, f"{_phase}_cumulated_cached_tokens", 0)
            setattr(self, f"{_phase}_cumulated_cache_creation_tokens", 0)
            setattr(self, f"{_phase}_cumulated_reasoning_tokens", 0)
            setattr(self, f"{_phase}_cumulated_failed_calls", 0)
            setattr(self, f"{_phase}_cumulated_fallback_activations", 0)
            setattr(self, f"{_phase}_cumulated_cache_hits", 0)
            setattr(self, f"{_phase}_cumulated_streaming_calls", 0)
            setattr(self, f"{_phase}_cumulated_streaming_ttft_s", 0.0)
            setattr(self, f"{_phase}_cumulated_streaming_ttlt_s", 0.0)
            setattr(self, f"{_phase}_cumulated_trajectory_calls", 0)
            setattr(self, f"{_phase}_cumulated_trajectory_ttft_s", 0.0)
            setattr(self, f"{_phase}_cumulated_details", {})
        # No state depends on the input shape, so mark built up-front and
        # skip Module's auto-build path (which would try to trace `call`).
        self.built = True

    def _record_event(self, suffix):
        """Bump an all-time + phase-scoped operational counter by 1 (used for
        `failed_calls` / `fallback_activations`). Phase follows the active
        `op_scope`, matching how successful-call counters are attributed.
        """
        op = current_op_scope()
        _accumulate(self, "", {suffix: 1}, None)
        if op is not None:
            _accumulate(self, f"{op}_", {suffix: 1}, None)

    async def call(
        self,
        messages,
        schema=None,
        tools=None,
        tool_schemas=None,
        streaming=False,
        **kwargs,
    ):
        """
        Call method to generate a response using the language model.

        Args:
            messages (dict): A formatted dict of chat messages.
            schema (dict): The target JSON schema for structed output (optional).
                If None, output a ChatMessage-like answer.
            tools (list | dict): Optional iterable or `{name: Tool}` mapping of
                `synalinks.modules.Tool` the LM may call. Mutually exclusive
                with `schema`: schema forces structured output, tools let the
                LM choose; they cannot both apply to the same call. In the
                function-calling agent pattern, the tool-call generator uses
                `tools` and the final generator uses `schema`.
            tool_schemas (list): Optional iterable of already-wire-formatted
                tool declaration dicts (OpenAI `{"type": "function", ...}`
                shape) passed through verbatim, appended after any converted
                `tools`. Use this to expose tools you already have a schema for
                without wrapping them in a `synalinks.modules.Tool`. Subject to
                the same mutual exclusivity with `schema` as `tools`.
            streaming (bool): Enable streaming (optional). Default to False.
                Can be enabled only if schema is None.
            **kwargs (keyword arguments): The additional keywords arguments
                forwarded to the LM call.
        Returns:
            (dict): The generated structured response.
        """
        if not messages:
            return None
        if schema and (tools or tool_schemas):
            raise ValueError(
                "`schema` and `tools` cannot be passed to the same LM call: "
                "schema forces structured output, while tools let the LM choose "
                "which to call. Split into two calls: typically the tool-call "
                "generator uses `tools` and the final generator uses `schema`."
            )
        # Cleared per call: a cache hit or a streamed call has no choice to
        # read it from and must not report the previous call's reason.
        self.last_call_finish_reason = None
        _CURRENT_CALL_USAGE.set(None)
        input_kwargs = copy.deepcopy(kwargs)
        # Merge instance-level defaults; per-call kwargs win.
        kwargs = {**self.default_kwargs, **kwargs}
        schema = copy.deepcopy(schema)
        provider = self.model.split("/")[0]

        # Single pass: messages → OpenAI wire shape (nested tool_call
        # envelopes, JSON-string arguments) and synalinks Tools → wire tool
        # declarations. The schema branches below may override `tools` /
        # `tool_choice` for providers that need structured-output-as-tool;
        # cache_control is applied to the system message after this.
        # `messages` here is typically a JsonDataModel from `ops.predict`;
        # wrap it in `ChatMessages` so its `before`-validator converts the
        # dict list into typed `ChatMessage` instances the converter needs.
        _typed_messages = (
            messages
            if hasattr(messages, "messages")
            else ChatMessages(messages=messages.get("messages", []))
        )
        formatted_messages = [_message_to_wire(m) for m in _typed_messages.messages]
        # Inline any deferred multimodal references (a dataset's image_url/
        # input_audio parts carrying a url/path) into base64 payloads, for this
        # batch only. Content already inlined at construction is left untouched.
        formatted_messages = await resolve_content_media(formatted_messages)
        # Media returned by tools (a sandbox plot, a `read_image`, a
        # `read_audio`): images stay inside the tool result for the models
        # that take them there (Claude's `tool_result`, Gemini 3's multimodal
        # function responses; litellm maps both), the rest goes to a
        # follow-up user message, the only role the other APIs take it in.
        media_kinds = {
            part.get("type")
            for message in formatted_messages
            if message["role"] == "tool" and isinstance(message["content"], list)
            for part in message["content"]
        }
        if media_kinds & {"image_url", "input_audio"}:
            formatted_messages = place_tool_result_media(
                formatted_messages,
                native=(
                    provider in ("anthropic", "vertex_ai", "bedrock")
                    and "claude" in self.model
                )
                or (provider in ("gemini", "vertex_ai") and "gemini-3" in self.model),
                vision="image_url" in media_kinds and self.supports_vision(),
                audio="input_audio" in media_kinds and self.supports_audio(),
            )
        if tools or tool_schemas:
            wire_tools = []
            if tools:
                if isinstance(tools, dict):
                    tools = tools.values()
                wire_tools.extend(_tool_to_wire(t) for t in tools)
            if tool_schemas:
                # Already in OpenAI wire shape; pass through verbatim.
                wire_tools.extend(tool_schemas)
            kwargs["tools"] = wire_tools

        # Handle reasoning_effort:
        #   "none"    -> send nothing; leave the model at its provider default.
        #   "disable" -> actively turn native reasoning OFF. This only needs an
        #                explicit flag where reasoning is ON by default, i.e.
        #                ollama's thinking models (qwen3, deepseek-r1, ...), which
        #                reason unless told not to. `think=False` is the ollama
        #                toggle and is safely ignored by non-thinking ollama
        #                models. The ChatGPT backend (`codex/`) also reasons by
        #                default: "disable" is handed to its transport, which
        #                knows the lowest effort the CLI accepts. Opt-in
        #                providers (OpenAI, Anthropic, Gemini, ...) reason only
        #                when enabled, so there is nothing to send.
        #   otherwise -> forward the effort to litellm when the model supports it.
        reasoning_effort = kwargs.pop("reasoning_effort", "none")
        schema_had_thinking = bool(schema) and "thinking" in (
            schema.get("properties") or {}
        )
        if reasoning_effort == "disable":
            if self.model.startswith("ollama"):
                kwargs["think"] = False
            elif self.model.startswith("codex"):
                kwargs["reasoning_effort"] = "disable"
        elif reasoning_effort != "none":
            if self.supports_reasoning():
                kwargs["reasoning_effort"] = reasoning_effort
                if schema_had_thinking and self.supports_reasoning_trace():
                    # The LM produces a native reasoning trace via
                    # `reasoning_content`; strip `thinking` from the LM
                    # schema to save tokens; we re-inject it after the call.
                    schema["properties"].pop("thinking", None)
                    required = schema.get("required")
                    if isinstance(required, list) and "thinking" in required:
                        required.remove("thinking")

        if schema:
            if self.model.startswith("cohere"):
                # Cohere's v2 chat constrains to `schema`; litellm refuses the param.
                extra_body = dict(kwargs.get("extra_body") or {})
                extra_body.setdefault(
                    "response_format", {"type": "json_object", "schema": schema}
                )
                kwargs["extra_body"] = extra_body
            elif self.model.startswith("anthropic"):
                # Use response_format for Anthropic - LiteLLM handles this correctly:
                # - For newer models (sonnet-4.5, opus-4.1): uses native output_format
                # - For older models: uses tool call with proper tool_choice handling
                #   (auto when thinking is enabled, forced otherwise)
                _set_if_unset(
                    kwargs,
                    {
                        "response_format": {
                            "type": "json_schema",
                            "json_schema": {
                                "schema": schema,
                            },
                        },
                    },
                )
            elif self.model.startswith("ollama") or self.model.startswith("mistral"):
                # Use constrained structured output for ollama/mistral
                _set_if_unset(
                    kwargs,
                    {
                        "response_format": {
                            "type": "json_schema",
                            "json_schema": {"schema": schema},
                            "strict": True,
                        },
                    },
                )
            elif (
                self.model.startswith("openai")
                or self.model.startswith("azure")
                or self.model.startswith("deepseek")
                or self.model.startswith("together_ai")
                or self.model.startswith("huggingface")
                or self.model.startswith("openrouter")
                or self.model.startswith("groq")
                or self.model.startswith("bedrock")
                or self.model.startswith("codex")
            ):
                # Use constrained structured output for openai/azure
                # plus deepseek, together_ai, huggingface (TGI and
                # Inference Providers), openrouter, groq and bedrock, which
                # honor the same payload (litellm maps it to bedrock's
                # `outputConfig.textFormat`). For a groq or bedrock model
                # without native support, litellm falls back to a forced tool
                # call itself.
                # OpenAI/Azure require the field  "additionalProperties"
                # Also OpenAI/Azure disallow the field "description" in $ref
                if "properties" in schema:
                    for prop_key, prop_value in schema["properties"].items():
                        if "$ref" in prop_value and "description" in prop_value:
                            del prop_value["description"]
                _set_if_unset(
                    kwargs,
                    {
                        "response_format": {
                            "type": "json_schema",
                            "json_schema": {
                                "name": "structured_output",
                                "strict": True,
                                "schema": schema,
                            },
                        }
                    },
                )
                if self.model.startswith("openrouter"):
                    # Else an endpoint ignoring `response_format` answers in prose.
                    provider = dict(kwargs.get("provider") or {})
                    provider.setdefault("require_parameters", True)
                    chosen = {
                        str(name).lower()
                        for key in ("order", "only")
                        for name in provider.get(key) or ()
                    }
                    ignored = list(provider.get("ignore") or ())
                    known = {str(name).lower() for name in ignored}
                    ignored += [
                        name
                        for name in OPENROUTER_IGNORED_PROVIDERS
                        if name not in known and name not in chosen
                    ]
                    if ignored:
                        provider["ignore"] = ignored
                    kwargs["provider"] = provider
            elif self.model.startswith("gemini"):
                _set_if_unset(
                    kwargs,
                    {
                        "response_format": {
                            "type": "json_schema",
                            "json_schema": {
                                "schema": schema,
                            },
                            "strict": True,
                        }
                    },
                )
            elif self.model.startswith("xai"):
                _set_if_unset(
                    kwargs,
                    {
                        "response_format": {
                            "type": "json_schema",
                            "json_schema": {
                                "schema": schema,
                            },
                            "strict": True,
                        }
                    },
                )
            elif self.model.startswith("hosted_vllm"):
                _set_if_unset(
                    kwargs,
                    {
                        "response_format": {
                            "type": "json_schema",
                            "json_schema": {
                                "name": "structured_output",
                                "schema": schema,
                            },
                            "strict": True,
                        }
                    },
                )
            else:
                provider = self.model.split("/")[0]
                raise ValueError(
                    f"LM provider '{provider}' not supported yet, please ensure that"
                    " they support constrained structured output and fill an issue."
                    f" Supported providers: {', '.join(SUPPORTED_PROVIDERS)}."
                )

        if self.api_base:
            kwargs.update(
                {
                    "api_base": self.api_base,
                }
            )
        if streaming and schema:
            streaming = False
        if streaming:
            kwargs.update({"stream": True})
        # Enable prompt caching for the system instructions
        # (that only change during training not inference)
        if provider in ("gemini", "anthropic"):
            system_message_with_cache_control = {
                **formatted_messages[0],
                "cache_control": {"type": "ephemeral"},
            }
            formatted_messages[0] = system_message_with_cache_control
        # Persistent on-disk cache: keyed by the fully-formatted request
        # (model, wire messages, schema and final kwargs), so any change to
        # prompts, tools or generation parameters yields a different entry.
        # The `stream` flag is excluded from the key: a streamed call stores
        # the same assistant ChatMessage a non-streamed schema-less call does
        # (streaming forces schema=None), so both share one entry; a hit is
        # replayed as a single-chunk stream when `streaming=True`.
        cache_key = None
        if self._file_cache is not None:
            cache_key = self._file_cache.make_key(
                {
                    "model": self.model,
                    "messages": formatted_messages,
                    "schema": schema,
                    "kwargs": {k: v for k, v in kwargs.items() if k != "stream"},
                }
            )
            if cache_key is not None:
                cached_json = self._file_cache.get(cache_key)
                if cached_json is not None:
                    self._record_event("cache_hits")
                    _CURRENT_CALL_USAGE.set({"cache_hit": True})
                    if streaming:
                        return StreamingIterator(_cached_message_to_chunks(cached_json))
                    return JsonDataModel(
                        json=cached_json,
                        schema=schema if schema else ChatMessage.get_schema(),
                        name=f"{self.name}_response",
                    )
        try:
            response = await self._call_with_retry(
                formatted_messages,
                schema,
                streaming,
                schema_had_thinking,
                **kwargs,
            )
            if cache_key is not None and response is not None:
                if streaming:
                    # Cache once the caller drains the stream; an abandoned
                    # stream is incomplete and is not cached.
                    file_cache = self._file_cache
                    response._on_complete = lambda json_instance: file_cache.set(
                        cache_key, json_instance
                    )
                else:
                    self._file_cache.set(cache_key, response.get_json())
            return response
        except Exception as e:
            warnings.warn(f"All retries failed for {self}: {e}")
            self._record_event("failed_calls")
            if self.fallback:
                self._record_event("fallback_activations")
                response = await self.fallback(
                    messages,
                    schema=schema,
                    streaming=streaming,
                    **input_kwargs,
                )
                # The fallback's usage is reported on its own span: this call
                # only reports its failure.
                _CURRENT_CALL_USAGE.set({"error": str(e), "fallback": True})
                return response
            else:
                _CURRENT_CALL_USAGE.set({"error": str(e)})
                return None

    async def _acompletion(self, formatted_messages, **kwargs):
        """Send one completion request and return the provider response.

        This is the transport seam: everything around it (message
        formatting, structured-output payload, cache, retry, fallback, usage
        accounting and response parsing) is provider-agnostic. A subclass
        that does not go through LiteLLM (e.g. `OAuthLanguageModel`)
        overrides this method and returns a LiteLLM-shaped response.

        Args:
            formatted_messages (list): The chat messages in OpenAI wire shape.
            **kwargs (keyword arguments): The request parameters.

        Returns:
            (ModelResponse): The provider response.
        """
        return await litellm.acompletion(
            model=self.model,
            messages=formatted_messages,
            timeout=self.timeout,
            caching=self.caching,
            **kwargs,
        )

    async def _call_with_retry(
        self, formatted_messages, schema, streaming, schema_had_thinking, **kwargs
    ):
        """Perform the LM call with tenacity retry logic."""
        logger = logging.getLogger(__name__)

        @retry(
            stop=stop_after_attempt(self.retry),
            # Honor a rate-limit `Retry-After` header (e.g. Azure OpenAI TPM
            # throttling); fall back to exponential backoff for other errors.
            wait=rate_limit_aware_wait(max_wait=self.retry_max_wait),
            retry=retry_if_not_exception_type(DeterministicStopError),
            before_sleep=before_sleep_log(logger, logging.WARNING),
            reraise=True,
        )
        async def _do_call():
            response_str = ""
            try:
                t0 = time.perf_counter()
                response = await self._acompletion(formatted_messages, **kwargs)
                elapsed_s = time.perf_counter() - t0
                op_scope = current_op_scope()
                response_cost = None
                if hasattr(response, "_hidden_params"):
                    if "response_cost" in response._hidden_params:
                        response_cost = response._hidden_params["response_cost"]
                        if response_cost is not None:
                            self.last_call_cost = response_cost
                # Streaming usage isn't known until the stream completes,
                # so skip counter updates in that case.
                if not streaming:
                    usage = response.get("usage") or {}
                    prompt_tokens = int(usage.get("prompt_tokens") or 0)
                    completion_tokens = int(usage.get("completion_tokens") or 0)
                    total_tokens = int(
                        usage.get("total_tokens") or (prompt_tokens + completion_tokens)
                    )
                    cached, cache_creation, reasoning, extras = _extract_lm_extras(
                        usage, response
                    )
                    self.last_call_prompt_tokens = prompt_tokens
                    self.last_call_completion_tokens = completion_tokens
                    self.last_call_tokens = total_tokens
                    self.last_call_elapsed_s = elapsed_s
                    self.last_call_cached_tokens = cached
                    self.last_call_cache_creation_tokens = cache_creation
                    self.last_call_reasoning_tokens = reasoning
                    flat_increments = {
                        "calls": 1,
                        "prompt_tokens": prompt_tokens,
                        "completion_tokens": completion_tokens,
                        "tokens": total_tokens,
                        "elapsed_s": elapsed_s,
                        "cached_tokens": cached,
                        "cache_creation_tokens": cache_creation,
                        "reasoning_tokens": reasoning,
                    }
                    if response_cost is not None:
                        flat_increments["cost"] = response_cost
                    _CURRENT_CALL_USAGE.set(
                        {
                            "input_tokens": prompt_tokens,
                            "output_tokens": completion_tokens,
                            "total_tokens": total_tokens,
                            "cache_read_input_tokens": cached,
                            "cache_creation_input_tokens": cache_creation,
                            "reasoning_tokens": reasoning,
                            "cost": response_cost,
                            "elapsed_s": elapsed_s,
                            "finish_reason": None,
                        }
                    )
                    _accumulate(self, "", flat_increments, extras)
                    if op_scope is not None:
                        _accumulate(self, f"{op_scope}_", flat_increments, extras)
                if streaming:
                    return StreamingIterator(
                        response,
                        language_model=self,
                        op_scope=op_scope,
                        request_start=t0,
                        trajectory_start=current_trajectory_start(),
                    )
                if not response.get("choices"):
                    raise ValueError(
                        "Empty response from the language model: no choices returned."
                    )
                response_message = response["choices"][0]["message"]
                finish_reason = _safe_get(response["choices"][0], "finish_reason")
                self.last_call_finish_reason = finish_reason
                if _CURRENT_CALL_USAGE.get():
                    _CURRENT_CALL_USAGE.get()["finish_reason"] = finish_reason
                wire_tool_calls = _safe_get(response_message, "tool_calls", None)
                refusal = _safe_get(response_message, "refusal")
                audio = _safe_get(response_message, "audio")
                if audio is not None and hasattr(audio, "model_dump"):
                    audio = audio.model_dump()
                if schema and wire_tool_calls:
                    # A reply carrying the structured output as a tool call
                    # (e.g. a caller's own tool payload): the JSON is the
                    # call's `arguments`, while `content` is empty or a
                    # short preamble. Read it wherever a tool call came
                    # back, whatever the provider.
                    arguments = _safe_get(
                        _safe_get(wire_tool_calls[0], "function"), "arguments", ""
                    )
                    if isinstance(arguments, (dict, list)):
                        arguments = orjson.dumps(arguments).decode()
                    response_str = arguments.strip() if arguments else ""
                    if not response_str:
                        raise ValueError(
                            "The language model returned a structured_output tool "
                            "call without arguments."
                        )
                else:
                    # Anthropic and other providers use response_format,
                    # which returns content in message["content"]
                    response_str = response_message["content"]
                    if not response_str and not wire_tool_calls and not audio:
                        if refusal:
                            raise ValueError(
                                "The language model refused to answer: " + refusal
                            )
                        if finish_reason in DETERMINISTIC_FINISH_REASONS:
                            raise DeterministicStopError(
                                _empty_response_message(finish_reason),
                                finish_reason,
                            )
                        raise ValueError(_empty_response_message(finish_reason))
                    response_str = response_str.strip() if response_str else ""
                reasoning_content = response_message.get("reasoning_content")
                thinking_blocks = response_message.get("thinking_blocks")
                if schema:
                    if not response_str and refusal:
                        raise ValueError(
                            "The language model refused to produce the "
                            "structured output: " + refusal
                        )
                    json_instance = orjson.loads(response_str)
                    if (
                        reasoning_content
                        and schema_had_thinking
                        and self.supports_reasoning_trace()
                    ):
                        # The trace replaces whatever the model wrote there, and
                        # comes first, where the schema holds it, as it would
                        # had the model written it.
                        json_instance.pop("thinking", None)
                        json_instance = {"thinking": reasoning_content, **json_instance}
                else:
                    # OpenAI's nested tool-call envelope is the synalinks
                    # `ToolCall` shape (`{id, type, function: {name,
                    # arguments}}`), the `arguments` the JSON-encoded string
                    # the model wrote: parsed when the tool is called.
                    parsed_tool_calls = None
                    if wire_tool_calls:
                        parsed_tool_calls = []
                        for tc in wire_tool_calls:
                            fn = _safe_get(tc, "function")
                            args = _safe_get(fn, "arguments", "")
                            if isinstance(args, (dict, list)):
                                args = orjson.dumps(args).decode()
                            parsed_tool_calls.append(
                                {
                                    "id": _safe_get(tc, "id"),
                                    "type": "function",
                                    "function": {
                                        "name": _safe_get(fn, "name"),
                                        # A call without arguments is sent back as `{}`.
                                        "arguments": args or "{}",
                                    },
                                }
                            )
                    json_instance = {
                        "role": ChatRole.ASSISTANT,
                        "content": response_str,
                    }
                    if reasoning_content:
                        json_instance["reasoning_content"] = reasoning_content
                    if thinking_blocks:
                        json_instance["thinking_blocks"] = thinking_blocks
                    if parsed_tool_calls:
                        json_instance["tool_calls"] = parsed_tool_calls
                    if refusal:
                        json_instance["refusal"] = refusal
                    if audio:
                        json_instance["audio"] = audio
                return JsonDataModel(
                    json=json_instance,
                    schema=schema if schema else ChatMessage.get_schema(),
                    name=f"{self.name}_response",
                )
            except Exception as e:
                warnings.warn(
                    f"Error occured while trying to call {self}: "
                    + str(e)
                    + f"\nReceived response={shorten_text(response_str)}"
                )
                raise

        return await _do_call()

    @classmethod
    def supported_providers(cls):
        """Returns the provider prefixes supported for structured output.

        These are the values accepted before the `/` in `model`, e.g.
        `"openai"` in `"openai/gpt-4o-mini"`. Some are routed to a different
        litellm provider internally (`ollama` to `ollama_chat`, `vllm` to
        `hosted_vllm`, `doubleword` to `openai`, `mirai` and `sglang` to
        `hosted_vllm`); the names listed here are the ones to write in `model`.
        Other litellm providers can still be used for plain chat completion, but
        structured output raises for them.

        ```python
        import synalinks

        print(synalinks.LanguageModel.supported_providers())
        ```

        Returns:
            (list): The sorted list of supported provider prefixes.
        """
        return list(SUPPORTED_PROVIDERS)

    @classmethod
    def provider_info(cls, provider):
        """Returns what a frontend needs to set a provider up.

        ```python
        import synalinks

        print(synalinks.LanguageModel.provider_info("ollama"))
        ```

        Args:
            provider (str): A provider prefix from `supported_providers()`.

        Returns:
            (dict): `provider`, `env` (the environment variables to set, each
                `{"name", "required", "secret"}`), `default_api_base` (or
                `None`) and `lists_models_from` (`"server"` when
                `supported_models()` asks the provider's server, else
                `"catalog"`).

        Raises:
            ValueError: If the provider is not supported.
        """
        model_catalog.check_provider(
            provider, cls.supported_providers(), "language model"
        )
        return model_catalog.provider_info(provider, model_catalog.PROVIDERS[provider])

    @classmethod
    async def supported_models(cls, provider, api_base=None):
        """Returns the language models a provider offers.

        Self-hosted providers and open catalogs (Ollama, vLLM, SGLang, Mirai,
        OpenRouter, Doubleword) are asked for the models they serve, keeping the Ollama
        models that can chat. The others are listed from litellm's model table;
        for OpenAI with `OPENAI_API_KEY` set, narrowed to the models the account
        can use (with its fine-tuned models), from OpenAI's live list.

        ```python
        import synalinks

        print(await synalinks.LanguageModel.supported_models("ollama"))
        ```

        Args:
            provider (str): A provider prefix from `supported_providers()`.
            api_base (str): Optional. The server to ask, for a provider listed
                from its server or from OpenAI's live list (defaults to its base
                URL variable, then to its default endpoint, see
                `provider_info()`).

        Returns:
            (list): The sorted model IDs, ready to pass as `model`
                (e.g. `"openai/gpt-4o-mini"`).

        Raises:
            ValueError: If the provider is not supported, or no endpoint is
                known to list its models from.
            ConnectionError: If the provider's server cannot be reached.
        """
        model_catalog.check_provider(
            provider, cls.supported_providers(), "language model"
        )
        info = model_catalog.PROVIDERS[provider]
        if info.get("server"):
            return await model_catalog.server_models(
                provider, info, api_base, ollama_capability="completion"
            )
        return await model_catalog.account_models(
            provider, model_catalog.catalog_models(provider, ("chat",)), api_base
        )

    @classmethod
    def model_info(cls, model):
        """Returns a model's info from litellm's model table.

        ```python
        import synalinks

        print(synalinks.LanguageModel.model_info("openai/gpt-4o-mini"))
        ```

        Args:
            model (str): The model, e.g. `"openai/gpt-4o-mini"`.

        Returns:
            (dict): `max_input_tokens`, `max_output_tokens`,
                `input_cost_per_token`, `output_cost_per_token` (each `None`
                when unknown) and `capabilities` (e.g. `["vision"]`), or
                `None` when litellm does not know the model (e.g. a
                self-hosted one).
        """
        return model_catalog.model_info(model)
    def supports_reasoning(self):
        """Whether `reasoning_effort` is forwarded to the provider.

        Looked up in litellm's model table: when True, the effort is sent
        with the request (see also `supports_reasoning_trace()`).

        Returns:
            (bool): True if the model takes a reasoning effort.
        """
        return litellm.supports_reasoning(model=self.model)

    def supports_reasoning_trace(self):
        """Whether the native reasoning trace fills the `thinking` field.

        When True and a reasoning effort is sent, the `thinking` field is left
        out of the schema handed to the model and filled with the
        `reasoning_content` of the response instead. When False, the model
        writes `thinking` like any other field.

        Returns:
            (bool): True if the reasoning trace replaces `thinking`.
        """
        return True

    def supports_vision(self):
        """Whether the model accepts images in its input.

        An Ollama model is asked for its capabilities (its server reports
        ``vision`` for the multimodal ones); any other model is looked up in
        litellm's model table, where an unlisted model counts as text-only.
        Register a self-hosted vision model with `litellm.register_model`
        (``supports_vision: True``) to make it pass.

        ```python
        import synalinks

        lm = synalinks.LanguageModel(model="openai/gpt-4o-mini")
        print(lm.supports_vision())  # True
        ```

        Returns:
            (bool): True if the model can read images.
        """
        if self.model.startswith("ollama"):
            return "vision" in self.ollama_capabilities()
        return bool(litellm.supports_vision(model=self.model))

    def supports_audio(self):
        """Whether the model accepts audio in its input.

        Looked up like `supports_vision`: an Ollama model reports ``audio``
        among its capabilities, any other model is looked up in litellm's
        model table (`litellm.register_model` with ``supports_audio_input:
        True`` makes an unlisted one pass).

        ```python
        import synalinks

        lm = synalinks.LanguageModel(model="gemini/gemini-2.5-flash")
        print(lm.supports_audio())  # True
        ```

        Returns:
            (bool): True if the model can listen to audio.
        """
        if self.model.startswith("ollama"):
            return "audio" in self.ollama_capabilities()
        return bool(litellm.supports_audio_input(model=self.model))

    def ollama_capabilities(self):
        """The capabilities an Ollama server reports for this model.

        Returns:
            (list): E.g. ``["completion", "vision", "tools"]``; empty when the
            server cannot be reached or does not know the model.
        """
        try:
            response = httpx.post(
                f"{self.api_base}/api/show",
                json={"model": self.model.split("/", 1)[1]},
                timeout=10,
            )
            response.raise_for_status()
        except httpx.HTTPError:
            return []
        return response.json().get("capabilities") or []

    def supports_softprompt(self):
        """Whether the provider accepts soft prompts (prompt embeddings).

        Only vLLM takes embeddings in place of prompt tokens (its
        ``prompt_embeds`` input, served with ``--enable-prompt-embeds``); the
        hosted APIs take text and media only.

        Returns:
            (bool): True if the model is served by vLLM.
        """
        return self.model.startswith("hosted_vllm")

    def _obj_type(self):
        return "LanguageModel"

    def get_config(self):
        config = {
            "model": self.model,
            "api_base": self.api_base,
            "timeout": self.timeout,
            "retry": self.retry,
            "retry_max_wait": self.retry_max_wait,
            "caching": self.caching,
            "cache_dir": self.cache_dir,
            "name": self.name,
            "description": self.description,
            **self.default_kwargs,
        }
        if self.fallback:
            fallback_config = {
                "fallback": serialization_lib.serialize_synalinks_object(
                    self.fallback,
                )
            }
            return {**fallback_config, **config}
        else:
            return config

    @classmethod
    def from_config(cls, config):
        if "fallback" in config:
            fallback = serialization_lib.deserialize_synalinks_object(
                config.pop("fallback")
            )
            return cls(fallback=fallback, **config)
        else:
            return cls(**config)

    def __repr__(self):
        api_base = f" api_base={self.api_base}" if self.api_base else ""
        return f"<LanguageModel model={self.model}{api_base}>"