658 lines
29 KiB
Python
658 lines
29 KiB
Python
"""Baustein-Lernen: Vertiefung, Bausteinchat und Prüfung zu einzelnen Guide-Sections.
|
||
|
||
Alle Aufrufe sind interaktiv (stdout-Antwort, lane "interactive") und stateless —
|
||
der Chat-/Prüfungs-Verlauf kommt vom Frontend, persistiert wird nur der
|
||
Prüfungs-Zähler (DB) und die Vertiefung (DB).
|
||
"""
|
||
|
||
import logging
|
||
import random
|
||
import re
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
|
||
from agents import run_agent
|
||
from config import DEFAULT_PROVIDER
|
||
from database import create_element, list_elements, get_baustein_huerden
|
||
from elements import generate_element
|
||
from jsonio import parse_json_text as _parse_json_text
|
||
from pipeline import _prompt, _probleme_schema
|
||
from textkit import _norm_titel
|
||
|
||
log = logging.getLogger("creator.lernen")
|
||
|
||
# Lernstufen je Baustein — relativ zum cap (Floor in % des Maximal-Scores):
|
||
# grün=Anfänger 20% · blau=Fortgeschritten 40% · lila=Experte 60% · gold=Meister 100%.
|
||
# Prüfungsform ist immer zufällig (5 Formen); der cap skaliert mit der Stoffmenge.
|
||
STUFEN = (("anfaenger", 0.2), ("fortgeschritten", 0.4), ("experte", 0.6), ("meister", 1.0))
|
||
|
||
|
||
PUNKT_BASIS = 25 # Punkte je Subbaustein. Meister-cap = (alle Subs) × 25.
|
||
|
||
|
||
def _ebenen(n_je_ebene: dict[int, int]) -> list[int]:
|
||
return [n_je_ebene.get(k, 0) for k in (1, 2, 3, 4)]
|
||
|
||
|
||
def schwellen(n_je_ebene: dict[int, int]) -> list[int]:
|
||
"""Kumulative Ebenen-Schwellen [S_1, S_2, S_3, S_4] = (n_1+…+n_k) × 25.
|
||
S_k ist der Score, ab dem Ebene k+1 (E→M→S→F) freigeschaltet ist; S_4 = cap_final."""
|
||
out, akk = [], 0
|
||
for n in _ebenen(n_je_ebene):
|
||
akk += n
|
||
out.append(akk * PUNKT_BASIS)
|
||
return out
|
||
|
||
|
||
def cap_final(n_je_ebene: dict[int, int]) -> int:
|
||
"""Maximal-Score (Meister) = alle Subbausteine × 25."""
|
||
return schwellen(n_je_ebene)[-1]
|
||
|
||
|
||
def freie_ebene(score: int, n_je_ebene: dict[int, int]) -> int:
|
||
"""Höchste freigeschaltete Sub-Ebene 1–4. Ebene k+1 frei, sobald score ≥ S_k.
|
||
Leere Ebenen (n_k=0) werden automatisch übersprungen (S_k == S_{k-1})."""
|
||
s = schwellen(n_je_ebene)
|
||
e = 1
|
||
for k in range(3): # S_1..S_3 schalten Ebene 2..4 frei
|
||
if score >= s[k]:
|
||
e = k + 2
|
||
return e
|
||
|
||
|
||
def cap_aktuell(score: int, n_je_ebene: dict[int, int]) -> int:
|
||
"""Erreichbarer cap der aktuell freigeschalteten Ebene = freigeschaltete Subs × 25."""
|
||
return schwellen(n_je_ebene)[freie_ebene(score, n_je_ebene) - 1]
|
||
|
||
|
||
def _schwelle(p: float, cap: int) -> int:
|
||
return round(p * cap)
|
||
|
||
|
||
def stufe_aus_score(score: int, cap_final_wert: int) -> str | None:
|
||
"""Höchste erreichte Lernstufe (None unter 20 %), relativ zum cap_final."""
|
||
erreicht = None
|
||
for key, p in STUFEN:
|
||
if score >= _schwelle(p, cap_final_wert):
|
||
erreicht = key
|
||
return erreicht
|
||
|
||
|
||
def progressiver_malus(basis: int, cap_akt: int) -> int:
|
||
"""Fehler-Strafe nach Fortschritt in der aktuellen Ebene (gegen cap_aktuell):
|
||
≤25 %→−5 · ≤50 %→−10 · ≤75 %→−15 · >75 %→−20."""
|
||
pct = (basis / cap_akt) if cap_akt else 0.0
|
||
if pct <= 0.25:
|
||
return -5
|
||
if pct <= 0.5:
|
||
return -10
|
||
if pct <= 0.75:
|
||
return -15
|
||
return -20
|
||
CHAT_TIMEOUT = 240
|
||
PRUEFUNG_TIMEOUT = 120 # kurze JSON-Turns; deckelt die Serien-Latenz pro Prüfungs-Schritt
|
||
GRUENDLICH_TIMEOUT = 600 # „Gründlich prüfen": starkes Modell (role guide) braucht länger
|
||
KRITIK_MAX_RUNDEN = 2 # Generator → Kritiker → ggf. Neu, höchstens so oft
|
||
|
||
# Fragetypen für Active Recall — pro Frage einer, zufällig gewählt. Schafft Vielfalt.
|
||
FRAGETYPEN = {
|
||
"abruf": "Free Recall: Lass den Lerner die Kernidee frei aus dem Kopf erklären (offene Verständnisfrage).",
|
||
"punkt": "Cued Recall: Frag gezielt EIN konkretes Detail oder eine Abgrenzung ab.",
|
||
"warum": "Warum-Frage: Frag nach dem Grund/Mechanismus — warum funktioniert oder gilt das so?",
|
||
"anwendung": "Anwendung: Lass das Konzept auf EIN kurzes, neues Beispiel/Szenario anwenden.",
|
||
"pruefen": "Bei Code-/Tool-Themen: kleinen Schnipsel zeigen — Output vorhersagen ODER den Fehler finden. Kein Code-Thema → stattdessen eine Anwendungsfrage.",
|
||
}
|
||
|
||
|
||
# Antwort-Niveau → Basis-Punkte (neue 25er-Skala). „kaum" = −1 ist nur das Signal für den
|
||
# progressiven Malus (echter Wert kommt aus progressiver_malus). Positive Werte werden bei
|
||
# Streak hochmoduliert und auf [10, 40] geklemmt.
|
||
NIVEAUS = {
|
||
"unbeantwortbar": 0, # Frage selbst kaputt → keine Änderung
|
||
"kaum": -1, # < 25 % richtig → Malus
|
||
"teilweise": 0, # 25–49 % → neutral
|
||
"solide": 16, # 50–74 %
|
||
"stark": 24, # 75–99 % (Quiz/Lücke-Treffer)
|
||
"komplett": 30, # 100 % (nur freies Erklären erreichbar)
|
||
}
|
||
|
||
# Reihenfolge schwach→stark (für den Nachfrage-Deckel).
|
||
_NIVEAU_RANG = ("kaum", "teilweise", "solide", "stark", "komplett")
|
||
|
||
|
||
def deckel_nachfrage(niveau: str, nachgefragt: bool) -> str:
|
||
"""Mit Nachfrage (Hinweis erhalten) höchstens „solide" — kein Voll-Score erschummeln."""
|
||
if nachgefragt and niveau in ("stark", "komplett"):
|
||
return "solide"
|
||
return niveau
|
||
|
||
|
||
def streak_punkte(basis_delta: int, streak_basis: int) -> int:
|
||
"""Positives Basis-Delta mit Streak hochmodulieren, auf [10, 40] geklemmt."""
|
||
faktor = min(1.33, 1 + 0.066 * min(streak_basis, 5))
|
||
return max(10, min(40, round(basis_delta * faktor)))
|
||
|
||
|
||
def punkte_delta(niveau: str, streak_basis: int, basis: int, cap_akt: int) -> tuple[int, int]:
|
||
"""Antwort-Niveau → (Punkt-Delta, neue Streak). Positiv: streak-moduliert, Streak +1.
|
||
Neutral (0): keine Änderung, Streak bleibt. Negativ: progressiver Malus, Streak-Reset 0."""
|
||
basis_delta = NIVEAUS.get(niveau, 0)
|
||
if basis_delta > 0:
|
||
return streak_punkte(basis_delta, streak_basis), streak_basis + 1
|
||
if basis_delta == 0:
|
||
return 0, streak_basis
|
||
return progressiver_malus(basis, cap_akt), 0
|
||
|
||
|
||
def score_berechnen(basis: int, delta: int, floor: int, cap_akt: int, cap_fin: int) -> int:
|
||
"""Neuer Score · driftfrei aus der Basis. Klemmt nach oben gegen `cap_akt` (Deckel der
|
||
aktuell freigeschalteten Ebene) und nach unten gegen `floor`. Eingefroren NUR am
|
||
absoluten Maximum (`basis ≥ cap_fin`) — sonst würde an jeder Ebenen-Schwelle blockiert."""
|
||
if basis >= cap_fin:
|
||
return basis
|
||
return max(floor, min(cap_akt, basis + delta))
|
||
|
||
|
||
def floor_aus_score(basis: int, cap_fin: int, s_schwellen: list[int]) -> int:
|
||
"""Untergrenze (kein Rückfall): höchste erreichte Lernstufen-Schwelle (über cap_final)
|
||
UND höchste erreichte Ebenen-Freischalt-Schwelle S_k. max beider Achsen."""
|
||
floor = 0
|
||
for _, p in STUFEN:
|
||
s = _schwelle(p, cap_fin)
|
||
if basis >= s:
|
||
floor = max(floor, s)
|
||
for s in s_schwellen:
|
||
if basis >= s:
|
||
floor = max(floor, s)
|
||
return floor
|
||
|
||
|
||
def _transcript(messages: list[dict]) -> str:
|
||
return "\n".join(
|
||
f"{'Nutzer' if m.get('role') == 'user' else 'Assistent'}: {m.get('content', '')}"
|
||
for m in messages
|
||
) or "(leer)"
|
||
|
||
|
||
async def baustein_chat(topic: str, baustein: str, section: str, kompakt: str | None, messages: list[dict], provider: str = DEFAULT_PROVIDER) -> str:
|
||
try:
|
||
prompt = _prompt(
|
||
"Baustein-Chat",
|
||
topic=topic, baustein=baustein,
|
||
section_block=section.strip() or "(keine Guide-Fassung übergeben)",
|
||
kompakt_block=(kompakt or "").strip() or "(keine)",
|
||
transcript=_transcript(messages),
|
||
)
|
||
returncode, stdout, _ = await run_agent(
|
||
"bausteinchat-" + str(uuid.uuid4()), prompt, CHAT_TIMEOUT,
|
||
provider=provider, role="fast", capabilities="none", lane="interactive",
|
||
)
|
||
if returncode != 0:
|
||
return "Entschuldigung, das hat nicht geklappt. Bitte versuche es erneut."
|
||
reply = stdout.strip()
|
||
return reply or "Entschuldigung, ich habe keine Antwort erhalten."
|
||
except Exception:
|
||
log.warning("[%s] Baustein-Chat fehlgeschlagen (%s)", topic, baustein, exc_info=True)
|
||
return "Entschuldigung, das hat nicht geklappt. Bitte versuche es erneut."
|
||
|
||
|
||
def _frage_schema(data) -> dict | None:
|
||
"""{"frage": str} · sonst None."""
|
||
if not isinstance(data, dict):
|
||
return None
|
||
frage = str(data.get("frage", "")).strip()
|
||
return {"frage": frage} if frage else None
|
||
|
||
|
||
def _bewertung_schema(data) -> dict | None:
|
||
"""{"feedback": str, "niveau": ∈ NIVEAUS} · sonst None."""
|
||
if not isinstance(data, dict):
|
||
return None
|
||
feedback = str(data.get("feedback", "")).strip()
|
||
niveau = data.get("niveau")
|
||
if not feedback or niveau not in NIVEAUS:
|
||
return None
|
||
return {"feedback": feedback, "niveau": niveau}
|
||
|
||
|
||
async def _gen_call(name: str, role: str, schema, provider: str, timeout: int = PRUEFUNG_TIMEOUT, lane: str = "interactive", **kwargs) -> dict | None:
|
||
"""Generator-Agent: Template füllen, laufen lassen, per schema parsen · None bei Fehler.
|
||
lane="batch" für Hintergrund (Vorladen, Genau-Bewertung) → eigene Slot-Schlange."""
|
||
returncode, stdout, _ = await run_agent(
|
||
name.lower() + "-" + str(uuid.uuid4()), _prompt(name, **kwargs), timeout,
|
||
provider=provider, role=role, capabilities="none", lane=lane,
|
||
)
|
||
return schema(_parse_json_text(stdout)) if returncode == 0 else None
|
||
|
||
|
||
async def _kritik_call(name: str, provider: str, role: str = "judge", timeout: int = PRUEFUNG_TIMEOUT, lane: str = "interactive", **kwargs) -> list[str]:
|
||
"""Kritiker-Agent (Default role judge): leere Liste = in Ordnung. Fail-open: Ausfall des
|
||
Kritikers darf den Turn nicht blockieren, also dann ebenfalls leere Liste."""
|
||
returncode, stdout, _ = await run_agent(
|
||
name.lower() + "-" + str(uuid.uuid4()), _prompt(name, **kwargs), timeout,
|
||
provider=provider, role=role, capabilities="none", lane=lane,
|
||
)
|
||
if returncode != 0:
|
||
return []
|
||
return _probleme_schema(_parse_json_text(stdout)) or []
|
||
|
||
|
||
def _kritik_block(vorversion: str, probleme: list[str]) -> str:
|
||
punkte = "\n".join(f"- {p}" for p in probleme)
|
||
return (
|
||
f"Deine vorige Fassung war:\n«{vorversion}»\n\n"
|
||
f"Der Prüfer bemängelt:\n{punkte}\n\nBehebe diese Punkte."
|
||
)
|
||
|
||
|
||
def _bewertung_text(bew: dict) -> str:
|
||
return f"Niveau: {bew['niveau']}\nFeedback: {bew['feedback']}"
|
||
|
||
|
||
# Deterministischer Guard gegen Doppelfragen — der KI-Kritiker übersieht „…, und welchen…".
|
||
_FRAGEWORT = r"(was|welche[rsnm]?|wie|wieso|warum|wofür|wozu|wann|wo|wer|wem|wen|nenne)"
|
||
_DOPPEL_RE = re.compile(r"[,;]?\s+(und|sowie|außerdem|bzw\.?)\s+" + _FRAGEWORT + r"\b", re.IGNORECASE)
|
||
|
||
|
||
def _doppelfrage_mangel(frage: str) -> str | None:
|
||
"""Erkennt zwei verkettete Fragen. None = ok. Flaggt NUR 'und/sowie' + Fragewort."""
|
||
if frage.count("?") > 1:
|
||
return "Mehr als ein Fragezeichen — stelle GENAU EINE Frage."
|
||
if _DOPPEL_RE.search(frage):
|
||
return "Zwei Fragen mit 'und'/'sowie' verkettet — stelle GENAU EINE Frage, eine Sache."
|
||
return None
|
||
|
||
|
||
async def _frage_mit_kritik(
|
||
topic: str, baustein: str, section_block: str, kompakt_block: str,
|
||
transcript: str, vermeide_block: str, typ_block: str, fokus_block: str,
|
||
niveau_block: str, provider: str,
|
||
) -> str | None:
|
||
"""Frage generieren, vom Kritiker prüfen lassen, bei Mängeln neu (max KRITIK_MAX_RUNDEN)."""
|
||
kritik_block = "(keine)"
|
||
frage = None
|
||
for _ in range(KRITIK_MAX_RUNDEN):
|
||
data = await _gen_call(
|
||
"Baustein-Frage", "guide", _frage_schema, provider, lane="batch",
|
||
topic=topic, baustein=baustein, section_block=section_block,
|
||
kompakt_block=kompakt_block, transcript=transcript, vermeide_block=vermeide_block,
|
||
typ_block=typ_block, fokus_block=fokus_block, niveau_block=niveau_block, kritik_block=kritik_block,
|
||
)
|
||
if data is None:
|
||
return None
|
||
frage = data["frage"]
|
||
probleme = await _kritik_call(
|
||
"Baustein-Frage-Kritik", provider, role="guide", lane="batch", # starke KI prüft die Regeln
|
||
topic=topic, baustein=baustein, section_block=section_block,
|
||
kompakt_block=kompakt_block, transcript=transcript, vermeide_block=vermeide_block,
|
||
typ_block=typ_block, fokus_block=fokus_block, frage=frage,
|
||
)
|
||
hart = _doppelfrage_mangel(frage) # erzwingt Neugenerierung, auch wenn der KI-Kritiker es übersah
|
||
if hart:
|
||
probleme = [hart, *(probleme or [])]
|
||
if not probleme:
|
||
return frage
|
||
kritik_block = _kritik_block(frage, probleme)
|
||
return frage # best-effort nach der letzten Runde
|
||
|
||
|
||
async def _bewertung_mit_kritik(
|
||
topic: str, baustein: str, section_block: str, kompakt_block: str,
|
||
frage: str, transcript: str, begruendung_block: str, provider: str, role: str = "judge",
|
||
) -> dict | None:
|
||
"""Antwort bewerten (Niveau), vom Kritiker prüfen lassen, bei Fehlurteil neu.
|
||
|
||
`frage` ankert die geprüfte Frage; der Dialog (transcript) liefert Antwort + Diskussion.
|
||
`begruendung_block` = optionale Unzufriedenheit des Lerners (nur bei „Gründlich prüfen").
|
||
`role` = "judge" (schnell) oder "guide" (gründlich, starkes Modell mit Thinking).
|
||
"""
|
||
timeout = GRUENDLICH_TIMEOUT if role == "guide" else PRUEFUNG_TIMEOUT
|
||
# Gründlich (role guide) = Nutzer wartet → interaktiv. Hintergrund-Genau (judge) → batch.
|
||
lane = "interactive" if role == "guide" else "batch"
|
||
kritik_block = "(keine)"
|
||
bew = None
|
||
for _ in range(KRITIK_MAX_RUNDEN):
|
||
bew = await _gen_call(
|
||
"Baustein-Bewertung", role, _bewertung_schema, provider, timeout, lane=lane,
|
||
topic=topic, baustein=baustein, section_block=section_block,
|
||
kompakt_block=kompakt_block, frage=frage, transcript=transcript,
|
||
begruendung_block=begruendung_block, kritik_block=kritik_block,
|
||
)
|
||
if bew is None:
|
||
return None
|
||
probleme = await _kritik_call(
|
||
"Baustein-Bewertung-Kritik", provider, role=role, timeout=timeout, lane=lane,
|
||
topic=topic, baustein=baustein, section_block=section_block,
|
||
kompakt_block=kompakt_block, frage=frage, transcript=transcript,
|
||
bewertung_block=_bewertung_text(bew),
|
||
)
|
||
if not probleme:
|
||
return bew
|
||
kritik_block = _kritik_block(_bewertung_text(bew), probleme)
|
||
return bew # best-effort nach der letzten Runde
|
||
|
||
|
||
def _bloecke(section: str, kompakt: str | None) -> tuple[str, str]:
|
||
return (
|
||
section.strip() or "(keine Guide-Fassung übergeben)",
|
||
(kompakt or "").strip() or "(keine)",
|
||
)
|
||
|
||
|
||
def _vermeide_block(vermeide: list[str] | None) -> str:
|
||
eintraege = [f.strip() for f in (vermeide or []) if f and f.strip()]
|
||
return "\n".join(f"- {f}" for f in eintraege) or "(keine)"
|
||
|
||
|
||
# Lerner-Niveau (aus dem Score abgeleitet) → Adressaten-Rolle für die Frage. So entsteht die
|
||
# Schwierigkeit: nicht „extra schwer machen", sondern „für einen Anfänger/Experten fragen".
|
||
# Je Stufe: Adressaten-Rolle + kognitive Anforderung (Bloom) + „frag so"-Cue. Ohne explizite Stufe nimmt
|
||
# das Modell den leichten Pfad (bloßer Abruf) — die Cues heben höhere Niveaus auf Anwenden/Analysieren/Transfer.
|
||
NIVEAU_ROLLE = {
|
||
"anfaenger": "Der Lerner ist ANFÄNGER. Kognitiv: ERINNERN/VERSTEHEN. Frage nach dem Grundverständnis — das Kernkonzept, einfach und direkt.",
|
||
"fortgeschritten": "Der Lerner ist FORTGESCHRITTEN. Kognitiv: ANWENDEN. Stelle eine kleine konkrete Situation und lass das Konzept darauf anwenden — frage nicht bloß die Definition ab.",
|
||
"experte": "Der Lerner ist EXPERTE. Kognitiv: ANALYSIEREN. Lass abgrenzen/vergleichen, einen Sonderfall einordnen oder eine typische Tücke (Hürde) aufdecken — nicht das Lehrbuch-Wissen abfragen.",
|
||
"meister": "Der Lerner ist auf MEISTER-Niveau. Kognitiv: BEWERTEN/TRANSFER. Lass das Konzept auf ein NEUES Problem übertragen, eine Entscheidung begründen oder einen Trade-off abwägen.",
|
||
}
|
||
|
||
|
||
def _niveau_block(niveau: str | None) -> str:
|
||
return NIVEAU_ROLLE.get(niveau or "", NIVEAU_ROLLE["anfaenger"])
|
||
|
||
|
||
async def pruefung_frage(
|
||
topic: str, baustein: str, section: str, kompakt: str | None,
|
||
messages: list[dict], subbausteine: list[str] | None = None,
|
||
vermeide: list[str] | None = None, niveau: str = "anfaenger", provider: str = DEFAULT_PROVIDER,
|
||
) -> str | None:
|
||
"""Aktion 'frage': eine Frage generieren — zufälliger Typ zu einem zufälligen Subbaustein,
|
||
in der Adressaten-Rolle des Niveaus, dann Kritiker (sequenziell) · None bei Fehler."""
|
||
try:
|
||
section_block, kompakt_block = _bloecke(section, kompakt)
|
||
transcript = _transcript(messages) if messages else "(leer)"
|
||
typ_block = FRAGETYPEN[random.choice(list(FRAGETYPEN))]
|
||
subs = [s for s in (subbausteine or []) if s and s.strip()]
|
||
fokus = random.choice(subs) if subs else ""
|
||
fokus_block = (
|
||
f"Konzentriere die Frage auf diesen Subbaustein: „{fokus}\"" if fokus
|
||
else "(ganzer Baustein — kein bestimmter Subbaustein)"
|
||
)
|
||
return await _frage_mit_kritik(
|
||
topic, baustein, section_block, kompakt_block, transcript,
|
||
_vermeide_block(vermeide), typ_block, fokus_block, _niveau_block(niveau), provider,
|
||
)
|
||
except Exception:
|
||
log.warning("[%s] Frage fehlgeschlagen (%s)", topic, baustein, exc_info=True)
|
||
return None
|
||
|
||
|
||
async def pruefung_frage_variante(
|
||
topic: str, baustein: str, section: str, kompakt: str | None,
|
||
muster: str, niveau: str = "anfaenger", provider: str = DEFAULT_PROVIDER,
|
||
) -> str | None:
|
||
"""Aktion 'frage' mit Muster: aus einem vordefinierten Muster eine konkrete Frage in der
|
||
Adressaten-Rolle des Niveaus formulieren. Kein Kritiker (Muster ist build-geprüft).
|
||
Stil-Guard bleibt als billiger Schutz gegen Doppelfragen · None bei Fehler."""
|
||
try:
|
||
section_block, kompakt_block = _bloecke(section, kompakt)
|
||
data = await _gen_call(
|
||
"Baustein-Frage-Variante", "guide", _frage_schema, provider, lane="batch",
|
||
topic=topic, baustein=baustein, section_block=section_block,
|
||
kompakt_block=kompakt_block, muster=muster, niveau_block=_niveau_block(niveau),
|
||
)
|
||
if data is None:
|
||
return None
|
||
return data["frage"]
|
||
except Exception:
|
||
log.warning("[%s] Frage-Variante fehlgeschlagen (%s)", topic, baustein, exc_info=True)
|
||
return None
|
||
|
||
|
||
def _optionen_schema(opts) -> list[dict] | None:
|
||
"""[{text, korrekt}]×4 → validierte Liste · sonst None."""
|
||
if not isinstance(opts, list) or len(opts) != 4:
|
||
return None
|
||
out = []
|
||
for o in opts:
|
||
if not isinstance(o, dict):
|
||
return None
|
||
text = str(o.get("text", "")).strip()
|
||
korrekt = o.get("korrekt")
|
||
if not text or not isinstance(korrekt, bool):
|
||
return None
|
||
out.append({"text": text, "korrekt": korrekt})
|
||
return out
|
||
|
||
|
||
def _quiz_schema(data) -> dict | None:
|
||
"""{"frage": str, "optionen": [{text, korrekt}]×4} → validiert · sonst None.
|
||
Single-Choice: genau 1 richtig. Die Schwierigkeit steckt im Niveau, nicht in der Anzahl."""
|
||
if not isinstance(data, dict):
|
||
return None
|
||
frage = str(data.get("frage", "")).strip()
|
||
out = _optionen_schema(data.get("optionen"))
|
||
if not frage or out is None:
|
||
return None
|
||
if sum(o["korrekt"] for o in out) != 1:
|
||
return None
|
||
return {"frage": frage, "optionen": out}
|
||
|
||
|
||
def _lueckwahl_schema(data) -> dict | None:
|
||
"""{"satz": str (mit ___), "optionen": [{text, korrekt}]×4} → genau 1 korrekt · sonst None."""
|
||
if not isinstance(data, dict):
|
||
return None
|
||
satz = str(data.get("satz", "")).strip()
|
||
out = _optionen_schema(data.get("optionen"))
|
||
if not satz or "___" not in satz or out is None or sum(o["korrekt"] for o in out) != 1:
|
||
return None
|
||
return {"satz": satz, "optionen": out}
|
||
|
||
|
||
async def huerden_distraktor_block(topic: str, baustein: str) -> str:
|
||
"""Typische Irrtümer (Fakten-Hürden) des Bausteins als Distraktor-Quelle für Quiz/Lückenwahl.
|
||
Leer, wenn keine vorhanden (Altbestand) → der Prompt-Platzhalter verschwindet rückstandslos."""
|
||
try:
|
||
huerden = await get_baustein_huerden(topic, _norm_titel(baustein))
|
||
except Exception:
|
||
return ""
|
||
if not huerden:
|
||
return ""
|
||
zeilen = "\n".join(f"- {h}" for h in huerden[:8])
|
||
return ("TYPISCHE IRRTÜMER zu diesem Baustein (nutze sie als Distraktoren, wenn sie zur Frage passen):\n"
|
||
+ zeilen + "\n")
|
||
|
||
|
||
async def quiz_generieren(
|
||
topic: str, baustein: str, section: str, kompakt: str | None,
|
||
muster: str, niveau: str = "anfaenger", provider: str = DEFAULT_PROVIDER,
|
||
distraktor_block: str = "",
|
||
) -> dict | None:
|
||
"""Aus einem Muster eine Single-Choice-Frage (genau 1 richtig), im Anspruch des Niveaus.
|
||
Starkes Modell (role guide) für korrekte Flags. → {frage, optionen} · None bei Fehler.
|
||
distraktor_block: optionale typische Irrtümer (aus den Fakten-Hürden) als Distraktor-Quelle."""
|
||
try:
|
||
section_block, kompakt_block = _bloecke(section, kompakt)
|
||
return await _gen_call(
|
||
"Baustein-Quiz", "guide", _quiz_schema, provider, lane="batch",
|
||
topic=topic, baustein=baustein, section_block=section_block,
|
||
kompakt_block=kompakt_block, muster=muster, niveau_block=_niveau_block(niveau),
|
||
distraktor_block=distraktor_block,
|
||
)
|
||
except Exception:
|
||
log.warning("[%s] Quiz-Frage fehlgeschlagen (%s)", topic, baustein, exc_info=True)
|
||
return None
|
||
|
||
|
||
async def lueckwahl_generieren(
|
||
topic: str, baustein: str, section: str, kompakt: str | None,
|
||
muster: str, niveau: str = "anfaenger", provider: str = DEFAULT_PROVIDER,
|
||
distraktor_block: str = "",
|
||
) -> dict | None:
|
||
"""Lückentext mit Auswahl: Satz mit ___ + 4 Begriffe, genau 1 richtig — im Anspruch des Niveaus.
|
||
→ {satz, optionen:[{text,korrekt}]} · None bei Fehler.
|
||
distraktor_block: optionale typische Irrtümer (aus den Fakten-Hürden) als Distraktor-Quelle."""
|
||
try:
|
||
section_block, kompakt_block = _bloecke(section, kompakt)
|
||
return await _gen_call(
|
||
"Baustein-Lueckwahl", "guide", _lueckwahl_schema, provider, lane="batch",
|
||
topic=topic, baustein=baustein, section_block=section_block,
|
||
kompakt_block=kompakt_block, muster=muster, niveau_block=_niveau_block(niveau),
|
||
distraktor_block=distraktor_block,
|
||
)
|
||
except Exception:
|
||
log.warning("[%s] Lückentext-Auswahl fehlgeschlagen (%s)", topic, baustein, exc_info=True)
|
||
return None
|
||
|
||
|
||
def _lueck_schema(data) -> dict | None:
|
||
"""{"satz": str (mit ___), "loesung": str, "alternativen": [str]} → validiert · sonst None."""
|
||
if not isinstance(data, dict):
|
||
return None
|
||
satz = str(data.get("satz", "")).strip()
|
||
loesung = str(data.get("loesung", "")).strip()
|
||
alt = data.get("alternativen", [])
|
||
if not satz or "___" not in satz or not loesung:
|
||
return None
|
||
alternativen = [str(a).strip() for a in alt if isinstance(a, str) and str(a).strip()] if isinstance(alt, list) else []
|
||
return {"satz": satz, "loesung": loesung, "alternativen": alternativen}
|
||
|
||
|
||
async def lueckentext_generieren(
|
||
topic: str, baustein: str, section: str, kompakt: str | None,
|
||
muster: str, niveau: str = "anfaenger", provider: str = DEFAULT_PROVIDER,
|
||
) -> dict | None:
|
||
"""Aus einem Muster eine Lückentext-Aufgabe (Satz mit ___, Lösung, Synonyme), im Anspruch
|
||
des Niveaus. → {satz, loesung, alternativen} · None bei Fehler."""
|
||
try:
|
||
section_block, kompakt_block = _bloecke(section, kompakt)
|
||
return await _gen_call(
|
||
"Baustein-Lueckentext", "guide", _lueck_schema, provider, lane="batch",
|
||
topic=topic, baustein=baustein, section_block=section_block,
|
||
kompakt_block=kompakt_block, muster=muster, niveau_block=_niveau_block(niveau),
|
||
)
|
||
except Exception:
|
||
log.warning("[%s] Lückentext-Frage fehlgeschlagen (%s)", topic, baustein, exc_info=True)
|
||
return None
|
||
|
||
|
||
def _norm_begriff(t: str) -> str:
|
||
return re.sub(r"[^\wäöüß]", "", str(t or "").lower())
|
||
|
||
|
||
def _richtig_schema(data) -> dict | None:
|
||
if not isinstance(data, dict) or not isinstance(data.get("richtig"), bool):
|
||
return None
|
||
return {"richtig": data["richtig"]}
|
||
|
||
|
||
async def lueckentext_pruefen(
|
||
topic: str, baustein: str, satz: str, loesung: str, alternativen: list[str],
|
||
eingabe: str, provider: str = DEFAULT_PROVIDER,
|
||
) -> bool:
|
||
"""Lückentext-Antwort prüfen: erst normalisierter Vergleich (Lösung + Synonyme),
|
||
sonst 1 KI-Call für Synonym-Toleranz. Fail-open zu RICHTIG nur bei exaktem Match."""
|
||
if not eingabe.strip():
|
||
return False
|
||
norm = _norm_begriff(eingabe)
|
||
if norm and norm in {_norm_begriff(loesung), *(_norm_begriff(a) for a in alternativen)}:
|
||
return True
|
||
data = await _gen_call(
|
||
"Baustein-Lueckentext-Pruefung", "fast", _richtig_schema, provider,
|
||
topic=topic, baustein=baustein, satz=satz, loesung=loesung,
|
||
alternativen=", ".join(alternativen) or "(keine)", eingabe=eingabe,
|
||
)
|
||
return bool(data and data["richtig"])
|
||
|
||
|
||
async def pruefung_bewertung_schnell(
|
||
topic: str, baustein: str, section: str, kompakt: str | None,
|
||
frage: str, messages: list[dict], provider: str = DEFAULT_PROVIDER,
|
||
) -> dict | None:
|
||
"""Aktion 'antwort' (Agent 1, schnell): nur Evaluator, kein Kritiker. → {feedback, niveau}."""
|
||
try:
|
||
section_block, kompakt_block = _bloecke(section, kompakt)
|
||
transcript = _transcript(messages) if messages else "(leer)"
|
||
return await _gen_call(
|
||
"Baustein-Bewertung", "judge", _bewertung_schema, provider,
|
||
topic=topic, baustein=baustein, section_block=section_block, kompakt_block=kompakt_block,
|
||
frage=frage.strip() or "(keine Frage übergeben)", transcript=transcript,
|
||
begruendung_block="(keine)", kritik_block="(keine)",
|
||
)
|
||
except Exception:
|
||
log.warning("[%s] Schnell-Bewertung fehlgeschlagen (%s)", topic, baustein, exc_info=True)
|
||
return None
|
||
|
||
|
||
async def pruefung_bewertung(
|
||
topic: str, baustein: str, section: str, kompakt: str | None,
|
||
frage: str, messages: list[dict], provider: str = DEFAULT_PROVIDER,
|
||
role: str = "judge", begruendung: str = "",
|
||
) -> dict | None:
|
||
"""Aktion 'antwort_pruefen' (Agent 2, genau): Evaluator + Kritiker. → {feedback, niveau}.
|
||
|
||
`role` = "guide" für „Gründlich prüfen" (starkes Modell). `begruendung` = optionale
|
||
Unzufriedenheit des Lerners mit einer früheren Bewertung.
|
||
"""
|
||
try:
|
||
section_block, kompakt_block = _bloecke(section, kompakt)
|
||
transcript = _transcript(messages) if messages else "(leer)"
|
||
begruendung_block = begruendung.strip() or "(keine)"
|
||
return await _bewertung_mit_kritik(
|
||
topic, baustein, section_block, kompakt_block,
|
||
frage.strip() or "(keine Frage übergeben)", transcript, begruendung_block, provider, role,
|
||
)
|
||
except Exception:
|
||
log.warning("[%s] Bewertung fehlgeschlagen (%s)", topic, baustein, exc_info=True)
|
||
return None
|
||
|
||
|
||
async def baustein_diskussion(
|
||
topic: str, baustein: str, section: str, kompakt: str | None,
|
||
frage: str, letzte_bewertung: str | None, messages: list[dict], provider: str = DEFAULT_PROVIDER,
|
||
) -> str | None:
|
||
"""Aktion 'diskussion': Tutor erklärt/diskutiert die Frage oder eine Bewertung.
|
||
|
||
Kein Bewerten, kein Kritiker — hier ist der Mensch der Prüfer. None bei Fehler.
|
||
"""
|
||
try:
|
||
section_block, kompakt_block = _bloecke(section, kompakt)
|
||
prompt = _prompt(
|
||
"Baustein-Pruefung-Diskussion",
|
||
topic=topic, baustein=baustein,
|
||
section_block=section_block, kompakt_block=kompakt_block,
|
||
frage=frage.strip() or "(keine Frage übergeben)",
|
||
letzte_bewertung_block=(letzte_bewertung or "").strip() or "(noch keine)",
|
||
transcript=_transcript(messages) if messages else "(leer)",
|
||
)
|
||
returncode, stdout, _ = await run_agent(
|
||
"pruefungdiskussion-" + str(uuid.uuid4()), prompt, CHAT_TIMEOUT,
|
||
provider=provider, role="fast", capabilities="none", lane="interactive",
|
||
)
|
||
if returncode != 0:
|
||
return None
|
||
return stdout.strip() or None
|
||
except Exception:
|
||
log.warning("[%s] Prüfungs-Diskussion fehlgeschlagen (%s)", topic, baustein, exc_info=True)
|
||
return None
|
||
|
||
|
||
async def baustein_element_anlegen(topic: str, baustein: str, section: str, provider: str = DEFAULT_PROVIDER) -> None:
|
||
"""Hintergrund-Task nach dem Absolvieren: Baustein als Element anlegen.
|
||
|
||
Dedup über normalisierte Titel — existiert schon ein Element zum Baustein,
|
||
passiert nichts. Darf nie eine Exception nach außen werfen.
|
||
"""
|
||
try:
|
||
vorhanden = {_norm_titel(e["title"]) for e in await list_elements(topic)}
|
||
if _norm_titel(baustein) in vorhanden:
|
||
return
|
||
fields = await generate_element(topic, hint=baustein, provider=provider, extra_context=section)
|
||
if _norm_titel(fields["title"]) in vorhanden:
|
||
return
|
||
now = datetime.now(timezone.utc).isoformat()
|
||
await create_element({"id": str(uuid.uuid4()), "topic": topic, **fields, "created_at": now, "updated_at": now})
|
||
log.info("[%s] Baustein als Element angelegt: %s", topic, fields["title"])
|
||
except Exception:
|
||
log.warning("[%s] Element-Anlage nach Prüfung fehlgeschlagen (%s)", topic, baustein, exc_info=True)
|