mirror of
https://github.com/Jeuners/Logpy-AgentOne.git
synced 2026-09-09 15:02:33 +02:00
90548944a2
Jede Karte im Leitstand bekommt Daumen hoch/runter fuer die Kategorisierung. Bei 👎 optional Korrektur (richtige Kategorie/Dringlichkeit). Gespeichert als bewertung.json pro Anruf-Ordner (gleiches Muster wie stapel.json, pipe/bewertung.py). Export als JSONL (pipe/export.py, neuer "📤 Bewertungen exportieren"-Knopf + automatisch beim "Tag archivieren"): ein Datensatz pro bewertetem Anruf mit Transkript, Modell-Ausgabe, Backend/Modell, Bewertung und Korrektur - fuer externe Auswertung (z.B. um Kategorisierungs-Fehler systematisch statt zufaellig zu finden, wie das Notfall-Sicherheitsnetz vom letzten Fund). Export-Knopf verlangt Bestaetigung ("Transkripte und Namen verlassen den Rechner") - bewusst kein automatischer Export ohne Hinweis, das steht sonst im Widerspruch zum On-Prem-Versprechen. training/ ist neu gitignored (volle Transkripte/Namen drin, siehe README-Hinweis zu Testdaten). Nebenbei: process_call.py legt jetzt auch backend/modell/dauer_s der Kategorisierung dauerhaft in meta.json ab (vorher nur im fluechtigen Log) - noetig, damit der Export nachvollziehbar ist, welches Modell lief. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015vtvcyNkapuDh583643hNs
69 lines
2.4 KiB
Python
69 lines
2.4 KiB
Python
"""Exportiert bewertete Anrufe als JSONL-Trainingsdaten (Transkript + Modell-
|
|
Ausgabe + Bewertung + ggf. Korrektur) - fuer externe Auswertung, z.B. um
|
|
Kategorisierungs-Fehler systematisch statt zufaellig zu finden.
|
|
|
|
python3 -m pipe.export
|
|
|
|
Nimmt jeden Anruf mit bewertung.json (unabhaengig vom Datum - kein
|
|
inkrementeller Export, einfach alle aktuell vorhandenen Bewertungen). Legt
|
|
unter training/ ab, das wie ablage/ nicht ins Git gehoert (siehe .gitignore) -
|
|
volle Transkripte + Namen sind bewusst drin (Testphase, siehe README-Hinweis
|
|
zu erfundenen Daten), fuer den Produktivbetrieb muss das neu entschieden
|
|
werden, siehe Warnhinweis beim Export-Knopf im Leitstand.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from . import bewertung, config
|
|
|
|
EXPORT_ORDNER = config.WURZEL / "training"
|
|
|
|
|
|
def sammle() -> list[dict]:
|
|
"""Alle aktuell bewerteten Anrufe, als flache Trainingsdatensaetze."""
|
|
eintraege = []
|
|
if not config.ABLAGE_LOKAL.is_dir():
|
|
return eintraege
|
|
for meta_pfad in sorted(config.ABLAGE_LOKAL.glob("*/*/meta.json")):
|
|
ordner = meta_pfad.parent
|
|
b = bewertung.lies(ordner)
|
|
if b is None:
|
|
continue
|
|
try:
|
|
daten = json.loads(meta_pfad.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
continue
|
|
eintraege.append({
|
|
"anruf_id": str(ordner.relative_to(config.ABLAGE_LOKAL)),
|
|
"empfangen": daten.get("empfangen"),
|
|
"transkript": daten.get("transkript"),
|
|
"modell_ausgabe": daten.get("auswertung"),
|
|
"kategorisierung": daten.get("kategorisierung"),
|
|
"bewertung": b["bewertung"],
|
|
"korrektur": b.get("korrektur"),
|
|
"bewertet_am": b.get("zeitpunkt"),
|
|
})
|
|
return eintraege
|
|
|
|
|
|
def exportiere() -> tuple[Path, int]:
|
|
"""Schreibt eine neue JSONL-Datei. Gibt (Pfad, Anzahl) zurueck."""
|
|
eintraege = sammle()
|
|
EXPORT_ORDNER.mkdir(parents=True, exist_ok=True)
|
|
ziel = EXPORT_ORDNER / f"export_{datetime.now():%Y%m%d-%H%M%S}.jsonl"
|
|
with ziel.open("w", encoding="utf-8") as f:
|
|
for e in eintraege:
|
|
f.write(json.dumps(e, ensure_ascii=False) + "\n")
|
|
return ziel, len(eintraege)
|
|
|
|
|
|
def main() -> None:
|
|
ziel, anzahl = exportiere()
|
|
print(f"{anzahl} bewertete Anrufe exportiert nach {ziel}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|