| """Resolve Gemma 4 12B Unified: direct vLLM, Hermes custom_providers, or Studio API.""" |
|
|
| from __future__ import annotations |
|
|
| import re |
| from dataclasses import dataclass |
| from pathlib import Path |
| from typing import Callable |
| from urllib.parse import urlparse |
|
|
| import httpx |
|
|
| from app.config import Settings |
|
|
| GEMMA_MODEL_MARKERS = ("gemma-4-12b", "gemma4-12b", "gemma-4-12B", "12b-it", "12B-it") |
|
|
|
|
| @dataclass(frozen=True) |
| class Brain: |
| kind: str |
| base_url: str |
| model: str |
| source: str |
|
|
|
|
| def _norm(url: str) -> str: |
| url = url.strip().rstrip("/") |
| if url and not url.endswith("/v1") and "/v1" not in urlparse(url).path: |
| url = url + "/v1" |
| return url |
|
|
|
|
| def looks_like_gemma(name: str) -> bool: |
| lower = name.lower() |
| return any(m.lower() in lower for m in GEMMA_MODEL_MARKERS) or "gemma" in lower |
|
|
|
|
| def parse_hermes_providers(text: str) -> list[Brain]: |
| """Pull OpenAI-compat custom_providers from a Hermes config.yaml.""" |
| found: list[Brain] = [] |
| block = text.split("custom_providers:", 1) |
| if len(block) < 2: |
| return found |
| body = block[1] |
| chunks = re.split(r"\n - name:", body) |
| for chunk in chunks[1:]: |
| name_m = re.match(r"\s*([^\n]+)", chunk) |
| url_m = re.search(r"base_url:\s*(\S+)", chunk) |
| if not url_m: |
| continue |
| name = (name_m.group(1).strip() if name_m else "") |
| base = url_m.group(1).strip().strip("\"'") |
| models = re.findall(r"\n - (\S+)", chunk) |
| if looks_like_gemma(name) or any(looks_like_gemma(m) for m in models): |
| model = next((m for m in models if looks_like_gemma(m)), models[0] if models else "google/gemma-4-12B-it") |
| found.append( |
| Brain(kind="hermes", base_url=_norm(base), model=model, source=f"hermes:{name}") |
| ) |
| return found |
|
|
|
|
| def brains_from_hermes_file(path: Path) -> list[Brain]: |
| if not path.is_file(): |
| return [] |
| try: |
| return parse_hermes_providers(path.read_text(encoding="utf-8")) |
| except OSError: |
| return [] |
|
|
|
|
| def probe_models(base_url: str, *, timeout_s: float = 1.5, client: httpx.Client | None = None) -> list[str]: |
| url = _norm(base_url) |
| own = client is None |
| http = client or httpx.Client(timeout=timeout_s) |
| try: |
| response = http.get(f"{url}/models") |
| if response.status_code >= 500: |
| return [] |
| payload = response.json() |
| rows = payload.get("data", payload if isinstance(payload, list) else []) |
| ids: list[str] = [] |
| for row in rows: |
| if isinstance(row, dict) and row.get("id"): |
| ids.append(str(row["id"])) |
| elif isinstance(row, str): |
| ids.append(row) |
| return ids |
| except (httpx.HTTPError, ValueError, TypeError): |
| return [] |
| finally: |
| if own: |
| http.close() |
|
|
|
|
| def probe_studio(studio_url: str, *, timeout_s: float = 1.5, client: httpx.Client | None = None) -> bool: |
| url = studio_url.rstrip("/") |
| own = client is None |
| http = client or httpx.Client(timeout=timeout_s) |
| try: |
| try: |
| response = http.get(f"{url}/api/health") |
| if response.status_code < 500: |
| return True |
| except httpx.HTTPError: |
| pass |
| try: |
| response = http.get(f"{url}/phone") |
| return response.status_code < 500 |
| except httpx.HTTPError: |
| return False |
| finally: |
| if own: |
| http.close() |
|
|
|
|
| ProbeFn = Callable[[str], list[str]] |
| StudioProbeFn = Callable[[str], bool] |
|
|
|
|
| def candidate_brains(settings: Settings) -> list[Brain]: |
| out: list[Brain] = [] |
| seen: set[str] = set() |
|
|
| def add(brain: Brain) -> None: |
| key = f"{brain.kind}|{brain.base_url}|{brain.model}" |
| if key not in seen: |
| seen.add(key) |
| out.append(brain) |
|
|
| model = settings.llm_model |
| add(Brain("direct", _norm(settings.llm_base_url), model, "RECEIPT_LLM_BASE_URL")) |
| if settings.hermes_base_url: |
| add(Brain("hermes", _norm(settings.hermes_base_url), settings.hermes_model or model, "RECEIPT_HERMES_BASE_URL")) |
| for brain in brains_from_hermes_file(settings.hermes_config_path): |
| add(brain) |
| add(Brain("direct", "http://127.0.0.1:8080/v1", model, "localhost:8080")) |
| host = (settings.gpu_host or "").strip() |
| if host: |
| add(Brain("direct", _norm(f"http://{host}:8080/v1"), model, f"RECEIPT_GPU_HOST:{host}")) |
| return out |
|
|
|
|
| def resolve_brain( |
| settings: Settings, |
| *, |
| probe: ProbeFn | None = None, |
| studio_probe: StudioProbeFn | None = None, |
| ) -> Brain: |
| """Pick Gemma 4 12B: studio (Lamp) vs direct vLLM vs Hermes-discovered provider.""" |
| route = settings.llm_route.lower().strip() |
| model = settings.llm_model |
|
|
| if route == "studio": |
| url = (settings.studio_url or "").rstrip("/") |
| if not url: |
| raise RuntimeError("RECEIPT_LLM_ROUTE=studio requires RECEIPT_STUDIO_URL") |
| return Brain("studio", url, model, "RECEIPT_STUDIO_URL") |
|
|
| if route == "direct": |
| return Brain("direct", _norm(settings.llm_base_url), model, "RECEIPT_LLM_BASE_URL") |
|
|
| if route == "hermes": |
| hermes = [ |
| b |
| for b in candidate_brains(settings) |
| if b.kind == "hermes" or b.source.startswith("hermes") or b.source == "RECEIPT_HERMES_BASE_URL" |
| ] |
| if settings.hermes_base_url: |
| hermes.insert( |
| 0, |
| Brain("hermes", _norm(settings.hermes_base_url), settings.hermes_model or model, "RECEIPT_HERMES_BASE_URL"), |
| ) |
| if not hermes: |
| hermes = brains_from_hermes_file(settings.hermes_config_path) |
| if not hermes: |
| raise RuntimeError( |
| "no Hermes Gemma 4 12B provider — run scripts/register-hermes-gemma.py " |
| "or set RECEIPT_HERMES_BASE_URL" |
| ) |
| check = probe or probe_models |
| for brain in hermes: |
| ids = check(brain.base_url) |
| if ids is None: |
| continue |
| if not ids or any(looks_like_gemma(i) for i in ids) or brain.model in ids: |
| return brain |
| return hermes[0] |
|
|
| |
| studio = (settings.studio_url or "").rstrip("/") |
| if studio: |
| ok = (studio_probe or (lambda u: probe_studio(u)))(studio) |
| if ok: |
| return Brain("studio", studio, model, "RECEIPT_STUDIO_URL") |
|
|
| check = probe or probe_models |
| for brain in candidate_brains(settings): |
| ids = check(brain.base_url) |
| if not ids: |
| continue |
| if any(looks_like_gemma(i) for i in ids) or brain.model in ids or brain.kind == "hermes": |
| if brain.model not in ids and ids: |
| gemma_id = next((i for i in ids if looks_like_gemma(i)), ids[0]) |
| return Brain(brain.kind, brain.base_url, gemma_id, brain.source) |
| return brain |
| |
| if brain.source == "RECEIPT_LLM_BASE_URL": |
| return brain |
|
|
| |
| return Brain("direct", _norm(settings.llm_base_url), model, "RECEIPT_LLM_BASE_URL") |
|
|
