246 lines
11 KiB
Python
246 lines
11 KiB
Python
"""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 <topic> [--llm] (oder: make qa-guide TOPIC=<topic> [LLM=1])
|
||
Report: storage/qa/<topic>/guide-<ts>.json + Konsolen-Digest.
|
||
"""
|
||
|
||
import asyncio
|
||
import logging
|
||
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, parse_facts
|
||
|
||
log = logging.getLogger("creator.guide_qa")
|
||
|
||
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"<!--\s*sub:\s*\w+\s*\|\s*(.*?)\s*-->")
|
||
|
||
|
||
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"<!--\s*ausführlich\s*-->", 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."""
|
||
async def _pass(kandidaten: list[dict], tag: str) -> list[str]:
|
||
items = [f"SECTION {c['block']}:\n{_ausfuehrlich(c['md'])[:LLM_SECTION_CHARS]}" for c in kandidaten]
|
||
v = await qa.judge_wave("QA-Guide-Fakten", topic, f"fakten{tag}", "sections", items,
|
||
chunk=5, prefix="qa-guide", label="Guide-QA")
|
||
return [c["block"] for k, c in enumerate(kandidaten, 1) if v.get(k) == "ja"]
|
||
|
||
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
|
||
subs_by_norm.setdefault(r["block_norm"], []).append(
|
||
{"relevance": r["relevance"], "facts": parse_facts(r["facts"])})
|
||
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 <topic> [--llm]")
|
||
sys.exit(1)
|
||
asyncio.run(main(args[0], "--llm" in sys.argv))
|