Spaces:
Runtime error
Runtime error
| """ | |
| ╔══════════════════════════════════════════════════════════════════════════╗ | |
| ║ MYTHICAL UNIVERSAL SYSTEM — app.py [v7.1] ║ | |
| ║ ║ | |
| ║ v7.0 fixes carried forward + ║ | |
| ║ [FIX-A] Watchdog container-aware (cgroup v1+v2): no more constant ║ | |
| ║ flush loop on shared HF Spaces hosts ║ | |
| ║ [FIX-B] Threshold sanity-check: env vars < 30% of total are ignored ║ | |
| ║ [FIX-C] Watchdog flush cooldown (30s) — reduces /slots/0 spam ║ | |
| ║ [FIX-D] THINKING_TIMEOUT (300s) — thinking mode no longer times out ║ | |
| ║ [FIX-E] Dynamic httpx timeout per-request (thinking vs normal) ║ | |
| ║ [FIX-F] Token factor uses wd.level, not hardcoded GB values ║ | |
| ║ [FIX-G] Health / metrics use container-aware RAM % ║ | |
| ╚══════════════════════════════════════════════════════════════════════════╝ | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| import base64 | |
| import collections | |
| import enum | |
| import gc | |
| import hashlib | |
| import io | |
| import logging | |
| import math | |
| import os | |
| import re | |
| import time | |
| import urllib.parse | |
| import uuid | |
| from contextlib import asynccontextmanager | |
| from pathlib import Path | |
| from typing import Any, AsyncIterator | |
| import httpx | |
| import orjson | |
| import psutil | |
| import uvloop | |
| from fastapi import FastAPI, File, Form, Request, UploadFile | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.responses import Response, StreamingResponse | |
| from model_manager import ModelManager | |
| uvloop.install() | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # CONFIG | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| LLAMA_HOST = os.getenv("LLAMA_HOST", "127.0.0.1") | |
| LLAMA_PORT = os.getenv("LLAMA_PORT", "8080") | |
| WHISPER_HOST = os.getenv("WHISPER_HOST", "127.0.0.1") | |
| WHISPER_PORT = os.getenv("WHISPER_PORT", "8081") | |
| LLAMA_URL = f"http://{LLAMA_HOST}:{LLAMA_PORT}" | |
| WHISPER_URL = f"http://{WHISPER_HOST}:{WHISPER_PORT}" | |
| API_KEY = os.getenv("API_KEY", "change-this-key") | |
| MAX_CTX_TOKENS = int(os.getenv("MAX_CTX_TOKENS", "14000")) | |
| MAX_NEW_TOKENS = int(os.getenv("MAX_NEW_TOKENS", "2048")) | |
| MAX_FILE_MB = int(os.getenv("MAX_FILE_MB", "50")) | |
| IMAGE_MAX_PX = int(os.getenv("IMAGE_MAX_PX", "1120")) | |
| VIDEO_MAX_FRAMES = int(os.getenv("VIDEO_MAX_FRAMES", "8")) | |
| CACHE_TTL = int(os.getenv("CACHE_TTL", "60")) | |
| RATE_LIMIT_RPM = int(os.getenv("RATE_LIMIT_RPM", "60")) | |
| RATE_BURST = int(os.getenv("RATE_LIMIT_BURST", "10")) | |
| RATE_VIP_IPS = set(os.getenv("RATE_VIP_IPS", "127.0.0.1").split(",")) | |
| # RAM thresholds — auto-calculated as % of total RAM if not set explicitly | |
| # RAM thresholds — computed dynamically in watchdog_task() based on actual machine RAM | |
| RAM_WARN_GB = float(os.getenv("RAM_WARN_GB", "0")) # 0 = auto 82% of total RAM | |
| RAM_REJECT_GB = float(os.getenv("RAM_REJECT_GB", "0")) # 0 = auto 90% of total RAM | |
| RAM_FLUSH_GB = float(os.getenv("RAM_FLUSH_GB", "0")) # 0 = auto 95% of total RAM | |
| REQUEST_TIMEOUT = float(os.getenv("REQUEST_TIMEOUT","90.0")) | |
| THINKING_TIMEOUT = float(os.getenv("THINKING_TIMEOUT","600.0")) # thinking mode — CPU needs time | |
| GENERATE_TIMEOUT = float(os.getenv("GENERATE_TIMEOUT","600.0")) # file generation (big code files) | |
| QUEUE_TIMEOUT = float(os.getenv("QUEUE_TIMEOUT", "30.0")) | |
| ENRICH_TIMEOUT = float(os.getenv("ENRICH_TIMEOUT", "12.0")) | |
| DOWNLOAD_TIMEOUT = float(os.getenv("DOWNLOAD_TIMEOUT","30.0")) | |
| # ── Single source of truth for system prompt ───────────────────────────────── | |
| # startup.sh reads this via: python3 -c "from app import DEFAULT_SYSTEM; print(DEFAULT_SYSTEM)" | |
| # NEVER duplicate this string anywhere else. | |
| DEFAULT_SYSTEM = ( | |
| "You are a universal AI assistant. " | |
| "Handle text, images, audio, video, PDFs, and web URLs. " | |
| "When tools are provided emit tool_calls precisely — never execute them. " | |
| "Think step by step when needed. Respond in the user language. " | |
| "Be precise and concise." | |
| ) | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", | |
| datefmt="%Y-%m-%dT%H:%M:%S", | |
| ) | |
| logger = logging.getLogger("mythical") | |
| # Ready flag: set True only after llama-server is confirmed healthy | |
| _INFERENCE_READY = False | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # CONNECTION MANAGER | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| class ConnManager: | |
| def __init__(self, base_url: str, name: str = "srv"): | |
| self._base = base_url; self._name = name | |
| self._c: httpx.AsyncClient | None = None | |
| self._lock = asyncio.Lock() | |
| self.reconnects = 0; self.healthy = False | |
| def _build(self) -> httpx.AsyncClient: | |
| return httpx.AsyncClient( | |
| base_url=self._base, | |
| timeout=httpx.Timeout(connect=10.0, read=REQUEST_TIMEOUT, | |
| write=10.0, pool=5.0), | |
| limits=httpx.Limits(max_connections=10, | |
| max_keepalive_connections=5, | |
| keepalive_expiry=60.0), | |
| ) | |
| async def _get(self) -> httpx.AsyncClient: | |
| if not self._c or self._c.is_closed: | |
| async with self._lock: | |
| if not self._c or self._c.is_closed: | |
| self._c = self._build() | |
| return self._c | |
| async def _reset(self): | |
| async with self._lock: | |
| if self._c and not self._c.is_closed: | |
| await self._c.aclose() | |
| self._c = None; self.healthy = False | |
| self.reconnects += 1 | |
| logger.info(f"[conn:{self._name}] reset #{self.reconnects}") | |
| async def req(self, method: str, path: str, **kw) -> httpx.Response: | |
| for attempt in range(2): | |
| c = await self._get() | |
| try: | |
| r = await c.request(method, path, **kw) | |
| self.healthy = True; return r | |
| except (httpx.RemoteProtocolError, httpx.ConnectError) as e: | |
| if attempt == 0: | |
| logger.warning(f"[conn:{self._name}] {type(e).__name__} — resetting") | |
| await self._reset(); await asyncio.sleep(0.5) | |
| else: | |
| raise | |
| except httpx.TimeoutException: | |
| raise | |
| def stream(self, path: str, data: bytes, hdrs: dict): | |
| mgr = self | |
| class _Ctx: | |
| async def __aenter__(s): | |
| c = await mgr._get() | |
| s._cm = c.stream("POST", path, content=data, headers=hdrs) | |
| return await s._cm.__aenter__() | |
| async def __aexit__(s, *a): | |
| return await s._cm.__aexit__(*a) | |
| return _Ctx() | |
| async def close(self): | |
| if self._c and not self._c.is_closed: | |
| await self._c.aclose() | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # [FIX-06] STREAM GUARD — replaces fragile _sem_released flag | |
| # Context manager that guarantees semaphore released exactly once | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| class StreamGuard: | |
| """Wraps a streaming generator, ensures semaphore released exactly once.""" | |
| def __init__(self, gen: AsyncIterator[bytes], sem: "ObsSem"): | |
| self._gen = gen; self._sem = sem; self._released = False | |
| def release(self): | |
| if not self._released: | |
| self._released = True | |
| self._sem.release() | |
| async def __aiter__(self) -> AsyncIterator[bytes]: | |
| try: | |
| async for chunk in self._gen: | |
| yield chunk | |
| finally: | |
| self.release() | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # OBSERVABLE SEMAPHORE | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| class ObsSem: | |
| def __init__(self, n: int): | |
| self._s = asyncio.Semaphore(n) | |
| self.active = 0; self.waiting = 0 | |
| async def acquire(self): | |
| self.waiting += 1 | |
| try: | |
| await self._s.acquire() | |
| finally: | |
| self.waiting -= 1 | |
| self.active += 1 | |
| def release(self): | |
| if self.active > 0: | |
| self.active -= 1 | |
| self._s.release() | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # RATE LIMITER | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| class RateLimiter: | |
| def __init__(self, rpm: int, burst: int): | |
| self._rpm = rpm; self._burst = burst | |
| self._w: dict[str, collections.deque] = {} | |
| self.blocked = 0 | |
| def check(self, ip: str) -> tuple[bool, float]: | |
| if ip in RATE_VIP_IPS: return True, 0.0 | |
| now = time.monotonic(); ws = now - 60.0 | |
| if ip not in self._w: self._w[ip] = collections.deque() | |
| dq = self._w[ip] | |
| while dq and dq[0] < ws: dq.popleft() | |
| if len(dq) >= self._rpm + self._burst: | |
| retry = round(60.0 - (now - dq[0]) + 0.5, 1) | |
| self.blocked += 1; return False, max(retry, 1.0) | |
| dq.append(now); return True, 0.0 | |
| def cleanup(self): | |
| now = time.monotonic() | |
| stale = [ip for ip, dq in self._w.items() | |
| if not dq or dq[-1] < now - 120] | |
| for ip in stale: del self._w[ip] | |
| return len(stale) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # EXACT CACHE | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| class ExactCache: | |
| def __init__(self, ttl: int, max_size: int = 200): | |
| self._s: dict[str, tuple[bytes, float]] = {} | |
| self._ttl = ttl; self._max = max_size | |
| self.hits = 0; self.misses = 0 | |
| def _key(self, msgs: list, tools: list, n: int) -> str: | |
| return hashlib.sha256( | |
| orjson.dumps({"m": msgs, "t": tools, "n": n}, | |
| option=orjson.OPT_SORT_KEYS) | |
| ).hexdigest()[:16] | |
| def get(self, k: str) -> bytes | None: | |
| if k not in self._s: self.misses += 1; return None | |
| data, exp = self._s[k] | |
| if time.monotonic() > exp: | |
| del self._s[k]; self.misses += 1; return None | |
| self.hits += 1; return data | |
| def set(self, k: str, data: bytes): | |
| if len(self._s) >= self._max: | |
| oldest = min(self._s, key=lambda x: self._s[x][1]) | |
| del self._s[oldest] | |
| self._s[k] = (data, time.monotonic() + self._ttl) | |
| def cleanup(self) -> int: | |
| now = time.monotonic() | |
| exp = [k for k, (_, t) in self._s.items() if now > t] | |
| for k in exp: del self._s[k] | |
| return len(exp) | |
| def hit_rate(self) -> float: | |
| total = self.hits + self.misses | |
| return (self.hits / total * 100) if total else 0.0 | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # SEMANTIC CACHE [FIX-05: uses CACHE_TTL env var] | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| class SemCache: | |
| THRESHOLD = float(os.getenv("SEMANTIC_THRESHOLD", "0.92")) | |
| def __init__(self, max_size: int = 200): | |
| self._e: list = [] | |
| self._max = max_size | |
| self._ttl = CACHE_TTL * 2 # uses env var | |
| self.hits = 0; self.misses = 0 | |
| def _vec(text: str) -> dict: | |
| t = text.lower().strip(); ng: dict[str, int] = {} | |
| for i in range(max(0, len(t) - 3)): | |
| g = t[i:i+4]; ng[g] = ng.get(g, 0) + 1 | |
| total = sum(ng.values()) or 1 | |
| return {k: v/total for k, v in ng.items()} | |
| def _cos(a: dict, b: dict) -> float: | |
| dot = sum(a.get(k, 0) * v for k, v in b.items()) | |
| ma = math.sqrt(sum(v*v for v in a.values())) | |
| mb = math.sqrt(sum(v*v for v in b.values())) | |
| return dot / (ma * mb) if ma and mb else 0.0 | |
| def _text(self, msgs: list) -> str: | |
| parts = [] | |
| for m in msgs: | |
| c = m.get("content", "") | |
| if isinstance(c, str): parts.append(c) | |
| elif isinstance(c, list): | |
| for b in c: | |
| if b.get("type") == "text": | |
| parts.append(b.get("text", "")) | |
| return " ".join(parts)[:2000] | |
| def lookup(self, msgs: list) -> bytes | None: | |
| now = time.monotonic(); text = self._text(msgs) | |
| if len(text) < 8: self.misses += 1; return None | |
| qv = self._vec(text); best_s = 0.0; best_r = None | |
| for vec, resp, ts in reversed(self._e): | |
| if now - ts > self._ttl: continue | |
| s = self._cos(qv, vec) | |
| if s > best_s: best_s, best_r = s, resp | |
| if best_s >= self.THRESHOLD and best_r: | |
| self.hits += 1 | |
| logger.info(f"[semantic] HIT sim={best_s:.3f}") | |
| return best_r | |
| self.misses += 1; return None | |
| def store(self, msgs: list, resp: bytes): | |
| text = self._text(msgs) | |
| if not text: return | |
| self._e.append((self._vec(text), resp, time.monotonic())) | |
| if len(self._e) > self._max: self._e.pop(0) | |
| def cleanup(self) -> int: | |
| now = time.monotonic(); before = len(self._e) | |
| self._e = [(v, r, t) for v, r, t in self._e if now - t <= self._ttl] | |
| return before - len(self._e) | |
| def hit_rate(self) -> float: | |
| total = self.hits + self.misses | |
| return (self.hits / total * 100) if total else 0.0 | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # DEDUPLICATOR | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| class Dedup: | |
| def __init__(self): | |
| self._f: dict[str, asyncio.Future] = {} | |
| self.count = 0 | |
| async def run_once(self, key: str, fn) -> tuple[Any, bool]: | |
| if key in self._f: | |
| self.count += 1 | |
| try: | |
| r = await asyncio.wait_for(asyncio.shield(self._f[key]), 120.0) | |
| return r, True | |
| except (asyncio.TimeoutError, asyncio.CancelledError): | |
| pass | |
| fut = asyncio.get_running_loop().create_future() | |
| self._f[key] = fut | |
| try: | |
| r = await fn(); fut.set_result(r); return r, False | |
| except Exception as e: | |
| if not fut.done(): fut.set_exception(e) | |
| raise | |
| finally: | |
| self._f.pop(key, None) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # JOB QUEUE [FIX-11: asyncio.Lock on submit] | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| class JobStatus(str, enum.Enum): | |
| PENDING = "pending"; RUNNING = "running" | |
| DONE = "done"; FAILED = "failed" | |
| class Job: | |
| __slots__ = ("id","status","created_at","started_at","finished_at", | |
| "result","error","payload") | |
| def __init__(self, jid: str, payload: dict): | |
| self.id = jid; self.status = JobStatus.PENDING | |
| self.created_at = time.monotonic() | |
| self.started_at = self.finished_at = None | |
| self.result = self.error = None | |
| self.payload = payload | |
| def to_dict(self) -> dict: | |
| return { | |
| "job_id": self.id, | |
| "status": self.status.value, | |
| "elapsed_s": round((self.finished_at or time.monotonic()) - self.created_at, 2), | |
| "error": self.error, | |
| } | |
| class JobQ: | |
| TTL = int(os.getenv("JOB_TTL_SECONDS", "600")) | |
| MAX = int(os.getenv("MAX_JOBS", "50")) | |
| def __init__(self): | |
| self._jobs: dict[str, Job] = {} | |
| self._q = asyncio.Queue() | |
| self._lock = asyncio.Lock() # [FIX-11] | |
| self._task: asyncio.Task | None = None | |
| self.submitted = self.done = self.failed = 0 | |
| def start(self): | |
| self._task = asyncio.create_task(self._worker(), name="job_worker") | |
| async def _worker(self): | |
| logger.info("[jobs] Worker started.") | |
| while True: | |
| try: | |
| jid = await self._q.get() | |
| job = self._jobs.get(jid) | |
| if not job: continue | |
| job.status = JobStatus.RUNNING | |
| job.started_at = time.monotonic() | |
| try: | |
| resp = await asyncio.wait_for( | |
| llama.req("POST", "/v1/chat/completions", | |
| content=orjson.dumps(job.payload), | |
| headers={"Content-Type": "application/json"}), | |
| timeout=300.0) | |
| if resp.status_code != 200: | |
| raise RuntimeError(f"HTTP {resp.status_code}: {resp.text[:120]}") | |
| job.result = resp.content | |
| job.status = JobStatus.DONE | |
| self.done += 1 | |
| except Exception as e: | |
| job.error = str(e) | |
| job.status = JobStatus.FAILED | |
| self.failed += 1 | |
| finally: | |
| job.finished_at = time.monotonic() | |
| elapsed = job.finished_at - (job.started_at or job.finished_at) | |
| logger.info(f"[jobs] {jid} → {job.status} ({elapsed:.1f}s)") | |
| except asyncio.CancelledError: | |
| break | |
| except Exception as e: | |
| logger.error(f"[jobs] worker error: {e}", exc_info=True) | |
| async def submit(self, payload: dict) -> str: | |
| async with self._lock: # [FIX-11] atomic check+insert | |
| self._evict() | |
| if len(self._jobs) >= self.MAX: | |
| raise RuntimeError(f"Job queue full ({self.MAX} max). Retry later.") | |
| jid = f"job-{uuid.uuid4().hex[:12]}" | |
| self._jobs[jid] = Job(jid, payload) | |
| await self._q.put(jid) | |
| self.submitted += 1 | |
| return jid | |
| def get(self, jid: str) -> Job | None: | |
| return self._jobs.get(jid) | |
| def _evict(self): | |
| now = time.monotonic() | |
| done = {JobStatus.DONE, JobStatus.FAILED} | |
| old = [k for k, j in self._jobs.items() | |
| if j.status in done and (now - (j.finished_at or 0)) > self.TTL] | |
| for k in old: del self._jobs[k] | |
| def stop(self): | |
| if self._task: self._task.cancel() | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # OOM WATCHDOG | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # ── Container-aware memory helpers ─────────────────────────────────────────── | |
| def _get_container_mem_limit_gb() -> float: | |
| """Read container memory limit from cgroup (Docker / HF Spaces). | |
| Returns 0.0 if running on bare metal or limit is 'unlimited'.""" | |
| checks = [ | |
| ("/sys/fs/cgroup/memory.max", "max"), # cgroup v2 | |
| ("/sys/fs/cgroup/memory/memory.limit_in_bytes", None), # cgroup v1 | |
| ] | |
| for path, unlimited_sentinel in checks: | |
| try: | |
| raw = Path(path).read_text().strip() | |
| if raw == unlimited_sentinel: | |
| continue | |
| val = int(raw) | |
| # Sanity: must be between 256 MB and 512 GB to be a real limit | |
| if 256 * 1024 * 1024 <= val <= 512 * (1024 ** 3): | |
| return val / (1024 ** 3) | |
| except Exception: | |
| pass | |
| return 0.0 | |
| def _get_mem_used_gb() -> float: | |
| """Current memory usage — reads from cgroup when in a container, | |
| falls back to psutil system-wide.""" | |
| for path in ( | |
| "/sys/fs/cgroup/memory.current", # cgroup v2 | |
| "/sys/fs/cgroup/memory/memory.usage_in_bytes", # cgroup v1 | |
| ): | |
| try: | |
| return int(Path(path).read_text()) / (1024 ** 3) | |
| except Exception: | |
| pass | |
| return psutil.virtual_memory().used / (1024 ** 3) | |
| class _WD: | |
| level = "ok"; ram_gb = 0.0; flushes = 0; rejects = 0 | |
| wd = _WD() | |
| _last_flush_attempt: float = 0.0 | |
| _flush_backoff: float = 10.0 # start at 10s, doubles on failure, max 120s | |
| async def _flush_kv() -> bool: | |
| global _last_flush_attempt, _flush_backoff | |
| now = time.monotonic() | |
| # Backoff: don't hammer llama-server on repeated failures | |
| if now - _last_flush_attempt < _flush_backoff: | |
| return False | |
| _last_flush_attempt = now | |
| # Try 1: /slots/0 {"action":"erase"} (llama.cpp >= v0.3.x) | |
| try: | |
| r = await llama.req("POST", "/slots/0", | |
| content=orjson.dumps({"action": "erase"}), | |
| headers={"Content-Type": "application/json"}, | |
| timeout=httpx.Timeout(5.0)) | |
| if r.status_code in (200, 204): | |
| logger.info("[wd] KV flushed via /slots/0") | |
| _flush_backoff = 10.0 # reset on success | |
| return True | |
| # 400 = endpoint not supported in this build | |
| if r.status_code == 400: | |
| logger.debug("[wd] /slots/0 not supported (400) — KV flush unavailable in this llama.cpp build") | |
| _flush_backoff = min(_flush_backoff * 2, 120.0) | |
| return False | |
| except Exception as e: | |
| logger.debug(f"[wd] /slots/0 failed: {e}") | |
| # Try 2: /cache/clear (older llama.cpp) | |
| try: | |
| r = await llama.req("POST", "/cache/clear", timeout=httpx.Timeout(5.0)) | |
| if r.status_code in (200, 204): | |
| logger.info("[wd] KV flushed via /cache/clear") | |
| _flush_backoff = 10.0 | |
| return True | |
| except Exception: | |
| pass | |
| _flush_backoff = min(_flush_backoff * 2, 120.0) | |
| return False | |
| async def watchdog_task(): | |
| # ── Step 1: Determine the "effective total" for threshold math ──────────── | |
| _sys_total = psutil.virtual_memory().total / (1024 ** 3) | |
| _container = _get_container_mem_limit_gb() | |
| _base = _container if _container else _sys_total | |
| # ── Step 2: Auto-thresholds as % of effective base ──────────────────────── | |
| _auto_warn = round(_base * 0.82, 1) | |
| _auto_reject = round(_base * 0.90, 1) | |
| _auto_flush = round(_base * 0.95, 1) | |
| # ── Step 3: Accept env-var overrides ONLY when they look sensible ───────── | |
| # Any env var < 30% of base is almost certainly a stale tiny value from an | |
| # old config (e.g. 11 / 13 / 14 GB on a 124 GB host) → ignore it. | |
| _floor = _base * 0.30 | |
| _warn = RAM_WARN_GB if RAM_WARN_GB >= _floor else _auto_warn | |
| _reject = RAM_REJECT_GB if RAM_REJECT_GB >= _floor else _auto_reject | |
| _flush = RAM_FLUSH_GB if RAM_FLUSH_GB >= _floor else _auto_flush | |
| logger.info( | |
| f"[wd] sys={_sys_total:.1f}G " | |
| f"{'container=' + f'{_container:.1f}G ' if _container else ''}" | |
| f"warn={_warn}G reject={_reject}G flush={_flush}G" | |
| ) | |
| _flush_cooldown = 30.0 # minimum seconds between flush attempts | |
| _last_flush_t = 0.0 | |
| while True: | |
| try: | |
| await asyncio.sleep(2.0) | |
| used = _get_mem_used_gb() | |
| wd.ram_gb = used | |
| if used >= _flush: | |
| prev = wd.level | |
| if prev != "flush": | |
| logger.warning(f"[wd] 🚨 FLUSH {used:.1f}/{_base:.0f}GB") | |
| wd.level = "flush"; wd.flushes += 1 | |
| # Only attempt GC + KV flush if cooldown has elapsed | |
| now = time.monotonic() | |
| if now - _last_flush_t >= _flush_cooldown: | |
| _last_flush_t = now | |
| gc.collect(2) | |
| exact_cache.cleanup(); sem_cache.cleanup() | |
| rl.cleanup(); jobs._evict() | |
| await _flush_kv() | |
| after = _get_mem_used_gb() | |
| freed = used - after | |
| if abs(freed) > 0.02: # only log if something actually moved | |
| logger.info(f"[wd] post-flush {after:.1f}GB (freed {freed:.1f}GB)") | |
| # Transition back if flush worked | |
| if _get_mem_used_gb() < _reject: | |
| wd.level = "ok" | |
| elif used >= _reject: | |
| if wd.level not in ("reject", "flush"): | |
| logger.warning(f"[wd] ⚠ REJECT {used:.1f}/{_base:.0f}GB") | |
| wd.level = "reject"; wd.rejects += 1; gc.collect(1) | |
| elif used >= _warn: | |
| if wd.level == "ok": | |
| logger.info(f"[wd] ⚡ WARN {used:.1f}/{_base:.0f}GB") | |
| wd.level = "warn" | |
| else: | |
| if wd.level != "ok": | |
| logger.info(f"[wd] ✅ OK {used:.1f}GB") | |
| wd.level = "ok" | |
| except asyncio.CancelledError: | |
| break | |
| except Exception as e: | |
| logger.error(f"[wd] {e}", exc_info=True) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # GLOBAL INSTANCES | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| llama = ConnManager(LLAMA_URL, "llama") | |
| whisper_c = ConnManager(WHISPER_URL, "whisper") | |
| sem = ObsSem(2) | |
| rl = RateLimiter(RATE_LIMIT_RPM, RATE_BURST) | |
| exact_cache = ExactCache(CACHE_TTL) | |
| sem_cache = SemCache() | |
| dedup = Dedup() | |
| jobs = JobQ() | |
| mgr = ModelManager() | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # MEDIA PROCESSING | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| def _resize_img(data: bytes) -> str | None: | |
| try: | |
| from PIL import Image | |
| img = Image.open(io.BytesIO(data)).convert("RGB") | |
| if max(img.size) > IMAGE_MAX_PX: | |
| img.thumbnail((IMAGE_MAX_PX, IMAGE_MAX_PX), Image.LANCZOS) | |
| buf = io.BytesIO() | |
| img.save(buf, "JPEG", quality=85, optimize=True) | |
| return base64.b64encode(buf.getvalue()).decode() | |
| except Exception as e: | |
| logger.warning(f"[img] {e}"); return None | |
| async def resize_img(data: bytes) -> str | None: | |
| loop = asyncio.get_running_loop() | |
| return await loop.run_in_executor(None, _resize_img, data) | |
| # [FIX-02] async chunked read to avoid blocking event loop | |
| async def read_upload_chunked(file: UploadFile, max_mb: int = MAX_FILE_MB) -> bytes | None: | |
| """Read upload in chunks on thread executor — never blocks asyncio loop.""" | |
| max_bytes = max_mb * 1024 * 1024 | |
| loop = asyncio.get_running_loop() | |
| chunks = [] | |
| total = 0 | |
| while True: | |
| chunk = await loop.run_in_executor(None, file.file.read, 65536) | |
| if not chunk: | |
| break | |
| total += len(chunk) | |
| if total > max_bytes: | |
| logger.warning(f"[upload] File exceeds {max_mb}MB limit") | |
| return None | |
| chunks.append(chunk) | |
| return b"".join(chunks) | |
| async def convert_audio(data: bytes, fmt: str = "mp3") -> str | None: | |
| if len(data) > MAX_FILE_MB * 1024 * 1024: return None | |
| Path("/tmp/media").mkdir(parents=True, exist_ok=True) | |
| inp = f"/tmp/media/{uuid.uuid4().hex}.{fmt}" | |
| out = f"/tmp/media/{uuid.uuid4().hex}.wav" | |
| try: | |
| Path(inp).write_bytes(data) | |
| p = await asyncio.create_subprocess_exec( | |
| "ffmpeg", "-y", "-i", inp, | |
| "-ar", "16000", "-ac", "1", "-c:a", "pcm_s16le", "-t", "60", | |
| out, | |
| stdout=asyncio.subprocess.DEVNULL, | |
| stderr=asyncio.subprocess.DEVNULL) | |
| await asyncio.wait_for(p.communicate(), timeout=45.0) | |
| if p.returncode != 0: return None | |
| loop = asyncio.get_running_loop() | |
| wav = await loop.run_in_executor(None, Path(out).read_bytes) | |
| return base64.b64encode(wav).decode() | |
| except Exception as e: | |
| logger.warning(f"[audio] {e}"); return None | |
| finally: | |
| for f in (inp, out): | |
| try: os.unlink(f) | |
| except: pass | |
| async def extract_frames(data: bytes) -> list[str]: | |
| if len(data) > MAX_FILE_MB * 1024 * 1024: return [] | |
| import shutil | |
| td = Path(f"/tmp/media/{uuid.uuid4().hex}") | |
| td.mkdir(parents=True) | |
| try: | |
| (td / "v.mp4").write_bytes(data) | |
| fps = VIDEO_MAX_FRAMES / 60.0 | |
| p = await asyncio.create_subprocess_exec( | |
| "ffmpeg", "-y", "-i", str(td / "v.mp4"), | |
| "-vf", f"fps={fps:.4f},scale=560:-1", "-q:v", "4", | |
| str(td / "f%04d.jpg"), | |
| stdout=asyncio.subprocess.DEVNULL, | |
| stderr=asyncio.subprocess.DEVNULL) | |
| await asyncio.wait_for(p.communicate(), timeout=60.0) | |
| frames: list[str] = [] | |
| loop = asyncio.get_running_loop() | |
| for f in sorted(td.glob("f*.jpg"))[:VIDEO_MAX_FRAMES]: | |
| raw = await loop.run_in_executor(None, f.read_bytes) | |
| b64 = await resize_img(raw) | |
| if b64: frames.append(f"data:image/jpeg;base64,{b64}") | |
| return frames | |
| except Exception as e: | |
| logger.warning(f"[video] {e}"); return [] | |
| finally: | |
| shutil.rmtree(td, ignore_errors=True) | |
| # [FIX-08] graceful handling for encrypted/corrupt PDFs | |
| def _pdf_text(data: bytes, max_chars: int = 20000) -> str | None: | |
| try: | |
| from pypdf import PdfReader | |
| try: | |
| reader = PdfReader(io.BytesIO(data)) | |
| except Exception: | |
| return "[PDF Error: file is encrypted, corrupted, or not a valid PDF]" | |
| if reader.is_encrypted: | |
| # try empty password | |
| try: | |
| reader.decrypt("") | |
| except Exception: | |
| return "[PDF Error: file is password-protected. Please provide an unlocked PDF]" | |
| text = "" | |
| for page in reader.pages: | |
| try: | |
| text += (page.extract_text() or "") | |
| except Exception: | |
| continue | |
| if len(text) > max_chars: | |
| break | |
| return text[:max_chars] if text.strip() else "[PDF: no extractable text found (may be image-based)]" | |
| except Exception as e: | |
| return f"[PDF processing error: {e}]" | |
| def _yt_id(url: str) -> str | None: | |
| p = urllib.parse.urlparse(url) | |
| if p.hostname == "youtu.be": return p.path[1:] | |
| if p.hostname in ("www.youtube.com", "youtube.com"): | |
| if p.path == "/watch": | |
| return urllib.parse.parse_qs(p.query).get("v", [None])[0] | |
| if p.path.startswith("/shorts/"): | |
| return p.path.split("/")[2] | |
| return None | |
| async def _yt_transcript(vid: str) -> str | None: | |
| try: | |
| from youtube_transcript_api import YouTubeTranscriptApi | |
| loop = asyncio.get_running_loop() | |
| t = await asyncio.wait_for( | |
| loop.run_in_executor(None, lambda: YouTubeTranscriptApi.get_transcript( | |
| vid, languages=["ar", "ar-SA", "en", "en-US"])), | |
| timeout=10.0) | |
| return " ".join(x["text"] for x in t)[:15000] | |
| except asyncio.TimeoutError: | |
| return "[YouTube transcript timed out]" | |
| except Exception: | |
| return None | |
| async def _fetch_url(url: str) -> str | None: | |
| try: | |
| import aiohttp | |
| from bs4 import BeautifulSoup | |
| async with aiohttp.ClientSession() as s: | |
| async with s.get(url, | |
| headers={"User-Agent": "Mozilla/5.0 (compatible; MythicalBot/7.0)"}, | |
| timeout=aiohttp.ClientTimeout(total=8), | |
| allow_redirects=True) as r: | |
| if r.status == 200: | |
| soup = BeautifulSoup(await r.text(errors="replace"), "html.parser") | |
| for t in soup(["script", "style", "nav", "footer", "aside"]): t.decompose() | |
| return soup.get_text(" ", strip=True)[:15000] | |
| except Exception: | |
| return None | |
| async def web_search(query: str, max_results: int = 5) -> list[dict]: | |
| """DuckDuckGo HTML search — free, no API key.""" | |
| try: | |
| import aiohttp | |
| from bs4 import BeautifulSoup | |
| url = f"https://html.duckduckgo.com/html/?q={urllib.parse.quote(query)}" | |
| async with aiohttp.ClientSession() as s: | |
| async with s.get(url, | |
| headers={"User-Agent": "Mozilla/5.0"}, | |
| timeout=aiohttp.ClientTimeout(total=10)) as resp: | |
| if resp.status != 200: return [] | |
| soup = BeautifulSoup(await resp.text(), "html.parser") | |
| results = [] | |
| for r in soup.select(".result__body")[:max_results]: | |
| title = r.select_one(".result__title") | |
| snippet = r.select_one(".result__snippet") | |
| link = r.select_one(".result__url") | |
| results.append({ | |
| "title": title.get_text(strip=True) if title else "", | |
| "snippet": snippet.get_text(strip=True) if snippet else "", | |
| "url": link.get_text(strip=True) if link else "", | |
| }) | |
| return results | |
| except Exception as e: | |
| logger.warning(f"[search] {e}"); return [] | |
| # [FIX-04] enrich_error creates user-visible message for empty search | |
| async def enrich(messages: list[dict]) -> list[dict]: | |
| """Process all media/URL content with hard timeout.""" | |
| try: | |
| return await asyncio.wait_for(_enrich_inner(messages), timeout=ENRICH_TIMEOUT) | |
| except asyncio.TimeoutError: | |
| logger.warning(f"[enrich] Timeout after {ENRICH_TIMEOUT}s") | |
| return messages | |
| async def _enrich_inner(messages: list[dict]) -> list[dict]: | |
| result = [] | |
| for msg in messages: | |
| content = msg.get("content") | |
| if isinstance(content, str): | |
| urls = re.findall(r"https?://\S+", content) | |
| if urls: | |
| url = urls[0] | |
| yt = _yt_id(url) | |
| if yt: | |
| t = await _yt_transcript(yt) | |
| if t: content += f"\n\n[YouTube Transcript]:\n{t}" | |
| elif any(url.lower().endswith(e) for e in | |
| [".jpg",".jpeg",".png",".webp",".gif"]): | |
| try: | |
| import aiohttp | |
| async with aiohttp.ClientSession() as s: | |
| async with s.get(url, timeout=aiohttp.ClientTimeout(total=10)) as r: | |
| if r.status == 200: | |
| b64 = await resize_img(await r.read()) | |
| if b64: | |
| result.append({**msg, "content": [ | |
| {"type": "text", "text": content}, | |
| {"type": "image_url", | |
| "image_url": {"url": f"data:image/jpeg;base64,{b64}"}} | |
| ]}) | |
| continue | |
| except Exception: | |
| pass | |
| else: | |
| web = await _fetch_url(url) | |
| if web: | |
| content += f"\n\n[Web Content from {url}]:\n{web}" | |
| result.append({**msg, "content": content}) | |
| continue | |
| if isinstance(content, list): | |
| new = [] | |
| for block in content: | |
| bt = block.get("type", "") | |
| if bt == "image_url": | |
| uv = block.get("image_url", {}).get("url", "") | |
| if uv.startswith("data:"): | |
| try: | |
| _, b = uv.split(",", 1) | |
| raw = base64.b64decode(b) | |
| if len(raw) > MAX_FILE_MB * 1024 * 1024: | |
| new.append({"type":"text","text":f"[Image rejected: exceeds {MAX_FILE_MB}MB limit]"}) | |
| continue | |
| b64 = await resize_img(raw) | |
| new.append({"type":"image_url","image_url": | |
| {"url":f"data:image/jpeg;base64,{b64 or b}"}}) | |
| except Exception as e: | |
| new.append({"type":"text","text":f"[Image error: {e}]"}) | |
| else: | |
| new.append({"type":"text","text":"[External image URLs not supported — please convert to base64 data URI]"}) | |
| elif bt == "input_audio": | |
| info = block.get("input_audio", {}) | |
| fmt = info.get("format", "wav") | |
| data_b64 = info.get("data", "") | |
| if data_b64 and fmt != "wav": | |
| wav = await convert_audio(base64.b64decode(data_b64), fmt) | |
| new.append({"type":"input_audio", | |
| "input_audio":{"data": wav or data_b64, "format":"wav"}}) | |
| else: | |
| new.append(block) | |
| elif bt == "video_url": | |
| uv = block.get("video_url", {}).get("url", "") | |
| if uv.startswith("data:"): | |
| try: | |
| _, b = uv.split(",", 1) | |
| frames = await extract_frames(base64.b64decode(b)) | |
| if frames: | |
| new.append({"type":"text","text":f"[Video: {len(frames)} frames extracted]"}) | |
| for i, f in enumerate(frames): | |
| new.append({"type":"text","text":f"Frame {i+1}/{len(frames)}:"}) | |
| new.append({"type":"image_url","image_url":{"url":f}}) | |
| else: | |
| new.append({"type":"text","text":"[Video: could not extract frames — check format/size]"}) | |
| except Exception as e: | |
| new.append({"type":"text","text":f"[Video error: {e}]"}) | |
| else: | |
| new.append(block) | |
| elif "pdf" in block.get("image_url", {}).get("url", ""): | |
| try: | |
| _, b = block["image_url"]["url"].split(",", 1) | |
| loop = asyncio.get_running_loop() | |
| text = await loop.run_in_executor(None, _pdf_text, base64.b64decode(b)) | |
| new.append({"type":"text","text":f"[PDF Content]:\n{text}"}) | |
| except Exception as e: | |
| new.append({"type":"text","text":f"[PDF error: {e}]"}) | |
| else: | |
| new.append(block) | |
| result.append({**msg, "content": new}) | |
| continue | |
| result.append(msg) | |
| return result | |
| def count_tokens(messages: list) -> int: | |
| total = 0 | |
| for m in messages: | |
| c = m.get("content", "") | |
| if isinstance(c, str): total += len(c) | |
| elif isinstance(c, list): | |
| for b in c: | |
| if b.get("type") == "text": total += len(b.get("text","")) | |
| elif b.get("type") == "image_url": total += 1500 | |
| elif b.get("type") == "input_audio": total += 3000 | |
| total += 16 | |
| return int(total / 3.5) | |
| def surgeon(messages: list, max_tok: int) -> tuple[list, int, int]: | |
| orig = count_tokens(messages) | |
| if orig <= max_tok: return messages, orig, orig | |
| sys_m = [m for m in messages if m.get("role") == "system"] | |
| conv = [m for m in messages if m.get("role") != "system"] | |
| while len(conv) > 2: | |
| conv.pop(len(conv) // 2) | |
| if count_tokens(sys_m + conv) <= max_tok: break | |
| if count_tokens(sys_m + conv) > max_tok and conv: | |
| conv = conv[-1:] | |
| res = sys_m + conv | |
| return res, orig, count_tokens(res) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # LIFESPAN | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| async def lifespan(app: FastAPI) -> AsyncIterator[None]: | |
| global _INFERENCE_READY | |
| logger.info("══ MYTHICAL UNIVERSAL SYSTEM v7.0 starting ══") | |
| wd_t = asyncio.create_task(watchdog_task(), name="oom_watchdog") | |
| jobs.start() | |
| mgr.start_updater() | |
| # [FIX-01] Poll until llama-server is actually healthy before accepting inference | |
| async def _wait_for_inference(): | |
| global _INFERENCE_READY | |
| for _ in range(120): # max 6 minutes | |
| try: | |
| r = await llama.req("GET", "/health", timeout=httpx.Timeout(3.0)) | |
| if r.status_code == 200: | |
| _INFERENCE_READY = True | |
| logger.info("✅ Inference engine ready — accepting all requests") | |
| return | |
| except Exception: | |
| pass | |
| await asyncio.sleep(3) | |
| logger.warning("⚠ Inference engine not ready after 6min — requests will fail gracefully") | |
| _INFERENCE_READY = True # allow through, will get 502 with proper error | |
| asyncio.create_task(_wait_for_inference(), name="inference_ready_probe") | |
| logger.info("Services: Watchdog ✓ | Jobs ✓ | ModelUpdater ✓ | ReadyProbe ✓") | |
| yield | |
| logger.info("Shutting down...") | |
| wd_t.cancel(); jobs.stop(); mgr.stop() | |
| try: await asyncio.wait_for(asyncio.shield(wd_t), 3.0) | |
| except (asyncio.CancelledError, asyncio.TimeoutError): pass | |
| await llama.close(); await whisper_c.close() | |
| logger.info("Clean shutdown complete.") | |
| app = FastAPI(title="Mythical Universal System", version="7.0.0", | |
| lifespan=lifespan, default_response_class=Response) | |
| app.add_middleware(CORSMiddleware, allow_origins=["*"], | |
| allow_methods=["*"], allow_headers=["*"], | |
| allow_credentials=True) | |
| def J(data: Any, code: int = 200) -> Response: | |
| return Response(orjson.dumps(data), code, media_type="application/json") | |
| def E(msg: str, code: int = 500, t: str = "error", rid: str | None = None) -> Response: | |
| return J({"error": {"message": msg, "type": t, "code": code}, | |
| "id": rid or f"err-{uuid.uuid4().hex[:8]}"}, code) | |
| def auth(r: Request) -> bool: | |
| return r.headers.get("Authorization","").replace("Bearer ","").strip() == API_KEY | |
| def client_ip(r: Request) -> str: | |
| return r.headers.get("X-Forwarded-For", | |
| r.client.host or "0.0.0.0").split(",")[0].strip() | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # HEALTH & MONITORING | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| async def health() -> Response: | |
| used = _get_mem_used_gb() | |
| _climit = _get_container_mem_limit_gb() | |
| _sys_total = psutil.virtual_memory().total / (1024 ** 3) | |
| _total = _climit if _climit else _sys_total | |
| ram_pct = round(used / _total * 100, 1) if _total > 0 else 0.0 | |
| ll_ok = False | |
| try: | |
| r = await llama.req("GET", "/health", timeout=httpx.Timeout(3.0)) | |
| ll_ok = r.status_code == 200 | |
| except Exception: | |
| pass | |
| ok = ll_ok and wd.level in ("ok","warn","flush") # flush = high RAM but still working | |
| return J({ | |
| "status": "healthy" if ok else "degraded", | |
| "inference_ready": _INFERENCE_READY, | |
| "llama_server": "ok" if ll_ok else "error", | |
| "ram_gb": round(used, 2), | |
| "ram_pct": ram_pct, | |
| "watchdog": wd.level, | |
| "queue_active": sem.active, | |
| "queue_waiting": sem.waiting, | |
| "cache_hit_rate": f"{exact_cache.hit_rate:.1f}%", | |
| "sem_hit_rate": f"{sem_cache.hit_rate:.1f}%", | |
| "conn_reconnects": llama.reconnects, | |
| "model": mgr.status()["active"], | |
| "capabilities": { | |
| "text":True, "images":True, "audio":True, "video":True, | |
| "pdf":True, "youtube":True, "urls":True, "web_search":True, | |
| "tool_calls":True, "streaming":True, "thinking_mode":True, | |
| "async_jobs":True, "file_upload":True, "file_generation":True, | |
| }, | |
| }, 200 if ok else 503) | |
| # [FIX-01] /ready endpoint — used by n8n/orchestrators to wait for inference | |
| async def ready() -> Response: | |
| """Returns 200 only when inference engine is confirmed healthy.""" | |
| if not _INFERENCE_READY: | |
| return J({"ready": False, "message": "Inference engine still loading..."}, 503) | |
| ll_ok = False | |
| try: | |
| r = await llama.req("GET", "/health", timeout=httpx.Timeout(2.0)) | |
| ll_ok = r.status_code == 200 | |
| except Exception: | |
| pass | |
| if ll_ok: | |
| return J({"ready": True}) | |
| return J({"ready": False, "message": "Inference engine not responding"}, 503) | |
| async def metrics() -> Response: | |
| used = _get_mem_used_gb() | |
| _cl = _get_container_mem_limit_gb() | |
| _tot = _cl if _cl else psutil.virtual_memory().total / (1024 ** 3) | |
| pct = round(used / _tot * 100, 1) if _tot > 0 else 0.0 | |
| lines = [ | |
| f"ram_gb {used:.3f}", f"ram_pct {pct}", | |
| f"watchdog_flushes {wd.flushes}", f"watchdog_rejects {wd.rejects}", | |
| f"queue_active {sem.active}", f"queue_waiting {sem.waiting}", | |
| f"exact_cache_hits {exact_cache.hits}", f"exact_cache_misses {exact_cache.misses}", | |
| f"sem_cache_hits {sem_cache.hits}", f"rate_blocked {rl.blocked}", | |
| f"jobs_submitted {jobs.submitted}", f"jobs_done {jobs.done}", | |
| f"jobs_failed {jobs.failed}", f"conn_reconnects {llama.reconnects}", | |
| f"dedup_count {dedup.count}", f"inference_ready {int(_INFERENCE_READY)}", | |
| ] | |
| return Response("\n".join(lines), media_type="text/plain") | |
| async def model_list() -> Response: | |
| return J({"object":"list","data":[ | |
| {"id":"mythical","object":"model","owned_by":"mythical-system", | |
| "capabilities":["text","vision","audio","function_calling"]}]}) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # CHAT COMPLETIONS | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| async def chat(request: Request) -> Response | StreamingResponse: | |
| if not auth(request): return E("Unauthorized",401,"auth_error") | |
| rid = f"chatcmpl-{uuid.uuid4().hex[:12]}"; t0 = time.monotonic() | |
| ip = client_ip(request) | |
| # Rate limit | |
| allowed, retry = rl.check(ip) | |
| if not allowed: | |
| return Response( | |
| orjson.dumps({"error":{"message":f"Rate limit exceeded. Retry in {retry}s."}}), | |
| 429, media_type="application/json", | |
| headers={"Retry-After": str(int(retry))}) | |
| # OOM gate | |
| if wd.level == "reject": | |
| return E(f"Memory pressure ({wd.ram_gb:.1f}GB). Retry in 30s.", 503, "server_overloaded") | |
| # [FIX-01] Inference ready gate | |
| if not _INFERENCE_READY: | |
| return E("Inference engine is still loading. Check /ready endpoint.", 503, "not_ready") | |
| try: | |
| body = orjson.loads(await request.body()) | |
| except Exception as e: | |
| return E(f"Invalid JSON: {e}", 400, "invalid_request") | |
| messages: list = body.get("messages", []) | |
| if not messages: | |
| return E("'messages' field is required", 400, "invalid_request") | |
| # Vision pre-check: reject image content early if model has no mmproj | |
| # (avoids llama-server returning a confusing 500) | |
| _has_image = any( | |
| isinstance(m.get("content"), list) and | |
| any(p.get("type") in ("image_url","image") for p in m["content"]) | |
| for m in messages | |
| ) | |
| if _has_image and not mgr.cfg.mmproj_path: | |
| return E( | |
| "Vision not available: this model was loaded without an mmproj file. " | |
| "Use a vision-capable model or send text-only messages.", | |
| 400, "unsupported_media_type" | |
| ) | |
| if not any(m.get("role") == "system" for m in messages): | |
| messages = [{"role":"system","content":DEFAULT_SYSTEM}] + messages | |
| messages = await enrich(messages) | |
| messages, orig, final = surgeon(messages, MAX_CTX_TOKENS) | |
| if orig > final: | |
| logger.info(f"[{rid}] surgeon {orig}→{final} tokens") | |
| req_max = int(body.get("max_tokens") or MAX_NEW_TOKENS) | |
| # Token budget: reduce under memory pressure (based on watchdog level, not hardcoded GBs) | |
| factor = 0.5 if wd.level in ("flush", "reject") else 0.75 if wd.level == "warn" else 1.0 | |
| max_tok = max(int(min(req_max, MAX_NEW_TOKENS) * factor), 128) | |
| is_stream= bool(body.get("stream", False)) | |
| temp = float(body.get("temperature", 0.6)) | |
| use_cache= (temp == 0.0 or bool(body.get("use_cache"))) and not is_stream | |
| logger.info(f"[{rid}] {ip} | {final}tok | " | |
| f"tools={len(body.get('tools',[]))} | stream={is_stream} | " | |
| f"thinking={body.get('thinking',False)} | q={sem.active+sem.waiting}") | |
| # Cache check | |
| cache_key: str | None = None | |
| if use_cache: | |
| cache_key = exact_cache._key(messages, body.get("tools",[]), max_tok) | |
| if (hit := exact_cache.get(cache_key)): | |
| logger.info(f"[{rid}] EXACT HIT ({exact_cache.hit_rate:.0f}%)") | |
| return Response(hit, 200, media_type="application/json", | |
| headers={"X-Request-ID":rid,"X-Cache":"EXACT-HIT"}) | |
| if (sh := sem_cache.lookup(messages)): | |
| return Response(sh, 200, media_type="application/json", | |
| headers={"X-Request-ID":rid,"X-Cache":"SEMANTIC-HIT"}) | |
| # Build payload | |
| # Thinking mode OFF by default — Qwen3 thinks for EVERY request otherwise | |
| # User must explicitly pass "thinking": true to enable reasoning | |
| thinking_requested = body.get("thinking", False) | |
| payload: dict = { | |
| "messages": messages, | |
| "cache_prompt": True, | |
| "id_slot": 0, | |
| "max_tokens": max_tok, | |
| "temperature": temp, | |
| "top_p": float(body.get("top_p", 0.95)), | |
| "top_k": int(body.get("top_k", 20)), | |
| "min_p": float(body.get("min_p", 0.0)), | |
| "stream": is_stream, | |
| # Always set thinking explicitly to avoid Qwen3 auto-enabling it | |
| "chat_template_kwargs": {"enable_thinking": bool(thinking_requested)}, | |
| } | |
| if thinking_requested: | |
| _think_budget = int(body.get("thinking_budget", 512)) # 512 default — 4096 is too slow on CPU | |
| payload["reasoning_budget"] = _think_budget | |
| # CRITICAL: max_tokens must cover thinking budget + actual answer | |
| # Without this, the <think> block eats all tokens and content is empty | |
| max_tok = max(max_tok, _think_budget + 256) | |
| payload["max_tokens"] = max_tok | |
| if body.get("tools"): | |
| payload["tools"] = body["tools"] | |
| payload["tool_choice"] = body.get("tool_choice", "auto") | |
| payload["parallel_tool_calls"] = body.get("parallel_tool_calls", True) | |
| for p in ("stop","presence_penalty","frequency_penalty","seed"): | |
| if p in body: payload[p] = body[p] | |
| # Semaphore | |
| try: | |
| await asyncio.wait_for(sem.acquire(), timeout=QUEUE_TIMEOUT) | |
| except asyncio.TimeoutError: | |
| return E(f"Queued {QUEUE_TIMEOUT:.0f}s — server busy. Retry shortly.", | |
| 503, "server_overloaded", rid) | |
| raw = orjson.dumps(payload) | |
| ct = {"Content-Type": "application/json"} | |
| try: | |
| # [FIX-06] Streaming uses StreamGuard — semaphore released exactly once | |
| if is_stream: | |
| async def _raw_stream() -> AsyncIterator[bytes]: | |
| async with llama.stream("/v1/chat/completions", raw, ct) as resp: | |
| if resp.status_code != 200: | |
| b = await resp.aread() | |
| yield b"data: " + orjson.dumps({ | |
| "error":{"message":"Upstream error","code":resp.status_code} | |
| }) + b"\n\n" | |
| return | |
| async for chunk in resp.aiter_bytes(256): | |
| if chunk: yield chunk | |
| guard = StreamGuard(_raw_stream(), sem) | |
| async def _guarded_stream() -> AsyncIterator[bytes]: | |
| try: | |
| async for chunk in guard: | |
| yield chunk | |
| except httpx.TimeoutException: | |
| yield b"data: [DONE]\n\n" | |
| except Exception as e: | |
| logger.error(f"[{rid}] stream error: {e}") | |
| yield b"data: [DONE]\n\n" | |
| finally: | |
| guard.release() # idempotent via StreamGuard | |
| logger.info(f"[{rid}] stream done {time.monotonic()-t0:.2f}s") | |
| return StreamingResponse(_guarded_stream(), | |
| media_type="text/event-stream", | |
| headers={"X-Request-ID":rid, | |
| "Cache-Control":"no-cache"}) | |
| # Blocking | |
| try: | |
| # Thinking mode needs a much longer budget — use THINKING_TIMEOUT | |
| _req_timeout = THINKING_TIMEOUT if thinking_requested else REQUEST_TIMEOUT | |
| _http_timeout = httpx.Timeout( | |
| connect=10.0, read=_req_timeout + 30.0, write=10.0, pool=5.0 | |
| ) | |
| async def _do(): | |
| return await asyncio.wait_for( | |
| llama.req("POST", "/v1/chat/completions", | |
| content=raw, headers=ct, timeout=_http_timeout), | |
| timeout=_req_timeout) | |
| resp, deduped = await dedup.run_once(cache_key or rid, _do) | |
| if deduped: logger.info(f"[{rid}] DEDUP HIT") | |
| except asyncio.TimeoutError: | |
| await llama._reset() # close broken connection before next request uses it | |
| return E(f"Inference timeout {_req_timeout:.0f}s", 504, "timeout", rid) | |
| except httpx.RequestError as e: | |
| return E(f"Upstream error: {e}",502,"server_error",rid) | |
| elapsed = time.monotonic() - t0 | |
| logger.info(f"[{rid}] done {elapsed:.2f}s HTTP {resp.status_code}") | |
| if use_cache and cache_key and resp.status_code == 200: | |
| exact_cache.set(cache_key, resp.content) | |
| sem_cache.store(messages, resp.content) | |
| return Response(resp.content, resp.status_code, | |
| media_type="application/json", | |
| headers={"X-Request-ID":rid, | |
| "X-Time":f"{elapsed:.3f}", | |
| "X-Cache":"MISS"}) | |
| except Exception as e: | |
| logger.error(f"[{rid}] unhandled: {e}", exc_info=True) | |
| return E(f"Internal: {type(e).__name__}", 500, "server_error", rid) | |
| finally: | |
| if not is_stream: | |
| sem.release() # safe: ObsSem.release guards active > 0 | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # FILE UPLOAD [FIX-02: chunked read] | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| async def file_upload( | |
| request: Request, | |
| file: UploadFile = File(...), | |
| prompt: str = Form(default="Analyze this file and describe its contents."), | |
| thinking: str = Form(default="false"), | |
| ) -> Response: | |
| """ | |
| Direct file upload — no base64 encoding needed. | |
| Supports: images, PDFs, audio, video, text/code files. | |
| curl -H "Authorization: Bearer KEY" \\ | |
| -F "file=@document.pdf" \\ | |
| -F "prompt=Summarize this document" \\ | |
| https://YOUR-SPACE.hf.space/v1/files | |
| """ | |
| if not auth(request): return E("Unauthorized",401) | |
| if not _INFERENCE_READY: return E("Inference engine still loading.",503,"not_ready") | |
| try: | |
| # [FIX-02] chunked read — never blocks asyncio loop | |
| data = await read_upload_chunked(file, MAX_FILE_MB) | |
| if data is None: | |
| return E(f"File too large. Maximum size: {MAX_FILE_MB}MB", 413, "file_too_large") | |
| ct_in = file.content_type or "" | |
| fname = file.filename or "upload" | |
| b64 = base64.b64encode(data).decode() | |
| if ct_in.startswith("image/") or any(fname.lower().endswith(e) | |
| for e in [".jpg",".jpeg",".png",".webp",".gif",".bmp"]): | |
| content = [ | |
| {"type":"text","text":prompt}, | |
| {"type":"image_url","image_url":{"url":f"data:{ct_in};base64,{b64}"}}, | |
| ] | |
| elif ct_in == "application/pdf" or fname.lower().endswith(".pdf"): | |
| content = [ | |
| {"type":"image_url","image_url":{"url":f"data:application/pdf;base64,{b64}"}}, | |
| {"type":"text","text":prompt}, | |
| ] | |
| elif ct_in.startswith("audio/") or any(fname.lower().endswith(e) | |
| for e in [".mp3",".wav",".ogg",".m4a",".webm",".flac"]): | |
| ext = fname.rsplit(".",1)[-1].lower() if "." in fname else "mp3" | |
| content = [ | |
| {"type":"input_audio","input_audio":{"data":b64,"format":ext}}, | |
| {"type":"text","text":prompt}, | |
| ] | |
| elif ct_in.startswith("video/") or any(fname.lower().endswith(e) | |
| for e in [".mp4",".mov",".avi",".mkv",".webm"]): | |
| content = [ | |
| {"type":"video_url","video_url":{"url":f"data:{ct_in};base64,{b64}"}}, | |
| {"type":"text","text":prompt}, | |
| ] | |
| else: | |
| # Text / code file | |
| try: | |
| text_content = data.decode("utf-8","replace")[:15000] | |
| except Exception: | |
| text_content = "[Binary file — cannot display as text]" | |
| content = [{"type":"text", | |
| "text":f"File: {fname} ({len(data)} bytes)\n\n{text_content}\n\n{prompt}"}] | |
| messages_raw = [ | |
| {"role":"system","content":DEFAULT_SYSTEM}, | |
| {"role":"user","content":content}, | |
| ] | |
| messages_enriched = await enrich(messages_raw) | |
| messages_final, _, _ = surgeon(messages_enriched, MAX_CTX_TOKENS) | |
| pl = { | |
| "messages": messages_final, | |
| "max_tokens": MAX_NEW_TOKENS, | |
| "temperature": 0.6, | |
| "cache_prompt": False, | |
| "stream": False, | |
| # Always set explicitly — without this Qwen3 may silently enter thinking mode | |
| "chat_template_kwargs": {"enable_thinking": thinking.lower() == "true"}, | |
| } | |
| if thinking.lower() == "true": | |
| _think_budget = 512 # same conservative default as chat endpoint | |
| pl["reasoning_budget"] = _think_budget | |
| pl["max_tokens"] = max(MAX_NEW_TOKENS, _think_budget + 256) | |
| _file_timeout = THINKING_TIMEOUT if thinking.lower() == "true" else REQUEST_TIMEOUT | |
| resp = await asyncio.wait_for( | |
| llama.req("POST","/v1/chat/completions", | |
| content=orjson.dumps(pl), | |
| headers={"Content-Type":"application/json"}, | |
| timeout=httpx.Timeout(connect=10.0, read=_file_timeout + 30.0, | |
| write=10.0, pool=5.0)), | |
| timeout=_file_timeout) | |
| return Response(resp.content, resp.status_code, | |
| media_type="application/json", | |
| headers={"X-File":fname,"X-Bytes":str(len(data))}) | |
| except asyncio.TimeoutError: | |
| await llama._reset() | |
| return E("File processing timed out", 504, "timeout") | |
| except Exception as e: | |
| logger.error(f"[files] {e}", exc_info=True) | |
| return E(f"File upload error: {e}", 500) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # FILE GENERATION [FIX-07: better fence removal] | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| VALID_FORMATS = {"py","md","html","json","txt","sh","csv","js","ts", | |
| "yaml","yml","sql","r","cpp","c","java","go","rs","jsx","vue"} | |
| async def generate_file(request: Request) -> Response: | |
| """ | |
| Ask the AI to generate a file and download it directly. | |
| { | |
| "prompt": "Write a Python script to download all images from a webpage", | |
| "format": "py", | |
| "filename": "image_downloader.py", | |
| "thinking": false | |
| } | |
| """ | |
| if not auth(request): return E("Unauthorized",401) | |
| if not _INFERENCE_READY: return E("Inference engine still loading.",503,"not_ready") | |
| try: body = orjson.loads(await request.body()) | |
| except Exception as e: return E(f"Invalid JSON: {e}",400) | |
| prompt = body.get("prompt","").strip() | |
| fmt = body.get("format","txt").lower().strip(".") | |
| filename = body.get("filename") or f"output_{uuid.uuid4().hex[:6]}.{fmt}" | |
| thinking = bool(body.get("thinking",False)) | |
| # Cap max_tokens: 1024 default (enough for most scripts; user can override up to 2048) | |
| gen_max_tokens = min(int(body.get("max_tokens", 1024)), MAX_NEW_TOKENS) | |
| if not prompt: return E("'prompt' required",400) | |
| if fmt not in VALID_FORMATS: | |
| return E(f"Invalid format '{fmt}'. Supported: {sorted(VALID_FORMATS)}",400) | |
| gen_system = ( | |
| f"You are a file generator. Output ONLY the raw content of a .{fmt} file. " | |
| f"No explanations, no markdown code blocks, no preamble. " | |
| f"Output the file content directly, ready to save." | |
| ) | |
| pl = { | |
| "messages": [ | |
| {"role":"system","content":gen_system}, | |
| {"role":"user","content":prompt}, | |
| ], | |
| "max_tokens": gen_max_tokens, | |
| "temperature": 0.3, | |
| "cache_prompt":False, | |
| "stream": False, | |
| # CRITICAL: always set thinking explicitly — without this Qwen3 may enter thinking | |
| # mode silently, consuming all tokens in <think> blocks and producing empty files | |
| "chat_template_kwargs": {"enable_thinking": bool(thinking)}, | |
| } | |
| if thinking: | |
| pl["reasoning_budget"] = 4096 | |
| try: | |
| resp = await asyncio.wait_for( | |
| llama.req("POST","/v1/chat/completions", | |
| content=orjson.dumps(pl), | |
| headers={"Content-Type":"application/json"}, | |
| timeout=httpx.Timeout(connect=10.0, read=GENERATE_TIMEOUT + 30.0, | |
| write=10.0, pool=5.0)), | |
| timeout=GENERATE_TIMEOUT) | |
| if resp.status_code != 200: | |
| return E(f"Generation failed: HTTP {resp.status_code}",500) | |
| data = orjson.loads(resp.content) | |
| raw_text = data["choices"][0]["message"]["content"].strip() | |
| # [FIX-07] Remove only first/last fence lines, preserve internal ``` | |
| lines = raw_text.splitlines() | |
| if lines and lines[0].startswith("```"): | |
| lines = lines[1:] | |
| if lines and lines[-1].strip() == "```": | |
| lines = lines[:-1] | |
| file_content = "\n".join(lines).strip() | |
| tokens_used = data.get("usage",{}).get("completion_tokens","?") | |
| # Guard: if content is empty the model likely entered thinking mode silently | |
| if not file_content: | |
| logger.warning(f"[generate] model returned empty content (tokens={tokens_used}) — " | |
| "possible silent thinking mode. Check enable_thinking=False is applied.") | |
| return E("Model returned empty file content — the generation produced no output. " | |
| "Try rephrasing the prompt.", 500) | |
| return Response( | |
| content=file_content.encode("utf-8"), | |
| media_type="application/octet-stream", | |
| headers={ | |
| "Content-Disposition": f'attachment; filename="{filename}"', | |
| "X-Format": fmt, | |
| "X-Tokens-Used": str(tokens_used), | |
| "X-Lines": str(len(file_content.splitlines())), | |
| }) | |
| except asyncio.TimeoutError: | |
| await llama._reset() # prevent broken connection on next request | |
| return E("Generation timed out",504) | |
| except Exception as e: | |
| logger.error(f"[generate] {e}",exc_info=True) | |
| return E(f"Generation error: {e}",500) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # WEB SEARCH [FIX-04: empty results shown clearly] | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| async def search_endpoint(request: Request) -> Response: | |
| """ | |
| Web search via DuckDuckGo — free, no API key. | |
| GET /v1/search?q=your+query&n=5&ask=summarize | |
| POST {"query":"...","max_results":5,"ask":"analyze these results"} | |
| """ | |
| if not auth(request): return E("Unauthorized",401) | |
| if request.method == "GET": | |
| params = dict(request.query_params) | |
| query = params.get("q","") | |
| max_res = int(params.get("n","5")) | |
| ask = params.get("ask","") | |
| else: | |
| try: body = orjson.loads(await request.body()) | |
| except: body = {} | |
| query = body.get("query","") | |
| max_res = int(body.get("max_results",5)) | |
| ask = body.get("ask","") | |
| if not query: return E("'query' or 'q' parameter required",400) | |
| results = await web_search(query, min(max_res, 10)) | |
| # [FIX-04] Explicit message when no results | |
| if not results: | |
| no_results_msg = ( | |
| f"No results found for '{query}'. " | |
| "This may be due to network restrictions or a very specific query. " | |
| "Try rephrasing or broadening your search." | |
| ) | |
| return J({"query":query,"results":[],"count":0, | |
| "message":no_results_msg}) | |
| if not ask: | |
| return J({"query":query,"results":results,"count":len(results)}) | |
| results_text = "\n\n".join( | |
| f"[{i+1}] {r['title']}\n{r['snippet']}\nURL: {r['url']}" | |
| for i, r in enumerate(results)) | |
| pl = { | |
| "messages":[ | |
| {"role":"system","content":"You are a research assistant. Analyze web search results accurately."}, | |
| {"role":"user","content":f"Query: {query}\n\nResults:\n{results_text}\n\nTask: {ask}"}, | |
| ], | |
| "max_tokens":1024,"temperature":0.3,"cache_prompt":False, | |
| } | |
| analysis = "" | |
| try: | |
| r = await asyncio.wait_for( | |
| llama.req("POST","/v1/chat/completions", | |
| content=orjson.dumps(pl), | |
| headers={"Content-Type":"application/json"}), | |
| timeout=60.0) | |
| if r.status_code == 200: | |
| analysis = orjson.loads(r.content)["choices"][0]["message"]["content"] | |
| except Exception: | |
| analysis = "[Analysis unavailable]" | |
| return J({"query":query,"results":results,"count":len(results),"analysis":analysis}) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # AUDIO TRANSCRIPTION | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| async def audio_transcription( | |
| request: Request, | |
| file: UploadFile = File(...), | |
| language: str = Form(default=""), | |
| prompt: str = Form(default=""), | |
| ) -> Response: | |
| if not auth(request): return E("Unauthorized",401) | |
| try: | |
| import aiohttp | |
| data = await read_upload_chunked(file, MAX_FILE_MB) # [FIX-02] | |
| if data is None: return E(f"Audio file too large (max {MAX_FILE_MB}MB)",413) | |
| form = aiohttp.FormData() | |
| form.add_field("file",data,filename=file.filename or "audio.wav", | |
| content_type=file.content_type or "audio/wav") | |
| form.add_field("model","whisper-base") | |
| if language: form.add_field("language",language) | |
| if prompt: form.add_field("prompt",prompt) | |
| async with aiohttp.ClientSession() as s: | |
| async with s.post(f"{WHISPER_URL}/v1/audio/transcriptions", | |
| data=form,timeout=aiohttp.ClientTimeout(total=120)) as r: | |
| ct_resp = r.headers.get("content-type","") | |
| if "json" in ct_resp: | |
| return J(await r.json(), r.status) | |
| text = await r.text() | |
| return J({"text": text}, r.status) | |
| except Exception as e: | |
| logger.error(f"[transcription] {e}",exc_info=True) | |
| return E(f"Transcription error: {e}",500,"audio_error") | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # JOB QUEUE ENDPOINTS | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| async def job_submit(request: Request) -> Response: | |
| if not auth(request): return E("Unauthorized",401) | |
| if not _INFERENCE_READY: return E("Inference engine still loading.",503,"not_ready") | |
| try: body = orjson.loads(await request.body()) | |
| except Exception as e: return E(f"Invalid JSON: {e}",400) | |
| body["cache_prompt"] = True | |
| body["max_tokens"] = min(int(body.get("max_tokens") or MAX_NEW_TOKENS), MAX_NEW_TOKENS) | |
| body.pop("model",None); body.pop("stream",None) | |
| msgs = body.get("messages",[]) | |
| if msgs and not any(m.get("role")=="system" for m in msgs): | |
| body["messages"] = [{"role":"system","content":DEFAULT_SYSTEM}] + msgs | |
| try: | |
| jid = await jobs.submit(body) | |
| return J({"job_id":jid,"status":"pending", | |
| "poll_url":f"/v1/jobs/{jid}", | |
| "result_url":f"/v1/jobs/{jid}/result"}) | |
| except RuntimeError as e: | |
| return E(str(e),503,"server_overloaded") | |
| async def job_status(job_id: str, request: Request) -> Response: | |
| if not auth(request): return E("Unauthorized",401) | |
| j = jobs.get(job_id) | |
| if not j: return E(f"Job '{job_id}' not found",404,"not_found") | |
| return J(j.to_dict()) | |
| async def job_result(job_id: str, request: Request) -> Response: | |
| if not auth(request): return E("Unauthorized",401) | |
| j = jobs.get(job_id) | |
| if not j: return E(f"Job '{job_id}' not found",404,"not_found") | |
| if j.status in (JobStatus.PENDING, JobStatus.RUNNING): | |
| return J({"job_id":job_id,"status":j.status.value, | |
| "elapsed_s":round(time.monotonic()-j.created_at,1), | |
| "message":"Job is still processing."}, 202) | |
| if j.status == JobStatus.FAILED: | |
| return E(f"Job failed: {j.error}",500,"job_failed") | |
| if j.result is None: | |
| return E("Job completed but result is empty",500,"job_empty") | |
| return Response(j.result,200,media_type="application/json", | |
| headers={"X-Job-ID":job_id, | |
| "X-Elapsed":str(j.to_dict()["elapsed_s"])}) | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # MODEL MANAGER ENDPOINTS | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| async def mm_status(request: Request) -> Response: | |
| if not auth(request): return E("Unauthorized",401) | |
| return J(mgr.status()) | |
| async def mm_catalog(request: Request) -> Response: | |
| if not auth(request): return E("Unauthorized",401) | |
| return J({"catalog":mgr.catalog(),"active":mgr.cfg.active_id}) | |
| async def mm_check(request: Request) -> Response: | |
| if not auth(request): return E("Unauthorized",401) | |
| try: | |
| await mgr._smart_upgrade() | |
| return J({"status":"checked","pending":mgr.cfg.pending_id or None}) | |
| except Exception as e: return E(str(e),500) | |
| async def mm_switch(request: Request) -> Response: | |
| if not auth(request): return E("Unauthorized",401) | |
| try: | |
| body = orjson.loads(await request.body()) | |
| mid = body.get("model_id","") | |
| from model_manager import CATALOG_BY_ID as _C | |
| if mid not in _C: | |
| return E(f"Unknown model_id: '{mid}'. Valid: {list(_C)}",400) | |
| m = _C[mid] | |
| path,mmproj = await mgr.download(m) | |
| if not path: return E(f"Failed to download '{mid}'",500) | |
| mgr.cfg.pending_id = mid | |
| mgr.cfg.pending_path = path | |
| mgr.cfg.save() | |
| return J({"status":"pending_restart","model":m.name, | |
| "message":"Downloaded. Will activate on next container restart."}) | |
| except Exception as e: return E(str(e),500) | |
| if __name__ == "__main__": | |
| import uvicorn | |
| uvicorn.run("app:app", host="0.0.0.0", | |
| port=int(os.getenv("API_PORT","7860")), | |
| loop="uvloop", http="h11", workers=1) |