update

backend/readability.py

@@ -0,0 +1,115 @@
+"""Deterministic readability gate for guide sections.
+
+A small German complexity model (DistilBERT, GermEval 2022, scale 1–7) rates the
+readability of the prose. guide.py feeds sections that are too hard into the existing
+read-exam/revision loop — no prompt, no guessing.
+
+Optional: if `transformers`/`torch` are missing or the model won't load, the gate is
+silently disabled (the backend keeps running unchanged). CPU is enough; the caller
+wraps the scoring in `asyncio.to_thread` (blocking model inference).
+"""
+
+import logging
+import re
+
+from config import (
+    READABILITY_ACTIVE, READABILITY_HARD, READABILITY_HARD_SHARE, READABILITY_MAX, READABILITY_MODEL,
+)
+
+log = logging.getLogger("creator.readability")
+
+_model_cache = None       # (tokenizer, model, torch) — singleton
+_load_attempt = False     # already tried to load?
+
+# Strip markup → plain prose (code does not count toward readability).
+_CODE_FENCE = re.compile(r"```.*?```", re.DOTALL)
+_COMMENT = re.compile(r"<!--.*?-->", re.DOTALL)
+_INLINE_CODE = re.compile(r"`[^`]*`")
+_LINK = re.compile(r"\[([^\]]*)\]\([^)]*\)")
+_MD_MARK = re.compile(r"^[ \t]*([#>]+|[-*+]\s)|[*_~|]", re.MULTILINE)
+_WS = re.compile(r"\s+")
+_SENTENCE = re.compile(r"(?<=[.!?])\s+")
+
+
+def _model():
+    """Load the model once. None = gate off (disabled or load error)."""
+    global _model_cache, _load_attempt
+    if _load_attempt:
+        return _model_cache
+    _load_attempt = True
+    if not READABILITY_ACTIVE:
+        return None
+    try:
+        import torch
+        from transformers import AutoModelForSequenceClassification, AutoTokenizer
+        tok = AutoTokenizer.from_pretrained(READABILITY_MODEL)
+        model = AutoModelForSequenceClassification.from_pretrained(READABILITY_MODEL)
+        model.eval()
+        _model_cache = (tok, model, torch)
+        log.info("readability model loaded: %s (num_labels=%d)", READABILITY_MODEL, model.config.num_labels)
+    except Exception as e:
+        log.warning("readability gate disabled (model not loadable): %s", e)
+        _model_cache = None
+    return _model_cache
+
+
+def _prose(md: str) -> str:
+    """Strip markdown/code → plain prose for scoring."""
+    t = _CODE_FENCE.sub(" ", md)
+    t = _COMMENT.sub(" ", t)
+    t = _INLINE_CODE.sub(" ", t)
+    t = _LINK.sub(r"\1", t)
+    t = _MD_MARK.sub(" ", t)
+    return _WS.sub(" ", t).strip()
+
+
+def _sentences(text: str) -> list[str]:
+    """Split prose into sentences; discard very short fragments."""
+    return [s.strip() for s in _SENTENCE.split(text) if len(s.strip()) >= 15]
+
+
+def _scores(sentences: list[str]) -> list[float]:
+    """Complexity per sentence (1–7). Regression (num_labels=1) or expectation over classes."""
+    tok, model, torch = _model_cache
+    values: list[float] = []
+    n = model.config.num_labels
+    for i in range(0, len(sentences), 16):
+        batch = sentences[i:i + 16]
+        enc = tok(batch, return_tensors="pt", truncation=True, max_length=256, padding=True)
+        with torch.no_grad():
+            logits = model(**enc).logits
+        if n == 1:
+            vals = logits.reshape(-1).tolist()
+        else:
+            probs = torch.softmax(logits, dim=-1)
+            levels = torch.arange(1, n + 1, dtype=probs.dtype)
+            vals = (probs * levels).sum(-1).reshape(-1).tolist()
+        values.extend(vals)
+    return values
+
+
+def rate_sections(md_by_num: dict[int, str]) -> dict[int, str]:
+    """{num: section_md} → {num: hint} only for sections that are too hard.
+
+    Empty dict if the gate is off. Blocking (CPU) — call inside to_thread.
+    """
+    if _model() is None:
+        return {}
+    out: dict[int, str] = {}
+    for num, md in md_by_num.items():
+        sentences = _sentences(_prose(md or ""))
+        if len(sentences) < 2:  # almost only code / too short → skip
+            continue
+        values = _scores(sentences)
+        if not values:
+            continue
+        mean = sum(values) / len(values)
+        hard = sum(1 for w in values if w > READABILITY_HARD) / len(values)
+        # Too hard = high mean OR too many hard individual sentences (outlier nests).
+        if mean > READABILITY_MAX or hard >= READABILITY_HARD_SHARE:
+            # German revision hint fed to the (German-writing) writer agent — kept German on purpose.
+            out[num] = (
+                f"Zu schwer lesbar (Ø {mean:.1f}/7, {hard * 100:.0f}% harte Sätze): "
+                "kürzere Sätze, einfachere Wörter, weniger Schachtelsätze, mehr Examples."
+            )
+    return out