docs+feat(platform): Pruefer-Matrix-Foundation einfrieren (Evidenz, Mapping, Checker-Library, AGB-Kalibrierung)
Know-how-Freeze der Website-Compliance-Runde (DSE/Cookie/Impressum/AGB). docs: platform_evidence_v1 (Evidenz-/Qualitaetsnachweis, echte Zahlen), nutzungsbedingungen_mapping (neues Modul = Mapping, empirisch belegt), platform_checker_matrix (Meta-Modell verification_method x decision_method), verification_method, platform_validation_v1. code: checkers/ (reusable Pruefer-Library base+reference+embedding+llm, im Container validiert), agb/ (decision_method-Routing + Checker-Prototypen, 71% FP -> ~0 validiert). Dev-only, kein Prod-Push; Benchmark-GTs/Korpora im internen Archiv (data-retention). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
@@ -0,0 +1,82 @@
|
||||
"""Pruefer-Library — gemeinsames Interface. Siehe docs platform_checker_matrix.md.
|
||||
|
||||
Ein Checker prueft EINEN Control gegen EIN Dokument und liefert: vorhanden / fehlt
|
||||
/ unklar (+ Evidence). Module (DSE/Impressum/AGB/...) liefern nur Control-Metadaten
|
||||
ueber `ControlSpec` (verification_method + decision_method + checker-spezifische
|
||||
Config); die Engine routet method-agnostisch zum passenden Checker.
|
||||
|
||||
Ziel der Plattform: 14k Controls -> 7 Pruefertypen -> wenige Pruefer. Ein neues
|
||||
Modul wird damit ein Klassifizierungs-, kein Forschungsproblem.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Optional, Protocol, runtime_checkable
|
||||
|
||||
|
||||
class VerificationMethod:
|
||||
"""Achse 1 — WELCHER Pruefer-Typ (Kategorie)."""
|
||||
FIELD = "FIELD"
|
||||
REFERENCE = "REFERENCE"
|
||||
BEHAVIOR = "BEHAVIOR"
|
||||
PRESENTATION = "PRESENTATION"
|
||||
CONTENT = "CONTENT"
|
||||
PROCESS = "PROCESS"
|
||||
TECHNICAL = "TECHNICAL"
|
||||
CONTRACTUAL = "CONTRACTUAL"
|
||||
|
||||
|
||||
class DecisionMethod:
|
||||
"""Achse 2 — WIE entschieden wird (konkreter Mechanismus)."""
|
||||
REGEX = "REGEX"
|
||||
EMBEDDING = "EMBEDDING"
|
||||
LLM = "LLM"
|
||||
LINK_RESOLVER = "LINK_RESOLVER"
|
||||
PLAYWRIGHT = "PLAYWRIGHT"
|
||||
AUDIT = "AUDIT"
|
||||
SCANNER = "SCANNER"
|
||||
|
||||
|
||||
@dataclass
|
||||
class ControlSpec:
|
||||
"""Routing-Metadaten + checker-spezifische Config eines Controls. Module fuellen
|
||||
nur die fuer ihren decision_method relevanten Felder."""
|
||||
control_id: str
|
||||
verification_method: str
|
||||
decision_method: str
|
||||
label: str = ""
|
||||
severity: str = "MEDIUM"
|
||||
patterns: list[str] = field(default_factory=list) # FIELD/REGEX, REFERENCE
|
||||
paraphrases: list[str] = field(default_factory=list) # CONTENT (EMBEDDING/LLM)
|
||||
embed_threshold: Optional[float] = None # EMBEDDING (per-Control)
|
||||
topic_regex: str = "" # LLM: Section-Retrieval
|
||||
question: str = "" # LLM: Pruef-Frage
|
||||
extra: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
|
||||
@dataclass
|
||||
class DocContext:
|
||||
"""Das zu pruefende Artefakt. `text` = Volltext; `url`/`rendered` fuer
|
||||
PRESENTATION/BEHAVIOR (Playwright) — spaeter."""
|
||||
text: str = ""
|
||||
url: str = ""
|
||||
rendered: Any = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class CheckResult:
|
||||
present: Optional[bool] # True=erfuellt, False=fehlt, None=unklar (fail-safe)
|
||||
evidence: str = ""
|
||||
confidence: float = 0.0
|
||||
source: str = "" # welcher Pruefer/Tier geantwortet hat
|
||||
detail: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class Checker(Protocol):
|
||||
"""Alle Pruefer haben dieselbe Signatur -> die Engine ist method-agnostisch und
|
||||
routet nur ueber ctrl.verification_method / ctrl.decision_method."""
|
||||
verification_method: str
|
||||
|
||||
async def check(self, ctrl: ControlSpec, doc: DocContext) -> CheckResult:
|
||||
...
|
||||
@@ -0,0 +1,51 @@
|
||||
"""CONTENT-Pruefer / decision_method=EMBEDDING.
|
||||
|
||||
Ist die Pflicht SEMANTISCH im Text vorhanden? Max-Cosinus (Doc-Chunks x Control-
|
||||
Paraphrasen) >= per-Control-Schwelle. Deterministisch (festes Embedding-Modell)
|
||||
und gecacht. Rettet Recall-FP (Klausel da, anders formuliert).
|
||||
|
||||
Faellt der Embedding-Service aus, liefert der Checker present=None (unklar) — der
|
||||
Aufrufer behaelt dann das Keyword-Ergebnis (kein Hang, kein Crash).
|
||||
(Validiert an AGB: 17 Items, per-Item-Schwelle, 0 Fehl-Rescue.)
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
from .base import CheckResult, ControlSpec, DocContext, VerificationMethod
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Paraphrasen-Vektoren je Control einmal einbetten + cachen.
|
||||
_PARA_CACHE: dict[str, list] = {}
|
||||
|
||||
|
||||
class EmbeddingChecker:
|
||||
verification_method = VerificationMethod.CONTENT
|
||||
|
||||
async def check(self, ctrl: ControlSpec, doc: DocContext) -> CheckResult:
|
||||
text = doc.text or ""
|
||||
paras = ctrl.paraphrases or []
|
||||
thr = ctrl.embed_threshold if ctrl.embed_threshold is not None else 0.60
|
||||
if not paras or len(text) < 100:
|
||||
return CheckResult(present=None, source="embedding")
|
||||
try:
|
||||
from compliance.services.mc_embedding_matcher import (
|
||||
DIM, _chunk_text, _cosine, _embed_texts,
|
||||
)
|
||||
if ctrl.control_id not in _PARA_CACHE:
|
||||
pv = await _embed_texts(paras)
|
||||
_PARA_CACHE[ctrl.control_id] = [v for v in pv if v and len(v) == DIM]
|
||||
pvecs = _PARA_CACHE[ctrl.control_id]
|
||||
chunks = _chunk_text(text)
|
||||
cvecs = [v for v in await asyncio.wait_for(
|
||||
_embed_texts(chunks), timeout=90.0) if v and len(v) == DIM]
|
||||
except (Exception, asyncio.TimeoutError) as e:
|
||||
logger.info("embedding checker inaktiv %s: %s", ctrl.control_id, str(e)[:80])
|
||||
return CheckResult(present=None, source="embedding")
|
||||
if not pvecs or not cvecs:
|
||||
return CheckResult(present=None, source="embedding")
|
||||
best = max((_cosine(p, c) for p in pvecs for c in cvecs), default=0.0)
|
||||
return CheckResult(present=best >= thr, confidence=round(best, 3),
|
||||
source="embedding")
|
||||
@@ -0,0 +1,73 @@
|
||||
"""CONTENT/CONTRACTUAL-Pruefer / decision_method=LLM.
|
||||
|
||||
present/absent ueber die LLM-Kaskade (`call_with_cascade`; prod: OVH-120b zuerst).
|
||||
Retrieval = GANZE Paragraph-Abschnitte zum Topic (nicht Top-k-Chunks — das war in
|
||||
der AGB-Validierung der Schluessel). KEIN DEFECT — Korrektheits-/Defekt-Pruefung
|
||||
ist ein separater Modus. present=None bei Fehler (fail-safe: Aufrufer behaelt
|
||||
Keyword-Ergebnis). (Validiert an AGB delivery/warranty.)
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
|
||||
from .base import CheckResult, ControlSpec, DocContext, VerificationMethod
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_SECTION = re.compile(r"(?m)(?=^\s*(?:§\s*)?\d+[\.\)]\s)")
|
||||
_SYS = (
|
||||
"Du bist deutscher Compliance-Rechtsexperte. Entscheide, ob die genannte "
|
||||
"Pflicht in den vorgelegten Abschnitten vorhanden ist. NUR die Abschnitte "
|
||||
'zaehlen. Antworte NUR JSON: {"verdict":"ERFUELLT|FEHLT","zitat":"woertlich '
|
||||
'oder leer","begruendung":"1 Satz"}.'
|
||||
)
|
||||
|
||||
|
||||
def _sections(text: str) -> list[str]:
|
||||
return [s.strip() for s in _SECTION.split(text) if s.strip()]
|
||||
|
||||
|
||||
def _parse(txt: str) -> dict:
|
||||
out = (txt or "").strip()
|
||||
if out.startswith("```"):
|
||||
out = out.split("```", 2)[1]
|
||||
out = out[4:] if out.startswith("json") else out
|
||||
a, b = out.find("{"), out.rfind("}")
|
||||
return json.loads(out[a:b + 1] if 0 <= a < b else out)
|
||||
|
||||
|
||||
class LLMChecker:
|
||||
verification_method = VerificationMethod.CONTENT
|
||||
|
||||
async def check(self, ctrl: ControlSpec, doc: DocContext) -> CheckResult:
|
||||
text = doc.text or ""
|
||||
if len(text) < 50:
|
||||
return CheckResult(present=None, source="llm")
|
||||
secs = _sections(text)
|
||||
if ctrl.topic_regex:
|
||||
rel = [s for s in secs if re.search(ctrl.topic_regex, s, re.I)][:6] or secs[:6]
|
||||
else:
|
||||
rel = secs[:6]
|
||||
question = ctrl.question or f"Ist die Pflicht '{ctrl.label}' im Text vorhanden?"
|
||||
try:
|
||||
from compliance.services.llm_cascade import call_with_cascade
|
||||
r = await call_with_cascade(
|
||||
_SYS,
|
||||
json.dumps({"frage": question, "abschnitte": rel}, ensure_ascii=False),
|
||||
min_confidence=0.6, max_tokens=500,
|
||||
)
|
||||
obj = _parse(r.get("text"))
|
||||
verdict = obj.get("verdict")
|
||||
zitat = (obj.get("zitat") or "")[:120]
|
||||
if verdict not in ("ERFUELLT", "FEHLT"):
|
||||
return CheckResult(present=None, evidence=zitat, source=r.get("source", "?"))
|
||||
return CheckResult(
|
||||
present=verdict == "ERFUELLT", evidence=zitat,
|
||||
confidence=float(r.get("confidence") or 0.0),
|
||||
source=r.get("source", "llm"),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.info("llm checker fail %s: %s", ctrl.control_id, str(e)[:80])
|
||||
return CheckResult(present=None, source="error")
|
||||
@@ -0,0 +1,41 @@
|
||||
"""REFERENCE-Pruefer (verification_method=REFERENCE, decision_method=LINK_RESOLVER).
|
||||
|
||||
Ist ein klarer Verweis auf ein anderes Pflichtdokument vorhanden (+ optional: loest
|
||||
der Link auf)? Deterministisch. Bsp: 'Details in unserer Datenschutzerklaerung'.
|
||||
KEIN LLM, kein juristisches Urteil. (Validiert an AGB data_protection: 7/7.)
|
||||
|
||||
Die tatsaechliche HTTP-Aufloesung des Links ist ein optionaler Runtime-Schritt
|
||||
(online), nicht Teil dieser deterministischen Text-Pruefung — die URL wird hier
|
||||
nur extrahiert und in `detail['link']` zurueckgegeben.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
from .base import CheckResult, ControlSpec, DocContext, VerificationMethod
|
||||
|
||||
_URL = re.compile(r"https?://[^\s)\]]+", re.I)
|
||||
|
||||
|
||||
class ReferenceChecker:
|
||||
verification_method = VerificationMethod.REFERENCE
|
||||
|
||||
async def check(self, ctrl: ControlSpec, doc: DocContext) -> CheckResult:
|
||||
text = doc.text or ""
|
||||
pats = ctrl.patterns or []
|
||||
if not pats or not text:
|
||||
return CheckResult(present=False, source="reference")
|
||||
for p in pats:
|
||||
m = re.search(p, text, re.I)
|
||||
if m:
|
||||
window = text[max(0, m.start() - 40): m.end() + 200]
|
||||
url = _URL.search(window) or _URL.search(text)
|
||||
link = url.group(0) if url else None
|
||||
return CheckResult(
|
||||
present=True,
|
||||
evidence=" ".join(m.group(0).split())[:120],
|
||||
confidence=1.0,
|
||||
source="reference",
|
||||
detail={"link": link},
|
||||
)
|
||||
return CheckResult(present=False, source="reference")
|
||||
Reference in New Issue
Block a user