1776 lines
77 KiB
Python
1776 lines
77 KiB
Python
"""Native Hermes dashboard API for managing a local Ollama instance."""
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import binascii
|
||
import io
|
||
import importlib.util
|
||
import ipaddress
|
||
import json
|
||
import mimetypes
|
||
import os
|
||
import re
|
||
import socket
|
||
import sqlite3
|
||
import subprocess
|
||
import threading
|
||
import time
|
||
import uuid
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from datetime import datetime, timedelta, timezone
|
||
from html import unescape
|
||
from html.parser import HTMLParser
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from urllib.error import HTTPError, URLError
|
||
from urllib.parse import unquote, urlencode, urlparse
|
||
from urllib.request import HTTPRedirectHandler, Request, build_opener, urlopen
|
||
from zoneinfo import ZoneInfo
|
||
|
||
from fastapi import APIRouter, HTTPException
|
||
from pydantic import BaseModel, Field
|
||
from hermes_constants import get_hermes_home
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
def _running_in_container() -> bool:
|
||
return Path("/.dockerenv").exists() or Path("/run/.containerenv").exists()
|
||
|
||
|
||
def _ollama_base_url() -> str:
|
||
configured = os.environ.get("OLLAMA_HOST", "").strip()
|
||
if configured:
|
||
return configured.rstrip("/")
|
||
return "http://ollama:11434" if _running_in_container() else "http://localhost:11434"
|
||
|
||
|
||
def _discover_ollama_endpoint() -> None:
|
||
global LOCAL_OLLAMA
|
||
if os.environ.get("OLLAMA_HOST", "").strip() or not _running_in_container():
|
||
return
|
||
candidates = ("http://ollama:11434", "http://host.docker.internal:11434")
|
||
for candidate in candidates:
|
||
try:
|
||
_json_request(candidate + "/api/version", timeout=2)
|
||
except (HTTPError, URLError, OSError, ValueError):
|
||
continue
|
||
LOCAL_OLLAMA = candidate
|
||
return
|
||
|
||
|
||
LOCAL_OLLAMA = _ollama_base_url()
|
||
REMOTE_OLLAMA = "https://ollama.com"
|
||
CATALOG_FILE = "catalog.json"
|
||
MODEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,190}$")
|
||
MELBOURNE = ZoneInfo("Australia/Melbourne")
|
||
MAX_ATTACHMENT_BYTES = 20 * 1024 * 1024
|
||
MAX_ATTACHMENT_TEXT = 80_000
|
||
MAX_URL_BYTES = 15 * 1024 * 1024
|
||
CHAT_KEEP_ALIVE = -1
|
||
|
||
_jobs: dict[str, dict[str, Any]] = {}
|
||
_jobs_lock = threading.Lock()
|
||
_model_loads: dict[str, dict[str, Any]] = {}
|
||
_model_loads_lock = threading.Lock()
|
||
_chat_requests: dict[str, dict[str, Any]] = {}
|
||
_chat_requests_lock = threading.Lock()
|
||
_catalog_lock = threading.Lock()
|
||
_chat_db_init_lock = threading.Lock()
|
||
_chat_db_ready = False
|
||
|
||
|
||
def _chat_db_path() -> Path:
|
||
return _home() / "chat.sqlite3"
|
||
|
||
|
||
def _chat_db() -> sqlite3.Connection:
|
||
global _chat_db_ready
|
||
path = _chat_db_path()
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
with _chat_db_init_lock:
|
||
connection = sqlite3.connect(path, timeout=30)
|
||
connection.row_factory = sqlite3.Row
|
||
connection.execute("PRAGMA journal_mode=WAL")
|
||
connection.execute("PRAGMA foreign_keys=ON")
|
||
if not _chat_db_ready:
|
||
connection.executescript(
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS conversations (
|
||
id TEXT PRIMARY KEY,
|
||
title TEXT NOT NULL DEFAULT 'New conversation',
|
||
model TEXT NOT NULL DEFAULT '',
|
||
models_json TEXT NOT NULL DEFAULT '[]',
|
||
created_at REAL NOT NULL,
|
||
updated_at REAL NOT NULL
|
||
);
|
||
CREATE TABLE IF NOT EXISTS messages (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
|
||
request_id TEXT NOT NULL,
|
||
role TEXT NOT NULL,
|
||
content TEXT NOT NULL DEFAULT '',
|
||
model TEXT NOT NULL DEFAULT '',
|
||
attachments_json TEXT NOT NULL DEFAULT '[]',
|
||
created_at REAL NOT NULL,
|
||
UNIQUE(request_id, role, model)
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_messages_conversation ON messages(conversation_id, id);
|
||
CREATE TABLE IF NOT EXISTS chat_metrics (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
conversation_id TEXT NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
|
||
request_id TEXT NOT NULL,
|
||
model TEXT NOT NULL,
|
||
status TEXT NOT NULL,
|
||
started_at REAL,
|
||
first_token_at REAL,
|
||
finished_at REAL,
|
||
prompt_eval_count INTEGER,
|
||
eval_count INTEGER,
|
||
total_duration_ns INTEGER,
|
||
load_duration_ns INTEGER,
|
||
prompt_eval_duration_ns INTEGER,
|
||
eval_duration_ns INTEGER,
|
||
error TEXT NOT NULL DEFAULT '',
|
||
created_at REAL NOT NULL,
|
||
UNIQUE(request_id, model)
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_metrics_conversation ON chat_metrics(conversation_id, id);
|
||
"""
|
||
)
|
||
connection.commit()
|
||
try:
|
||
path.chmod(0o600)
|
||
except OSError:
|
||
pass
|
||
_chat_db_ready = True
|
||
return connection
|
||
|
||
|
||
def _conversation_id(value: str | None = None) -> str:
|
||
value = str(value or uuid.uuid4().hex).strip()
|
||
if not re.fullmatch(r"[A-Za-z0-9._-]{1,80}", value):
|
||
raise HTTPException(400, "Invalid conversation id")
|
||
return value
|
||
|
||
|
||
def _ensure_conversation(conversation_id: str, model: str, models: list[str], title: str = "") -> None:
|
||
now = time.time()
|
||
title = re.sub(r"\\s+", " ", title.strip())[:100] or "New conversation"
|
||
db = _chat_db()
|
||
try:
|
||
db.execute(
|
||
"INSERT INTO conversations(id,title,model,models_json,created_at,updated_at) VALUES(?,?,?,?,?,?) "
|
||
"ON CONFLICT(id) DO UPDATE SET model=excluded.model, models_json=excluded.models_json, updated_at=excluded.updated_at",
|
||
(conversation_id, title, model, json.dumps(models[:12]), now, now),
|
||
)
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def _persist_message(conversation_id: str, request_id: str, role: str, content: str, model: str, attachments: list[dict[str, Any]] | None = None) -> None:
|
||
db = _chat_db()
|
||
try:
|
||
db.execute(
|
||
"INSERT OR IGNORE INTO messages(conversation_id,request_id,role,content,model,attachments_json,created_at) VALUES(?,?,?,?,?,?,?)",
|
||
(conversation_id, request_id, role, content, model, json.dumps(attachments or [])[:10000], time.time()),
|
||
)
|
||
db.execute("UPDATE conversations SET updated_at=? WHERE id=?", (time.time(), conversation_id))
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def _metric_values(state: dict[str, Any], status: str | None = None, error: str = "") -> dict[str, Any]:
|
||
started = float(state.get("started_at") or time.time())
|
||
finished = float(state.get("finished_at") or time.time())
|
||
first = state.get("first_token_at")
|
||
prompt_count = state.get("prompt_eval_count")
|
||
eval_count = state.get("eval_count")
|
||
prompt_duration = state.get("prompt_eval_duration")
|
||
eval_duration = state.get("eval_duration")
|
||
total_duration = state.get("total_duration")
|
||
load_duration = state.get("load_duration")
|
||
return {
|
||
"status": status or str(state.get("state") or "unknown"),
|
||
"started_at": started,
|
||
"first_token_at": first,
|
||
"finished_at": finished,
|
||
"time_to_first_token_ms": round((float(first) - started) * 1000, 2) if first else None,
|
||
"total_latency_ms": round((finished - started) * 1000, 2),
|
||
"prompt_eval_count": prompt_count,
|
||
"eval_count": eval_count,
|
||
"total_tokens": (int(prompt_count) + int(eval_count)) if prompt_count is not None and eval_count is not None else None,
|
||
"prompt_eval_duration_ns": prompt_duration,
|
||
"eval_duration_ns": eval_duration,
|
||
"total_duration_ns": total_duration,
|
||
"load_duration_ns": load_duration,
|
||
"prompt_tokens_per_second": round(int(prompt_count) / (int(prompt_duration) / 1e9), 2) if prompt_count and prompt_duration else None,
|
||
"eval_tokens_per_second": round(int(eval_count) / (int(eval_duration) / 1e9), 2) if eval_count and eval_duration else None,
|
||
"error": error,
|
||
}
|
||
|
||
|
||
def _persist_metric(conversation_id: str, request_id: str, model: str, state: dict[str, Any], status: str | None = None, error: str = "") -> dict[str, Any]:
|
||
values = _metric_values(state, status=status, error=error)
|
||
db = _chat_db()
|
||
try:
|
||
db.execute(
|
||
"INSERT INTO chat_metrics(conversation_id,request_id,model,status,started_at,first_token_at,finished_at,prompt_eval_count,eval_count,total_duration_ns,load_duration_ns,prompt_eval_duration_ns,eval_duration_ns,error,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) "
|
||
"ON CONFLICT(request_id,model) DO UPDATE SET status=excluded.status, started_at=excluded.started_at, first_token_at=excluded.first_token_at, finished_at=excluded.finished_at, prompt_eval_count=excluded.prompt_eval_count, eval_count=excluded.eval_count, total_duration_ns=excluded.total_duration_ns, load_duration_ns=excluded.load_duration_ns, prompt_eval_duration_ns=excluded.prompt_eval_duration_ns, eval_duration_ns=excluded.eval_duration_ns, error=excluded.error",
|
||
(conversation_id, request_id, model, values["status"], values["started_at"], values["first_token_at"], values["finished_at"], values["prompt_eval_count"], values["eval_count"], values["total_duration_ns"], values["load_duration_ns"], values["prompt_eval_duration_ns"], values["eval_duration_ns"], values["error"], time.time()),
|
||
)
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
return values
|
||
|
||
|
||
def _row_metric(row: sqlite3.Row) -> dict[str, Any]:
|
||
value = dict(row)
|
||
started = value.get("started_at")
|
||
first = value.get("first_token_at")
|
||
finished = value.get("finished_at")
|
||
value["time_to_first_token_ms"] = round((first - started) * 1000, 2) if first and started else None
|
||
value["total_latency_ms"] = round((finished - started) * 1000, 2) if finished and started else None
|
||
prompt = value.get("prompt_eval_count")
|
||
output = value.get("eval_count")
|
||
value["total_tokens"] = (prompt + output) if prompt is not None and output is not None else None
|
||
value["prompt_tokens_per_second"] = round(prompt / (value["prompt_eval_duration_ns"] / 1e9), 2) if prompt and value.get("prompt_eval_duration_ns") else None
|
||
value["eval_tokens_per_second"] = round(output / (value["eval_duration_ns"] / 1e9), 2) if output and value.get("eval_duration_ns") else None
|
||
return value
|
||
|
||
CAPABILITY_INFO = {
|
||
"completion": "Text generation and chat completion.",
|
||
"tools": "Tool/function calling for agent workflows.",
|
||
"thinking": "Explicit reasoning/thinking output support.",
|
||
"vision": "Image input and visual understanding.",
|
||
"audio": "Audio input or audio-aware inference.",
|
||
"video": "Video input or video-aware inference.",
|
||
}
|
||
|
||
FAMILY_STRENGTHS = {
|
||
"qwen": ["general reasoning", "coding", "multilingual work", "tool use"],
|
||
"qwen35": ["general reasoning", "coding", "long-context work", "tool use"],
|
||
"gemma": ["general assistance", "reasoning", "tool use", "efficient local inference"],
|
||
"nemotron": ["reasoning", "agent workflows", "long-context work", "technical tasks"],
|
||
"deepseek": ["coding", "mathematical reasoning", "technical analysis"],
|
||
"gpt-oss": ["general reasoning", "coding", "agent workflows"],
|
||
"mistral": ["general assistance", "multilingual work", "coding"],
|
||
"kimi": ["long-context work", "reasoning", "coding"],
|
||
"minimax": ["agent workflows", "reasoning", "long-context work"],
|
||
"glm": ["reasoning", "coding", "multilingual work"],
|
||
"lfm2": ["fast local assistants", "low-resource inference", "general chat"],
|
||
}
|
||
|
||
|
||
_PLUGIN_HOME_ROOT = Path(os.environ.get("HERMES_HOME", "").strip()).expanduser().resolve() if os.environ.get("HERMES_HOME", "").strip() else Path(get_hermes_home()).resolve()
|
||
|
||
|
||
def _home() -> Path:
|
||
path = _PLUGIN_HOME_ROOT / "ollama-manager"
|
||
path.mkdir(parents=True, exist_ok=True)
|
||
return path
|
||
|
||
|
||
CONNECTIONS_FILE = "connections.json"
|
||
|
||
|
||
def _valid_ollama_url(value: str) -> str:
|
||
value = str(value or "").strip().rstrip("/")
|
||
parsed = urlparse(value)
|
||
if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password:
|
||
raise HTTPException(400, "Ollama URL must be an http(s) URL without credentials")
|
||
if parsed.path not in {"", "/"} or parsed.query or parsed.fragment:
|
||
raise HTTPException(400, "Ollama URL must be a base URL without a path, query, or fragment")
|
||
try:
|
||
if parsed.port is not None and not 1 <= parsed.port <= 65535:
|
||
raise ValueError
|
||
except ValueError as exc:
|
||
raise HTTPException(400, "Ollama URL has an invalid port") from exc
|
||
return value
|
||
|
||
|
||
def _read_connections() -> dict[str, str]:
|
||
try:
|
||
value = json.loads((_home() / CONNECTIONS_FILE).read_text(encoding="utf-8"))
|
||
except (OSError, ValueError):
|
||
value = {}
|
||
return {key: str(value.get(key) or "").strip().rstrip("/") for key in ("active_url", "local_url", "remote_url")}
|
||
|
||
|
||
def _write_connections(value: dict[str, str]) -> None:
|
||
path = _home() / CONNECTIONS_FILE
|
||
temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
|
||
temporary.write_text(json.dumps(value, indent=2) + "\n", encoding="utf-8")
|
||
try:
|
||
temporary.chmod(0o600)
|
||
os.replace(temporary, path)
|
||
finally:
|
||
try:
|
||
temporary.unlink()
|
||
except FileNotFoundError:
|
||
pass
|
||
|
||
|
||
def _sync_native_ollama_providers(connections: dict[str, str] | None = None) -> None:
|
||
"""Keep Hermes' native model picker aligned with this plugin's endpoints."""
|
||
try:
|
||
helper_path = Path(__file__).resolve().parent.parent / "ollama_provider.py"
|
||
spec = importlib.util.spec_from_file_location("_ollama_manager_provider_sync", helper_path)
|
||
if spec is None or spec.loader is None:
|
||
return
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
module.sync_hermes_ollama_providers(connections)
|
||
except Exception:
|
||
# Provider synchronization must never make the Ollama dashboard route fail.
|
||
return
|
||
|
||
|
||
|
||
def _apply_saved_connection() -> None:
|
||
global LOCAL_OLLAMA
|
||
saved = _read_connections().get("active_url")
|
||
if saved:
|
||
LOCAL_OLLAMA = saved
|
||
return
|
||
configured = os.environ.get("OLLAMA_HOST", "").strip()
|
||
LOCAL_OLLAMA = configured.rstrip("/") if configured else _ollama_base_url()
|
||
|
||
|
||
def _target_endpoint(target: str = "local") -> str:
|
||
target = str(target or "local").strip().lower()
|
||
if target not in {"local", "remote"}:
|
||
raise HTTPException(400, "Target must be local or remote")
|
||
saved = _read_connections()
|
||
if target == "remote":
|
||
endpoint = saved.get("remote_url")
|
||
if not endpoint:
|
||
raise HTTPException(400, "No remote Ollama URL is configured")
|
||
return _valid_ollama_url(endpoint)
|
||
_apply_saved_connection()
|
||
return _valid_ollama_url(saved.get("local_url") or LOCAL_OLLAMA)
|
||
|
||
|
||
def _probe_endpoint(url: str, timeout: int = 5) -> dict[str, Any]:
|
||
url = _valid_ollama_url(url)
|
||
version_payload = _json_request(url + "/api/version", timeout=timeout)
|
||
tags_payload = _json_request(url + "/api/tags", timeout=timeout)
|
||
models = tags_payload.get("models", [])
|
||
model_names = [str(item.get("name") or item.get("model")) for item in models if isinstance(item, dict) and (item.get("name") or item.get("model"))]
|
||
return {"available": True, "url": url, "version": str(version_payload.get("version") or "unknown"), "models": len(model_names), "model_names": model_names[:100]}
|
||
|
||
|
||
def _connection_snapshot() -> list[dict[str, Any]]:
|
||
saved = _read_connections()
|
||
configured_local = saved.get("local_url") or os.environ.get("OLLAMA_HOST", "").strip()
|
||
candidates: list[tuple[str, str, str]] = []
|
||
if configured_local:
|
||
candidates.append(("local", "Configured local endpoint", configured_local))
|
||
elif _running_in_container():
|
||
candidates.extend((("local", "Docker Ollama service", "http://ollama:11434"), ("local", "Docker host Ollama", "http://host.docker.internal:11434")))
|
||
else:
|
||
candidates.append(("local", "Physical host Ollama", "http://localhost:11434"))
|
||
if saved.get("remote_url"):
|
||
candidates.append(("remote", "Configured remote endpoint", saved["remote_url"]))
|
||
results: list[dict[str, Any]] = []
|
||
seen: set[tuple[str, str]] = set()
|
||
for kind, label, url in candidates:
|
||
try:
|
||
normalized = _valid_ollama_url(url)
|
||
except HTTPException:
|
||
continue
|
||
key = (kind, normalized)
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
try:
|
||
result = _probe_endpoint(normalized, timeout=3)
|
||
result.update({"kind": kind, "label": label, "saved": saved.get(f"{kind}_url") == normalized})
|
||
except Exception as exc:
|
||
result = {"available": False, "kind": kind, "label": label, "url": normalized, "saved": saved.get(f"{kind}_url") == normalized, "error": str(exc)[:240]}
|
||
results.append(result)
|
||
local_urls = {row.get("url") for row in results if row.get("kind") == "local" and row.get("available")}
|
||
for row in results:
|
||
if row.get("kind") == "remote" and row.get("url") in local_urls:
|
||
row["same_endpoint"] = True
|
||
return results
|
||
|
||
|
||
def _json_request(url: str, method: str = "GET", payload: Any = None, timeout: int = 30) -> dict[str, Any]:
|
||
data = None if payload is None else json.dumps(payload).encode("utf-8")
|
||
headers = {"Accept": "application/json"}
|
||
if data is not None:
|
||
headers["Content-Type"] = "application/json"
|
||
request = Request(url, data=data, headers=headers, method=method)
|
||
with urlopen(request, timeout=timeout) as response:
|
||
raw = response.read()
|
||
value = json.loads(raw.decode("utf-8")) if raw else {}
|
||
return value if isinstance(value, dict) else {}
|
||
|
||
|
||
def _valid_name(name: str) -> str:
|
||
name = str(name or "").strip()
|
||
if not MODEL_RE.fullmatch(name):
|
||
raise HTTPException(400, "Invalid Ollama model name")
|
||
return name
|
||
|
||
|
||
def _local_tags() -> list[dict[str, Any]]:
|
||
_apply_saved_connection()
|
||
_discover_ollama_endpoint()
|
||
try:
|
||
payload = _json_request(LOCAL_OLLAMA + "/api/tags", timeout=15)
|
||
models = payload.get("models", [])
|
||
return [item for item in models if isinstance(item, dict)]
|
||
except (HTTPError, URLError, OSError, ValueError):
|
||
return []
|
||
|
||
|
||
def _local_ps() -> list[dict[str, Any]]:
|
||
try:
|
||
payload = _json_request(LOCAL_OLLAMA + "/api/ps", timeout=10)
|
||
models = payload.get("models", [])
|
||
return [item for item in models if isinstance(item, dict)]
|
||
except (HTTPError, URLError, OSError, ValueError):
|
||
return []
|
||
|
||
|
||
def _local_ps() -> list[dict[str, Any]]:
|
||
try:
|
||
payload = _json_request(LOCAL_OLLAMA + "/api/ps", timeout=10)
|
||
models = payload.get("models", [])
|
||
return [item for item in models if isinstance(item, dict)]
|
||
except (HTTPError, URLError, OSError, ValueError):
|
||
return []
|
||
|
||
|
||
def _model_load_update(name: str, **values: Any) -> None:
|
||
with _model_loads_lock:
|
||
current = _model_loads.setdefault(name, {"name": name, "state": "queued", "stage": "Queued for Ollama", "started_at": time.time()})
|
||
current.update(values, updated_at=time.time())
|
||
|
||
|
||
def _model_load_snapshot() -> list[dict[str, Any]]:
|
||
now = time.time()
|
||
with _model_loads_lock:
|
||
rows = [dict(value) for value in _model_loads.values()]
|
||
for row in rows:
|
||
row["elapsed"] = round(max(0.0, now - float(row.get("started_at") or now)), 1)
|
||
return sorted(rows, key=lambda row: row.get("started_at") or 0)
|
||
|
||
|
||
def _read_meminfo() -> dict[str, int]:
|
||
values: dict[str, int] = {}
|
||
try:
|
||
for line in Path("/proc/meminfo").read_text(encoding="utf-8").splitlines():
|
||
key, _, raw = line.partition(":")
|
||
match = re.search(r"([0-9]+)", raw)
|
||
if match:
|
||
values[key] = int(match.group(1)) * 1024
|
||
except OSError:
|
||
return {}
|
||
return values
|
||
|
||
|
||
def _host_ram_gib() -> float | None:
|
||
"""Return installed system RAM as GiB, detected from the running host."""
|
||
total = _read_meminfo().get("MemTotal", 0)
|
||
return round(total / (1024 ** 3), 1) if total else None
|
||
|
||
|
||
def _gpu_snapshot() -> dict[str, Any]:
|
||
"""Return NVIDIA GPU telemetry when available, without requiring CUDA."""
|
||
query = "name,memory.total,memory.used,memory.free"
|
||
try:
|
||
result = subprocess.run(
|
||
["nvidia-smi", f"--query-gpu={query}", "--format=csv,noheader,nounits"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=4,
|
||
check=False,
|
||
)
|
||
except (OSError, subprocess.SubprocessError):
|
||
result = None
|
||
if result and result.returncode == 0:
|
||
gpus = []
|
||
for line in result.stdout.splitlines():
|
||
parts = [part.strip() for part in line.split(",")]
|
||
if len(parts) != 4:
|
||
continue
|
||
try:
|
||
total, used, free = (int(float(value)) * 1024 * 1024 for value in parts[1:])
|
||
except ValueError:
|
||
continue
|
||
gpus.append({"name": parts[0], "total_bytes": total, "used_bytes": used, "free_bytes": free})
|
||
if gpus:
|
||
return {"detected": True, "telemetry_available": True, "gpus": gpus}
|
||
nvidia_present = False
|
||
for vendor in Path("/sys/class/drm").glob("card*/device/vendor"):
|
||
try:
|
||
nvidia_present = nvidia_present or vendor.read_text().strip().lower() == "0x10de"
|
||
except OSError:
|
||
continue
|
||
return {"detected": nvidia_present, "telemetry_available": False, "gpus": []}
|
||
|
||
|
||
def _runtime_snapshot() -> dict[str, Any]:
|
||
mem = _read_meminfo()
|
||
total = mem.get("MemTotal", 0)
|
||
available = mem.get("MemAvailable", mem.get("MemFree", 0))
|
||
swap_total = mem.get("SwapTotal", 0)
|
||
swap_free = mem.get("SwapFree", 0)
|
||
ps_rows = _local_ps()
|
||
tag_rows = {str(row.get("name") or row.get("model")): row for row in _local_tags()}
|
||
load_rows = _model_load_snapshot()
|
||
active_loads = [row for row in load_rows if row.get("active")]
|
||
model_memory = []
|
||
for row in ps_rows:
|
||
name = str(row.get("name") or row.get("model") or "")
|
||
total_bytes = int(row.get("size") or 0)
|
||
gpu_bytes = int(row.get("size_vram") or 0)
|
||
capability_view = _model_view(tag_rows.get(name, {"name": name}), row)
|
||
model_memory.append({
|
||
"name": name,
|
||
"total_bytes": total_bytes,
|
||
"gpu_bytes": gpu_bytes,
|
||
"ram_bytes": max(0, total_bytes - gpu_bytes),
|
||
"gpu_offload_percent": round(gpu_bytes * 100 / total_bytes, 1) if total_bytes else 0,
|
||
"capabilities": capability_view["capabilities"],
|
||
"capability_breakdown": capability_view["capability_breakdown"],
|
||
"input_modalities": capability_view["input_modalities"],
|
||
"family": capability_view["family"],
|
||
"context_length": capability_view["context_length"],
|
||
"parameter_size": capability_view["parameter_size"],
|
||
"quantization": capability_view["quantization"],
|
||
"permanent": True,
|
||
})
|
||
gpu = _gpu_snapshot()
|
||
ollama_model_bytes = sum(int(row.get("total_bytes") or 0) for row in model_memory)
|
||
ollama_model_vram_bytes = sum(int(row.get("gpu_bytes") or 0) for row in model_memory)
|
||
model_loading = []
|
||
for row in active_loads:
|
||
tag = tag_rows.get(str(row.get("name")), {})
|
||
model_loading.append({
|
||
**row,
|
||
"estimated_bytes": int(tag.get("size") or 0),
|
||
"estimated_vram_bytes": 0,
|
||
})
|
||
ollama_target_model_bytes = ollama_model_bytes + sum(int(row.get("estimated_bytes") or 0) for row in model_loading)
|
||
return {
|
||
"captured_at": time.time(),
|
||
"memory_total_bytes": total,
|
||
"memory_used_bytes": max(0, total - available),
|
||
"memory_available_bytes": available,
|
||
"swap_total_bytes": swap_total,
|
||
"swap_used_bytes": max(0, swap_total - swap_free),
|
||
"model_memory": model_memory,
|
||
"model_loading": model_loading,
|
||
"model_loads": load_rows[-12:],
|
||
"ollama_model_bytes": ollama_model_bytes,
|
||
"ollama_model_vram_bytes": ollama_model_vram_bytes,
|
||
"ollama_target_model_bytes": ollama_target_model_bytes,
|
||
"gpu": gpu,
|
||
}
|
||
|
||
|
||
def _validate_public_url(value: str) -> str:
|
||
parsed = urlparse(value.strip())
|
||
if parsed.scheme not in {"http", "https"} or not parsed.hostname:
|
||
raise HTTPException(400, "URL attachments must use http:// or https://")
|
||
host = parsed.hostname
|
||
try:
|
||
addresses = {info[4][0] for info in socket.getaddrinfo(host, parsed.port or 443, type=socket.SOCK_STREAM)}
|
||
except (OSError, ValueError) as exc:
|
||
raise HTTPException(400, f"Could not resolve URL host: {exc}") from exc
|
||
for address in addresses:
|
||
ip = ipaddress.ip_address(address)
|
||
if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_multicast or ip.is_unspecified:
|
||
raise HTTPException(400, "Private or local URL targets are not allowed")
|
||
return value.strip()
|
||
|
||
|
||
class _SafeRedirectHandler(HTTPRedirectHandler):
|
||
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
||
_validate_public_url(newurl)
|
||
return super().redirect_request(req, fp, code, msg, headers, newurl)
|
||
|
||
|
||
_SAFE_URL_OPENER = build_opener(_SafeRedirectHandler)
|
||
|
||
|
||
def _fetch_attachment_url(value: str) -> tuple[bytes, str, str]:
|
||
value = _validate_public_url(value)
|
||
request = Request(value, headers={"Accept": "text/html, text/plain, application/pdf, image/*", "User-Agent": "Hermes-Ollama-Manager/1.3"})
|
||
try:
|
||
with _SAFE_URL_OPENER.open(request, timeout=30) as response:
|
||
final_url = _validate_public_url(response.geturl())
|
||
content_type = response.headers.get_content_type() if response.headers else "application/octet-stream"
|
||
data = response.read(MAX_URL_BYTES + 1)
|
||
except (HTTPError, URLError, OSError, ValueError) as exc:
|
||
raise HTTPException(400, f"Could not fetch URL: {exc}") from exc
|
||
if len(data) > MAX_URL_BYTES:
|
||
raise HTTPException(413, "URL attachment is larger than 15 MiB")
|
||
return data, content_type, final_url
|
||
|
||
|
||
def _extract_pdf_text(data: bytes, label: str) -> str:
|
||
try:
|
||
from pypdf import PdfReader
|
||
except ImportError as exc:
|
||
raise HTTPException(500, "PDF support requires the pypdf package") from exc
|
||
try:
|
||
reader = PdfReader(io.BytesIO(data))
|
||
text = "\n\n".join(page.extract_text() or "" for page in reader.pages)
|
||
except Exception as exc:
|
||
raise HTTPException(400, f"Could not extract text from PDF {label}: {exc}") from exc
|
||
return text[:MAX_ATTACHMENT_TEXT]
|
||
|
||
|
||
class _PageTextParser(HTMLParser):
|
||
def __init__(self) -> None:
|
||
super().__init__()
|
||
self.parts: list[str] = []
|
||
self._skip = 0
|
||
|
||
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
||
if tag.lower() in {"script", "style", "noscript", "svg"}:
|
||
self._skip += 1
|
||
|
||
def handle_endtag(self, tag: str) -> None:
|
||
if tag.lower() in {"script", "style", "noscript", "svg"} and self._skip:
|
||
self._skip -= 1
|
||
|
||
def handle_data(self, data: str) -> None:
|
||
if not self._skip and data.strip():
|
||
self.parts.append(data.strip())
|
||
|
||
|
||
def _extract_page_text(data: bytes, content_type: str) -> str:
|
||
decoded = data.decode("utf-8", errors="replace")
|
||
if "html" in content_type.lower() or re.search(r"<html|<body|<article", decoded, re.I):
|
||
parser = _PageTextParser()
|
||
parser.feed(decoded)
|
||
decoded = "\n".join(parser.parts)
|
||
return unescape(decoded)[:MAX_ATTACHMENT_TEXT]
|
||
|
||
|
||
def _decode_data_url(data_url: str, fallback_mime: str, label: str) -> tuple[bytes, str]:
|
||
match = re.match(r"data:([^;,]+)?;base64,(.*)", data_url or "", re.S)
|
||
if not match:
|
||
raise HTTPException(400, f"Attachment {label} is not a valid base64 data URL")
|
||
mime = (match.group(1) or fallback_mime or "application/octet-stream").lower()
|
||
try:
|
||
data = base64.b64decode(match.group(2), validate=True)
|
||
except (binascii.Error, ValueError) as exc:
|
||
raise HTTPException(400, f"Attachment {label} has invalid base64 data") from exc
|
||
if len(data) > MAX_ATTACHMENT_BYTES:
|
||
raise HTTPException(413, f"Attachment {label} is larger than 20 MiB")
|
||
return data, mime
|
||
|
||
|
||
def _attachment_parts(attachment: "ChatAttachment") -> tuple[str | None, str | None]:
|
||
label = attachment.name or attachment.url or "attachment"
|
||
if attachment.url:
|
||
data, mime, _ = _fetch_attachment_url(attachment.url)
|
||
elif attachment.data_url:
|
||
data, mime = _decode_data_url(attachment.data_url, attachment.mime_type, label)
|
||
else:
|
||
raise HTTPException(400, f"Attachment {label} has no data or URL")
|
||
if mime == "application/pdf" or label.lower().endswith(".pdf"):
|
||
return f"[PDF: {label}]\n{_extract_pdf_text(data, label)}", None
|
||
if mime.startswith("image/"):
|
||
return f"[Image attached: {label}]", base64.b64encode(data).decode("ascii")
|
||
return f"[Text attachment: {label}]\n{_extract_page_text(data, mime)}", None
|
||
|
||
|
||
def _is_mlx(raw: dict[str, Any] | str) -> bool:
|
||
text = str(raw if isinstance(raw, str) else {
|
||
"name": raw.get("name") or raw.get("model"),
|
||
"format": (raw.get("details") or {}).get("format"),
|
||
"capabilities": raw.get("capabilities"),
|
||
}).lower()
|
||
return "mlx" in text or bool(re.search(r"(?:^|[-:])mlx(?:$|[-:])", text))
|
||
|
||
|
||
def _family_key(name: str) -> str:
|
||
return str(name or "").split(":", 1)[0].strip()
|
||
|
||
|
||
def _known_ram_fit(model: dict[str, Any]) -> bool:
|
||
"""Return True when known size/RAM estimates fit installed host RAM."""
|
||
size_gb = model.get("size_gb")
|
||
ram_gb = model.get("expected_ram_gb")
|
||
host_ram_gib = _host_ram_gib()
|
||
return (
|
||
isinstance(size_gb, (int, float))
|
||
and isinstance(ram_gb, (int, float))
|
||
and host_ram_gib is not None
|
||
and float(ram_gb) <= host_ram_gib
|
||
)
|
||
|
||
|
||
def _popular_fit_models(
|
||
raw_rows: list[dict[str, Any]],
|
||
family_rows: dict[str, list[dict[str, Any]]],
|
||
catalog_view,
|
||
installed_names: set[str],
|
||
loaded: dict[str, Any],
|
||
) -> list[dict[str, Any]]:
|
||
"""Select popular models that are usable within this host's RAM budget.
|
||
|
||
Oversized popular entries may be represented by the largest known smaller
|
||
family variant that fits. Unknown entries are never shown.
|
||
"""
|
||
result: list[dict[str, Any]] = []
|
||
seen: set[str] = set()
|
||
seen_footprints: set[tuple[str, float, float]] = set()
|
||
seen_families: set[str] = set()
|
||
for raw in raw_rows:
|
||
if _is_mlx(raw):
|
||
continue
|
||
original = catalog_view(raw, "popular")
|
||
if _known_ram_fit(original):
|
||
candidate = original
|
||
else:
|
||
original_size = original.get("size_gb")
|
||
if not isinstance(original_size, (int, float)):
|
||
continue
|
||
family = _family_key(original["name"])
|
||
variants: list[dict[str, Any]] = []
|
||
for variant_raw in family_rows.get(family, []):
|
||
if _is_mlx(variant_raw):
|
||
continue
|
||
variant = _model_view(variant_raw, loaded.get(str(variant_raw.get("name") or variant_raw.get("model"))), source="popular")
|
||
variant["installed"] = variant["name"] in installed_names
|
||
variant["loaded"] = variant["name"] in loaded
|
||
if _known_ram_fit(variant) and float(variant.get("size_gb", 0)) < float(original_size):
|
||
variants.append(variant)
|
||
if not variants:
|
||
continue
|
||
candidate = max(variants, key=lambda item: (float(item.get("size_gb", 0)), item["name"]))
|
||
candidate["popular_origin"] = original["name"]
|
||
candidate["popular_note"] = f"Smaller fit variant for {original['name']}"
|
||
if candidate["name"] in seen:
|
||
continue
|
||
candidate_family = _family_key(candidate["name"])
|
||
if candidate.get("popular_origin") and candidate_family in seen_families:
|
||
continue
|
||
footprint = (
|
||
candidate_family,
|
||
round(float(candidate.get("size_gb", 0)), 2),
|
||
round(float(candidate.get("expected_ram_gb", 0)), 1),
|
||
)
|
||
if footprint in seen_footprints:
|
||
continue
|
||
seen.add(candidate["name"])
|
||
seen_families.add(candidate_family)
|
||
seen_footprints.add(footprint)
|
||
result.append(candidate)
|
||
for rank, row in enumerate(result, 1):
|
||
row["popularity_rank"] = rank
|
||
return result
|
||
|
||
|
||
def _parse_number(text: Any) -> float | None:
|
||
match = re.search(r"([0-9]+(?:\.[0-9]+)?)", str(text or ""))
|
||
return float(match.group(1)) if match else None
|
||
|
||
|
||
def _ram_estimate(size_bytes: Any, parameter_size: Any, quantization: Any, context_length: Any) -> tuple[float | None, str]:
|
||
size = float(size_bytes or 0)
|
||
if size > 0:
|
||
base = size / (1024 ** 3)
|
||
# Ollama's runtime needs allocator/graph overhead beyond the GGUF blob.
|
||
estimate = base * 1.15 + 0.5
|
||
basis = "disk size × 1.15 + 0.5 GiB runtime overhead"
|
||
else:
|
||
params = _parse_number(parameter_size)
|
||
if params is None:
|
||
return None, "Unavailable: source did not publish a model size"
|
||
bits = 4.5 if "Q4" in str(quantization).upper() else 8.0
|
||
estimate = params * (bits / 8.0) + 0.8
|
||
basis = "parameter/quantization estimate; source size unavailable"
|
||
context = int(context_length or 0)
|
||
if context >= 524288:
|
||
estimate += 2.0
|
||
basis += "; includes large-context allowance"
|
||
elif context >= 131072:
|
||
estimate += 1.0
|
||
basis += "; includes long-context allowance"
|
||
return round(estimate, 1), basis
|
||
|
||
|
||
def _infer_capabilities(name: str, family: str, details: dict[str, Any], advertised: Any) -> list[str]:
|
||
caps = [str(item).lower() for item in advertised or [] if item]
|
||
text = f"{name} {family}".lower()
|
||
if not caps or "completion" not in caps:
|
||
caps.append("completion")
|
||
if any(token in text for token in ("vision", "vl", "gemma4", "muse-glimmer", "qwen3.8")) and "vision" not in caps:
|
||
caps.append("vision")
|
||
if any(token in text for token in ("audio", "omni")) and "audio" not in caps:
|
||
caps.append("audio")
|
||
if "video" in text and "video" not in caps:
|
||
caps.append("video")
|
||
if any(token in text for token in ("thinking", "reasoning", "nemotron", "deepseek", "qwen", "kimi")) and "thinking" not in caps:
|
||
caps.append("thinking")
|
||
if any(token in text for token in ("tool", "agent", "qwen", "gemma", "nemotron", "deepseek", "gpt-oss")) and "tools" not in caps:
|
||
caps.append("tools")
|
||
order = ["completion", "tools", "thinking", "vision", "audio", "video"]
|
||
return [cap for cap in order if cap in set(caps)]
|
||
|
||
|
||
def _architecture(name: str, family: str, details: dict[str, Any]) -> tuple[str, bool]:
|
||
text = f"{name} {family} {details.get('parent_model', '')}".lower()
|
||
moe = any(token in text for token in ("moe", "mixture", "_h_", "a3b"))
|
||
label = "Mixture of Experts (MoE)" if moe else "Dense / single-expert"
|
||
if family:
|
||
label += f" · {family}"
|
||
return label, moe
|
||
|
||
|
||
def _strengths(name: str, family: str, capabilities: list[str], context_length: Any) -> list[str]:
|
||
text = f"{name} {family}".lower()
|
||
result: list[str] = []
|
||
for key, values in FAMILY_STRENGTHS.items():
|
||
if key in text:
|
||
result.extend(values)
|
||
break
|
||
if "tools" in capabilities:
|
||
result.append("tool-enabled automation")
|
||
if "vision" in capabilities:
|
||
result.append("image-aware tasks")
|
||
if int(context_length or 0) >= 131072:
|
||
result.append("long documents and large codebases")
|
||
if not result:
|
||
result = ["general local inference"]
|
||
return list(dict.fromkeys(result))
|
||
|
||
|
||
def _model_view(raw: dict[str, Any], loaded: dict[str, Any] | None = None, source: str = "local") -> dict[str, Any]:
|
||
details = raw.get("details") if isinstance(raw.get("details"), dict) else {}
|
||
name = str(raw.get("name") or raw.get("model") or "")
|
||
family = str(details.get("family") or (details.get("families") or [""])[0] or "")
|
||
capabilities = _infer_capabilities(name, family, details, raw.get("capabilities"))
|
||
context_length = details.get("context_length") or raw.get("context_length")
|
||
architecture, is_moe = _architecture(name, family, details)
|
||
size_bytes = raw.get("size") or 0
|
||
ram_gb, ram_basis = _ram_estimate(size_bytes, details.get("parameter_size"), details.get("quantization_level"), context_length)
|
||
loaded = loaded or {}
|
||
return {
|
||
"name": name,
|
||
"source": source,
|
||
"downloadable": source == "catalog",
|
||
"installed": source == "local",
|
||
"loaded": bool(loaded),
|
||
"size_bytes": int(size_bytes or 0),
|
||
"size_gb": round(float(size_bytes or 0) / (1024 ** 3), 2) if size_bytes else None,
|
||
"size_label": raw.get("size_label") or (f"{round(float(size_bytes or 0) / (1024 ** 3), 2)} GiB" if size_bytes else "Unknown"),
|
||
"loaded_bytes": int(loaded.get("size") or 0),
|
||
"loaded_vram_bytes": int(loaded.get("size_vram") or 0),
|
||
"digest": raw.get("digest", ""),
|
||
"modified_at": raw.get("modified_at"),
|
||
"family": family or "unknown",
|
||
"architecture": architecture,
|
||
"is_moe": is_moe,
|
||
"parameter_size": details.get("parameter_size") or "unknown",
|
||
"quantization": details.get("quantization_level") or "unknown",
|
||
"format": details.get("format") or "unknown",
|
||
"context_length": context_length,
|
||
"input_modalities": raw.get("input_modalities") or (["Text", "Image"] if "vision" in capabilities else ["Text"]),
|
||
"embedding_length": details.get("embedding_length"),
|
||
"capabilities": capabilities,
|
||
"capability_breakdown": {cap: CAPABILITY_INFO[cap] for cap in capabilities if cap in CAPABILITY_INFO},
|
||
"strengths": _strengths(name, family, capabilities, context_length),
|
||
"expected_ram_gb": ram_gb,
|
||
"expected_ram_label": f"{ram_gb:.1f} GiB baseline" if ram_gb is not None else "Unknown",
|
||
"expected_ram_basis": ram_basis,
|
||
}
|
||
|
||
|
||
class _VariantPageParser(HTMLParser):
|
||
"""Extract the public Ollama tag rows without depending on third-party HTML packages."""
|
||
|
||
def __init__(self) -> None:
|
||
super().__init__()
|
||
self._depth = 0
|
||
self._parts: list[str] = []
|
||
self._href = ""
|
||
self.rows: list[dict[str, Any]] = []
|
||
|
||
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
||
attrs_map = dict(attrs)
|
||
classes = attrs_map.get("class") or ""
|
||
if tag == "div" and "group" in classes.split() and "px-4" in classes.split():
|
||
self._depth = 1
|
||
self._parts = []
|
||
self._href = ""
|
||
return
|
||
if self._depth:
|
||
if tag == "div":
|
||
self._depth += 1
|
||
if tag == "a" and (attrs_map.get("href") or "").startswith("/library/") and not self._href:
|
||
self._href = attrs_map["href"] or ""
|
||
|
||
def handle_data(self, data: str) -> None:
|
||
if self._depth:
|
||
self._parts.append(data)
|
||
|
||
def handle_endtag(self, tag: str) -> None:
|
||
if not self._depth or tag != "div":
|
||
return
|
||
self._depth -= 1
|
||
if self._depth == 0 and self._href:
|
||
self.rows.append({"href": self._href, "text": " ".join("".join(self._parts).split())})
|
||
self._parts = []
|
||
self._href = ""
|
||
|
||
|
||
def _text_request(url: str, timeout: int = 30) -> str:
|
||
request = Request(url, headers={"Accept": "text/html"})
|
||
with urlopen(request, timeout=timeout) as response:
|
||
return response.read().decode("utf-8", errors="replace")
|
||
|
||
|
||
def _size_bytes(label: str) -> int:
|
||
match = re.search(r"([0-9]+(?:\.[0-9]+)?)\s*(KB|MB|GB|TB)", label.upper())
|
||
if not match:
|
||
return 0
|
||
multipliers = {"KB": 10**3, "MB": 10**6, "GB": 10**9, "TB": 10**12}
|
||
return int(float(match.group(1)) * multipliers[match.group(2)])
|
||
|
||
|
||
def _parse_variant_page(html: str, family: str) -> list[dict[str, Any]]:
|
||
parser = _VariantPageParser()
|
||
parser.feed(html)
|
||
rows: list[dict[str, Any]] = []
|
||
seen: set[str] = set()
|
||
for item in parser.rows:
|
||
href = unescape(item["href"])
|
||
name = unquote(href.rsplit("/", 1)[-1])
|
||
if not name or name in seen or not name.startswith(family + ":"):
|
||
continue
|
||
seen.add(name)
|
||
text = item["text"]
|
||
size_match = re.search(r"•\s*([0-9]+(?:\.[0-9]+)?(?:KB|MB|GB|TB))\s*•", text, re.I)
|
||
context_match = re.search(r"([0-9]+)K\s+context window", text, re.I)
|
||
input_match = re.search(r"([^•]+?)\s+input\s+•", text, re.I)
|
||
size_label = size_match.group(1).upper() if size_match else ("cloud" if ("-cloud" in name or ":cloud" in name) else "Unknown")
|
||
modalities = [part.strip() for part in (input_match.group(1).split(",") if input_match else []) if part.strip()]
|
||
caps = ["vision"] if any(part.lower() == "image" for part in modalities) else []
|
||
rows.append({
|
||
"name": name,
|
||
"model": name,
|
||
"size": _size_bytes(size_label),
|
||
"size_label": size_label,
|
||
"modified_at": None,
|
||
"digest": "",
|
||
"details": {"family": family, "context_length": int(context_match.group(1)) * 1024 if context_match else None, "format": "gguf"},
|
||
"capabilities": caps,
|
||
"input_modalities": modalities or ["Text"],
|
||
"is_mlx": _is_mlx(name) or bool(re.search(r"\bMLX\b", text, re.I)),
|
||
})
|
||
return [row for row in rows if not row["is_mlx"]]
|
||
|
||
|
||
def _fetch_family_variants(family: str) -> list[dict[str, Any]]:
|
||
try:
|
||
html = _text_request(f"{REMOTE_OLLAMA}/library/{family}/tags", timeout=30)
|
||
return _parse_variant_page(html, family)
|
||
except (HTTPError, URLError, OSError, ValueError):
|
||
return []
|
||
|
||
|
||
def _catalog_path() -> Path:
|
||
return _home() / CATALOG_FILE
|
||
|
||
|
||
def _read_catalog() -> dict[str, Any]:
|
||
try:
|
||
value = json.loads(_catalog_path().read_text(encoding="utf-8"))
|
||
return value if isinstance(value, dict) else {}
|
||
except (OSError, ValueError):
|
||
return {}
|
||
|
||
|
||
def _write_catalog(value: dict[str, Any]) -> None:
|
||
path = _catalog_path()
|
||
temp = path.with_suffix(".tmp")
|
||
temp.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||
temp.replace(path)
|
||
|
||
|
||
def _refresh_family_variants(rows: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
|
||
families = sorted({_family_key(str(row.get("name") or row.get("model"))) for row in rows if not _is_mlx(row)})
|
||
return {family: _fetch_family_variants(family) for family in families if family}
|
||
|
||
|
||
def refresh_catalog(force: bool = True) -> dict[str, Any]:
|
||
with _catalog_lock:
|
||
try:
|
||
query = urlencode({"limit": 100, "sort": "popular"})
|
||
payload = _json_request(f"{REMOTE_OLLAMA}/api/tags?{query}", timeout=30)
|
||
rows = [item for item in payload.get("models", []) if isinstance(item, dict) and not _is_mlx(item)]
|
||
catalog = {
|
||
"fetched_at": datetime.now(timezone.utc).isoformat(),
|
||
"source": f"{REMOTE_OLLAMA}/api/tags",
|
||
"models": rows,
|
||
"families": _refresh_family_variants(_local_tags()),
|
||
}
|
||
_write_catalog(catalog)
|
||
return catalog
|
||
except (HTTPError, URLError, OSError, ValueError) as exc:
|
||
cached = _read_catalog()
|
||
if cached:
|
||
cached["last_error"] = str(exc)
|
||
return cached
|
||
return {"fetched_at": None, "source": f"{REMOTE_OLLAMA}/api/tags", "models": [], "last_error": str(exc)}
|
||
|
||
|
||
def _catalog_stale(catalog: dict[str, Any]) -> bool:
|
||
raw = catalog.get("fetched_at")
|
||
if not raw:
|
||
return True
|
||
try:
|
||
fetched = datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
|
||
return datetime.now(timezone.utc) - fetched > timedelta(hours=20)
|
||
except ValueError:
|
||
return True
|
||
|
||
|
||
def _ensure_catalog() -> dict[str, Any]:
|
||
catalog = _read_catalog()
|
||
if not catalog.get("models"):
|
||
return refresh_catalog()
|
||
if _catalog_stale(catalog) and not _catalog_lock.locked():
|
||
threading.Thread(target=refresh_catalog, kwargs={"force": True}, daemon=True, name="ollama-catalog-refresh").start()
|
||
return catalog
|
||
|
||
|
||
def _next_refresh() -> str:
|
||
now = datetime.now(MELBOURNE)
|
||
target = now.replace(hour=1, minute=0, second=0, microsecond=0)
|
||
if now >= target:
|
||
target += timedelta(days=1)
|
||
return target.isoformat()
|
||
|
||
|
||
def _job_snapshot() -> list[dict[str, Any]]:
|
||
with _jobs_lock:
|
||
return [dict(item) for item in _jobs.values()]
|
||
|
||
|
||
def _set_job(job_id: str, **values: Any) -> None:
|
||
with _jobs_lock:
|
||
if job_id in _jobs:
|
||
_jobs[job_id].update(values, updated_at=time.time())
|
||
|
||
|
||
def _run_pull(job_id: str, name: str, action: str, target: str) -> None:
|
||
try:
|
||
endpoint = _target_endpoint(target)
|
||
payload = json.dumps({"name": name, "stream": True}).encode("utf-8")
|
||
request = Request(endpoint + "/api/pull", data=payload, headers={"Content-Type": "application/json"}, method="POST")
|
||
with urlopen(request, timeout=3600) as response:
|
||
for raw_line in response:
|
||
try:
|
||
event = json.loads(raw_line.decode("utf-8"))
|
||
except ValueError:
|
||
continue
|
||
status = str(event.get("status") or "working")
|
||
completed = int(event.get("completed") or 0)
|
||
total = int(event.get("total") or 0)
|
||
percent = round(completed * 100 / total, 1) if total else None
|
||
_set_job(job_id, status=status, completed=completed, total=total, percent=percent, digest=event.get("digest"))
|
||
if event.get("error"):
|
||
raise RuntimeError(str(event["error"]))
|
||
_set_job(job_id, state="completed", status="success", percent=100)
|
||
except Exception as exc:
|
||
_set_job(job_id, state="failed", status="error", error=str(exc))
|
||
|
||
|
||
def _run_delete(job_id: str, name: str, target: str) -> None:
|
||
try:
|
||
endpoint = _target_endpoint(target)
|
||
_json_request(endpoint + "/api/delete", method="DELETE", payload={"name": name}, timeout=120)
|
||
_set_job(job_id, state="completed", status="deleted", percent=100)
|
||
except Exception as exc:
|
||
_set_job(job_id, state="failed", status="error", error=str(exc))
|
||
|
||
|
||
def _new_job(name: str, action: str, target: str = "local") -> str:
|
||
target = str(target or "local").strip().lower()
|
||
if target not in {"local", "remote"}:
|
||
raise HTTPException(400, "Target must be local or remote")
|
||
job_id = uuid.uuid4().hex
|
||
with _jobs_lock:
|
||
_jobs[job_id] = {"id": job_id, "name": name, "action": action, "target": target, "state": "running", "status": "starting", "percent": 0, "created_at": time.time(), "updated_at": time.time()}
|
||
target_fn = _run_delete if action == "delete" else _run_pull
|
||
args = (job_id, name, target) if action == "delete" else (job_id, name, action, target)
|
||
threading.Thread(target=target_fn, args=args, daemon=True, name=f"ollama-{action}-{job_id[:8]}").start()
|
||
return job_id
|
||
|
||
|
||
class ModelRequest(BaseModel):
|
||
name: str
|
||
target: str = "local"
|
||
|
||
|
||
class ConnectionRequest(BaseModel):
|
||
url: str
|
||
role: str = "local"
|
||
|
||
|
||
class ModelsRequest(BaseModel):
|
||
names: list[str] = Field(default_factory=list)
|
||
|
||
|
||
class ChatAttachment(BaseModel):
|
||
name: str = ""
|
||
mime_type: str = ""
|
||
data_url: str | None = None
|
||
url: str | None = None
|
||
|
||
|
||
class ChatRequest(BaseModel):
|
||
model: str = ""
|
||
models: list[str] = Field(default_factory=list)
|
||
message: str = ""
|
||
history: list[dict[str, Any]] = Field(default_factory=list)
|
||
attachments: list[ChatAttachment] = Field(default_factory=list)
|
||
request_id: str = ""
|
||
conversation_id: str = ""
|
||
|
||
|
||
class ChatStopRequest(BaseModel):
|
||
request_id: str
|
||
|
||
def _installed_model_names() -> set[str]:
|
||
return {
|
||
str(row.get("name") or row.get("model"))
|
||
for row in _local_tags()
|
||
if not _is_mlx(row)
|
||
}
|
||
|
||
|
||
def _require_installed_model(name: str) -> str:
|
||
name = _valid_name(name)
|
||
if name not in _installed_model_names():
|
||
raise HTTPException(400, f"Model '{name}' is not installed locally")
|
||
return name
|
||
|
||
|
||
def _ollama_error(exc: HTTPError) -> HTTPException:
|
||
try:
|
||
detail = exc.read().decode("utf-8", errors="replace")[:1000]
|
||
payload = json.loads(detail)
|
||
detail = str(payload.get("error") or detail)
|
||
except (OSError, ValueError):
|
||
detail = str(exc)
|
||
return HTTPException(502, f"Ollama request failed: {detail}")
|
||
|
||
|
||
def _load_model(name: str) -> dict[str, Any]:
|
||
name = _require_installed_model(name)
|
||
try:
|
||
result = _json_request(
|
||
LOCAL_OLLAMA + "/api/generate",
|
||
method="POST",
|
||
payload={"model": name, "prompt": "", "stream": False, "keep_alive": CHAT_KEEP_ALIVE, "options": {"num_predict": 1}},
|
||
timeout=900,
|
||
)
|
||
except HTTPError as exc:
|
||
raise _ollama_error(exc) from exc
|
||
return {"ok": True, "model": name, "response": result.get("response", ""), "runtime": _runtime_snapshot()}
|
||
|
||
|
||
def _chat_payload(body: ChatRequest, model_name: str | None = None) -> dict[str, Any]:
|
||
model = _require_installed_model(model_name or body.model)
|
||
messages: list[dict[str, Any]] = []
|
||
for item in body.history[-24:]:
|
||
role = str(item.get("role") or "")
|
||
content = str(item.get("content") or "").strip()
|
||
if role in {"user", "assistant"} and content:
|
||
messages.append({"role": role, "content": content[:MAX_ATTACHMENT_TEXT]})
|
||
text_parts = [body.message.strip()] if body.message.strip() else []
|
||
images: list[str] = []
|
||
for attachment in body.attachments[:12]:
|
||
text, image = _attachment_parts(attachment)
|
||
if text:
|
||
text_parts.append(text)
|
||
if image:
|
||
images.append(image)
|
||
if not text_parts and not images:
|
||
raise HTTPException(400, "Enter a message or attach a file/URL")
|
||
user_message: dict[str, Any] = {"role": "user", "content": "\n\n".join(text_parts)[:MAX_ATTACHMENT_TEXT]}
|
||
if images:
|
||
user_message["images"] = images
|
||
messages.append(user_message)
|
||
return {"model": model, "messages": messages, "stream": False, "keep_alive": CHAT_KEEP_ALIVE}
|
||
|
||
|
||
class _ChatStopped(Exception):
|
||
pass
|
||
|
||
|
||
def _valid_chat_request_id(value: str) -> str:
|
||
value = str(value or "").strip()
|
||
if not re.fullmatch(r"[A-Za-z0-9._-]{1,80}", value):
|
||
raise HTTPException(400, "Invalid chat request id")
|
||
return value
|
||
|
||
|
||
def _chat_state(request_id: str, **values: Any) -> dict[str, Any]:
|
||
with _chat_requests_lock:
|
||
state = _chat_requests.setdefault(request_id, {"request_id": request_id, "cancel": threading.Event(), "state": "starting", "stage": "Preparing request", "started_at": time.time(), "chunks": 0, "thinking_chars": 0, "response_chars": 0, "first_token_at": None, "prompt_eval_count": None, "eval_count": None, "total_duration": None, "load_duration": None, "prompt_eval_duration": None, "eval_duration": None})
|
||
state.update(values, updated_at=time.time())
|
||
return {key: value for key, value in state.items() if key not in {"cancel", "response"}}
|
||
|
||
|
||
def _stream_chat_request(payload: dict[str, Any], request_id: str, cancel_event: threading.Event | None = None, parent_id: str | None = None) -> dict[str, Any]:
|
||
if cancel_event is not None or parent_id is not None:
|
||
with _chat_requests_lock:
|
||
state = _chat_requests.get(request_id)
|
||
if state:
|
||
if cancel_event is not None: state["cancel"] = cancel_event
|
||
if parent_id is not None: state["parent_id"] = parent_id
|
||
data = json.dumps({**payload, "stream": True}).encode("utf-8")
|
||
request = Request(LOCAL_OLLAMA + "/api/chat", data=data, headers={"Accept": "application/x-ndjson", "Content-Type": "application/json"}, method="POST")
|
||
response_text: list[str] = []
|
||
thinking_text: list[str] = []
|
||
_chat_state(request_id, state="connecting", stage="Connecting to Ollama")
|
||
try:
|
||
with urlopen(request, timeout=1800) as response:
|
||
_chat_state(request_id, response=response, state="generating", stage="Ollama is generating")
|
||
while True:
|
||
with _chat_requests_lock:
|
||
cancelled = bool(_chat_requests.get(request_id, {}).get("cancel", threading.Event()).is_set())
|
||
if cancelled:
|
||
response.close()
|
||
raise _ChatStopped()
|
||
try:
|
||
raw_line = response.readline()
|
||
except (socket.timeout, TimeoutError, OSError, ValueError) as exc:
|
||
with _chat_requests_lock:
|
||
cancelled = bool(_chat_requests.get(request_id, {}).get("cancel", threading.Event()).is_set())
|
||
if cancelled:
|
||
raise _ChatStopped() from exc
|
||
raise
|
||
if not raw_line:
|
||
break
|
||
try:
|
||
event = json.loads(raw_line.decode("utf-8", errors="replace"))
|
||
except ValueError:
|
||
continue
|
||
message = event.get("message") if isinstance(event.get("message"), dict) else {}
|
||
chunk = str(message.get("content") or event.get("response") or "")
|
||
thinking_chunk = str(message.get("thinking") or event.get("thinking") or "")
|
||
if chunk:
|
||
response_text.append(chunk)
|
||
if thinking_chunk:
|
||
thinking_text.append(thinking_chunk)
|
||
if chunk and not _chat_requests.get(request_id, {}).get("first_token_at"):
|
||
_chat_state(request_id, first_token_at=time.time())
|
||
_chat_state(
|
||
request_id,
|
||
state="generating",
|
||
stage="Ollama is generating the response" if chunk else "Ollama is processing model thinking",
|
||
chunks=int(_chat_requests.get(request_id, {}).get("chunks", 0)) + 1,
|
||
thinking_chars=sum(map(len, thinking_text)),
|
||
response_chars=sum(map(len, response_text)),
|
||
prompt_eval_count=event.get("prompt_eval_count", _chat_requests.get(request_id, {}).get("prompt_eval_count")),
|
||
eval_count=event.get("eval_count", _chat_requests.get(request_id, {}).get("eval_count")),
|
||
total_duration=event.get("total_duration", _chat_requests.get(request_id, {}).get("total_duration")),
|
||
load_duration=event.get("load_duration", _chat_requests.get(request_id, {}).get("load_duration")),
|
||
prompt_eval_duration=event.get("prompt_eval_duration", _chat_requests.get(request_id, {}).get("prompt_eval_duration")),
|
||
eval_duration=event.get("eval_duration", _chat_requests.get(request_id, {}).get("eval_duration")),
|
||
)
|
||
if event.get("done"):
|
||
break
|
||
except _ChatStopped:
|
||
_chat_state(request_id, state="stopped", stage="Stopped by user", finished_at=time.time())
|
||
raise
|
||
except Exception:
|
||
_chat_state(request_id, state="failed", stage="Ollama request failed", finished_at=time.time())
|
||
raise
|
||
_chat_state(request_id, state="completed", stage="Response complete", finished_at=time.time())
|
||
with _chat_requests_lock:
|
||
final_state = dict(_chat_requests.get(request_id, {}))
|
||
return {"message": {"role": "assistant", "content": "".join(response_text)}, "done": True, "metrics": _metric_values(final_state, status="completed")}
|
||
|
||
|
||
@router.get("/chat/status/{request_id}")
|
||
def chat_status(request_id: str) -> dict[str, Any]:
|
||
request_id = _valid_chat_request_id(request_id)
|
||
with _chat_requests_lock:
|
||
state = _chat_requests.get(request_id)
|
||
if not state:
|
||
raise HTTPException(404, "Chat request not found")
|
||
result = {key: value for key, value in state.items() if key not in {"cancel", "response"}}
|
||
result["elapsed"] = round(max(0.0, time.time() - float(state.get("started_at") or time.time())), 1)
|
||
return result
|
||
|
||
|
||
@router.post("/chat/stop")
|
||
def chat_stop(body: ChatStopRequest) -> dict[str, Any]:
|
||
request_id = _valid_chat_request_id(body.request_id)
|
||
with _chat_requests_lock:
|
||
state = _chat_requests.get(request_id)
|
||
if not state:
|
||
return {"ok": True, "request_id": request_id, "state": "not_found"}
|
||
state["cancel"].set()
|
||
responses = []
|
||
for child_id, child in _chat_requests.items():
|
||
if child_id == request_id or child.get("parent_id") == request_id:
|
||
child["cancel"].set()
|
||
response = child.get("response")
|
||
if response is not None: responses.append(response)
|
||
child["state"] = "stopping"
|
||
child["stage"] = "Stopping Ollama request"
|
||
child["updated_at"] = time.time()
|
||
state["state"] = "stopping"
|
||
state["stage"] = "Stopping Ollama request"
|
||
state["updated_at"] = time.time()
|
||
for response in responses:
|
||
try: response.close()
|
||
except Exception: pass
|
||
return {"ok": True, "request_id": request_id, "state": "stopping"}
|
||
|
||
|
||
@router.get("/conversations")
|
||
def conversations() -> dict[str, Any]:
|
||
db = _chat_db()
|
||
try:
|
||
rows = db.execute("SELECT id,title,model,models_json,created_at,updated_at,(SELECT COUNT(*) FROM messages m WHERE m.conversation_id=c.id) AS message_count FROM conversations c ORDER BY updated_at DESC LIMIT 100").fetchall()
|
||
result = []
|
||
for row in rows:
|
||
item = dict(row)
|
||
item["models"] = json.loads(item.pop("models_json") or "[]")
|
||
result.append(item)
|
||
return {"conversations": result}
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
@router.get("/conversations/{conversation_id}")
|
||
def conversation(conversation_id: str) -> dict[str, Any]:
|
||
conversation_id = _conversation_id(conversation_id)
|
||
db = _chat_db()
|
||
try:
|
||
row = db.execute("SELECT id,title,model,models_json,created_at,updated_at FROM conversations WHERE id=?", (conversation_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "Conversation not found")
|
||
item = dict(row)
|
||
item["models"] = json.loads(item.pop("models_json") or "[]")
|
||
messages = []
|
||
for message in db.execute("SELECT id,request_id,role,content,model,attachments_json,created_at FROM messages WHERE conversation_id=? ORDER BY id", (conversation_id,)).fetchall():
|
||
value = dict(message)
|
||
value["attachments"] = json.loads(value.pop("attachments_json") or "[]")
|
||
messages.append(value)
|
||
metrics = [_row_metric(metric) for metric in db.execute("SELECT * FROM chat_metrics WHERE conversation_id=? ORDER BY id", (conversation_id,)).fetchall()]
|
||
return {"conversation": item, "messages": messages, "metrics": metrics}
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
@router.delete("/conversations/{conversation_id}")
|
||
def delete_conversation(conversation_id: str) -> dict[str, Any]:
|
||
conversation_id = _conversation_id(conversation_id)
|
||
db = _chat_db()
|
||
try:
|
||
db.execute("DELETE FROM conversations WHERE id=?", (conversation_id,))
|
||
db.commit()
|
||
return {"ok": True, "conversation_id": conversation_id}
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
@router.get("/metrics")
|
||
def metrics(limit: int = 100) -> dict[str, Any]:
|
||
limit = max(1, min(int(limit), 500))
|
||
db = _chat_db()
|
||
try:
|
||
rows = [_row_metric(row) for row in db.execute("SELECT * FROM chat_metrics ORDER BY id DESC LIMIT ?", (limit,)).fetchall()]
|
||
completed = [row for row in rows if row["status"] == "completed"]
|
||
def average(key: str) -> float | None:
|
||
values = [float(row[key]) for row in completed if row.get(key) is not None]
|
||
return round(sum(values) / len(values), 2) if values else None
|
||
aggregate = {
|
||
"sample_count": len(rows),
|
||
"completed_count": len(completed),
|
||
"error_count": sum(row["status"] == "failed" for row in rows),
|
||
"stopped_count": sum(row["status"] == "stopped" for row in rows),
|
||
"avg_time_to_first_token_ms": average("time_to_first_token_ms"),
|
||
"avg_total_latency_ms": average("total_latency_ms"),
|
||
"avg_eval_tokens_per_second": average("eval_tokens_per_second"),
|
||
"avg_prompt_tokens_per_second": average("prompt_tokens_per_second"),
|
||
"total_output_tokens": sum(int(row["eval_count"]) for row in completed if row.get("eval_count") is not None),
|
||
}
|
||
return {"metrics": rows, "aggregate": aggregate}
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
@router.get("/runtime")
|
||
def runtime() -> dict[str, Any]:
|
||
return _runtime_snapshot()
|
||
|
||
|
||
@router.post("/chat/load")
|
||
def chat_load(body: ModelRequest) -> dict[str, Any]:
|
||
return _load_model(body.name)
|
||
|
||
|
||
@router.post("/models/load")
|
||
def models_load(body: ModelsRequest) -> dict[str, Any]:
|
||
names = list(dict.fromkeys(_valid_name(name) for name in body.names if str(name).strip()))[:12]
|
||
if not names:
|
||
raise HTTPException(400, "Select at least one model to load")
|
||
results = []
|
||
for name in names:
|
||
_model_load_update(name, active=True, state="queued", stage="Waiting for Ollama", started_at=time.time(), error="")
|
||
for name in names:
|
||
_model_load_update(name, state="loading", stage="Loading model into Ollama memory")
|
||
try:
|
||
results.append({"name": name, "ok": True, "result": _load_model(name)})
|
||
_model_load_update(name, state="checking", stage="Checking Ollama resident state")
|
||
except Exception as exc:
|
||
results.append({"name": name, "ok": False, "error": str(exc)})
|
||
_model_load_update(name, active=False, state="failed", stage="Ollama load failed", finished_at=time.time(), error=str(exc))
|
||
resident_rows = _local_ps()
|
||
resident_names = {str(row.get("name") or row.get("model")) for row in resident_rows}
|
||
for item in results:
|
||
item["resident"] = item["name"] in resident_names
|
||
not_resident = [name for name in names if name not in resident_names]
|
||
for item in results:
|
||
if item["name"] in resident_names:
|
||
_model_load_update(item["name"], active=False, state="resident", stage="Model is resident in Ollama", finished_at=time.time())
|
||
elif item["ok"]:
|
||
_model_load_update(item["name"], active=False, state="evicted", stage="Ollama did not retain this model", finished_at=time.time())
|
||
runtime = _runtime_snapshot()
|
||
return {
|
||
"ok": bool(results) and not not_resident and all(item["ok"] for item in results),
|
||
"requested": names,
|
||
"resident": sorted(resident_names),
|
||
"not_resident": not_resident,
|
||
"results": results,
|
||
"runtime": runtime,
|
||
"keep_alive": "permanent",
|
||
"message": (
|
||
"All selected models are resident."
|
||
if not not_resident
|
||
else "Ollama did not keep every selected model resident; see not_resident."
|
||
),
|
||
}
|
||
|
||
|
||
@router.post("/models/unload")
|
||
def models_unload(body: ModelsRequest) -> dict[str, Any]:
|
||
names = list(dict.fromkeys(_valid_name(name) for name in body.names if str(name).strip()))[:12]
|
||
if not names:
|
||
raise HTTPException(400, "Select at least one model to unload")
|
||
results = []
|
||
for name in names:
|
||
try:
|
||
_require_installed_model(name)
|
||
_json_request(LOCAL_OLLAMA + "/api/generate", method="POST", payload={"model": name, "prompt": "", "stream": False, "keep_alive": 0}, timeout=120)
|
||
results.append({"name": name, "ok": True})
|
||
except Exception as exc:
|
||
results.append({"name": name, "ok": False, "error": str(exc)})
|
||
return {"ok": all(item["ok"] for item in results), "results": results, "runtime": _runtime_snapshot()}
|
||
|
||
|
||
@router.post("/chat")
|
||
def chat(body: ChatRequest) -> dict[str, Any]:
|
||
request_id = _valid_chat_request_id(body.request_id or uuid.uuid4().hex)
|
||
conversation_id = _conversation_id(body.conversation_id)
|
||
selected = list(dict.fromkeys(_valid_name(name) for name in (body.models or ([body.model] if body.model else [])) if str(name).strip()))[:12]
|
||
if not selected:
|
||
raise HTTPException(400, "Select at least one loaded model")
|
||
_ensure_conversation(conversation_id, selected[0], selected, body.message or "New conversation")
|
||
attachment_meta = [{"name": item.name, "mime_type": item.mime_type, "url": item.url} for item in body.attachments[:12]]
|
||
_persist_message(conversation_id, request_id, "user", body.message.strip() or "[Attachments]", selected[0], attachment_meta)
|
||
_chat_state(request_id, state="preparing", stage="Preparing attachments", models=selected, conversation_id=conversation_id)
|
||
with _chat_requests_lock:
|
||
cancel_event = _chat_requests[request_id]["cancel"]
|
||
if len(selected) == 1:
|
||
payload = _chat_payload(body, selected[0])
|
||
try:
|
||
result = _stream_chat_request(payload, request_id, cancel_event=cancel_event)
|
||
except _ChatStopped as exc:
|
||
with _chat_requests_lock:
|
||
state = dict(_chat_requests.get(request_id, {}))
|
||
_persist_metric(conversation_id, request_id, selected[0], state, status="stopped")
|
||
raise HTTPException(499, "Chat stopped by user") from exc
|
||
except HTTPError as exc:
|
||
with _chat_requests_lock:
|
||
state = dict(_chat_requests.get(request_id, {}))
|
||
_persist_metric(conversation_id, request_id, selected[0], state, status="failed", error=str(exc))
|
||
raise _ollama_error(exc) from exc
|
||
except Exception as exc:
|
||
with _chat_requests_lock:
|
||
state = dict(_chat_requests.get(request_id, {}))
|
||
_persist_metric(conversation_id, request_id, selected[0], state, status="failed", error=str(exc))
|
||
raise
|
||
message = result.get("message") if isinstance(result.get("message"), dict) else {}
|
||
content = str(message.get("content") or "")
|
||
_persist_message(conversation_id, request_id, "assistant", content, selected[0])
|
||
with _chat_requests_lock:
|
||
state = dict(_chat_requests.get(request_id, {}))
|
||
persisted_metrics = _persist_metric(conversation_id, request_id, selected[0], state, status="completed")
|
||
return {"ok": True, "request_id": request_id, "conversation_id": conversation_id, "model": selected[0], "models": selected, "message": {"role": "assistant", "content": content}, "done": True, "metrics": persisted_metrics, "runtime": _runtime_snapshot()}
|
||
|
||
_chat_state(request_id, state="generating", stage=f"Querying {len(selected)} models in parallel")
|
||
results: dict[str, dict[str, Any]] = {}
|
||
errors: dict[str, str] = {}
|
||
|
||
def run_model(index: int, name: str):
|
||
child_id = f"{request_id}-{index}"
|
||
_chat_state(child_id, state="preparing", stage=f"Preparing {name}", model=name, parent_id=request_id, conversation_id=conversation_id)
|
||
payload = _chat_payload(body, name)
|
||
return name, _stream_chat_request(payload, child_id, cancel_event=cancel_event, parent_id=request_id)
|
||
|
||
try:
|
||
with ThreadPoolExecutor(max_workers=len(selected), thread_name_prefix="ollama-chat") as pool:
|
||
futures = [pool.submit(run_model, index, name) for index, name in enumerate(selected)]
|
||
for future in as_completed(futures):
|
||
try:
|
||
name, result = future.result()
|
||
results[name] = result
|
||
child_id = f"{request_id}-{selected.index(name)}"
|
||
with _chat_requests_lock:
|
||
child_state = dict(_chat_requests.get(child_id, {}))
|
||
_persist_metric(conversation_id, child_id, name, child_state, status="completed")
|
||
_chat_state(request_id, stage=f"Received response from {len(results)} of {len(selected)} models", response_chars=sum(len(str((r.get("message") or {}).get("content") or "")) for r in results.values()))
|
||
except _ChatStopped:
|
||
raise
|
||
except HTTPError as exc:
|
||
errors[str(exc)] = str(exc)
|
||
except Exception as exc:
|
||
errors[type(exc).__name__] = str(exc)
|
||
except _ChatStopped as exc:
|
||
raise HTTPException(499, "Chat stopped by user") from exc
|
||
|
||
if not results and errors:
|
||
raise HTTPException(502, "All selected Ollama models failed: " + "; ".join(errors.values()))
|
||
sections = []
|
||
for name in selected:
|
||
if name in results:
|
||
message = results[name].get("message") if isinstance(results[name].get("message"), dict) else {}
|
||
sections.append(f"[{name}]\n{str(message.get('content') or '').strip()}")
|
||
else:
|
||
sections.append(f"[{name}]\nModel failed: {errors.get(name, 'No response received')}")
|
||
combined = "\n\n".join(sections)
|
||
_chat_state(request_id, state="completed", stage="Combined model responses", finished_at=time.time(), response_chars=len(combined))
|
||
_persist_message(conversation_id, request_id, "assistant", combined, selected[0])
|
||
return {"ok": True, "request_id": request_id, "conversation_id": conversation_id, "model": selected[0], "models": selected, "message": {"role": "assistant", "content": combined}, "model_responses": {name: str((results.get(name, {}).get("message") or {}).get("content") or "") for name in selected if name in results}, "metrics": [results[name].get("metrics") for name in selected if name in results], "errors": errors, "done": True, "runtime": _runtime_snapshot()}
|
||
|
||
|
||
@router.get("/connections")
|
||
def connections() -> dict[str, Any]:
|
||
saved = _read_connections()
|
||
active = saved.get("active_url") or os.environ.get("OLLAMA_HOST", "").strip() or LOCAL_OLLAMA
|
||
return {"active_url": active, "connections": _connection_snapshot(), "containerized": _running_in_container()}
|
||
|
||
|
||
@router.post("/connections/test")
|
||
def connections_test(body: ConnectionRequest) -> dict[str, Any]:
|
||
result = _probe_endpoint(body.url, timeout=8)
|
||
result["role"] = str(body.role or "local").strip().lower()
|
||
return result
|
||
|
||
|
||
@router.post("/connections/configure")
|
||
def connections_configure(body: ConnectionRequest) -> dict[str, Any]:
|
||
url = _valid_ollama_url(body.url)
|
||
role = str(body.role or "local").strip().lower()
|
||
if role not in {"local", "remote"}:
|
||
raise HTTPException(400, "Connection role must be local or remote")
|
||
saved = _read_connections()
|
||
saved[f"{role}_url"] = url
|
||
if role == "local":
|
||
saved["active_url"] = url
|
||
_write_connections(saved)
|
||
_sync_native_ollama_providers(saved)
|
||
if role == "local":
|
||
global LOCAL_OLLAMA
|
||
LOCAL_OLLAMA = url
|
||
result = _probe_endpoint(url, timeout=8)
|
||
result.update({"ok": True, "role": role, "message": f"Saved {role} Ollama endpoint"})
|
||
return result
|
||
|
||
|
||
@router.delete("/connections/{role}")
|
||
def connections_delete(role: str) -> dict[str, Any]:
|
||
role = str(role or "").strip().lower()
|
||
if role not in {"local", "remote"}:
|
||
raise HTTPException(400, "Connection role must be local or remote")
|
||
saved = _read_connections()
|
||
key = f"{role}_url"
|
||
removed = saved.get(key, "")
|
||
if not removed:
|
||
raise HTTPException(404, f"No saved {role} Ollama endpoint")
|
||
saved[key] = ""
|
||
if role == "local":
|
||
saved["active_url"] = ""
|
||
_write_connections(saved)
|
||
_sync_native_ollama_providers(saved)
|
||
if role == "local":
|
||
_apply_saved_connection()
|
||
return {"ok": True, "role": role, "removed_url": removed, "message": f"Removed saved {role} Ollama endpoint"}
|
||
|
||
|
||
@router.get("/status")
|
||
def status() -> dict[str, Any]:
|
||
tags = _local_tags()
|
||
ps_rows = _local_ps()
|
||
loaded = {str(row.get("name") or row.get("model")): row for row in ps_rows}
|
||
local = [
|
||
_model_view(row, loaded.get(str(row.get("name") or row.get("model"))))
|
||
for row in tags
|
||
if not _is_mlx(row)
|
||
]
|
||
catalog = _ensure_catalog()
|
||
installed_names = {row["name"] for row in local}
|
||
|
||
def catalog_view(raw: dict[str, Any], source: str) -> dict[str, Any]:
|
||
name = str(raw.get("name") or raw.get("model"))
|
||
view = _model_view(raw, loaded.get(name), source=source)
|
||
view["installed"] = name in installed_names
|
||
view["loaded"] = name in loaded
|
||
return view
|
||
|
||
family_rows = catalog.get("families") if isinstance(catalog.get("families"), dict) else {}
|
||
catalog_rows = list(catalog.get("models", []))
|
||
catalog_names = {str(row.get("name") or row.get("model") or "") for row in catalog_rows}
|
||
candidate_rows = catalog_rows + [
|
||
variant
|
||
for rows in family_rows.values()
|
||
for variant in rows
|
||
if str(variant.get("name") or variant.get("model") or "") not in catalog_names
|
||
]
|
||
downloadable = []
|
||
seen_downloads: set[str] = set()
|
||
for rank, row in enumerate(candidate_rows, 1):
|
||
name = str(row.get("name") or row.get("model") or "")
|
||
if _is_mlx(row) or not name or name in installed_names or name in seen_downloads:
|
||
continue
|
||
view = catalog_view(row, "catalog")
|
||
if not _known_ram_fit(view):
|
||
continue
|
||
if name in catalog_names:
|
||
view["popularity_rank"] = rank
|
||
seen_downloads.add(name)
|
||
downloadable.append(view)
|
||
|
||
popular = _popular_fit_models(
|
||
[row for row in catalog.get("models", []) if not _is_mlx(row)],
|
||
family_rows,
|
||
catalog_view,
|
||
installed_names,
|
||
loaded,
|
||
)
|
||
for row in local:
|
||
family = _family_key(row["name"])
|
||
variants = []
|
||
for raw in family_rows.get(family, []):
|
||
if _is_mlx(raw):
|
||
continue
|
||
variant = _model_view(raw, loaded.get(raw["name"]), source="variant")
|
||
variant["installed"] = variant["name"] in installed_names
|
||
variant["current"] = variant["name"] == row["name"]
|
||
variants.append(variant)
|
||
row["variants"] = sorted(variants, key=lambda item: (item.get("size_bytes") or 0, item["name"]))
|
||
|
||
ollama_version = _ollama_version()
|
||
connection_rows = _connection_snapshot()
|
||
return {
|
||
"ollama": {
|
||
"available": bool(tags or ps_rows or ollama_version),
|
||
"version": ollama_version,
|
||
"endpoint": LOCAL_OLLAMA,
|
||
"containerized": _running_in_container(),
|
||
"configured_endpoint": bool(os.environ.get("OLLAMA_HOST", "").strip()),
|
||
"connection_hint": (
|
||
"Set OLLAMA_HOST to a reachable Ollama service, such as "
|
||
"http://ollama:11434 or http://host.docker.internal:11434."
|
||
if _running_in_container() and not (tags or ps_rows or ollama_version)
|
||
else ""
|
||
),
|
||
},
|
||
"connections": connection_rows,
|
||
"models": local,
|
||
"popular": popular,
|
||
"popular_filter": {
|
||
"max_expected_ram_gib": _host_ram_gib(),
|
||
"basis": "detected MemTotal from the running host",
|
||
"requires_known_size": True,
|
||
"requires_known_ram": True,
|
||
"smaller_fit_variants_substituted": True,
|
||
},
|
||
"catalog": downloadable,
|
||
"catalog_filter": {
|
||
"max_expected_ram_gib": _host_ram_gib(),
|
||
"basis": "detected MemTotal from the running host",
|
||
"requires_known_size": True,
|
||
"requires_known_ram": True,
|
||
},
|
||
"catalog_filter_options": {
|
||
"types": ["all", "moe", "dense"],
|
||
"capabilities": sorted({cap for row in downloadable for cap in row.get("capabilities", [])}),
|
||
},
|
||
"catalog_source": catalog.get("source"),
|
||
"catalog_updated_at": catalog.get("fetched_at"),
|
||
"catalog_error": catalog.get("last_error"),
|
||
"next_catalog_refresh": _next_refresh(),
|
||
"jobs": _job_snapshot(),
|
||
"generated_at": time.time(),
|
||
}
|
||
|
||
|
||
def _ollama_version() -> str | None:
|
||
try:
|
||
return str(_json_request(LOCAL_OLLAMA + "/api/version", timeout=5).get("version") or "unknown")
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
@router.post("/catalog/refresh")
|
||
def catalog_refresh() -> dict[str, Any]:
|
||
catalog = refresh_catalog()
|
||
return {"ok": bool(catalog.get("models")), "updated_at": catalog.get("fetched_at"), "count": len(catalog.get("models", [])), "error": catalog.get("last_error")}
|
||
|
||
|
||
@router.post("/pull")
|
||
def pull_model(body: ModelRequest) -> dict[str, Any]:
|
||
name = _valid_name(body.name)
|
||
target = str(body.target or "local").strip().lower()
|
||
endpoint = _target_endpoint(target)
|
||
job_id = _new_job(name, "download", target)
|
||
return {"ok": True, "job_id": job_id, "endpoint": endpoint, "target": target, "message": f"Downloading or updating {name} on {target} ({endpoint})"}
|
||
|
||
|
||
@router.post("/redownload")
|
||
def redownload_model(body: ModelRequest) -> dict[str, Any]:
|
||
name = _valid_name(body.name)
|
||
target = str(body.target or "local").strip().lower()
|
||
endpoint = _target_endpoint(target)
|
||
job_id = _new_job(name, "redownload", target)
|
||
return {"ok": True, "job_id": job_id, "endpoint": endpoint, "target": target, "message": f"Re-downloading or updating {name} on {target} ({endpoint})"}
|
||
|
||
|
||
@router.delete("/model")
|
||
def delete_model(body: ModelRequest) -> dict[str, Any]:
|
||
name = _valid_name(body.name)
|
||
target = str(body.target or "local").strip().lower()
|
||
endpoint = _target_endpoint(target)
|
||
job_id = _new_job(name, "delete", target)
|
||
return {"ok": True, "job_id": job_id, "endpoint": endpoint, "target": target, "message": f"Removing {name} from {target} ({endpoint})"}
|
||
|
||
|
||
def create_ollama_routes(app) -> None:
|
||
_sync_native_ollama_providers()
|
||
app.include_router(router, prefix="/api/plugins/ollama-manager")
|