Hitonabi/mission-control-v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor: Zentrales Logging + robuste Token-Erfassung (Phase 2)

Browse Source 
- app.py: logging.basicConfig (MC_LOG_LEVEL, INFO default) als eine
  Konfiguration für alle Module.
- Neuer services/gateway_stream.py: SSE-/Non-Stream-usage-Parsing aus dem
  gateway_proxy-Router extrahiert; robuster Zeilenparser mit Debug-Logging
  statt verschluckter Exceptions. Router ist jetzt dünn.
- token_stats.py: In-Memory-Cache + gedrosseltes Flushen (5s) + atexit-Flush
  statt Write-pro-Request; atomarer Write (.tmp -> replace); thread-safe.
- agent.py/discover.py: stille `except Exception: pass` durch gezieltes
  log.debug/warning ersetzt; ungenutzten yaml-Import entfernt.

Verifiziert: Stream-Parsing (Summen + per-Modell), malformed-Chunk übersteht,
flush schreibt; app importiert sauber.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Hitonabi
2026-06-26 14:32:15 +02:00
parent a7c3f8f516
commit 341ea870bb
6 changed files with 163 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app.py
+10
View File
@@ -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.
backend/routers/gateway_proxy.py
+3 -31
View File
@@ -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)
backend/services/agent.py
+10 -4
View File
@@ -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
backend/services/discover.py
+8 -2
View File
@@ -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
backend/services/gateway_stream.py
+41
View File
@@ -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])
backend/services/token_stats.py
+89 -46
View File
@@ -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
"""Aktueller Stand (inkl. noch nicht geflushter Inkremente) als Kopie."""
with _lock:
return json.loads(json.dumps(_ensure_loaded()))
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": {}}
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 = 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 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 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)