276 lines
13 KiB
Python
276 lines
13 KiB
Python
"""Struktur-Ebene: Band-Schnitt, Judge-Ordnung mit Quellpositions-Prior,
|
||
Kapitel — mit synthetischen Atomen."""
|
||
|
||
import db
|
||
import fake_agents
|
||
import llm
|
||
import struktur
|
||
from conftest import run_anlegen, topic_anlegen
|
||
|
||
|
||
def _atom(topic, titel, ziel_id=None, soll_id=1):
|
||
return db.insert("atome", topic=topic, titel=titel, typ="begriff",
|
||
definition=f"Definition {titel}", status="neu",
|
||
soll_id=soll_id, ziel_id=ziel_id)
|
||
|
||
|
||
def _baustein_groessen(topic):
|
||
from collections import Counter
|
||
c = Counter(a["baustein_id"] for a in struktur._atome(topic))
|
||
return sorted(c.values())
|
||
|
||
|
||
async def test_band_split_43_atome():
|
||
"""Große Gruppe gleichverteilt splitten: jeder Baustein im Band 4–8."""
|
||
topic = topic_anlegen("split43")
|
||
run = run_anlegen(topic)
|
||
soll = db.insert("soll", topic=topic, punkt="P", status="bestaetigt", belege=db.j([]))
|
||
ziel = db.insert("lernziele", topic=topic, text="Kann P", soll_id=soll, status="aktiv")
|
||
for i in range(43):
|
||
_atom(topic, f"A{i}", ziel, soll)
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
ctx.ebene = "struktur"
|
||
await struktur._bausteine_schneiden(ctx)
|
||
groessen = _baustein_groessen(topic)
|
||
assert sum(groessen) == 43
|
||
assert all(struktur.BAUSTEIN_MIN_ATOME <= n <= struktur.BAUSTEIN_MAX_ATOME
|
||
for n in groessen), groessen
|
||
assert "band" not in {b["art"] for b in struktur.messen(ctx)}
|
||
|
||
|
||
async def test_band_messen_konsistent_mit_schnitt():
|
||
"""Kleine Gruppe mergt trotz Summe > MAX (Split re-balanciert) → kein band-Befund;
|
||
eine einsame Kleingruppe ohne Level-Partner meldet ebenfalls nichts."""
|
||
topic = topic_anlegen("bandkon")
|
||
run = run_anlegen(topic)
|
||
soll = db.insert("soll", topic=topic, punkt="P", status="bestaetigt", belege=db.j([]))
|
||
z1 = db.insert("lernziele", topic=topic, text="Z1", soll_id=soll, status="aktiv")
|
||
z2 = db.insert("lernziele", topic=topic, text="Z2", soll_id=soll, status="aktiv")
|
||
for i in range(3):
|
||
_atom(topic, f"K{i}", z1, soll)
|
||
for i in range(13):
|
||
_atom(topic, f"G{i}", z2, soll)
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
ctx.ebene = "struktur"
|
||
await struktur._bausteine_schneiden(ctx)
|
||
assert all(4 <= n <= 8 for n in _baustein_groessen(topic))
|
||
assert "band" not in {b["art"] for b in struktur.messen(ctx)}
|
||
|
||
|
||
async def test_einsame_kleingruppe_kein_band():
|
||
topic = topic_anlegen("lone")
|
||
run = run_anlegen(topic)
|
||
soll = db.insert("soll", topic=topic, punkt="P", status="bestaetigt", belege=db.j([]))
|
||
ziel = db.insert("lernziele", topic=topic, text="Z", soll_id=soll, status="aktiv")
|
||
for i in range(2):
|
||
_atom(topic, f"A{i}", ziel, soll)
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
ctx.ebene = "struktur"
|
||
await struktur._bausteine_schneiden(ctx)
|
||
assert "band" not in {b["art"] for b in struktur.messen(ctx)}
|
||
|
||
|
||
async def test_band_split():
|
||
topic = topic_anlegen("band")
|
||
run = run_anlegen(topic)
|
||
soll = db.insert("soll", topic=topic, punkt="P", status="bestaetigt", belege=db.j([]))
|
||
ziel = db.insert("lernziele", topic=topic, text="Kann P", soll_id=soll, status="aktiv")
|
||
for i in range(18): # 18 Atome in EINEM Ziel → muss in ≤8er-Teile splitten
|
||
_atom(topic, f"A{i}", ziel, soll)
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
await struktur._bausteine_schneiden(ctx)
|
||
bausteine = db.query("SELECT * FROM bausteine WHERE topic=?", (topic,))
|
||
assert len(bausteine) >= 3
|
||
for bs in bausteine:
|
||
n = db.one("SELECT COUNT(*) n FROM atome WHERE baustein_id=?", (bs["id"],))["n"]
|
||
assert 4 <= n <= 8
|
||
befunde = struktur.messen(ctx)
|
||
assert not [x for x in befunde if x["art"] in ("band", "partition")]
|
||
|
||
|
||
async def test_merge_ohne_soll_schranke_und_kapitel():
|
||
# Entkopplung: kleines Ziel merged über Soll-Punkt-Grenzen (Partner per
|
||
# Quell-Nähe); Kapitel entstehen danach als kontiguierliche Segmente
|
||
topic = topic_anlegen("kapitel")
|
||
run = run_anlegen(topic)
|
||
s1 = db.insert("soll", topic=topic, punkt="P1", status="bestaetigt", belege=db.j([]))
|
||
s2 = db.insert("soll", topic=topic, punkt="P2", status="bestaetigt", belege=db.j([]))
|
||
z1 = db.insert("lernziele", topic=topic, text="Kann P1", soll_id=s1, status="aktiv")
|
||
z2 = db.insert("lernziele", topic=topic, text="Kann P2", soll_id=s2, status="aktiv")
|
||
_atom(topic, "A", z1, s1) # 1-Atom-Ziel, anderer Soll-Punkt als z2
|
||
for i in (0, 1, 2, 3):
|
||
_atom(topic, f"B{i}", z2, s2)
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
ctx.ebene = "struktur"
|
||
await struktur._bausteine_schneiden(ctx)
|
||
atome = struktur._atome(topic)
|
||
assert len({x["baustein_id"] for x in atome}) == 1 # gemerged trotz fremdem Soll
|
||
await struktur._kapitel_bilden(ctx)
|
||
bausteine = db.query("SELECT * FROM bausteine WHERE topic=?", (topic,))
|
||
assert bausteine and all(x["kapitel_id"] for x in bausteine)
|
||
assert struktur.messen(ctx) == []
|
||
|
||
|
||
async def test_ordnung_judge_permutation(monkeypatch):
|
||
"""Der Ordnungs-Judge bestimmt die Lehr-Reihenfolge; Prior ist die
|
||
Quellposition (Eingabe-Nummerierung)."""
|
||
topic = topic_anlegen("ordjudge")
|
||
run = run_anlegen(topic)
|
||
soll = db.insert("soll", topic=topic, punkt="P", status="bestaetigt", belege=db.j([]))
|
||
ziele = [db.insert("lernziele", topic=topic, text=f"Kann Z{i}", soll_id=soll,
|
||
status="aktiv") for i in range(2)]
|
||
q = db.insert("quellen", topic=topic, art="datei", titel="s.txt", snapshot="s",
|
||
rolle="stoff", status="atome")
|
||
for zi, ziel in enumerate(ziele):
|
||
for i in range(4):
|
||
a = _atom(topic, f"Z{zi}A{i}", ziel, soll)
|
||
db.insert("anker", atom_id=a, quelle_id=q, start=zi * 1000 + i, ende=0,
|
||
zitat="x")
|
||
monkeypatch.setitem(fake_agents._HANDLER, "Baustein-Ordnung",
|
||
lambda p: list(reversed(fake_agents._baustein_ordnung(p))))
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
ctx.ebene = "struktur"
|
||
await struktur._bausteine_schneiden(ctx)
|
||
bausteine = db.query("SELECT * FROM bausteine WHERE topic=? ORDER BY ord", (topic,))
|
||
# Prior wäre Z0 vor Z1 — der Judge hat umgedreht, und das gilt
|
||
assert [b["ziel_id"] for b in bausteine] == [ziele[1], ziele[0]]
|
||
assert all(b["ordnung"] == "judge" for b in bausteine)
|
||
assert "ordnung_fallback" not in {x["art"] for x in struktur.messen(ctx)}
|
||
|
||
|
||
async def test_ordnung_retry_und_fallback(monkeypatch):
|
||
topic = topic_anlegen("ordfall")
|
||
run = run_anlegen(topic)
|
||
soll = db.insert("soll", topic=topic, punkt="P", status="bestaetigt", belege=db.j([]))
|
||
z1 = db.insert("lernziele", topic=topic, text="Z1", soll_id=soll, status="aktiv")
|
||
z2 = db.insert("lernziele", topic=topic, text="Z2", soll_id=soll, status="aktiv")
|
||
for ziel in (z1, z2):
|
||
for i in range(4):
|
||
_atom(topic, f"{ziel}A{i}", ziel, soll)
|
||
aufrufe = {"n": 0}
|
||
|
||
def judge(prompt):
|
||
aufrufe["n"] += 1
|
||
if aufrufe["n"] == 1:
|
||
return [1, 1] # keine Permutation
|
||
return [2, 1]
|
||
monkeypatch.setitem(fake_agents._HANDLER, "Baustein-Ordnung", judge)
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
ctx.ebene = "struktur"
|
||
await struktur._bausteine_schneiden(ctx)
|
||
assert aufrufe["n"] == 2 # Retry hat gegriffen
|
||
assert all(b["ordnung"] == "judge" for b in
|
||
db.query("SELECT * FROM bausteine WHERE topic=?", (topic,)))
|
||
|
||
# beide Versuche ungültig → Prior-Ordnung + Befund
|
||
monkeypatch.setitem(fake_agents._HANDLER, "Baustein-Ordnung", lambda p: ["x"])
|
||
await struktur._bausteine_schneiden(ctx)
|
||
assert all(b["ordnung"] == "fallback" for b in
|
||
db.query("SELECT * FROM bausteine WHERE topic=?", (topic,)))
|
||
assert "ordnung_fallback" in {x["art"] for x in struktur.messen(ctx)}
|
||
|
||
|
||
def test_baustein_rang_median_gegen_merge_gift():
|
||
"""ETH-Muster: EIN importiertes Atom vom Quellanfang darf den Baustein nicht
|
||
nach vorn ziehen — der Median liegt bei den echten Mitgliedern."""
|
||
bausteine = [{"id": 1}, {"id": 2}]
|
||
atome = [{"id": 10, "baustein_id": 1}, {"id": 11, "baustein_id": 1},
|
||
{"id": 12, "baustein_id": 1}, # Merge-Import mit Rang vom Dateianfang
|
||
{"id": 20, "baustein_id": 2}, {"id": 21, "baustein_id": 2}]
|
||
rang = {10: (1, 47000), 11: (1, 48000), 12: (1, 464),
|
||
20: (1, 5000), 21: (1, 6000)}
|
||
b_rang = struktur._baustein_rang(bausteine, atome, rang)
|
||
assert b_rang[2] < b_rang[1] # trotz 464-Import bleibt Baustein 1 hinten
|
||
|
||
|
||
async def test_kapitel_retry_und_fallback(monkeypatch):
|
||
topic = topic_anlegen("kapfall")
|
||
run = run_anlegen(topic)
|
||
ziel = db.insert("lernziele", topic=topic, text="Kann X", status="aktiv")
|
||
b = [db.insert("bausteine", topic=topic, ziel_id=ziel, titel=f"B{i}", ord=i,
|
||
status="neu") for i in range(4)]
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
ctx.ebene = "struktur"
|
||
|
||
aufrufe = {"n": 0}
|
||
def judge(prompt):
|
||
aufrufe["n"] += 1
|
||
if aufrufe["n"] == 1:
|
||
return [{"titel": "kaputt", "bis": 999999}] # ungültige Grenze
|
||
return [{"titel": "Gutes Kapitel", "bis": b[-1]}]
|
||
monkeypatch.setitem(fake_agents._HANDLER, "Kapitel-Schnitt", judge)
|
||
await struktur._kapitel_bilden(ctx)
|
||
kaps = db.query("SELECT * FROM kapitel WHERE topic=?", (topic,))
|
||
assert aufrufe["n"] == 2 and len(kaps) == 1 and kaps[0]["art"] == "judge"
|
||
|
||
monkeypatch.setitem(fake_agents._HANDLER, "Kapitel-Schnitt",
|
||
lambda p: [{"titel": "x", "bis": 999999}])
|
||
await struktur._kapitel_bilden(ctx) # beide Versuche ungültig → Fallback
|
||
kaps = db.query("SELECT * FROM kapitel WHERE topic=?", (topic,))
|
||
assert kaps and all(k["art"] == "fallback" for k in kaps)
|
||
assert "kapitel_fallback" in {bf["art"] for bf in struktur.messen(ctx)}
|
||
|
||
|
||
async def test_level_kalibrierung(monkeypatch):
|
||
topic = topic_anlegen("level")
|
||
run = run_anlegen(topic)
|
||
ziel = db.insert("lernziele", topic=topic, text="Kann X", status="aktiv")
|
||
a1 = db.insert("atome", topic=topic, titel="Kern", typ="begriff", definition="d",
|
||
level="M", status="neu", ziel_id=ziel)
|
||
a2 = db.insert("atome", topic=topic, titel="Detail", typ="aussage", definition="d",
|
||
level="M", status="neu", ziel_id=ziel)
|
||
|
||
def judge(prompt):
|
||
return [{"atom": a1, "level": "E"}, {"atom": a2, "level": "S"},
|
||
{"atom": 99999, "level": "E"}, {"atom": a2, "level": "X"}] # Müll ignorieren
|
||
monkeypatch.setitem(fake_agents._HANDLER, "Level-Kalibrierung", judge)
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
ctx.ebene = "struktur"
|
||
await struktur._level_kalibrieren(ctx)
|
||
assert db.one("SELECT level FROM atome WHERE id=?", (a1,))["level"] == "E"
|
||
assert db.one("SELECT level FROM atome WHERE id=?", (a2,))["level"] == "S"
|
||
|
||
|
||
async def test_level_split_ordnung_kapitel():
|
||
# Ein Ziel mit 4×E + 4×M → zwei Bausteine mit level, E-ord < M-ord,
|
||
# Kapitel je Durchgang, kein level_mix/kapitel_level
|
||
topic = topic_anlegen("split")
|
||
run = run_anlegen(topic)
|
||
soll = db.insert("soll", topic=topic, punkt="P", status="bestaetigt", belege=db.j([]))
|
||
ziel = db.insert("lernziele", topic=topic, text="Kann P", soll_id=soll, status="aktiv")
|
||
for i in range(4):
|
||
db.insert("atome", topic=topic, titel=f"E{i}", typ="begriff", definition="d",
|
||
level="E", status="neu", soll_id=soll, ziel_id=ziel)
|
||
db.insert("atome", topic=topic, titel=f"M{i}", typ="begriff", definition="d",
|
||
level="M", status="neu", soll_id=soll, ziel_id=ziel)
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
ctx.ebene = "struktur"
|
||
await struktur._bausteine_schneiden(ctx)
|
||
bausteine = db.query("SELECT * FROM bausteine WHERE topic=? ORDER BY ord", (topic,))
|
||
assert [b["level"] for b in bausteine] == ["E", "M"]
|
||
await struktur._kapitel_bilden(ctx)
|
||
kaps = db.query("SELECT * FROM kapitel WHERE topic=? ORDER BY ord", (topic,))
|
||
assert [k["level"] for k in kaps] == ["E", "M"]
|
||
arten = {b["art"] for b in struktur.messen(ctx)}
|
||
assert not arten & {"level_mix", "kapitel_level", "band", "partition"}
|
||
|
||
|
||
async def test_kein_merge_ueber_level():
|
||
# kleine E-Gruppe darf NICHT mit kleiner M-Gruppe mergen
|
||
topic = topic_anlegen("levelmerge")
|
||
run = run_anlegen(topic)
|
||
soll = db.insert("soll", topic=topic, punkt="P", status="bestaetigt", belege=db.j([]))
|
||
z1 = db.insert("lernziele", topic=topic, text="Kann P1", soll_id=soll, status="aktiv")
|
||
z2 = db.insert("lernziele", topic=topic, text="Kann P2", soll_id=soll, status="aktiv")
|
||
for i in range(2):
|
||
db.insert("atome", topic=topic, titel=f"E{i}", typ="begriff", definition="d",
|
||
level="E", status="neu", soll_id=soll, ziel_id=z1)
|
||
db.insert("atome", topic=topic, titel=f"M{i}", typ="begriff", definition="d",
|
||
level="M", status="neu", soll_id=soll, ziel_id=z2)
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
await struktur._bausteine_schneiden(ctx)
|
||
for b in db.query("SELECT * FROM bausteine WHERE topic=?", (topic,)):
|
||
levels = {a["level"] for a in struktur._atome(topic)
|
||
if a["baustein_id"] == b["id"]}
|
||
assert levels == {b["level"]}
|