feat: add adaptive Ollama catalog filtering
This commit is contained in:
Hermes Agent
+453
-41
@@ -10,10 +10,12 @@ 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
|
||||
@@ -34,17 +36,180 @@ 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")
|
||||
POPULAR_RAM_LIMIT_GIB = 30.0
|
||||
MAX_ATTACHMENT_BYTES = 20 * 1024 * 1024
|
||||
MAX_ATTACHMENT_TEXT = 80_000
|
||||
MAX_URL_BYTES = 15 * 1024 * 1024
|
||||
CHAT_KEEP_ALIVE = "10m"
|
||||
CHAT_KEEP_ALIVE = -1
|
||||
|
||||
_jobs: dict[str, dict[str, Any]] = {}
|
||||
_jobs_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.",
|
||||
@@ -135,6 +300,12 @@ def _read_meminfo() -> dict[str, int]:
|
||||
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"
|
||||
@@ -177,17 +348,27 @@ def _runtime_snapshot() -> dict[str, Any]:
|
||||
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()}
|
||||
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,
|
||||
})
|
||||
return {
|
||||
"captured_at": time.time(),
|
||||
@@ -325,13 +506,15 @@ def _family_key(name: str) -> str:
|
||||
|
||||
|
||||
def _known_ram_fit(model: dict[str, Any]) -> bool:
|
||||
"""Return True only when both size and RAM are known and fit this host."""
|
||||
"""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 float(ram_gb) <= POPULAR_RAM_LIMIT_GIB
|
||||
and host_ram_gib is not None
|
||||
and float(ram_gb) <= host_ram_gib
|
||||
)
|
||||
|
||||
|
||||
@@ -735,6 +918,10 @@ class ModelRequest(BaseModel):
|
||||
name: str
|
||||
|
||||
|
||||
class ModelsRequest(BaseModel):
|
||||
names: list[str] = Field(default_factory=list)
|
||||
|
||||
|
||||
class ChatAttachment(BaseModel):
|
||||
name: str = ""
|
||||
mime_type: str = ""
|
||||
@@ -743,11 +930,13 @@ class ChatAttachment(BaseModel):
|
||||
|
||||
|
||||
class ChatRequest(BaseModel):
|
||||
model: str
|
||||
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):
|
||||
@@ -792,8 +981,8 @@ def _load_model(name: str) -> dict[str, Any]:
|
||||
return {"ok": True, "model": name, "response": result.get("response", ""), "runtime": _runtime_snapshot()}
|
||||
|
||||
|
||||
def _chat_payload(body: ChatRequest) -> dict[str, Any]:
|
||||
model = _require_installed_model(body.model)
|
||||
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 "")
|
||||
@@ -830,12 +1019,18 @@ def _valid_chat_request_id(value: str) -> str:
|
||||
|
||||
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})
|
||||
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 != "cancel"}
|
||||
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) -> dict[str, Any]:
|
||||
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] = []
|
||||
@@ -843,10 +1038,7 @@ def _stream_chat_request(payload: dict[str, Any], request_id: str) -> dict[str,
|
||||
_chat_state(request_id, state="connecting", stage="Connecting to Ollama")
|
||||
try:
|
||||
with urlopen(request, timeout=1800) as response:
|
||||
sock = getattr(getattr(getattr(response, "fp", None), "raw", None), "_sock", None)
|
||||
if sock is not None:
|
||||
sock.settimeout(1.0)
|
||||
_chat_state(request_id, state="generating", stage="Ollama is generating")
|
||||
_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())
|
||||
@@ -855,8 +1047,12 @@ def _stream_chat_request(payload: dict[str, Any], request_id: str) -> dict[str,
|
||||
raise _ChatStopped()
|
||||
try:
|
||||
raw_line = response.readline()
|
||||
except (socket.timeout, TimeoutError):
|
||||
continue
|
||||
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:
|
||||
@@ -870,6 +1066,8 @@ def _stream_chat_request(payload: dict[str, Any], request_id: str) -> dict[str,
|
||||
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",
|
||||
@@ -877,7 +1075,12 @@ def _stream_chat_request(payload: dict[str, Any], request_id: str) -> dict[str,
|
||||
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)),
|
||||
eval_count=event.get("eval_count"),
|
||||
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
|
||||
@@ -888,7 +1091,9 @@ def _stream_chat_request(payload: dict[str, Any], request_id: str) -> dict[str,
|
||||
_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())
|
||||
return {"message": {"role": "assistant", "content": "".join(response_text)}, "done": True}
|
||||
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}")
|
||||
@@ -898,7 +1103,7 @@ def chat_status(request_id: str) -> dict[str, Any]:
|
||||
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"}}
|
||||
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
|
||||
|
||||
@@ -911,12 +1116,98 @@ def chat_stop(body: ChatStopRequest) -> dict[str, Any]:
|
||||
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()
|
||||
@@ -927,26 +1218,120 @@ 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:
|
||||
try:
|
||||
results.append({"name": name, "ok": True, "result": _load_model(name)})
|
||||
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(), "keep_alive": "permanent"}
|
||||
|
||||
|
||||
@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)
|
||||
_chat_state(request_id, state="preparing", stage="Preparing attachments")
|
||||
payload = _chat_payload(body)
|
||||
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:
|
||||
result = _stream_chat_request(payload, request_id)
|
||||
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
|
||||
except HTTPError as exc:
|
||||
raise _ollama_error(exc) from exc
|
||||
message = result.get("message") if isinstance(result.get("message"), dict) else {}
|
||||
return {
|
||||
"ok": True,
|
||||
"request_id": request_id,
|
||||
"model": payload["model"],
|
||||
"message": {"role": "assistant", "content": str(message.get("content") or "")},
|
||||
"done": True,
|
||||
"runtime": _runtime_snapshot(),
|
||||
}
|
||||
|
||||
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("/status")
|
||||
@@ -969,13 +1354,29 @@ def status() -> dict[str, Any]:
|
||||
view["loaded"] = name in loaded
|
||||
return view
|
||||
|
||||
downloadable = [
|
||||
catalog_view(row, "catalog")
|
||||
for row in catalog.get("models", [])
|
||||
if not _is_mlx(row) and str(row.get("name") or row.get("model")) not in installed_names
|
||||
]
|
||||
|
||||
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,
|
||||
@@ -1000,14 +1401,25 @@ def status() -> dict[str, Any]:
|
||||
"models": local,
|
||||
"popular": popular,
|
||||
"popular_filter": {
|
||||
"max_expected_ram_gib": POPULAR_RAM_LIMIT_GIB,
|
||||
"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_updated_at": catalog.get("fetched_at"),
|
||||
"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(),
|
||||
|
||||
Reference in New Issue
Block a user