Hitonabi
47f7a85510
Die Ampel-Nachruestung deckte 317/337 vorbestehende ruff-Verstoesse im ganzen Repo auf. Aufgeraeumt: - ruff.toml: intentionale Muster als Projekt-Politik ausgenommen (BLE001 blind-except, S110/S112 try-except-pass/continue, PLW1510 subprocess-best-effort, B008 FastAPI- Depends/File-Idiom, EXE001 Shebang, + wenige Stil-Regeln). __init__.py-Re-Exports geschuetzt (F401). - ruff --fix: 128 mechanische (Import-Sortierung, PEP585/604-Annotationen, tote Imports, ueberfluessige noqa) auto-behoben. - 12 echte Reste von Hand: PERF402/102, PLC3002 (Lambda->walrus), ISC004 (String-Concat geklammert), F841/RUF059 (ungenutzte Vars), PIE810 (startswith-Tuple), UP031 (f-string), UP035 (veraltete typing-Imports). Ergebnis: 'ruff check .' = 0, 'compileall' grün. Kein Verhaltenswechsel (nur Stil/Modernisierung). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
124 lines
4.1 KiB
Python
124 lines
4.1 KiB
Python
"""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 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
|
|
# mtime der Datei beim letzten eigenen Laden/Schreiben — seit dem Gateway-Auszug
|
|
# (UMBAU v3 P1) schreibt der mc2-gateway-Prozess die Datei, das Steuerpult liest nur
|
|
# noch: ohne mtime-Vergleich zeigte es ab Prozessstart eingefrorene Zahlen.
|
|
_disk_mtime: float | None = None
|
|
|
|
|
|
def _stat_mtime() -> float | None:
|
|
try:
|
|
return STATS_FILE.stat().st_mtime
|
|
except OSError:
|
|
return None
|
|
|
|
|
|
def _load_from_disk() -> dict:
|
|
global _disk_mtime
|
|
_disk_mtime = _stat_mtime()
|
|
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:
|
|
global _disk_mtime
|
|
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)
|
|
_disk_mtime = _stat_mtime() # eigener Write ist kein Fremd-Update
|
|
except OSError:
|
|
log.warning("token_stats: Schreiben fehlgeschlagen", exc_info=True)
|
|
|
|
|
|
def get_stats() -> dict:
|
|
"""Aktueller Stand (inkl. noch nicht geflushter Inkremente) als Kopie.
|
|
|
|
Multi-Prozess-fähig: Hat ein ANDERER Prozess (mc2-gateway) die Datei inzwischen
|
|
geschrieben und liegen hier keine ungeflushten Inkremente, wird frisch geladen.
|
|
Im Gateway-Prozess selbst ist nach jedem Flush Datei == Speicher → der
|
|
mtime-Vergleich lädt dort nie unnötig nach."""
|
|
global _stats
|
|
with _lock:
|
|
if _stats is not None and not _dirty:
|
|
mtime = _stat_mtime()
|
|
if mtime is not None and mtime != _disk_mtime:
|
|
_stats = _load_from_disk()
|
|
return json.loads(json.dumps(_ensure_loaded()))
|
|
|
|
|
|
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)
|