"""LLM routing config: where models live and in what order to try them.
Before this, the answer to "which model runs where" was spread across three
places that could not see each other: ``OLLAMA_HOST``/``OLLAMA_API_KEY`` in the
environment, an unused ``ollama_backends.json``, and a per-profile ``model``
list that named models without saying which host serves them. The fallback
engine then tried *every* model on *every* server, so a model that exists only
on the cloud was also attempted on the LAN box (404) and a local model was
attempted on the cloud (404) — and a single-server setup disabled every
blacklist, which is exactly how production was running.
``llm.json`` is the one place that answers it:
{
"targets": {
"cloud": {"host": "https://ollama.com", "api_key_env": "OLLAMA_API_KEY"},
"home": {"host": "http://192.168.1.129:11434"}
},
"routes": {
"default": [
{"model": "glm-5.3-flash:cloud", "target": "cloud"},
{"model": "qwen3.6:35b", "target": "home", "context_window": 49152}
]
}
}
A route is an ordered chain; the first entry is the primary one, and the
engine returns to it as soon as the background probe says it is healthy
again. Keys are never inlined: ``api_key_env`` names an environment variable,
so the file is safe to commit and the deploy branch's ``.env`` does not have
to change.
"""
from __future__ import annotations
import json
import os
from dataclasses import dataclass, field
from pathlib import Path
import structlog
log = structlog.get_logger()
_PROJECT_ROOT = Path(__file__).resolve().parents[2]
class LLMConfigError(ValueError):
"""``llm.json`` is present but cannot be used. Raised at startup."""
def _resolve_secret(env_name: str) -> str:
"""Look a key up by environment-variable name.
``os.environ`` alone is not enough: pydantic-settings reads ``.env`` into
the *settings object*, it does not export those values to the process
environment. On production the ollama key lives in ``.env`` and nowhere
else, so a plain ``os.environ.get`` would silently hand the cloud target
an empty key and every request would 401.
"""
value = os.environ.get(env_name, "")
if value:
return value
from navi.config import settings
known = {
"OLLAMA_API_KEY": "ollama_api_key",
"EMBEDDING_OLLAMA_API_KEY": "embedding_ollama_api_key",
}
attr = known.get(env_name)
return getattr(settings, attr, "") if attr else ""
@dataclass(frozen=True)
class Target:
"""One Ollama host (cloud or LAN)."""
name: str
host: str
api_key: str = ""
@dataclass(frozen=True)
class RouteEntry:
"""One "model on a target" step of a route's chain."""
model: str
target: Target
# Window this model is served with. Drives num_ctx and the compressor's
# window, which used to come from the profile even when the entry that
# actually answered had a smaller one.
context_window: int | None = None
# Extra Ollama options for this model only (e.g. presence_penalty), merged
# over the defaults. Lets a local model be tuned without editing Navi.
options: dict = field(default_factory=dict)
@dataclass(frozen=True)
class Route:
"""An ordered chain of entries: primary first, backups after."""
id: str
entries: tuple[RouteEntry, ...]
@property
def primary(self) -> RouteEntry:
return self.entries[0]
@dataclass(frozen=True)
class LLMConfig:
targets: dict[str, Target]
routes: dict[str, Route]
probe_interval_sec: float = 60.0
probe_timeout_sec: float = 5.0
source: str = "" # path the config came from, "" for the legacy fallback
def route(self, route_id: str | None) -> Route | None:
if not route_id:
return None
return self.routes.get(route_id)
def route_ids(self) -> list[str]:
return list(self.routes)
def entry_for_model(self, model: str) -> RouteEntry | None:
"""First entry anywhere that names this model.
Used by callers that only pass a model name (embeddings, AIHelper,
legacy profiles): the config still knows which host serves it.
"""
for route in self.routes.values():
for entry in route.entries:
if entry.model == model:
return entry
return None
def fallback_entry(self, model: str) -> RouteEntry:
"""Entry for a model the config has never heard of.
Keeps the old "model list on one host" behaviour working: the name is
tried on the first target. Prefer a configured route; this is the
compatibility path for a deployment without ``llm.json``.
"""
target = next(iter(self.targets.values()))
return RouteEntry(model=model, target=target)
def resolve_config_path(path: str | Path) -> Path:
"""Resolve ``llm_config_file`` relative to CWD, then to the repo root."""
p = Path(path).expanduser()
if p.is_absolute():
return p
if p.exists():
return p
return _PROJECT_ROOT / p
def load_llm_config(path: str | Path) -> LLMConfig | None:
"""Load and validate ``llm.json``; ``None`` when the file is absent.
A missing file is not an error — the backend then falls back to a single
target built from ``OLLAMA_HOST``, which is what a bare dev checkout has.
A *broken* file is an error: silently routing to the wrong host is worse
than refusing to start.
"""
if not str(path).strip():
return None # explicitly disabled (LLM_CONFIG_FILE="")
resolved = resolve_config_path(path)
try:
raw = json.loads(resolved.read_text(encoding="utf-8"))
except FileNotFoundError:
return None
except json.JSONDecodeError as e:
raise LLMConfigError(f"{resolved}: invalid JSON — {e}") from e
if not isinstance(raw, dict):
raise LLMConfigError(f"{resolved}: expected a JSON object")
targets_raw = raw.get("targets") or {}
if not isinstance(targets_raw, dict) or not targets_raw:
raise LLMConfigError(f"{resolved}: 'targets' must be a non-empty object")
targets: dict[str, Target] = {}
for name, spec in targets_raw.items():
if not isinstance(spec, dict) or not spec.get("host"):
raise LLMConfigError(f"{resolved}: target '{name}' needs a 'host'")
api_key = spec.get("api_key", "")
if not api_key and spec.get("api_key_env"):
api_key = _resolve_secret(spec["api_key_env"])
if not api_key:
log.warning(
"llm.config.api_key_missing",
target=name,
env=spec["api_key_env"],
detail="target will answer 401 until the variable is set",
)
targets[name] = Target(name=name, host=str(spec["host"]).rstrip("/"), api_key=api_key)
routes_raw = raw.get("routes") or {}
if not isinstance(routes_raw, dict) or not routes_raw:
raise LLMConfigError(f"{resolved}: 'routes' must be a non-empty object")
routes: dict[str, Route] = {}
for route_id, entries_raw in routes_raw.items():
if not isinstance(entries_raw, list) or not entries_raw:
raise LLMConfigError(f"{resolved}: route '{route_id}' must be a non-empty list")
entries: list[RouteEntry] = []
for i, spec in enumerate(entries_raw):
if not isinstance(spec, dict) or not spec.get("model"):
raise LLMConfigError(
f"{resolved}: route '{route_id}' entry #{i} needs a 'model'"
)
target_name = spec.get("target")
target = targets.get(target_name)
if target is None:
raise LLMConfigError(
f"{resolved}: route '{route_id}' entry #{i} names unknown target "
f"{target_name!r} (known: {', '.join(targets)})"
)
entries.append(
RouteEntry(
model=str(spec["model"]),
target=target,
context_window=spec.get("context_window"),
options=dict(spec.get("options") or {}),
)
)
routes[route_id] = Route(id=route_id, entries=tuple(entries))
probe = raw.get("probe") or {}
return LLMConfig(
targets=targets,
routes=routes,
probe_interval_sec=float(probe.get("interval_sec", 60)),
probe_timeout_sec=float(probe.get("timeout_sec", 5)),
source=str(resolved),
)
def config_from_host(host: str, api_key: str = "") -> LLMConfig:
"""Legacy single-target config for a deployment without ``llm.json``.
No routes: a profile's ``model`` list keeps driving model selection, one
model at a time on this host, exactly as before.
"""
target = Target(name="default", host=host.rstrip("/"), api_key=api_key)
return LLMConfig(targets={target.name: target}, routes={}, source="")
