201 lines
8.5 KiB
Python
201 lines
8.5 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 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
|
|
|
|
LAENGE_MIN = 150 # Zeichen je relevantem Sub im ausführlich-Teil (Untergrenze)
|
|
LAENGE_MAX = 1200 # Obergrenze — außerhalb = Tiefen-Lotterie statt Zerlegungs-Signal
|
|
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 _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 = {_norm_title(m) for m in _MARKER.findall(c["md"] or "")}
|
|
for sn in sorted(subs_rel.get(c["block_norm"], set())):
|
|
if sn not in marker and not any(m.startswith(sn) or sn.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
|
|
|
|
|
|
def laengen_ausreisser(cards: list[dict], subs_rel: dict[str, set]) -> list[dict]:
|
|
out = []
|
|
for c in cards:
|
|
n = max(len(subs_rel.get(c["block_norm"], set())), 1)
|
|
pro_sub = len(_ausfuehrlich(c["md"])) / n
|
|
if not (LAENGE_MIN <= pro_sub <= LAENGE_MAX):
|
|
out.append({"block": c["block"], "zeichen_pro_sub": round(pro_sub)})
|
|
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? (eigenes Template)."""
|
|
from agents import run_agent
|
|
from jsonio import parse_json_text
|
|
from pipeline import _yesno_schema
|
|
out = []
|
|
for lo in range(0, len(cards), 5):
|
|
chunk = cards[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-{lo}", qa._qa_prompt("QA-Guide-Fakten", topic=topic, extra="", sections=listing),
|
|
600, role="judge", capabilities="none", scope=topic, label=f"Guide-QA Fakten {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
|
|
|
|
|
|
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] = {}
|
|
for r in await db.list_subblocks(topic):
|
|
if r["status"] == "consensus" and 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, subs_rel)
|
|
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))
|