dschlueter
6f2f8aff6c
eval_prompt_tests.py measures the objective half of docs/EVAL_RUBRIC.md over the manual test archive: word count against the target stated in each prompt, truncation suspicion, and — for the coding domain — it writes the generated module and tests to a temp dir and actually runs pytest against them. Deriving the module's filename is the delicate part: a name taken from a test's `import sqlite3` would shadow the stdlib and fail the run for a reason the model is not responsible for. Names now come from the last *.py mention before the block, then from `from X import`, and anything in sys.stdlib_module_names is rejected. A module that no test imports is reported as such, since that is a finding about test quality rather than a guess the runner got wrong. run_prompt_suite.sh drives one prompt domain against a running profile and stores the outputs under the archive's naming convention. Both scripts join the ruff gate. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
457 lines
17 KiB
Python
Executable file
457 lines
17 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""eval_prompt_tests.py
|
|
|
|
Auswertung der manuellen Modell-Prompt-Testlaeufe unter
|
|
``~/llamacppctl_prompt_tests/`` gegen die Prompts in ``example_user_prompts/``.
|
|
|
|
Misst die objektive Ebene der Rubrik (docs/EVAL_RUBRIC.md):
|
|
|
|
* prosa/reden: Wortzahl und Abweichung vom im Prompt genannten Zielwert,
|
|
Trunkierungsverdacht.
|
|
* coding: extrahiert die Python-Codebloecke, schreibt Modul und Tests in
|
|
ein temporaeres Verzeichnis und **fuehrt pytest tatsaechlich aus**.
|
|
|
|
WARNUNG: ``--exec`` fuehrt modellgenerierten Code aus. Der Runner nutzt ein
|
|
temporaeres Verzeichnis, ein Zeitlimit und verweigert den Start als root, bietet
|
|
aber keine Netz- oder Dateisystem-Isolation. Siehe docs/EVAL_RUBRIC.md.
|
|
|
|
Beispiele
|
|
---------
|
|
scripts/eval_prompt_tests.py # messen + Tests ausfuehren
|
|
scripts/eval_prompt_tests.py --no-exec # nur statisch messen
|
|
scripts/eval_prompt_tests.py --json report.json # maschinenlesbar
|
|
scripts/eval_prompt_tests.py --anonymize blind/ # blinde Bewertung vorbereiten
|
|
|
|
Exit-Codes
|
|
----------
|
|
0 Auswertung durchgelaufen (auch wenn einzelne Testlaeufe rot sind)
|
|
1 Eingabeverzeichnis fehlt oder unsichere Ausfuehrungsumgebung
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import random
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from dataclasses import asdict, dataclass, field
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
DEFAULT_RESULTS_DIR = Path.home() / "llamacppctl_prompt_tests"
|
|
DEFAULT_PROMPTS_DIR = Path(__file__).resolve().parents[1] / "example_user_prompts"
|
|
|
|
DOMAINS = ("prosa", "reden", "coding")
|
|
# Dateien ohne Modellpraefix stammen vom Default-Modell aus llama.cpp.config.
|
|
KNOWN_MODELS = ("carnice", "qwen27b", "qwopus", "ornith", "qwen35base")
|
|
DEFAULT_MODEL_LABEL = "default"
|
|
VARIANT_MARKERS = frozenset({"r2", "highbudget", "explizit"})
|
|
|
|
# "etwa 1200 Wörtern", "ca. 700 Wörter", "rund 900 Worten"
|
|
LENGTH_TARGET_RE = re.compile(
|
|
r"(?:etwa|ca\.|circa|rund)\s*(\d{2,5})\s*(?:Wörter|Wörtern|Worte|Worten)",
|
|
re.IGNORECASE,
|
|
)
|
|
FENCE_RE = re.compile(r"^```([A-Za-z+#]*)\s*$(.*?)^```\s*$", re.MULTILINE | re.DOTALL)
|
|
PY_FILENAME_RE = re.compile(r"`([A-Za-z_][A-Za-z0-9_]*)\.py`")
|
|
FROM_IMPORT_RE = re.compile(r"^\s*from\s+([A-Za-z_][A-Za-z0-9_]*)\s+import", re.MULTILINE)
|
|
PLAIN_IMPORT_RE = re.compile(r"^\s*import\s+([A-Za-z_][A-Za-z0-9_]*)", re.MULTILINE)
|
|
PYTEST_COUNT_RE = re.compile(r"(\d+)\s+(passed|failed|error|errors)")
|
|
|
|
# Endet der Text sauber? Fehlendes Satzzeichen deutet auf finish_reason=length.
|
|
SENTENCE_END = tuple('.!?"»“’—-)')
|
|
|
|
# Ein Modul darf niemals nach einem Stdlib-Modul benannt werden: eine Datei
|
|
# sqlite3.py im Arbeitsverzeichnis ueberschattet die echte Stdlib und laesst den
|
|
# Testlauf mit einem Importfehler scheitern, der faelschlich dem Modell
|
|
# angelastet wuerde. sys.stdlib_module_names ist exakt (Python >= 3.10).
|
|
RESERVED_MODULE_NAMES = frozenset(sys.stdlib_module_names) | {"pytest", "conftest"}
|
|
|
|
|
|
@dataclass
|
|
class Run:
|
|
"""Ein einzelner Testlauf, abgeleitet aus dem Dateinamen."""
|
|
|
|
path: Path
|
|
model: str
|
|
domain: str
|
|
case: str
|
|
variant: str = ""
|
|
|
|
@property
|
|
def label(self) -> str:
|
|
base = f"{self.model}/{self.domain}_{self.case}"
|
|
return f"{base}[{self.variant}]" if self.variant else base
|
|
|
|
|
|
@dataclass
|
|
class Metrics:
|
|
label: str
|
|
model: str
|
|
domain: str
|
|
case: str
|
|
variant: str
|
|
words: int = 0
|
|
target_words: Optional[int] = None
|
|
deviation_pct: Optional[float] = None
|
|
looks_truncated: bool = False
|
|
# coding
|
|
executed: bool = False
|
|
exec_status: str = "" # ok | failed | timeout | no-tests | skipped-<grund>
|
|
tests_passed: int = 0
|
|
tests_failed: int = 0
|
|
detail: str = ""
|
|
code_langs: list = field(default_factory=list)
|
|
|
|
|
|
def parse_run(path: Path) -> Optional[Run]:
|
|
"""Zerlegt ``[<modell>_]<domaene>_<nr>[_<variante>].out.txt``.
|
|
|
|
Die Namenskonvention ist historisch uneinheitlich (mal steht der Fallname
|
|
hinter der Nummer, mal eine Variante wie ``r2``), deshalb wird tolerant
|
|
geparst statt streng validiert.
|
|
"""
|
|
stem = path.name
|
|
for suffix in (".out.txt", ".txt", ".md"):
|
|
if stem.endswith(suffix):
|
|
stem = stem[: -len(suffix)]
|
|
break
|
|
|
|
tokens = stem.split("_")
|
|
if not tokens:
|
|
return None
|
|
|
|
model = DEFAULT_MODEL_LABEL
|
|
if tokens[0] in KNOWN_MODELS:
|
|
model = tokens[0]
|
|
tokens = tokens[1:]
|
|
|
|
domain = next((t for t in tokens if t in DOMAINS), None)
|
|
if domain is None:
|
|
return None
|
|
rest = tokens[tokens.index(domain) + 1 :]
|
|
|
|
case = next((t for t in rest if t.isdigit()), None)
|
|
if case is None:
|
|
return None
|
|
|
|
# Der Fallname (z. B. "jsonl_validator") ist keine Variante. Nur bekannte
|
|
# Marker zaehlen -- sie muessen erhalten bleiben, sonst kollidieren zwei
|
|
# verschiedene Laeufe (prosa_04_dialog vs. prosa_04_dialog_highbudget) unter
|
|
# demselben Label.
|
|
variant = "_".join(t for t in rest[rest.index(case) + 1 :] if t in VARIANT_MARKERS)
|
|
|
|
return Run(path=path, model=model, domain=domain, case=case, variant=variant)
|
|
|
|
|
|
def load_length_targets(prompts_dir: Path) -> dict:
|
|
"""Mappt (domaene, nr) auf die im Prompt genannte Ziel-Wortzahl."""
|
|
targets: dict = {}
|
|
if not prompts_dir.is_dir():
|
|
return targets
|
|
for prompt in sorted(prompts_dir.glob("*.md")):
|
|
tokens = prompt.stem.split("_")
|
|
if len(tokens) < 2 or tokens[0] not in DOMAINS or not tokens[1].isdigit():
|
|
continue
|
|
match = LENGTH_TARGET_RE.search(prompt.read_text(encoding="utf-8"))
|
|
if match:
|
|
targets[(tokens[0], tokens[1])] = int(match.group(1))
|
|
return targets
|
|
|
|
|
|
def count_words(text: str) -> int:
|
|
"""Woerter ohne Markdown-Auszeichnung und ohne Codebloecke."""
|
|
without_code = FENCE_RE.sub("", text)
|
|
cleaned = re.sub(r"[#*_>`]", " ", without_code)
|
|
return len(cleaned.split())
|
|
|
|
|
|
def looks_truncated(text: str) -> bool:
|
|
stripped = text.strip()
|
|
if not stripped:
|
|
return True
|
|
return not stripped.endswith(SENTENCE_END)
|
|
|
|
|
|
def extract_code_blocks(text: str) -> list:
|
|
"""Liefert (sprache, code, start_offset) je Fence-Block.
|
|
|
|
Der Offset wird gebraucht, um den im Fliesstext *vor* einem Block genannten
|
|
Dateinamen zu finden; ein Textvergleich waere mehrdeutig, wenn derselbe Code
|
|
mehrfach vorkommt.
|
|
"""
|
|
return [
|
|
(match.group(1).lower(), match.group(2), match.start())
|
|
for match in FENCE_RE.finditer(text)
|
|
]
|
|
|
|
|
|
def is_test_block(code: str) -> bool:
|
|
return "def test_" in code or "import pytest" in code
|
|
|
|
|
|
def _usable(name: str) -> bool:
|
|
return bool(name) and name not in RESERVED_MODULE_NAMES and not name.startswith("test_")
|
|
|
|
|
|
def infer_module_name(markdown: str, block_start: int, tests: list) -> str:
|
|
"""Ermittelt den Dateinamen des Moduls.
|
|
|
|
Reihenfolge: (1) letzte im Fliesstext *vor* dem Block genannte ``*.py``-Datei,
|
|
(2) ``from X import`` in den Tests -- die Tests importieren das Modul unter
|
|
genau diesem Namen, (3) ``import X``. Stdlib-Namen und ``test_*`` werden in
|
|
jeder Stufe verworfen.
|
|
"""
|
|
for name in reversed(PY_FILENAME_RE.findall(markdown[:block_start])):
|
|
if _usable(name):
|
|
return name
|
|
|
|
for pattern in (FROM_IMPORT_RE, PLAIN_IMPORT_RE):
|
|
for test_code in tests:
|
|
for candidate in pattern.findall(test_code):
|
|
if _usable(candidate):
|
|
return candidate
|
|
|
|
return "generated_module"
|
|
|
|
|
|
def _parse_pytest_counts(output: str) -> tuple:
|
|
passed = failed = 0
|
|
for count, kind in PYTEST_COUNT_RE.findall(output):
|
|
if kind == "passed":
|
|
passed = int(count)
|
|
else:
|
|
failed += int(count)
|
|
return passed, failed
|
|
|
|
|
|
def run_python_case(markdown: str, timeout: int, keep_dir: Optional[Path]) -> dict:
|
|
"""Schreibt Modul + Tests in ein Temp-Verzeichnis und ruft pytest auf."""
|
|
blocks = extract_code_blocks(markdown)
|
|
py_blocks = [(body, start) for lang, body, start in blocks if lang in ("python", "py")]
|
|
if not py_blocks:
|
|
return {"exec_status": "no-code", "detail": "kein Python-Block gefunden"}
|
|
|
|
tests = [body for body, _ in py_blocks if is_test_block(body)]
|
|
modules = [(body, start) for body, start in py_blocks if not is_test_block(body)]
|
|
if not tests:
|
|
return {"exec_status": "no-tests", "detail": f"{len(modules)} Modul-Block(loecke), keine Tests"}
|
|
|
|
workdir = Path(tempfile.mkdtemp(prefix="llamaeval_"))
|
|
warnings = []
|
|
try:
|
|
used: set = set()
|
|
for module_body, block_start in modules:
|
|
name = infer_module_name(markdown, block_start, tests)
|
|
if name == "generated_module":
|
|
# Weder der Fliesstext nennt einen Dateinamen noch importieren die
|
|
# Tests das Modul: ein echter Mangel der Testqualitaet, kein
|
|
# Ratefehler des Runners.
|
|
warnings.append("Tests importieren das Modul nicht")
|
|
while name in used: # zwei Modul-Bloecke, gleicher geratener Name
|
|
name += "_x"
|
|
used.add(name)
|
|
(workdir / f"{name}.py").write_text(module_body, encoding="utf-8")
|
|
for index, test_body in enumerate(tests):
|
|
suffix = "" if index == 0 else f"_{index}"
|
|
(workdir / f"test_generated{suffix}.py").write_text(test_body, encoding="utf-8")
|
|
|
|
try:
|
|
proc = subprocess.run(
|
|
[sys.executable, "-m", "pytest", "-q", "--tb=no", "-p", "no:cacheprovider"],
|
|
cwd=workdir,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
return {"exec_status": "timeout", "detail": f"pytest > {timeout}s"}
|
|
except FileNotFoundError:
|
|
return {"exec_status": "skipped-no-pytest", "detail": "pytest nicht gefunden"}
|
|
|
|
output = proc.stdout + proc.stderr
|
|
passed, failed = _parse_pytest_counts(output)
|
|
status = "ok" if proc.returncode == 0 else "failed"
|
|
last = [ln for ln in output.strip().splitlines() if ln.strip()]
|
|
detail = "; ".join(warnings + ([last[-1]] if last else []))
|
|
return {
|
|
"exec_status": status,
|
|
"tests_passed": passed,
|
|
"tests_failed": failed,
|
|
"detail": detail[:120],
|
|
}
|
|
finally:
|
|
if keep_dir is not None:
|
|
shutil.move(str(workdir), str(keep_dir / workdir.name))
|
|
else:
|
|
shutil.rmtree(workdir, ignore_errors=True)
|
|
|
|
|
|
def evaluate(run: Run, targets: dict, do_exec: bool, timeout: int, keep_dir) -> Metrics:
|
|
text = run.path.read_text(encoding="utf-8", errors="replace")
|
|
metrics = Metrics(
|
|
label=run.label,
|
|
model=run.model,
|
|
domain=run.domain,
|
|
case=run.case,
|
|
variant=run.variant,
|
|
words=count_words(text),
|
|
looks_truncated=looks_truncated(text),
|
|
code_langs=sorted({lang for lang, _, _ in extract_code_blocks(text) if lang}),
|
|
)
|
|
|
|
target = targets.get((run.domain, run.case))
|
|
if target:
|
|
metrics.target_words = target
|
|
metrics.deviation_pct = round((metrics.words - target) / target * 100, 1)
|
|
|
|
if run.domain != "coding":
|
|
return metrics
|
|
|
|
if not do_exec:
|
|
metrics.exec_status = "skipped-no-exec"
|
|
return metrics
|
|
if "typescript" in metrics.code_langs and "python" not in metrics.code_langs:
|
|
metrics.exec_status = "skipped-typescript"
|
|
metrics.detail = "kein tsc im PATH; TypeScript-Lauf nicht implementiert"
|
|
return metrics
|
|
|
|
result = run_python_case(text, timeout, keep_dir)
|
|
metrics.executed = result["exec_status"] in ("ok", "failed")
|
|
metrics.exec_status = result["exec_status"]
|
|
metrics.tests_passed = result.get("tests_passed", 0)
|
|
metrics.tests_failed = result.get("tests_failed", 0)
|
|
metrics.detail = result.get("detail", "")
|
|
return metrics
|
|
|
|
|
|
def anonymize(runs: list, out_dir: Path) -> Path:
|
|
"""Kopiert die Laeufe unter Zufalls-IDs und schreibt die Aufloesungstabelle.
|
|
|
|
Fuer blinde Bewertung: der Bewertende darf das Modell nicht kennen.
|
|
"""
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
ids = [f"text_{i:03d}" for i in range(len(runs))]
|
|
random.shuffle(ids)
|
|
mapping = {}
|
|
for run, blind_id in zip(runs, ids):
|
|
shutil.copyfile(run.path, out_dir / f"{blind_id}.txt")
|
|
mapping[blind_id] = {"model": run.model, "domain": run.domain, "case": run.case,
|
|
"variant": run.variant, "source": run.path.name}
|
|
key_path = out_dir / "AUFLOESUNG.json"
|
|
key_path.write_text(json.dumps(mapping, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
return key_path
|
|
|
|
|
|
def _fmt_dev(metrics: Metrics) -> str:
|
|
if metrics.deviation_pct is None:
|
|
return "—"
|
|
return f"{metrics.deviation_pct:+.1f}%"
|
|
|
|
|
|
def print_report(all_metrics: list) -> None:
|
|
for domain in DOMAINS:
|
|
rows = [m for m in all_metrics if m.domain == domain]
|
|
if not rows:
|
|
continue
|
|
print(f"\n=== {domain} ===")
|
|
if domain == "coding":
|
|
print(f"{'Lauf':<28} {'Status':<18} {'passed':>6} {'failed':>6} Detail")
|
|
for m in sorted(rows, key=lambda r: (r.model, r.case)):
|
|
print(f"{m.label:<28} {m.exec_status:<18} {m.tests_passed:>6} "
|
|
f"{m.tests_failed:>6} {m.detail[:60]}")
|
|
else:
|
|
print(f"{'Lauf':<28} {'Wörter':>7} {'Ziel':>6} {'Abw.':>8} {'trunkiert':<9}")
|
|
for m in sorted(rows, key=lambda r: (r.model, r.case)):
|
|
target = str(m.target_words) if m.target_words else "—"
|
|
trunc = "ja" if m.looks_truncated else ""
|
|
print(f"{m.label:<28} {m.words:>7} {target:>6} {_fmt_dev(m):>8} {trunc:<9}")
|
|
|
|
coding = [m for m in all_metrics if m.domain == "coding" and m.executed]
|
|
if coding:
|
|
green = sum(1 for m in coding if m.exec_status == "ok")
|
|
total_p = sum(m.tests_passed for m in coding)
|
|
total_f = sum(m.tests_failed for m in coding)
|
|
print(f"\nCoding gesamt: {green}/{len(coding)} Läufe grün, "
|
|
f"{total_p} Tests bestanden, {total_f} durchgefallen.")
|
|
|
|
lengths = [m for m in all_metrics if m.deviation_pct is not None]
|
|
if lengths:
|
|
within5 = sum(1 for m in lengths if abs(m.deviation_pct) <= 5)
|
|
worst = max(lengths, key=lambda m: abs(m.deviation_pct))
|
|
print(f"Längentreue: {within5}/{len(lengths)} Läufe im ±5%-Fenster; "
|
|
f"größte Abweichung {_fmt_dev(worst)} ({worst.label}).")
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__.splitlines()[1] if __doc__ else "")
|
|
parser.add_argument("--results-dir", type=Path, default=DEFAULT_RESULTS_DIR)
|
|
parser.add_argument("--prompts-dir", type=Path, default=DEFAULT_PROMPTS_DIR)
|
|
parser.add_argument("--no-exec", dest="exec_code", action="store_false", default=True,
|
|
help="Generierten Code NICHT ausführen, nur statisch messen")
|
|
parser.add_argument("--timeout", type=int, default=60, help="Sekunden pro pytest-Lauf")
|
|
parser.add_argument("--json", type=Path, help="Report zusätzlich als JSON schreiben")
|
|
parser.add_argument("--keep-workdirs", type=Path,
|
|
help="Temp-Verzeichnisse der Testläufe hierhin retten (Debugging)")
|
|
parser.add_argument("--anonymize", type=Path,
|
|
help="Anonymisierte Kopien + Auflösungstabelle für blinde Bewertung")
|
|
args = parser.parse_args()
|
|
|
|
if not args.results_dir.is_dir():
|
|
print(f"Ergebnisverzeichnis nicht gefunden: {args.results_dir}", file=sys.stderr)
|
|
return 1
|
|
|
|
if args.exec_code and hasattr(os, "geteuid") and os.geteuid() == 0:
|
|
print("Verweigert: modellgenerierten Code nicht als root ausführen "
|
|
"(--no-exec erzwingt die statische Auswertung).", file=sys.stderr)
|
|
return 1
|
|
|
|
runs = []
|
|
skipped = []
|
|
for path in sorted(args.results_dir.iterdir()):
|
|
if not path.is_file():
|
|
continue
|
|
run = parse_run(path)
|
|
(runs.append(run) if run else skipped.append(path.name))
|
|
|
|
if not runs:
|
|
print(f"Keine auswertbaren Läufe in {args.results_dir}", file=sys.stderr)
|
|
return 1
|
|
|
|
if args.anonymize:
|
|
key = anonymize(runs, args.anonymize)
|
|
print(f"{len(runs)} Texte anonymisiert nach {args.anonymize}; Auflösung: {key}")
|
|
|
|
if args.keep_workdirs:
|
|
args.keep_workdirs.mkdir(parents=True, exist_ok=True)
|
|
|
|
targets = load_length_targets(args.prompts_dir)
|
|
if not targets:
|
|
print(f"Warnung: keine Ziel-Wortzahlen aus {args.prompts_dir} gelesen", file=sys.stderr)
|
|
|
|
all_metrics = [
|
|
evaluate(run, targets, args.exec_code, args.timeout, args.keep_workdirs)
|
|
for run in runs
|
|
]
|
|
|
|
print_report(all_metrics)
|
|
if skipped:
|
|
print(f"\nÜbersprungen ({len(skipped)}): {', '.join(skipped)}")
|
|
|
|
if args.json:
|
|
args.json.write_text(
|
|
json.dumps([asdict(m) for m in all_metrics], indent=2, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
print(f"JSON-Report: {args.json}")
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|