472 lines
22 KiB
Python
472 lines
22 KiB
Python
"""Independent quality audit over a FINISHED generation run — read-only.
|
||
|
||
Measures the MECE goal ("no duplicates, no gaps") with detectors that deliberately
|
||
do NOT reuse the pipeline's heuristics (_canonical_key/_relation_conflict/_evidence_pack)
|
||
— shared blind spots would make the audit worthless. Shared infra only: DB access,
|
||
embedding.py, the agent runner (--llm sampling), atomic_write_json.
|
||
|
||
CLI: python3 qa.py <topic> [--llm] (or: make qa TOPIC=<topic> [LLM=1])
|
||
Report: storage/qa/<topic>/<run_id|timestamp>.json + console digest + diff to the
|
||
previous report of the same topic.
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import re
|
||
import sys
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
import database as db
|
||
import embedding
|
||
from config import STORAGE_DIR, SUB_DUP_KANDIDAT_COS
|
||
from fsutil import atomic_write_json
|
||
from jsonio import read_json_file as _json_file
|
||
from paths import arbeit_dir
|
||
from textkit import _norm_title
|
||
|
||
QA_DIR = STORAGE_DIR / "qa"
|
||
JACCARD_FLOOR = 0.5 # title token overlap that makes a pair suspicious
|
||
EMB_FLOOR = 0.82 # casefolded title cosine (own threshold, NOT the pipeline's 0.65)
|
||
SECTION_CHARS = 4000 # own paragraph splitter — independent of _text_sections
|
||
COVER_MIN_TOKENS = 2 # distinctive block tokens a section must share to count as covered
|
||
FREMD_MIN_TOKENS = 1 # distinctive title tokens that must appear in the corpus
|
||
LLM_SAMPLE = 12 # pairs/sections per judge call with --llm
|
||
# Note 0-10, deterministisch aus den Quoten (transparent, diffbar — keine LLM-"Gefühlsnote").
|
||
# Lücken/Fremd wiegen am schwersten (fehlender/falscher Stoff); Dubletten-VERDACHT enthält
|
||
# bewusst Rauschen und wiegt daher wenig.
|
||
NOTE_GEWICHTE = {"luecken": 3.0, "fremd": 2.5, "unechte_bloecke": 2.5, "hygiene": 0.5}
|
||
# subs/artefacts only exist after board 2 — at gate time these quotas would always be 0
|
||
# and water down the inventory score, hence a separate score.
|
||
# sub_dubletten counts only with --llm (confirmed pairs); the bare candidate list is
|
||
# suspicion (sub_dubletten_verdacht, weightless — like dubletten_verdacht).
|
||
NOTE_GEWICHTE_ARTEFAKTE = {"subs_ohne_beleg": 2.0, "verwaiste": 1.0, "sub_dubletten": 1.0}
|
||
|
||
_WORD = re.compile(r"\w{3,}")
|
||
_PAREN = re.compile(r"^\s*(.*?)\s*\(([^()]{2,60})\)\s*$")
|
||
_STOP = {"der", "die", "das", "und", "oder", "für", "mit", "von", "des", "den", "dem",
|
||
"ein", "eine", "the", "and", "for", "with", "als", "auf", "bei", "aus",
|
||
"problem", "algorithmus", "algorithm", "definition", "satz", "lemma"}
|
||
|
||
|
||
def _tokens(s: str) -> set[str]:
|
||
return {t for t in _WORD.findall((s or "").casefold()) if t not in _STOP}
|
||
|
||
|
||
def _distinctive(s: str) -> set[str]:
|
||
"""Tokens that can anchor a title in a corpus (stopword-free, ≥3 chars)."""
|
||
return _tokens(s)
|
||
|
||
|
||
def _ascii(t: str) -> str:
|
||
return "".join(c for c in t if c.isascii())
|
||
|
||
|
||
def _jaccard(a: set[str], b: set[str]) -> float:
|
||
return len(a & b) / len(a | b) if a | b else 0.0
|
||
|
||
|
||
def _sections(text: str, goal: int | None = None) -> list[str]:
|
||
"""Own paragraph-boundary splitter (NOT blocks._text_sections — independence)."""
|
||
goal = goal or SECTION_CHARS
|
||
out, buf = [], ""
|
||
for para in re.split(r"\n\s*\n", text.strip()):
|
||
para = para.strip()
|
||
if not para:
|
||
continue
|
||
if buf and len(buf) + len(para) > goal:
|
||
out.append(buf)
|
||
buf = para
|
||
else:
|
||
buf = f"{buf}\n\n{para}" if buf else para
|
||
if buf.strip():
|
||
out.append(buf)
|
||
return out
|
||
|
||
|
||
def _corpus_texts(topic: str) -> dict[str, str]:
|
||
from blocks import source_folder # lazy: blocks pulls heavy deps
|
||
folder = source_folder(topic)
|
||
if not folder or not folder.is_dir():
|
||
return {}
|
||
out = {}
|
||
for f in sorted(folder.glob("*.txt")):
|
||
try:
|
||
out[f.name] = f.read_text(encoding="utf-8")
|
||
except OSError:
|
||
continue
|
||
return out
|
||
|
||
|
||
# ── Detectors ───────────────────────────────────────────────────────────────────────
|
||
|
||
def dubletten(blocks: list[dict], emb_on: bool = True) -> list[dict]:
|
||
"""Suspicious pairs via signal UNION: token jaccard, casefolded-title embedding
|
||
cosine, paren acronym == other title. Every signal is independent of the pipeline."""
|
||
titles = [b["title"] for b in blocks]
|
||
toks = [_tokens(t) for t in titles]
|
||
sims = None
|
||
if emb_on and titles and embedding.available():
|
||
arr = embedding.embed([t.casefold() for t in titles])
|
||
if arr is not None:
|
||
sims = arr @ arr.T
|
||
ops = [bool(re.search(r"[≤⪯≥⊆⊊→⇒⟹⇔←]", t)) for t in titles]
|
||
out = []
|
||
for i in range(len(titles)):
|
||
for j in range(i + 1, len(titles)):
|
||
# relation vs. its operand ("Subset Sum" ⊂ "3-SAT ≤ Subset Sum"): by design
|
||
# separate entities — token containment there is expected, not suspicious
|
||
if ops[i] != ops[j] and (toks[i] <= toks[j] or toks[j] <= toks[i]):
|
||
continue
|
||
signals = {}
|
||
jac = _jaccard(toks[i], toks[j])
|
||
if jac >= JACCARD_FLOOR:
|
||
signals["jaccard"] = round(jac, 2)
|
||
if sims is not None and float(sims[i][j]) >= EMB_FLOOR:
|
||
signals["emb_cos"] = round(float(sims[i][j]), 2)
|
||
for a, b in ((i, j), (j, i)):
|
||
m = _PAREN.match(titles[a])
|
||
if m and _norm_title(titles[b]) in (_norm_title(m.group(1)), _norm_title(m.group(2))):
|
||
signals["akronym"] = True
|
||
if signals:
|
||
out.append({"a": titles[i], "b": titles[j], "signale": signals})
|
||
return out
|
||
|
||
|
||
def sub_dubletten(sub_rows: list[dict], emb_on: bool = True) -> list[dict]:
|
||
"""Suspicious SUB pairs, in-block AND cross-block: casefolded title cosine ≥
|
||
SUB_DUP_KANDIDAT_COS. The pipeline's own merge paths act from 0.90 upward — the
|
||
measured bulk of real paraphrase duplicates sits in the band below, so everything
|
||
above the floor is a candidate. The verdict falls with --llm; without it this is
|
||
a suspicion list only (weightless)."""
|
||
cons = [r for r in sub_rows if r["status"] == "consensus"]
|
||
if len(cons) < 2 or not emb_on or not embedding.available():
|
||
return []
|
||
arr = embedding.embed([r["sub_title"].casefold() for r in cons])
|
||
if arr is None:
|
||
return []
|
||
sims = arr @ arr.T
|
||
out = []
|
||
for i in range(len(cons)):
|
||
for j in range(i + 1, len(cons)):
|
||
v = float(sims[i][j])
|
||
if v >= SUB_DUP_KANDIDAT_COS:
|
||
out.append({"a": f"[{cons[i]['block']}] {cons[i]['sub_title']}",
|
||
"b": f"[{cons[j]['block']}] {cons[j]['sub_title']}",
|
||
"cos": round(v, 2),
|
||
"cross": cons[i]["block_norm"] != cons[j]["block_norm"]})
|
||
return sorted(out, key=lambda p: -p["cos"])
|
||
|
||
|
||
def luecken(blocks: list[dict], subs_by_norm: dict[str, list[str]], corpus: dict[str, str]) -> list[dict]:
|
||
"""Corpus sections no block (title+description+subs tokens) sufficiently anchors.
|
||
Description tokens matter at the QA GATE: board 2 has not run yet, so titles alone
|
||
under-cover and inflate the quota."""
|
||
anchors: list[set[str]] = []
|
||
for b in blocks:
|
||
t = _distinctive(b["title"]) | _distinctive(b.get("description") or "")
|
||
for s in subs_by_norm.get(_norm_title(b["title"]), []):
|
||
t |= _distinctive(s)
|
||
anchors.append(t)
|
||
out = []
|
||
for fname, text in corpus.items():
|
||
for k, sec in enumerate(_sections(text), 1):
|
||
sec_toks = _tokens(sec)
|
||
covered = any(len(a & sec_toks) >= COVER_MIN_TOKENS for a in anchors)
|
||
if not covered:
|
||
preview = " ".join(sec.split())[:120]
|
||
out.append({"datei": fname, "abschnitt": k, "vorschau": preview})
|
||
return out
|
||
|
||
|
||
def fremd(blocks: list[dict], corpus: dict[str, str]) -> list[str]:
|
||
"""Blocks whose distinctive title tokens never appear in the corpus (scope creep).
|
||
Token/stem match, NOT raw substring — 'bergang' ⊂ 'Übergang' had whitewashed the
|
||
garbage title 'αÜbergang'. The ASCII form only bridges symbol variants (Δ/∆)."""
|
||
ctoks = set(_WORD.findall("\n".join(corpus.values()).casefold()))
|
||
|
||
def _hit(t: str) -> bool:
|
||
forms = {t} | ({a} if len(a := _ascii(t)) >= 3 else set())
|
||
# digit-suffix fallback: '∆TSP1' → 'tsp1' misses the corpus token 'tsp' ('∆' is no \w)
|
||
forms |= {f2 for f in list(forms) if len(f2 := f.rstrip("0123456789")) >= 3}
|
||
return any(ct == f or ct.startswith(f) for f in forms for ct in ctoks)
|
||
|
||
out = []
|
||
for b in blocks:
|
||
dist = _distinctive(b["title"])
|
||
if dist and sum(1 for t in dist if _hit(t)) < FREMD_MIN_TOKENS:
|
||
out.append(b["title"])
|
||
return out
|
||
|
||
|
||
def beleg(blocks: list[dict], sub_rows: list[dict]) -> dict:
|
||
ohne_quelle = [b["title"] for b in blocks if not b.get("sources")]
|
||
ohne_mention = [f"{r['block']} · {r['sub_title']}" for r in sub_rows
|
||
if r["status"] != "variant" and not r["mentions"]]
|
||
return {"bloecke_ohne_quelle": ohne_quelle, "subs_ohne_beleg": ohne_mention}
|
||
|
||
|
||
def hygiene(blocks: list[dict]) -> list[dict]:
|
||
out = []
|
||
for b in blocks:
|
||
t = b["title"]
|
||
probleme = []
|
||
if "**" in t or "`" in t:
|
||
probleme.append("markdown")
|
||
if re.search(r"\(\d+\)\s*$", t):
|
||
probleme.append("kollisions-suffix")
|
||
if not (b.get("description") or "").strip():
|
||
probleme.append("leere-beschreibung")
|
||
if probleme:
|
||
out.append({"titel": t, "probleme": probleme})
|
||
return out
|
||
|
||
|
||
def _zaehlbare_luecken(lk: list[dict], llm: bool) -> list[dict]:
|
||
"""With --llm only non-refuted gaps count ('?' = unjudged stays, conservative) — refuted
|
||
ones dragged the note although the judge cleared them (aak: 5 of 8, weight 3.0)."""
|
||
return [x for x in lk if x.get("llm") != "nein"] if llm else lk
|
||
|
||
|
||
def note(quoten: dict, gewichte: dict = NOTE_GEWICHTE) -> float:
|
||
"""10 = alle gewichteten Quoten 0. Gewicht = Punktabzug bei 100 % Quote (keine Normierung,
|
||
sonst staucht die Gewichtssumme die Skala nach oben). Ungemessene Quoten zählen nicht —
|
||
unechte_bloecke existiert nur mit --llm; dubletten_verdacht ist Verdachtsliste, kein Urteil."""
|
||
da = {k: w for k, w in gewichte.items() if k in quoten}
|
||
schaden = sum(w * min(float(quoten[k]), 1.0) for k, w in da.items())
|
||
return round(max(0.0, 10.0 * (1 - schaden)), 1)
|
||
|
||
|
||
def artefakte(sub_rows: list[dict], art_rows: list[dict], fragen: list[dict]) -> dict:
|
||
"""Coverage + orphans of the learning artefacts. Nenner = consensus-Subs (verworfene
|
||
zählen nicht als abzudeckendes Material). Waise = Ziel weder lebend (consensus/variant)
|
||
noch eindeutig als Kurztitel-Präfix von „kurztitel: beschreibung" auflösbar."""
|
||
if not art_rows and not fragen:
|
||
return {"status": "nicht generiert"}
|
||
cons = {(r["block_norm"], r["sub_norm"]) for r in sub_rows if r["status"] == "consensus"}
|
||
lebt = {(r["block_norm"], r["sub_norm"]) for r in sub_rows if r["status"] != "discarded"}
|
||
|
||
def _ziel(bn: str, sn: str):
|
||
if (bn, sn) in lebt:
|
||
return (bn, sn)
|
||
treffer = [k for k in lebt if k[0] == bn and k[1].startswith(sn + ":")]
|
||
if len(treffer) == 1:
|
||
return treffer[0]
|
||
# mehrere Treffer = meist ein consensus-Sub plus seine gefalteten Varianten
|
||
haupt = [k for k in treffer if k in cons]
|
||
return haupt[0] if len(haupt) == 1 else None
|
||
|
||
deck: dict[str, set] = {}
|
||
verwaist = []
|
||
for typ, bn, sn in ([(r["type"], r["block_norm"], r["sub_norm"]) for r in art_rows]
|
||
+ [("frage", r["block_norm"], r["sub_norm"]) for r in fragen]):
|
||
z = _ziel(bn, sn)
|
||
if z is None:
|
||
verwaist.append(f"{typ}: {bn} · {sn}")
|
||
else:
|
||
deck.setdefault(typ, set()).add(z)
|
||
n = max(len(cons), 1)
|
||
return {"status": "ok",
|
||
"frage_abdeckung": round(len(deck.get("frage", set()) & cons) / n, 3),
|
||
"flashcard_abdeckung": round(len(deck.get("flashcard", set()) & cons) / n, 3),
|
||
"beispiel_abdeckung": round(len(deck.get("example", set()) & cons) / n, 3),
|
||
"verwaiste": sorted(verwaist)}
|
||
|
||
|
||
# ── LLM sampling (optional, own prompts under templates/QA/) ────────────────────────
|
||
|
||
def _qa_prompt(name: str, **kwargs) -> str:
|
||
"""Own template dir (templates/QA/) — deliberately separate from the pipeline prompts."""
|
||
from config import TEMPLATES_DIR
|
||
return (TEMPLATES_DIR / "QA" / f"{name}.md").read_text(encoding="utf-8").format(**kwargs)
|
||
|
||
|
||
async def _llm_verdicts(template: str, topic: str, key: str, items: list[str]) -> dict[int, str]:
|
||
from agents import run_agent
|
||
from pipeline import _yesno_schema
|
||
from jsonio import parse_json_text
|
||
listing = "\n\n".join(f"{k}. {it}" for k, it in enumerate(items, 1))
|
||
slot = {"Dubletten": "pairs", "Luecken": "sections", "Bausteine": "blocks", "Sub": "pairs"}[template.split("-")[1]]
|
||
rc, out, _err = await run_agent(f"qa-{topic}-{key}", _qa_prompt(template, topic=topic, extra="", **{slot: listing}),
|
||
600, role="judge", capabilities="none", scope=topic, label=f"QA {key}")
|
||
return (_yesno_schema(parse_json_text(out)) or {}) if rc == 0 else {}
|
||
|
||
|
||
# ── Report ──────────────────────────────────────────────────────────────────────────
|
||
|
||
async def qa_report(topic: str, llm: bool = False) -> dict | None:
|
||
cards = await db.kanban_cards(topic, board="inventory", stage="done_block")
|
||
if not cards:
|
||
print(f"Keine fertigen Blöcke für '{topic}' — Tippfehler im Namen oder Lauf nicht durch?")
|
||
return None
|
||
blocks = [{"title": c["payload"].get("title", ""), "description": c["payload"].get("description", ""),
|
||
"sources": c["payload"].get("sources") or []} for c in cards]
|
||
sub_rows = [dict(r) for bn in {_norm_title(b["title"]) for b in blocks}
|
||
for r in await db.list_subblocks(topic, bn)]
|
||
subs_by_norm: dict[str, list[str]] = {}
|
||
for r in sub_rows:
|
||
if r["status"] != "variant":
|
||
subs_by_norm.setdefault(r["block_norm"], []).append(r["sub_title"])
|
||
corpus = _corpus_texts(topic)
|
||
|
||
d = dubletten(blocks)
|
||
sd = sub_dubletten(sub_rows)
|
||
lk = luecken(blocks, subs_by_norm, corpus) if corpus else []
|
||
fr = fremd(blocks, corpus) if corpus else []
|
||
bl = beleg(blocks, sub_rows)
|
||
hy = hygiene(blocks)
|
||
n_sections = sum(len(_sections(t)) for t in corpus.values()) or 1
|
||
|
||
if llm and d:
|
||
v = await _llm_verdicts("QA-Dubletten", topic, "dubletten",
|
||
[f"A: {p['a']}\nB: {p['b']}" for p in d[:LLM_SAMPLE]])
|
||
for k, p in enumerate(d[:LLM_SAMPLE], 1):
|
||
p["llm"] = v.get(k, "?")
|
||
if llm and lk:
|
||
v = await _llm_verdicts("QA-Luecken", topic, "luecken",
|
||
[f"[{x['datei']} #{x['abschnitt']}] {x['vorschau']}" for x in lk[:LLM_SAMPLE]])
|
||
for k, x in enumerate(lk[:LLM_SAMPLE], 1):
|
||
x["llm"] = v.get(k, "?")
|
||
if llm and sd: # full coverage in chunks — a sampled quota would mislead the note
|
||
for lo in range(0, len(sd), 40):
|
||
chunk = sd[lo:lo + 40]
|
||
v = await _llm_verdicts("QA-Sub-Dubletten", topic, f"sub-dubletten-{lo}",
|
||
[f"A: {p['a']}\nB: {p['b']}" for p in chunk])
|
||
for k, p in enumerate(chunk, 1):
|
||
p["llm"] = v.get(k, "?")
|
||
unecht: list[str] | None = None
|
||
if llm and blocks:
|
||
verdacht = []
|
||
for lo in range(0, len(blocks), 80): # ein Call je 80 Titel
|
||
chunk = blocks[lo:lo + 80]
|
||
v = await _llm_verdicts("QA-Bausteine", topic, f"bausteine-{lo}",
|
||
[f"{b['title']} — {b['description'] or '(ohne Beschreibung)'}" for b in chunk])
|
||
verdacht += [b for k, b in enumerate(chunk, 1) if v.get(k) == "nein"]
|
||
# Bestätiger-Pass nur über die Geflaggten: der Einzel-Judge flaggte pro Lauf ANDERE
|
||
# Blöcke (gemessen aak: Note pendelte 9.3↔10.0 bei identischem Bestand) — nur
|
||
# doppelt-„nein" zählt; Repair hat als dritte Sicherung die eigene Zweitmeinung
|
||
unecht = []
|
||
if verdacht:
|
||
v2 = await _llm_verdicts("QA-Bausteine", topic, "bausteine-b2",
|
||
[f"{b['title']} — {b['description'] or '(ohne Beschreibung)'}" for b in verdacht])
|
||
unecht = [b["title"] for k, b in enumerate(verdacht, 1) if v2.get(k) == "nein"]
|
||
|
||
art_rows = [dict(r) for r in await db.get_sub_artefakte(topic)]
|
||
fragen = [dict(r) for r in await db.list_question_pattern(topic)]
|
||
art = artefakte(sub_rows, art_rows, fragen)
|
||
quoten_art: dict[str, float] = {}
|
||
if sub_rows:
|
||
quoten_art["subs_ohne_beleg"] = round(len(bl["subs_ohne_beleg"]) / len(sub_rows), 3)
|
||
if art.get("status") == "ok":
|
||
quoten_art["verwaiste"] = round(len(art["verwaiste"]) / max(len(art_rows) + len(fragen), 1), 3)
|
||
n_cons = sum(1 for r in sub_rows if r["status"] == "consensus")
|
||
if n_cons:
|
||
quoten_art["sub_dubletten_verdacht"] = round(len(sd) / n_cons, 3)
|
||
if llm: # confirmed pairs only — the bare candidate list is suspicion, not damage
|
||
quoten_art["sub_dubletten"] = round(sum(1 for p in sd if p.get("llm") == "ja") / n_cons, 3)
|
||
summary = _json_file(arbeit_dir(topic) / "lauf-summary.json") or {}
|
||
report = {
|
||
"topic": topic, "erstellt": datetime.now(timezone.utc).isoformat(),
|
||
"run_id": summary.get("run_id", ""), "bloecke": len(blocks),
|
||
"quoten": {
|
||
"dubletten_verdacht": round(len(d) / max(len(blocks), 1), 3),
|
||
"luecken": round(len(_zaehlbare_luecken(lk, llm)) / n_sections, 3),
|
||
"fremd": round(len(fr) / max(len(blocks), 1), 3),
|
||
"hygiene": round(len(hy) / max(len(blocks), 1), 3),
|
||
**({"unechte_bloecke": round(len(unecht) / max(len(blocks), 1), 3)} if unecht is not None else {}),
|
||
},
|
||
"quoten_artefakte": quoten_art,
|
||
**({"unecht": unecht} if unecht is not None else {}),
|
||
"dubletten": d, "sub_dubletten": sd, "luecken": lk, "fremd": fr, "beleg": bl, "hygiene": hy,
|
||
"artefakte": art,
|
||
"lauf": summary,
|
||
}
|
||
report["note"] = note(report["quoten"])
|
||
# None statt 10.0, solange Board 2 nichts geliefert hat — nichts gemessen ist keine Bestnote
|
||
report["note_artefakte"] = note(quoten_art, NOTE_GEWICHTE_ARTEFAKTE) if quoten_art else None
|
||
report["note_gewichte"] = {"inventar": NOTE_GEWICHTE, "artefakte": NOTE_GEWICHTE_ARTEFAKTE}
|
||
return report
|
||
|
||
|
||
def _diff(prev: dict | None, cur: dict) -> dict:
|
||
if not prev:
|
||
return {}
|
||
# ältere Reports führten die Artefakt-Quoten noch unter "quoten"
|
||
alt = {**prev.get("quoten", {}), **prev.get("quoten_artefakte", {})}
|
||
neu = {**cur["quoten"], **cur.get("quoten_artefakte", {})}
|
||
return {k: round(v - alt.get(k, 0), 3) for k, v in neu.items()}
|
||
|
||
|
||
def _write_report(report: dict) -> Path:
|
||
tdir = QA_DIR / report["topic"]
|
||
tdir.mkdir(parents=True, exist_ok=True)
|
||
# by mtime: run-id names (…-1311-5e5c) and timestamp names don't sort lexicographically.
|
||
# guide-* reports share the directory but are a SEPARATE series (guide_qa.py).
|
||
older = sorted((p for p in tdir.glob("*.json") if not p.name.startswith("guide-")),
|
||
key=lambda p: p.stat().st_mtime)
|
||
prev = _json_file(older[-1]) if older else None
|
||
report["diff_zum_vorlauf"] = _diff(prev, report)
|
||
name = report["run_id"] or datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
|
||
path = tdir / f"{name}.json"
|
||
atomic_write_json(path, report, indent=1)
|
||
return path
|
||
|
||
|
||
async def write_report(report: dict) -> Path:
|
||
"""_write_report + kompaktes kind='qa'-Event. Die Report-JSONs liegen nur auf der
|
||
Lauf-Maschine (storage/qa/) — ein DB-Pull reichte nicht, um Note/Quoten eines Runs
|
||
zu rekonstruieren (Analyse 20260704-1452-b223). Nur die Kennzahlen, kein Volltext;
|
||
run_id stempelt add_event aus der Registry (gesetzt im Lauf, leer bei manueller QA)."""
|
||
path = await asyncio.to_thread(_write_report, report)
|
||
try: # Event ist Komfort — ein DB-Fehler darf den Report nicht kosten (fail-open)
|
||
await db.add_event(report["topic"], "qa", key=path.stem, meta={
|
||
"note": report["note"], "note_artefakte": report.get("note_artefakte"),
|
||
"quoten": report["quoten"], "quoten_artefakte": report.get("quoten_artefakte", {})})
|
||
except Exception:
|
||
pass
|
||
return path
|
||
|
||
|
||
def _digest(report: dict, path: Path):
|
||
na = report.get("note_artefakte")
|
||
print(f"QA {report['topic']} — {report['bloecke']} Blöcke (run {report['run_id'] or '—'})"
|
||
f" — Inventar {report['note']}/10 · Artefakte {f'{na}/10' if na is not None else '—'}")
|
||
for k, v in {**report["quoten"], **report.get("quoten_artefakte", {})}.items():
|
||
delta = report.get("diff_zum_vorlauf", {}).get(k)
|
||
d = f" ({'+' if delta > 0 else ''}{delta})" if delta else ""
|
||
print(f" {k:20} {v:6.1%}{d}")
|
||
for p in report["dubletten"][:8]:
|
||
print(f" DUBLETTE? {p['a']} <-> {p['b']} {p['signale']}{' LLM:' + p['llm'] if 'llm' in p else ''}")
|
||
for p in report.get("sub_dubletten", [])[:8]:
|
||
print(f" SUB-DUP? {p['a']} <-> {p['b']} cos={p['cos']}{' LLM:' + p['llm'] if 'llm' in p else ''}")
|
||
for t in report["fremd"][:8]:
|
||
print(f" FREMD? {t}")
|
||
for t in report.get("unecht", [])[:8]:
|
||
print(f" UNECHT {t}")
|
||
art = report["artefakte"]
|
||
if art.get("status") == "ok":
|
||
print(f" Artefakte: Frage {art['frage_abdeckung']:.0%} · Flashcard {art['flashcard_abdeckung']:.0%}"
|
||
f" · Beispiel {art['beispiel_abdeckung']:.0%} · verwaist {len(art['verwaiste'])}")
|
||
else:
|
||
print(" Artefakte: nicht generiert (Board 2 nicht gelaufen)")
|
||
print(f"Report: {path}")
|
||
|
||
|
||
async def main(topic: str, llm: bool):
|
||
await db.init_db()
|
||
try:
|
||
report = await qa_report(topic, llm=llm)
|
||
if report is None:
|
||
sys.exit(1)
|
||
_digest(report, await write_report(report))
|
||
finally:
|
||
await db.close_db()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
args = [a for a in sys.argv[1:] if not a.startswith("--")]
|
||
if not args:
|
||
print("Nutzung: python3 qa.py <topic> [--llm]")
|
||
sys.exit(1)
|
||
asyncio.run(main(args[0], "--llm" in sys.argv))
|