Spaces:
Paused
Paused
File size: 4,221 Bytes
ec17af3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 | from __future__ import annotations
from typing import Callable
import requests
from app.config import settings
RECOMMENDED_LOCAL_MODELS = (
"nvidia/parakeet-tdt-0.6b-v3",
"nvidia/parakeet-ctc-1.1b",
"mistralai/Voxtral-Mini-4B-Realtime-2602",
"Qwen/Qwen3-ASR-1.7B",
)
ENGINE_LABELS = {
"ollama": "Ollama",
"lmstudio": "LM Studio",
"llamacpp": "llama.cpp",
}
def _default_engine_url(engine_id: str) -> str:
if engine_id == "ollama":
return settings.ollama_base_url
if engine_id == "lmstudio":
return settings.lmstudio_base_url
if engine_id == "llamacpp":
return settings.llama_cpp_base_url
raise ValueError(f"Unsupported local inference engine '{engine_id}'")
def _join_url(base_url: str, suffix: str) -> str:
return f"{base_url.rstrip('/')}/{suffix.lstrip('/')}"
def _normalize_engine_url(engine_id: str, base_url: str | None = None) -> str:
selected = (base_url or _default_engine_url(engine_id)).strip()
return selected.rstrip("/")
def _request_json(url: str) -> dict:
response = requests.get(url, timeout=settings.local_engine_timeout_seconds)
response.raise_for_status()
return response.json()
def _request_ok(url: str) -> bool:
response = requests.get(url, timeout=settings.local_engine_timeout_seconds)
response.raise_for_status()
return True
def _check_with_candidates(
engine_id: str,
base_url: str,
candidates: list[tuple[str, Callable[[dict], list[str]]]],
probe: Callable[[str], dict],
) -> dict:
last_error = "Connection failed"
for suffix, parser in candidates:
url = _join_url(base_url, suffix)
try:
payload = probe(url)
return {
"id": engine_id,
"label": ENGINE_LABELS[engine_id],
"base_url": base_url,
"live": True,
"detail": "Connected",
"available_models": parser(payload),
}
except requests.RequestException as exc:
last_error = str(exc)
return {
"id": engine_id,
"label": ENGINE_LABELS[engine_id],
"base_url": base_url,
"live": False,
"detail": last_error,
"available_models": [],
}
def _parse_ollama_models(payload: dict) -> list[str]:
return [item["name"] for item in payload.get("models", []) if item.get("name")]
def _parse_openai_models(payload: dict) -> list[str]:
return [item["id"] for item in payload.get("data", []) if item.get("id")]
def _parse_empty(_payload: dict) -> list[str]:
return []
def check_engine(engine_id: str, base_url: str | None = None) -> dict:
normalized_url = _normalize_engine_url(engine_id, base_url)
if engine_id == "ollama":
return _check_with_candidates(
engine_id,
normalized_url,
[
("api/tags", _parse_ollama_models),
("v1/models", _parse_openai_models),
],
_request_json,
)
if engine_id == "lmstudio":
return _check_with_candidates(
engine_id,
normalized_url,
[("models", _parse_openai_models), ("v1/models", _parse_openai_models)],
_request_json,
)
if engine_id == "llamacpp":
status = _check_with_candidates(
engine_id,
normalized_url,
[("v1/models", _parse_openai_models)],
_request_json,
)
if status["live"]:
return status
try:
_request_ok(_join_url(normalized_url, "health"))
return {
"id": engine_id,
"label": ENGINE_LABELS[engine_id],
"base_url": normalized_url,
"live": True,
"detail": "Connected",
"available_models": [],
}
except requests.RequestException as exc:
status["detail"] = str(exc)
return status
raise ValueError(f"Unsupported local inference engine '{engine_id}'")
def get_all_engine_statuses() -> list[dict]:
return [check_engine(engine_id) for engine_id in ("ollama", "lmstudio", "llamacpp")]
|