"""`kode auswertung`: die Protokolle zu Zahlen verdichtet, um Schwachstellen zu finden - welches Modell verfehlt edit_file, welche Karte startet nicht, wo laeuft der Kontext ueber. """ import json import statistics import time from collections import Counter, defaultdict from typing import Any from rich.console import Console from rich.markup import escape from rich.table import Table from . import protokoll con = Console(highlight=False) def _ereignisse(tage: int) -> list[dict[str, Any]]: grenze = time.time() - tage * 86400 aus = [] for p in sorted(protokoll.ORDNER.glob("*.jsonl")): if p.stat().st_mtime < grenze: continue with open(p, encoding="utf-8") as f: for zeile in f: try: e = json.loads(zeile) except ValueError: continue if e.get("t", 0) >= grenze: e["_datei"] = p.stem aus.append(e) aus.sort(key=lambda e: e["t"]) return aus def _median(werte: list[float]) -> str: werte = [w for w in werte if w is not None] return f"{statistics.median(werte):.1f}" if werte else "–" def _quote(teil: int, ganz: int) -> str: return f"{teil}/{ganz} ({100 * teil / ganz:.0f} %)" if ganz else "–" def zeigen(tage: int = 30) -> None: ev = _ereignisse(tage) if not ev: con.print(f"Keine Protokolle der letzten {tage} Tage in {protokoll.ORDNER}.") return nach_art: dict[str, list[dict]] = defaultdict(list) for e in ev: nach_art[e["art"]].append(e) sitzungen = {e["_datei"] for e in nach_art["sitzung_start"]} ausgang = Counter(e.get("ausgang") for e in nach_art["auftrag_ende"]) con.print(f"[bold]Auswertung der letzten {tage} Tage[/] · {len(sitzungen)} Sitzungen · " f"{len(nach_art['eingabe'])} Aufträge · Ausgang: " + ", ".join(f"{k} {v}" for k, v in ausgang.most_common())) # --- Modelle modell_der_datei: dict[str, str] = {} t = Table(title="Modelle", header_style="bold") for sp in ("Modell", "Antworten", "tok/s", "1. Zeichen s", "Prompt tok/s", "Aufrufe", "aus Text", "kaputte Args", "Kontext ≥95 %", "Werkzeugfehler"): t.add_column(sp) je_modell: dict[str, list[dict]] = defaultdict(list) for e in nach_art["antwort"]: je_modell[e.get("modell")].append(e) modell_der_datei[e["_datei"]] = e.get("modell") werkzeug_je_modell: dict[str, list[dict]] = defaultdict(list) for e in nach_art["werkzeug"]: werkzeug_je_modell[modell_der_datei.get(e["_datei"], "?")].append(e) for m, liste in sorted(je_modell.items(), key=lambda x: -len(x[1])): wz = werkzeug_je_modell.get(m, []) t.add_row(m or "?", str(len(liste)), _median([e.get("tok_s") for e in liste]), _median([e.get("erstes_zeichen_s") for e in liste]), _median([e.get("prompt_tok_s") for e in liste]), str(sum(e.get("aufrufe_echt") or 0 for e in liste)), str(sum(e.get("aufrufe_aus_text") or 0 for e in liste)), str(sum(e.get("argumente_kaputt") or 0 for e in liste)), str(sum(1 for e in liste if (e.get("kontext_anteil") or 0) >= 0.95)), _quote(sum(1 for e in wz if e.get("fehler") and e.get("entscheidung") not in ("nein", "stopp")), len(wz))) if je_modell: con.print(t) # --- Werkzeuge t = Table(title="Werkzeuge", header_style="bold") for sp in ("Werkzeug", "Aufrufe", "Fehler", "abgelehnt", "Exit≠0", "Zeitüberschr.", "Median s", "häufigster Fehler"): t.add_column(sp) je_wz: dict[str, list[dict]] = defaultdict(list) for e in nach_art["werkzeug"]: je_wz[e.get("name")].append(e) for name, liste in sorted(je_wz.items(), key=lambda x: -len(x[1])): fehlertexte = Counter((e.get("ergebnis") or "")[:70] for e in liste if e.get("fehler") and e.get("entscheidung") not in ("nein", "stopp")) t.add_row(name or "?", str(len(liste)), str(sum(1 for e in liste if e.get("fehler") and e.get("entscheidung") not in ("nein", "stopp"))), str(sum(1 for e in liste if e.get("entscheidung") in ("nein", "stopp"))), str(sum(1 for e in liste if e.get("exit_code"))), str(sum(1 for e in liste if e.get("zeitueberschreitung"))), _median([e.get("dauer_s") for e in liste]), escape(fehlertexte.most_common(1)[0][0]) if fehlertexte else "") if je_wz: con.print(t) fr = Counter(e.get("antwort") for e in nach_art["freigabe"]) if fr: con.print("Freigaben: " + ", ".join(f"{k} {v}" for k, v in fr.most_common())) # --- Karten starts = nach_art["karte_start"] if starts: t = Table(title="Kartenstarts", header_style="bold") for sp in ("Anbieter", "Karte", "Starts", "bereit", "Median bis bereit s", "Gewichte MB/s", "häufigster Grund"): t.add_column(sp) # Ereignisse eines Starts: vom karte_start bis zum naechsten Start derselben Datei gruppen: list[list[dict]] = [] offen: dict[str, list[dict]] = {} for e in ev: if e["art"] == "karte_start": offen[e["_datei"]] = [e] gruppen.append(offen[e["_datei"]]) elif e["_datei"] in offen and e["art"].startswith(("karte_", "gewichte_")): offen[e["_datei"]].append(e) je_karte: dict[tuple, list[list[dict]]] = defaultdict(list) for g in gruppen: k = g[0].get("karte") or {} je_karte[(k.get("anbieter"), k.get("name"))].append(g) for (anb, name), gs in sorted(je_karte.items(), key=lambda x: -len(x[1])): bereit = [x for g in gs for x in g if x["art"] == "karte_bereit"] gruende = Counter(x.get("grund") for g in gs for x in g if x["art"] in ("karte_gescheitert",) or (x["art"] == "karte_mieten" and not x.get("kennung"))) mbs = [x.get("mb_s") for g in gs for x in g if x["art"] == "gewichte_fertig"] t.add_row(anb or "?", name or "?", str(len(gs)), str(len(bereit)), _median([x.get("dauer_s") for x in bereit]), _median(mbs), escape(str(gruende.most_common(1)[0][0])[:80]) if gruende else "") con.print(t) loe = nach_art["karte_loeschen"] if loe: nicht = [e for e in loe if not e.get("geloescht")] con.print(f"Löschungen: {len(loe)}, davon [{'red' if nicht else 'green'}]{len(nicht)} unbestätigt[/] · " f"Median Löschdauer {_median([e.get('loeschdauer_s') for e in loe])} s · " f"Summe {sum(e.get('kosten') or 0 for e in loe):.4f} USD · Gründe: " + ", ".join(f"{k} {v}" for k, v in Counter(e.get('grund') for e in loe).most_common())) if nach_art["waechter_waise"]: con.print(f"[yellow]Waisen vom Wächter gefunden: {len(nach_art['waechter_waise'])}[/]") # --- Kontext kz, vd = nach_art["kontext_kuerzen"], nach_art["verdichten"] if kz or vd: con.print(f"Kontext: {len(kz)}× alte Ergebnisse gekürzt, {len(vd)}× Verlauf zusammengefasst") # --- Fehler fe = nach_art["fehler"] if fe: con.print(f"\n[bold red]Fehler ({len(fe)})[/] - die letzten 8:") for e in fe[-8:]: con.print(f" {e['zeit']} [bold]{e.get('wo')}[/] {e.get('typ')}: {escape(str(e.get('text'))[:160])}" f" [dim]({e['_datei']})[/]") con.print(f"\n[dim]Rohdaten: {protokoll.ORDNER}/*.jsonl (eine JSON-Zeile je Ereignis)[/]")