diff --git a/backend/app.py b/backend/app.py index 4be5df9..f1bb3de 100644 --- a/backend/app.py +++ b/backend/app.py @@ -6,6 +6,9 @@ setzt eine no-cache-Middleware. Im Dev läuft das Frontend über den Vite-Dev- Server (proxyt /api hierher), daher CORS für localhost offen. """ +import logging +import os + from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse @@ -15,6 +18,13 @@ from starlette.requests import Request from config import FRONTEND_DIST, VERSION from routers import agent, connect, gateway_proxy, health, maintenance, memory, models, routing, system +# Zentrales Logging — Level via MC_LOG_LEVEL (INFO default). Eine Konfiguration +# für alle Module (logging.getLogger(__name__)). +logging.basicConfig( + level=os.environ.get("MC_LOG_LEVEL", "INFO").upper(), + format="%(asctime)s %(levelname)-7s %(name)s: %(message)s", +) + app = FastAPI(title="Mission Control 2.0", version=VERSION) # Dev: Vite-Dev-Server (5173) ruft das Backend per /api auf. diff --git a/backend/routers/gateway_proxy.py b/backend/routers/gateway_proxy.py index b7a6efe..e9f0f5a 100644 --- a/backend/routers/gateway_proxy.py +++ b/backend/routers/gateway_proxy.py @@ -1,11 +1,10 @@ -import json import httpx from fastapi import APIRouter, Request from fastapi.responses import JSONResponse, StreamingResponse from config import LLAMA_SWAP_URL +from services.gateway_stream import record_stream_chunk, record_usage from services.router_logic import FAST, FAST_NO_THINK, choose_model -from services.token_stats import increment_tokens router = APIRouter(prefix="/v1") @@ -37,41 +36,14 @@ async def _proxy(path: str, request: Request): async with httpx.AsyncClient(timeout=None) as c: async with c.stream("POST", url, json=body) as r: async for chunk in r.aiter_raw(): - try: - chunk_str = chunk.decode("utf-8", errors="ignore") - if '"usage":' in chunk_str: - for line in chunk_str.splitlines(): - if line.startswith("data:"): - data_str = line[5:].strip() - if data_str == "[DONE]": - continue - try: - data_json = json.loads(data_str) - usage = data_json.get("usage") - if usage: - prompt = usage.get("prompt_tokens", 0) - completion = usage.get("completion_tokens", 0) - if prompt or completion: - increment_tokens(prompt, completion, model=alias) - except Exception: - pass - except Exception: - pass + record_stream_chunk(chunk, alias) yield chunk return StreamingResponse(gen(), media_type="text/event-stream", headers=routed) async with httpx.AsyncClient(timeout=600) as c: r = await c.post(url, json=body) resp_json = r.json() - try: - usage = resp_json.get("usage") - if usage: - prompt = usage.get("prompt_tokens", 0) - completion = usage.get("completion_tokens", 0) - if prompt or completion: - increment_tokens(prompt, completion, model=alias) - except Exception: - pass + record_usage(resp_json.get("usage") if isinstance(resp_json, dict) else None, alias) return JSONResponse(resp_json, status_code=r.status_code, headers=routed) diff --git a/backend/services/agent.py b/backend/services/agent.py index a85e3c6..dde7ac3 100644 --- a/backend/services/agent.py +++ b/backend/services/agent.py @@ -4,16 +4,20 @@ Status + verlinkt das standalone hermes-webui. Voller Zugriff + Tools/MCP werden Hermes' eigener Config verdrahtet (siehe docs/HERMES_SETUP.md). """ +import logging + import httpx -from config import HERMES_API_URL, HERMES_HOME, HERMES_WEBUI_URL, yaml +from config import HERMES_API_URL, HERMES_HOME, HERMES_WEBUI_URL + +log = logging.getLogger(__name__) def _reach(url: str, path: str = "") -> bool: try: with httpx.Client(timeout=3.0) as c: return c.get(f"{url}{path}").status_code < 500 - except Exception: + except httpx.HTTPError: return False @@ -31,7 +35,7 @@ def agent_status() -> dict: if isinstance(cfg, dict): brain_model = cfg.get("model", {}).get("model", "auto") except Exception: - pass + log.debug("agent_status: Hermes-config.yaml nicht lesbar", exc_info=True) return { @@ -64,6 +68,7 @@ def update_brain_model(new_model: str) -> bool: with config_path.open("r", encoding="utf-8") as f: cfg = r_yaml.load(f) or {} except Exception: + log.debug("update_brain_model: bestehende config.yaml nicht lesbar", exc_info=True) cfg = {} if not isinstance(cfg, dict): @@ -85,8 +90,9 @@ def update_brain_model(new_model: str) -> bool: import services.maintenance as maintenance maintenance.restart_service("hermes-gateway") except Exception: - pass - + log.warning("update_brain_model: hermes-gateway-Restart fehlgeschlagen", exc_info=True) + return True except Exception: + log.warning("update_brain_model: Schreiben der config.yaml fehlgeschlagen", exc_info=True) return False diff --git a/backend/services/discover.py b/backend/services/discover.py index 4d10921..9b3666f 100644 --- a/backend/services/discover.py +++ b/backend/services/discover.py @@ -14,11 +14,15 @@ import time import httpx +import logging + from config import DISCOVER_CACHE_PATH, DISCOVER_TTL from services.caps import capabilities from services.fit import evaluate_fit, extract_params_b, max_ctx_for from services.sources import CATEGORIES, SKIP_TOKENS, TRUSTED_AUTHORS +log = logging.getLogger(__name__) + _FIT_ORDER = {"perfect": 0, "marginal": 1, "too_tight": 2} @@ -38,6 +42,7 @@ def _fetch_author_models(author: str) -> list: data = c.get(url).json() return data if isinstance(data, list) else [] except Exception: + log.debug("discover: Abfrage für Autor %s fehlgeschlagen", author, exc_info=True) return [] @@ -109,7 +114,7 @@ def refresh_discover(ram_gb: float) -> dict: tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") os.replace(tmp, DISCOVER_CACHE_PATH) except Exception: - pass # Cache ist nur Beschleunigung + log.debug("discover: Cache-Schreiben fehlgeschlagen (nur Beschleunigung)", exc_info=True) return data @@ -118,7 +123,7 @@ def load_discover() -> dict | None: if DISCOVER_CACHE_PATH.exists(): return json.loads(DISCOVER_CACHE_PATH.read_text(encoding="utf-8")) except Exception: - pass + log.debug("discover: Cache-Lesen fehlgeschlagen", exc_info=True) return None @@ -130,4 +135,5 @@ def safe_discover(ram_gb: float) -> dict | None: try: return refresh_discover(ram_gb) except Exception: + log.warning("discover: Live-Refresh fehlgeschlagen, nutze Cache", exc_info=True) return cached diff --git a/backend/services/gateway_stream.py b/backend/services/gateway_stream.py new file mode 100644 index 0000000..6c05268 --- /dev/null +++ b/backend/services/gateway_stream.py @@ -0,0 +1,41 @@ +"""Token-Erfassung für den Builtin-Gateway. + +Parst die `usage`-Felder aus llama-swap-Antworten (Stream + Non-Stream) und meldet +sie an token_stats. Hält den gateway_proxy-Router dünn und ersetzt die zuvor inline +verstreute, still scheiternde String-Suche durch einen testbaren SSE-Zeilenparser. +""" + +import json +import logging + +from services.token_stats import increment_tokens + +log = logging.getLogger(__name__) + + +def record_usage(usage: dict | None, model: str) -> None: + """Ein usage-Objekt verbuchen (no-op bei None/leer).""" + if not usage: + return + prompt = usage.get("prompt_tokens", 0) + completion = usage.get("completion_tokens", 0) + if prompt or completion: + increment_tokens(prompt, completion, model=model) + + +def record_stream_chunk(chunk: bytes, model: str) -> None: + """Rohen SSE-Chunk auf `usage` prüfen und Tokens verbuchen. Fehler werden + geloggt (debug) statt verschluckt — ein defekter Chunk bricht den Stream nicht.""" + if b'"usage"' not in chunk: + return + text = chunk.decode("utf-8", errors="ignore") + for line in text.splitlines(): + if not line.startswith("data:"): + continue + data_str = line[5:].strip() + if not data_str or data_str == "[DONE]": + continue + try: + record_usage(json.loads(data_str).get("usage"), model) + except json.JSONDecodeError: + log.debug("gateway stream: usage-Parsing fehlgeschlagen: %s", data_str[:120]) diff --git a/backend/services/token_stats.py b/backend/services/token_stats.py index 8907c1b..66fc5e5 100644 --- a/backend/services/token_stats.py +++ b/backend/services/token_stats.py @@ -1,56 +1,99 @@ +"""Token-Statistik (Verbrauch je Modell) mit gedrosseltem Persistieren. + +Früher wurde bei JEDEM Request die komplette JSON-Datei gelesen und geschrieben +(Disk-Thrash). Jetzt: einmaliges Laden in einen In-Memory-Cache, Inkremente laufen +gegen den Cache, Persistieren passiert höchstens alle FLUSH_INTERVAL Sekunden sowie +beim Prozess-Ende (atexit). Lesen liefert immer den aktuellen (auch ungeflushten) Stand. +""" + +import atexit import json +import logging +import threading +import time from pathlib import Path + from config import HERMES_HOME STATS_FILE = HERMES_HOME / "token_stats.json" +FLUSH_INTERVAL = 5.0 # Sekunden zwischen Disk-Writes +# Baseline (repräsentiert Verbrauch vor dem modellspezifischen Logging). +_BASELINE = {"prompt_tokens": 718400, "completion_tokens": 324200, "models": {}} + +log = logging.getLogger(__name__) + +_lock = threading.Lock() +_stats: dict | None = None +_dirty = False +_last_flush = 0.0 + + +def _load_from_disk() -> dict: + if not STATS_FILE.exists(): + return dict(_BASELINE) + try: + with open(STATS_FILE, "r", encoding="utf-8") as f: + data = json.load(f) + data.setdefault("prompt_tokens", 0) + data.setdefault("completion_tokens", 0) + data.setdefault("models", {}) + return data + except (OSError, json.JSONDecodeError): + log.warning("token_stats: Laden fehlgeschlagen, nutze Baseline", exc_info=True) + return dict(_BASELINE) + + +def _ensure_loaded() -> dict: + global _stats + if _stats is None: + _stats = _load_from_disk() + return _stats + + +def _write(stats: dict) -> None: + try: + STATS_FILE.parent.mkdir(parents=True, exist_ok=True) + tmp = STATS_FILE.with_suffix(".tmp") + with open(tmp, "w", encoding="utf-8") as f: + json.dump(stats, f) + tmp.replace(STATS_FILE) + except OSError: + log.warning("token_stats: Schreiben fehlgeschlagen", exc_info=True) + def get_stats() -> dict: - if not STATS_FILE.exists(): - # Initialize stats with a nice baseline (e.g., representing previous usage) - STATS_FILE.parent.mkdir(parents=True, exist_ok=True) - default_stats = { - "prompt_tokens": 718400, - "completion_tokens": 324200 - } - try: - with open(STATS_FILE, "w") as f: - json.dump(default_stats, f) - except Exception: - return default_stats - return default_stats - - try: - with open(STATS_FILE, "r") as f: - data = json.load(f) - # Ensure keys exist - if "prompt_tokens" not in data: - data["prompt_tokens"] = 0 - if "completion_tokens" not in data: - data["completion_tokens"] = 0 - if "models" not in data: - data["models"] = {} - return data - except Exception: - return {"prompt_tokens": 0, "completion_tokens": 0, "models": {}} + """Aktueller Stand (inkl. noch nicht geflushter Inkremente) als Kopie.""" + with _lock: + return json.loads(json.dumps(_ensure_loaded())) -def save_stats(stats: dict): - try: - STATS_FILE.parent.mkdir(parents=True, exist_ok=True) - with open(STATS_FILE, "w") as f: - json.dump(stats, f) - except Exception: - pass -def increment_tokens(prompt: int, completion: int, model: str = None): - stats = get_stats() - stats["prompt_tokens"] += prompt - stats["completion_tokens"] += completion - if model: - model = model.lower() - if "models" not in stats: - stats["models"] = {} - if model not in stats["models"]: - stats["models"][model] = {"prompt": 0, "completion": 0} - stats["models"][model]["prompt"] += prompt - stats["models"][model]["completion"] += completion - save_stats(stats) +def increment_tokens(prompt: int, completion: int, model: str | None = None) -> None: + """Tokens im Cache verbuchen; gedrosselt auf Disk persistieren.""" + global _dirty, _last_flush + with _lock: + stats = _ensure_loaded() + stats["prompt_tokens"] += prompt + stats["completion_tokens"] += completion + if model: + m = stats.setdefault("models", {}).setdefault( + model.lower(), {"prompt": 0, "completion": 0}) + m["prompt"] += prompt + m["completion"] += completion + _dirty = True + now = time.monotonic() + if now - _last_flush >= FLUSH_INTERVAL: + _write(stats) + _dirty = False + _last_flush = now + + +def flush() -> None: + """Ungeschriebene Inkremente sofort persistieren (z.B. beim Shutdown).""" + global _dirty + with _lock: + if _dirty and _stats is not None: + _write(_stats) + _dirty = False + + +atexit.register(flush)