"""Unabhängiges Guide-Audit über einen FERTIGEN Guide — read-only. Misst den gebauten Guide (guide_cards) gegen Lernziele und Sub-Satz mit Detektoren, die bewusst NICHT die Pipeline-Gates wiederverwenden (covered-Flag, Fakten-Gate) — geteilte blinde Flecken machen das Audit wertlos. Geteilt nur Infra: DB, readability, Agent-Runner (--llm), Note-Formel aus qa.py. CLI: python3 guide_qa.py [--llm] (oder: make qa-guide TOPIC= [LLM=1]) Report: storage/qa//guide-.json + Konsolen-Digest. """ import asyncio import json import re import sys from datetime import datetime, timezone import database as db import qa import readability from fsutil import atomic_write_json from textkit import _norm_title JACCARD_ABSATZ = 0.6 # Wort-Jaccard, ab dem zwei Absätze als Doppel gelten ABSATZ_MIN_CHARS = 200 # kürzere Absätze sind Übergänge — kein Dubletten-Signal LLM_SECTION_CHARS = 2500 # Section-Auszug je Judge-Item # fachliche Fehler wiegen am schwersten; Anker-lose Ziele = Coverage-Behauptung ohne Text. NOTE_GEWICHTE_GUIDE = {"fachlich_falsch": 3.0, "ziel_ohne_anker": 2.0, "marker_fehlend": 1.5, "redundanz": 1.0, "laengen_ausreisser": 0.5, "lesbarkeit": 0.5} _MARKER = re.compile(r"") def _mnorm(s: str) -> str: """Marker-/Sub-Norm ohne Backslashes — escapte Titel (`h\\~2\\~o`) erzeugten falsch-positive „Marker fehlt"-Befunde, weil Writer und DB verschieden escapen.""" return _norm_title(s.replace("\\", "")) def _ausfuehrlich(md: str) -> str: """Der Lern-Fließtext einer Karte (hinter dem ausführlich-Marker, sonst alles).""" teile = re.split(r"", md or "", maxsplit=1) return teile[1] if len(teile) == 2 else (md or "") def marker_fehlend(cards: list[dict], subs_rel: dict[str, set]) -> list[str]: """Relevante Subs ohne Sub-Marker in der Section — der Level-Filter verliert sie.""" out = [] for c in cards: marker = {_mnorm(m) for m in _MARKER.findall(c["md"] or "")} for sn in sorted(subs_rel.get(c["block_norm"], set())): mn = _mnorm(sn) if mn not in marker and not any(m.startswith(mn) or mn.startswith(m) for m in marker): out.append(f"{c['block']} · {sn}") return out def ziel_ohne_anker(cards: list[dict], ziele: list[dict]) -> list[str]: """Lernziele, deren distinktive Tokens im Section-Text fehlen — eigener Anker-Check, NICHT das covered-Flag der Pipeline (das hat der Coverage-Judge selbst gesetzt).""" text_by_norm = {c["block_norm"]: qa._tokens(_ausfuehrlich(c["md"])) for c in cards} out = [] for z in ziele: toks = qa._distinctive(z["text"]) st = text_by_norm.get(z["block_norm"]) if st is None or not toks: continue if len(toks & st) < min(2, len(toks)): out.append(f"{z['block_norm']} · ({z['ziel_id']}) {z['text'][:60]}") return out # Längenbudget je Sub aus der Inventar-Substanz — ersetzt den festen Rahmen 150–1200/Sub: # ein dichter Sub (viele key_points, Fakten, Beispiel) trägt mehr Text als ein Einzeiler. # Die Pipeline (Writer-Vorgabe, Prüfer-Trigger, Fix-Ziel) nutzt DIESELBE Formel mit engerem # Band — Messlatte und Fix-Auftrag müssen übereinstimmen, sonst sind Befunde unfixbar. BUDGET_BASIS = 200 # Einstieg/Übergang je Sub BUDGET_KEY_POINT = 160 # ~1–2 Sätze Erklärung je key_point BUDGET_FAKT = 60 # zitierter Fakt, in den Text eingewoben BUDGET_BEISPIEL = 250 # ausgearbeitetes Beispiel LAENGE_BAND = (0.35, 1.5) # QA-Toleranz um das Blockbudget def sub_budget(facts: dict) -> int: """Zeichenbudget für den ausführlich-Teil EINES Subs (facts = Inventar-JSON des Subs).""" kp = len(facts.get("key_points") or []) cf = len(facts.get("cited_facts") or []) ex = 1 if str(facts.get("example_idea") or "").strip() else 0 return BUDGET_BASIS + BUDGET_KEY_POINT * kp + BUDGET_FAKT * cf + BUDGET_BEISPIEL * ex def block_budget(subs: list[dict]) -> int: """Budget einer Section: Summe über die relevanten Subs ({relevance, facts}-Dicts).""" return max(BUDGET_BASIS, sum(sub_budget(s.get("facts") or {}) for s in subs if s.get("relevance") != "peripheral")) def laengen_ausreisser(cards: list[dict], budget_by_norm: dict[str, int]) -> list[dict]: out = [] for c in cards: budget = budget_by_norm.get(c["block_norm"]) if not budget: continue zeichen = len(_ausfuehrlich(c["md"])) if not (LAENGE_BAND[0] * budget <= zeichen <= LAENGE_BAND[1] * budget): out.append({"block": c["block"], "zeichen": zeichen, "budget": budget}) return out def redundanz(cards: list[dict]) -> list[dict]: """Absatz-Paare topic-weit mit hoher Token-Überlappung — derselbe Stoff doppelt erklärt.""" absaetze = [] for c in cards: for a in _ausfuehrlich(c["md"]).split("\n\n"): a = a.strip() if len(a) >= ABSATZ_MIN_CHARS: absaetze.append((c["block"], a, qa._tokens(a))) out = [] for i in range(len(absaetze)): for j in range(i + 1, len(absaetze)): if qa._jaccard(absaetze[i][2], absaetze[j][2]) >= JACCARD_ABSATZ: out.append({"a": f"{absaetze[i][0]}: {absaetze[i][1][:60]}", "b": f"{absaetze[j][0]}: {absaetze[j][1][:60]}"}) return out def lesbarkeit(cards: list[dict]) -> list[str]: """Deterministisches externes Rating; Modell nicht ladbar → nicht gemessen (zählt nicht).""" try: hints = readability.rate_sections({i: _ausfuehrlich(c["md"]) for i, c in enumerate(cards, 1)}) except Exception: return [] return [f"{cards[i - 1]['block']}: {h}" for i, h in sorted(hints.items()) if h] async def _fachlich_falsch(topic: str, cards: list[dict]) -> list[str]: """LLM-Stichprobe: Section enthält eine fachlich falsche Aussage? Zwei unabhängige Durchgänge, nur DOPPELT bestätigte zählen — ein Einzel-Judge schwankte zwischen 0 und 5 Befunden am selben Guide und kippte die Note (Gewicht 3.0) auf 0.""" from agents import run_agent from jsonio import parse_json_text from pipeline import _yesno_schema async def _pass(kandidaten: list[dict], tag: str) -> list[str]: out = [] for lo in range(0, len(kandidaten), 5): chunk = kandidaten[lo:lo + 5] listing = "\n\n".join(f"{k}. SECTION {c['block']}:\n{_ausfuehrlich(c['md'])[:LLM_SECTION_CHARS]}" for k, c in enumerate(chunk, 1)) rc, txt, _err = await run_agent( f"qa-guide-{topic}-fakten{tag}-{lo}", qa._qa_prompt("QA-Guide-Fakten", topic=topic, extra="", sections=listing), 600, role="judge", capabilities="none", scope=topic, label=f"Guide-QA Fakten{tag} {lo}") v = (_yesno_schema(parse_json_text(txt)) or {}) if rc == 0 else {} out += [c["block"] for k, c in enumerate(chunk, 1) if v.get(k) == "ja"] return out verdacht = await _pass(cards, "") if not verdacht: return [] # ZWEI unabhängige Bestätiger, beide müssen zustimmen — mit nur einem sprang die # Note desselben Guides weiter zwischen 2.0 und 6.6 (ein Zufalls-ja kostet 1.5 Punkte) kandidaten = [c for c in cards if c["block"] in set(verdacht)] b1 = set(await _pass(kandidaten, "-2")) if not b1: return [] b2 = set(await _pass([c for c in kandidaten if c["block"] in b1], "-3")) return [b for b in verdacht if b in b1 and b in b2] async def guide_qa_report(topic: str, llm: bool = False) -> dict | None: cards = [dict(r) for r in await db.list_guide_cards(topic)] cards = [c for c in cards if (c.get("md") or "").strip()] if not cards: print(f"Keine Guide-Karten für '{topic}' — Guide noch nicht gebaut?") return None subs_rel: dict[str, set] = {} subs_by_norm: dict[str, list[dict]] = {} for r in await db.list_subblocks(topic): if r["status"] != "consensus": continue try: facts = json.loads(r["facts"]) if r["facts"] else {} except (ValueError, TypeError): facts = {} subs_by_norm.setdefault(r["block_norm"], []).append( {"relevance": r["relevance"], "facts": facts if isinstance(facts, dict) else {}}) if r["relevance"] != "peripheral": subs_rel.setdefault(r["block_norm"], set()).add(r["sub_norm"]) ziele = [dict(r) for r in await db.list_lernziele(topic)] mf = marker_fehlend(cards, subs_rel) za = ziel_ohne_anker(cards, ziele) la = laengen_ausreisser(cards, {n: block_budget(s) for n, s in subs_by_norm.items()}) rd = redundanz(cards) lb = lesbarkeit(cards) falsch = await _fachlich_falsch(topic, cards) if llm else None n_subs = max(sum(len(s) for s in subs_rel.values()), 1) n_abs = max(sum(len([a for a in _ausfuehrlich(c["md"]).split("\n\n") if len(a.strip()) >= ABSATZ_MIN_CHARS]) for c in cards), 1) quoten = { "marker_fehlend": round(len(mf) / n_subs, 3), "ziel_ohne_anker": round(len(za) / max(len(ziele), 1), 3), "laengen_ausreisser": round(len(la) / len(cards), 3), "redundanz": round(len(rd) / n_abs, 3), "lesbarkeit": round(len(lb) / len(cards), 3), **({"fachlich_falsch": round(len(falsch) / len(cards), 3)} if falsch is not None else {}), } report = { "topic": topic, "erstellt": datetime.now(timezone.utc).isoformat(), "art": "guide", "bloecke": len(cards), "ziele": len(ziele), "quoten": quoten, "note_guide": qa.note(quoten, NOTE_GEWICHTE_GUIDE), "marker_fehlend": mf, "ziel_ohne_anker": za, "laengen_ausreisser": la, "redundanz": rd[:20], "lesbarkeit": lb, **({"fachlich_falsch": falsch} if falsch is not None else {}), "note_gewichte": NOTE_GEWICHTE_GUIDE, } return report def _write_report(report: dict): tdir = qa.QA_DIR / report["topic"] tdir.mkdir(parents=True, exist_ok=True) path = tdir / f"guide-{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}.json" atomic_write_json(path, report, indent=1) return path def _digest(report: dict, path): print(f"Guide-QA {report['topic']} — {report['bloecke']} Sections, {report['ziele']} Ziele" f" — Note {report['note_guide']}/10") for k, v in report["quoten"].items(): print(f" {k:20} {v:6.1%}") for k in ("marker_fehlend", "ziel_ohne_anker", "lesbarkeit", "fachlich_falsch"): for x in report.get(k, [])[:5]: print(f" {k.upper():16} {str(x)[:90]}") for p in report.get("redundanz", [])[:5]: print(f" DOPPELT? {p['a'][:55]} <-> {p['b'][:55]}") print(f"Report: {path}") async def main(topic: str, llm: bool): await db.init_db() try: report = await guide_qa_report(topic, llm=llm) if report is None: sys.exit(1) _digest(report, _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 guide_qa.py [--llm]") sys.exit(1) asyncio.run(main(args[0], "--llm" in sys.argv))