117 lines
5.0 KiB
Python
117 lines
5.0 KiB
Python
"""Verifier: deterministische Prüfungen + optionaler LLM-Judge."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any
|
|
|
|
import jsonschema
|
|
from pydantic import BaseModel, Field
|
|
|
|
from .llm import ChatClient, LLMError
|
|
from .problems import extract_json
|
|
from .registry import Registry
|
|
from .skills import Skill
|
|
|
|
JUDGE_SYSTEM = (
|
|
"Du bist ein strenger Prüfer. Bewerte, ob die Antwort die Aufgabe gemäß den Kriterien erfüllt. "
|
|
"Antworte mit genau einer Zeile: 'PASS' oder 'FAIL: <kurzer Grund>'."
|
|
)
|
|
|
|
|
|
class CheckResult(BaseModel):
|
|
type: str
|
|
passed: bool | None # None = nicht entscheidbar (z. B. Judge nicht erreichbar)
|
|
detail: str = ""
|
|
|
|
|
|
class VerifyResult(BaseModel):
|
|
passed: bool
|
|
inconclusive: bool = False
|
|
checks: list[CheckResult] = Field(default_factory=list)
|
|
|
|
@property
|
|
def failed_checks(self) -> list[CheckResult]:
|
|
return [c for c in self.checks if c.passed is False]
|
|
|
|
|
|
class Verifier:
|
|
def __init__(self, registry: Registry | None = None, chat: ChatClient | None = None,
|
|
default_judge_slot: str = "verifier-small"):
|
|
self.registry = registry
|
|
self.chat = chat
|
|
self.default_judge_slot = default_judge_slot
|
|
|
|
def verify(self, skill: Skill | None, text: str, output: str) -> VerifyResult:
|
|
if skill is None:
|
|
return VerifyResult(passed=True, checks=[])
|
|
checks: list[CheckResult] = []
|
|
specs = list(skill.meta.verifier)
|
|
schema = skill.meta.output.schema_
|
|
if skill.meta.output.format == "json":
|
|
checks.append(self._json(output, schema))
|
|
specs = [s for s in specs if s.get("type") != "json_schema"]
|
|
for spec in specs:
|
|
checks.append(self._run(spec, skill, text, output))
|
|
conclusive = [c for c in checks if c.passed is not None]
|
|
return VerifyResult(
|
|
passed=all(c.passed for c in conclusive),
|
|
inconclusive=len(conclusive) != len(checks),
|
|
checks=checks,
|
|
)
|
|
|
|
def _run(self, spec: dict[str, Any], skill: Skill, text: str, output: str) -> CheckResult:
|
|
kind = spec.get("type")
|
|
if kind == "json_schema":
|
|
return self._json(output, skill.meta.output.schema_)
|
|
if kind == "regex":
|
|
ok = re.search(spec["pattern"], output, re.MULTILINE) is not None
|
|
return CheckResult(type=kind, passed=ok, detail="" if ok else f"Muster fehlt: {spec['pattern']}")
|
|
if kind == "contains":
|
|
ok = spec["value"] in output
|
|
return CheckResult(type=kind, passed=ok, detail="" if ok else f"fehlt: {spec['value']}")
|
|
if kind == "not_contains":
|
|
ok = spec["value"] not in output
|
|
return CheckResult(type=kind, passed=ok, detail="" if ok else f"enthält: {spec['value']}")
|
|
if kind == "max_chars":
|
|
ok = len(output) <= int(spec["value"])
|
|
return CheckResult(type=kind, passed=ok, detail=f"{len(output)} Zeichen")
|
|
if kind == "judge":
|
|
return self._judge(spec, text, output)
|
|
return CheckResult(type=str(kind), passed=False, detail="Unbekannter Prüftyp")
|
|
|
|
@staticmethod
|
|
def _json(output: str, schema: dict[str, Any] | None) -> CheckResult:
|
|
try:
|
|
data = extract_json(output)
|
|
except ValueError as exc:
|
|
return CheckResult(type="json", passed=False, detail=f"kein gültiges JSON: {exc}")
|
|
if schema:
|
|
try:
|
|
jsonschema.validate(data, schema)
|
|
except jsonschema.ValidationError as exc:
|
|
return CheckResult(type="json_schema", passed=False, detail=exc.message)
|
|
return CheckResult(type="json_schema" if schema else "json", passed=True)
|
|
|
|
def _judge(self, spec: dict[str, Any], text: str, output: str) -> CheckResult:
|
|
if self.chat is None or self.registry is None:
|
|
return CheckResult(type="judge", passed=None, detail="kein Judge konfiguriert")
|
|
slot_name = spec.get("slot", self.default_judge_slot)
|
|
slot = self.registry.slot(slot_name)
|
|
if not slot.available:
|
|
return CheckResult(type="judge", passed=None, detail=f"Judge-Slot {slot_name} nicht verfügbar")
|
|
user = f"Aufgabe:\n{text}\n\nKriterien:\n{spec.get('rubric', 'Korrekt und vollständig.')}\n\nAntwort:\n{output}"
|
|
try:
|
|
res = self.chat.chat(slot.spec.litellm_model,
|
|
[{"role": "system", "content": JUDGE_SYSTEM}, {"role": "user", "content": user}],
|
|
max_tokens=60, temperature=0)
|
|
except LLMError as exc:
|
|
return CheckResult(type="judge", passed=None, detail=f"Judge-Fehler: {exc}")
|
|
verdict = res.content.strip()
|
|
first = verdict.split(maxsplit=1)[0].upper().rstrip(":") if verdict else ""
|
|
if first == "PASS":
|
|
return CheckResult(type="judge", passed=True)
|
|
if first == "FAIL":
|
|
return CheckResult(type="judge", passed=False, detail=verdict[:200])
|
|
return CheckResult(type="judge", passed=None, detail=f"Unklare Judge-Antwort: {verdict[:100]}")
|