feat(scripts): add prompt-test evaluator and suite runner
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>
This commit is contained in:
parent
3e3dd1c150
commit
6f2f8aff6c
4 changed files with 556 additions and 2 deletions
457
scripts/eval_prompt_tests.py
Executable file
457
scripts/eval_prompt_tests.py
Executable file
|
|
@ -0,0 +1,457 @@
|
|||
#!/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())
|
||||
Loading…
Add table
Add a link
Reference in a new issue