llm/endpoints.py
Rama main · réplica de lectura del repositorio original.
# SPDX-License-Identifier: BUSL-1.1
# Copyright (c) 2024-2026 Ali Sadeghi Aghili
"""Named LLM endpoints and per-task fallback routes, built from :mod:`config`.
The trivial case — one endpoint for every task — needs no configuration
beyond ``OPENAI_BASE_URL`` / ``OPENAI_MODEL`` / ``OPENAI_API_KEY``: that
alone gives every :class:`~llm.router.TaskType` a one-entry chain pointing
at a single :class:`~llm.providers.OpenAIBackend`, exactly as if this
module didn't exist. A deployment that wants a real router — a fast local
model for one task, a fallback for when it's down, a different model
entirely for another task — declares additional named endpoints in
``LLM_ENDPOINTS`` (a JSON array) and, optionally, which endpoints answer
which task and in what order in ``LLM_ROUTES`` (a JSON object). Nothing
about the trivial case changes when neither is set.
Example
-------
::
LLM_ENDPOINTS=[
{"name": "local", "base_url": "http://localhost:8000/v1", "model": "gpt-oss-20b"},
{"name": "hosted", "base_url": "https://api.openai.com/v1", "model": "gpt-4o-mini", "api_key": "sk-..."}
]
LLM_ROUTES={"sql_generation": ["local", "hosted"], "interpretation": ["local"]}
Every :class:`~llm.router.TaskType` not mentioned in ``LLM_ROUTES`` still
defaults to ``["default"]`` — the single endpoint built from the plain
``OPENAI_*`` variables.
"""
from __future__ import annotations
import json
from dataclasses import dataclass
import config as cfg
from llm.base import LLMBackend
from llm.providers import OpenAIBackend
#: The always-present endpoint name, built from OPENAI_BASE_URL / OPENAI_MODEL
#: / OPENAI_API_KEY / LLM_TRUSTED — the trivial single-endpoint case.
DEFAULT_ENDPOINT_NAME = "default"
#: Every ``TaskType`` value string, duplicated here as plain strings (rather
#: than importing ``llm.router.TaskType``) so this module has no dependency
#: on :mod:`llm.router` at all -- :mod:`llm.router` is the one that depends
#: on THIS module, inside ``LLMRouter.from_settings``, and a two-way import
#: would be a cycle.
_ALL_TASK_NAMES: tuple[str, ...] = ("sql_generation", "interpretation", "assumption_extraction")
__all__ = [
"DEFAULT_ENDPOINT_NAME",
"EndpointConfig",
"build_backend",
"load_endpoints",
"load_routes",
]
@dataclass(frozen=True)
class EndpointConfig:
"""One named LLM endpoint.
Parameters
----------
name:
Unique key used by ``LLM_ROUTES`` to reference this endpoint.
base_url, model, api_key:
As :class:`~llm.providers.OpenAIBackend`.
trusted:
Explicit override for whether this endpoint may see
schema/business-rule/row data. ``None`` (the default) defers to
:func:`~llm.trust.default_trust_for_url` at backend-construction
time — see :meth:`~llm.providers.OpenAIBackend.trusted`.
"""
name: str
base_url: str
model: str
api_key: str = ""
trusted: bool | None = None
def _default_endpoint() -> EndpointConfig:
"""The always-present ``"default"`` endpoint, built from plain OPENAI_* settings."""
return EndpointConfig(
name=DEFAULT_ENDPOINT_NAME,
base_url=cfg.settings.openai_base_url,
model=cfg.settings.openai_model,
api_key=cfg.settings.openai_api_key,
trusted=cfg.settings.llm_trusted,
)
def load_endpoints() -> dict[str, EndpointConfig]:
"""Build the named-endpoint registry from :mod:`config`.
Always includes :data:`DEFAULT_ENDPOINT_NAME`. Additional endpoints
come from ``LLM_ENDPOINTS`` — a JSON array of objects, each requiring
at least ``"name"``, ``"base_url"``, and ``"model"`` — and, if one of
them is itself named ``"default"``, it overrides the plain-settings
default entirely (letting a deployment fold everything into
``LLM_ENDPOINTS`` if it prefers one config surface over two).
Returns
-------
dict[str, EndpointConfig]
Raises
------
ValueError
If ``LLM_ENDPOINTS`` is set but is not valid JSON, or an entry is
missing a required key.
Examples
--------
>>> import config as cfg
>>> with cfg.override_settings(llm_endpoints_json=""):
... endpoints = load_endpoints()
>>> list(endpoints)
['default']
"""
endpoints: dict[str, EndpointConfig] = {DEFAULT_ENDPOINT_NAME: _default_endpoint()}
raw = cfg.settings.llm_endpoints_json.strip()
if not raw:
return endpoints
try:
declared = json.loads(raw)
except json.JSONDecodeError as exc:
raise ValueError(f"LLM_ENDPOINTS is not valid JSON: {exc}") from exc
if not isinstance(declared, list):
raise ValueError("LLM_ENDPOINTS must be a JSON array of endpoint objects")
for item in declared:
try:
name = item["name"]
base_url = item["base_url"]
model = item["model"]
except (KeyError, TypeError) as exc:
raise ValueError(
f"LLM_ENDPOINTS entry missing a required key (name/base_url/model): {item!r}"
) from exc
endpoints[name] = EndpointConfig(
name=name,
base_url=base_url,
model=model,
api_key=item.get("api_key", ""),
trusted=item.get("trusted"),
)
return endpoints
def load_routes() -> dict[str, list[str]]:
"""Per-task endpoint fallback chains, keyed by ``TaskType`` value string.
Every known task defaults to ``["default"]``. ``LLM_ROUTES`` — a JSON
object mapping a task name to a list of endpoint names, tried in order
— overrides only the tasks it mentions; any task it doesn't mention
keeps the ``["default"]`` fallback.
Returns
-------
dict[str, list[str]]
Raises
------
ValueError
If ``LLM_ROUTES`` is set but is not valid JSON.
Examples
--------
>>> import config as cfg
>>> with cfg.override_settings(llm_routes_json=""):
... routes = load_routes()
>>> routes["sql_generation"]
['default']
"""
routes: dict[str, list[str]] = {name: [DEFAULT_ENDPOINT_NAME] for name in _ALL_TASK_NAMES}
raw = cfg.settings.llm_routes_json.strip()
if not raw:
return routes
try:
declared = json.loads(raw)
except json.JSONDecodeError as exc:
raise ValueError(f"LLM_ROUTES is not valid JSON: {exc}") from exc
if not isinstance(declared, dict):
raise ValueError("LLM_ROUTES must be a JSON object mapping task name to endpoint list")
for task_name, names in declared.items():
routes[task_name] = list(names)
return routes
def build_backend(endpoint: EndpointConfig) -> LLMBackend:
"""Construct the :class:`~llm.base.LLMBackend` for one named endpoint.
Examples
--------
>>> backend = build_backend(EndpointConfig(name="d", base_url="http://x/v1", model="m"))
>>> backend.name
'openai:m'
"""
return OpenAIBackend(
model=endpoint.model,
api_key=endpoint.api_key,
base_url=endpoint.base_url,
trusted=endpoint.trusted,
)