116 lines
4.4 KiB
Python
116 lines
4.4 KiB
Python
"""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
|