115 lines
3.5 KiB
Python
115 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
hourly_finetune_check.py
|
|
|
|
Ebene 2 des Skills: läuft stündlich (oder öfter), prüft NUR Deltas
|
|
gegen die von daily_model_check.py gebaute Prioritätsliste:
|
|
|
|
- Ist ein Agent über sein Tagesbudget (hard_stop_eur_per_day)?
|
|
-> Kill-Switch: Agent wird auf lokalen Fallback gezwungen.
|
|
- Wurde ein Modell inzwischen als nicht mehr verfügbar gemeldet?
|
|
-> Wird aus der aktiven Leiter entfernt (bis zum nächsten Tages-Lauf).
|
|
|
|
Erwartet, dass der jeweilige Agent selbst fortlaufend seinen
|
|
Tagesverbrauch nach state/spend_today.json schreibt, Format:
|
|
{ "agent_a": 0.42, "agent_b": 1.10, ... }
|
|
|
|
Falls diese Datei fehlt (z.B. noch nicht angebunden), läuft der Check
|
|
trotzdem durch und markiert das als Hinweis statt Fehler.
|
|
|
|
Aufruf:
|
|
python3 hourly_finetune_check.py
|
|
"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
CONFIG_PATH = Path(__file__).parent.parent / "config" / "agents.yaml"
|
|
STATE_DIR = Path(__file__).parent.parent / "state"
|
|
PRIORITY_PATH = STATE_DIR / "model_priority.json"
|
|
SPEND_PATH = STATE_DIR / "spend_today.json"
|
|
KILL_SWITCH_PATH = STATE_DIR / "kill_switches.json"
|
|
|
|
|
|
def load_config():
|
|
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
|
|
return yaml.safe_load(f)
|
|
|
|
|
|
def load_priority():
|
|
if not PRIORITY_PATH.exists():
|
|
raise FileNotFoundError(
|
|
f"{PRIORITY_PATH} fehlt - bitte zuerst daily_model_check.py laufen lassen"
|
|
)
|
|
with open(PRIORITY_PATH, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def load_spend():
|
|
if not SPEND_PATH.exists():
|
|
return {}, ["spend_today.json fehlt noch - Budget-Check übersprungen, bitte Anbindung nachholen"]
|
|
with open(SPEND_PATH, "r", encoding="utf-8") as f:
|
|
return json.load(f), []
|
|
|
|
|
|
def check_budgets(config, spend):
|
|
kill_switches = {}
|
|
notes = []
|
|
|
|
for agent_name, agent_cfg in config["agents"].items():
|
|
limit = agent_cfg.get("hard_stop_eur_per_day")
|
|
spent = spend.get(agent_name)
|
|
|
|
if spent is None:
|
|
continue
|
|
|
|
if limit is None or limit == "BUDGET_PLACEHOLDER":
|
|
notes.append(
|
|
f"{agent_name}: hard_stop_eur_per_day ist noch Platzhalter - "
|
|
f"kein Kill-Switch aktiv, bitte echten Wert eintragen"
|
|
)
|
|
continue
|
|
|
|
if spent >= float(limit):
|
|
kill_switches[agent_name] = {
|
|
"reason": "daily_budget_exceeded",
|
|
"spent_eur": spent,
|
|
"limit_eur": limit,
|
|
"action": "postpone_to_next_day",
|
|
"basic_questions_fallback": "vps_ollama",
|
|
}
|
|
notes.append(
|
|
f"KILL-SWITCH: {agent_name} hat {spent}€ von {limit}€ Tageslimit erreicht - "
|
|
f"restliche Tasks werden auf morgen verschoben. Grundfragen laufen bis dahin "
|
|
f"weiter über das lokale VPS-Ollama-Modell, sofern das inhaltlich passt."
|
|
)
|
|
|
|
return kill_switches, notes
|
|
|
|
|
|
def main():
|
|
config = load_config()
|
|
priority = load_priority()
|
|
spend, spend_notes = load_spend()
|
|
|
|
kill_switches, budget_notes = check_budgets(config, spend)
|
|
|
|
result = {
|
|
"checked_at": priority.get("generated_at"),
|
|
"kill_switches": kill_switches,
|
|
"notes": spend_notes + budget_notes,
|
|
}
|
|
|
|
with open(KILL_SWITCH_PATH, "w", encoding="utf-8") as f:
|
|
json.dump(result, f, indent=2, ensure_ascii=False)
|
|
|
|
print(f"Feintuning-Check abgeschlossen, Ergebnis in {KILL_SWITCH_PATH}")
|
|
for note in result["notes"]:
|
|
print(f" - {note}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|