This commit is contained in:
team3
2026-06-27 00:58:46 +02:00
parent 2a2026c9ae
commit f018ca5048
44 changed files with 3070 additions and 379 deletions

View File

@@ -13,7 +13,7 @@ from datetime import datetime, timezone
from agents import run_agent
from config import DEFAULT_PROVIDER
from database import create_element, list_elements
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
@@ -27,34 +27,68 @@ log = logging.getLogger("creator.lernen")
STUFEN = (("anfaenger", 0.2), ("fortgeschritten", 0.4), ("experte", 0.6), ("meister", 1.0))
def cap_fuer(n_subs: int) -> int:
"""Maximal-Score eines Bausteins: 4 Punkte je relevantem Subbaustein, gedeckelt [10, 50]."""
return max(10, min(50, 4 * n_subs))
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 14. 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: int) -> str | None:
"""Höchste erreichte Stufe (None unter 20 %)."""
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):
if score >= _schwelle(p, cap_final_wert):
erreicht = key
return erreicht
def progressiver_malus(basis: int, cap: int) -> int:
"""Fehler-Strafe nach Fortschritt: 020 %→0 · 2140 %1 · 4160 %2 · 61100 %3."""
pct = (basis / cap) if cap else 0.0
if pct <= 0.2:
return 0
if pct <= 0.4:
return -1
if pct <= 0.6:
return -2
return -3
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
@@ -70,34 +104,66 @@ FRAGETYPEN = {
}
# Antwort-Niveau (wie sehr der Kern getroffen ist) → Punkt-Delta.
# 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 (Randthema/Verweis auf Nichtgezeigtes) → kein Malus
"kaum": -1, # < 25 % richtig
"teilweise": 0, # 2549 %
"solide": 1, # 5074 %
"stark": 2, # 7599 %
"komplett": 3, # 100 %
"unbeantwortbar": 0, # Frage selbst kaputt → keine Änderung
"kaum": -1, # < 25 % richtig → Malus
"teilweise": 0, # 2549 % → neutral
"solide": 16, # 5074 %
"stark": 24, # 7599 % (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 score_berechnen(basis: int, delta: int, floor: int, cap: int) -> int:
"""Neuer Score · driftfrei aus der Basis. `delta` liefert die Route je Prüfungsform
(Quiz +1/+3, Lückentext +1/+3, Erklären 1…+3; falsch = progressiver_malus). Auf
[floor, cap] geklemmt. floor = höchste erreichte Stufenschwelle (kein Rückfall);
cap = 4× relevante Subbausteine [10, 50]. Am Deckel (basis ≥ cap) eingefroren („Max")."""
if basis >= cap:
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, basis + delta))
return max(floor, min(cap_akt, basis + delta))
def floor_aus_score(basis: int, cap: int) -> int:
"""Untergrenze: höchste erreichte Stufenschwelle — kein Rückfall unter die Stufe."""
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)
s = _schwelle(p, cap_fin)
if basis >= s:
floor = s
floor = max(floor, s)
for s in s_schwellen:
if basis >= s:
floor = max(floor, s)
return floor
@@ -199,7 +265,8 @@ def _doppelfrage_mangel(frage: str) -> str | 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, provider: 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)"
@@ -209,7 +276,7 @@ async def _frage_mit_kritik(
"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, kritik_block=kritik_block,
typ_block=typ_block, fokus_block=fokus_block, niveau_block=niveau_block, kritik_block=kritik_block,
)
if data is None:
return None
@@ -277,13 +344,27 @@ def _vermeide_block(vermeide: list[str] | None) -> str:
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".
NIVEAU_ROLLE = {
"anfaenger": "Der Lerner ist ANFÄNGER. Frage nach dem Grundverständnis — das Kernkonzept, einfach und direkt.",
"fortgeschritten": "Der Lerner ist FORTGESCHRITTEN. Frage etwas tiefer — Zusammenhänge und die typische Anwendung.",
"experte": "Der Lerner ist EXPERTE. Frage anspruchsvoll — Sonderfälle, Abgrenzungen, typische Tücken.",
"meister": "Der Lerner ist auf MEISTER-Niveau. Frage sehr anspruchsvoll — Edge-Cases, präzise Begründung, Transfer auf Neues.",
}
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, provider: str = DEFAULT_PROVIDER,
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,
dann Kritiker (Generator → Kritiker sequenziell) · None bei Fehler."""
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)"
@@ -296,7 +377,7 @@ async def pruefung_frage(
)
return await _frage_mit_kritik(
topic, baustein, section_block, kompakt_block, transcript,
_vermeide_block(vermeide), typ_block, fokus_block, provider,
_vermeide_block(vermeide), typ_block, fokus_block, _niveau_block(niveau), provider,
)
except Exception:
log.warning("[%s] Frage fehlgeschlagen (%s)", topic, baustein, exc_info=True)
@@ -305,17 +386,17 @@ async def pruefung_frage(
async def pruefung_frage_variante(
topic: str, baustein: str, section: str, kompakt: str | None,
muster: str, provider: str = DEFAULT_PROVIDER,
muster: str, niveau: str = "anfaenger", provider: str = DEFAULT_PROVIDER,
) -> str | None:
"""Aktion 'frage' mit Muster: aus einem vordefinierten Muster eine konkrete, im
Wortlaut leicht andere Frage formulieren. Kein Kritiker (Muster ist build-geprüft).
"""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,
kompakt_block=kompakt_block, muster=muster, niveau_block=_niveau_block(niveau),
)
if data is None:
return None
@@ -341,17 +422,16 @@ def _optionen_schema(opts) -> list[dict] | None:
return out
def _quiz_schema(data, schwer: bool = False) -> dict | None:
def _quiz_schema(data) -> dict | None:
"""{"frage": str, "optionen": [{text, korrekt}]×4} → validiert · sonst None.
leicht: genau 1 richtig (Single) · schwer: mindestens 2 richtig (mehrdeutig)."""
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
n_korrekt = sum(o["korrekt"] for o in out)
if (n_korrekt < 2 if schwer else n_korrekt != 1):
if sum(o["korrekt"] for o in out) != 1:
return None
return {"frage": frage, "optionen": out}
@@ -367,24 +447,35 @@ def _lueckwahl_schema(data) -> dict | 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, schwer: bool = False, provider: str = DEFAULT_PROVIDER,
muster: str, niveau: str = "anfaenger", provider: str = DEFAULT_PROVIDER,
distraktor_block: str = "",
) -> dict | None:
"""Aus einem Muster eine Multiple-Choice-Frage. schwer: 04 richtig (Multi).
leicht: genau 1 richtig (Single). Starkes Modell (role guide) für korrekte Flags.
{frage, optionen:[{text,korrekt}]} · None bei Fehler."""
"""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)
anzahl_block = (
"MINDESTENS ZWEI Optionen sind richtig (gern auch drei) — NIEMALS nur eine. Mische bewusst, welche."
if schwer else
"GENAU EINE Option ist richtig, die anderen drei sind klar falsch."
)
return await _gen_call(
"Baustein-Quiz", "guide", (lambda d: _quiz_schema(d, schwer=schwer)), provider, lane="batch",
"Baustein-Quiz", "guide", _quiz_schema, provider, lane="batch",
topic=topic, baustein=baustein, section_block=section_block,
kompakt_block=kompakt_block, muster=muster, anzahl_block=anzahl_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)
@@ -393,16 +484,19 @@ async def quiz_generieren(
async def lueckwahl_generieren(
topic: str, baustein: str, section: str, kompakt: str | None,
muster: str, provider: str = DEFAULT_PROVIDER,
muster: str, niveau: str = "anfaenger", provider: str = DEFAULT_PROVIDER,
distraktor_block: str = "",
) -> dict | None:
"""Lückentext mit Auswahl (leichte Variante): Satz mit ___ + 4 Begriffe, genau 1 richtig.
{satz, optionen:[{text,korrekt}]} · None bei Fehler."""
"""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,
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)
@@ -424,16 +518,16 @@ def _lueck_schema(data) -> dict | None:
async def lueckentext_generieren(
topic: str, baustein: str, section: str, kompakt: str | None,
muster: str, provider: str = DEFAULT_PROVIDER,
muster: str, niveau: str = "anfaenger", provider: str = DEFAULT_PROVIDER,
) -> dict | None:
"""Aus einem Muster eine Lückentext-Aufgabe (Satz mit ___, Lösung, Synonyme).
{satz, loesung, alternativen} · None bei Fehler."""
"""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,
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)