marek/creator2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
team3
2026-07-13 12:31:25 +02:00
parent 32d7ad9ea1
commit 9cd8e02e22
34 changed files with 747 additions and 606 deletions
Download Patch File Download Diff File Expand all files Collapse all files

208
tests/test_struktur.py
View File

@@ -1,7 +1,8 @@
"""Struktur-Ebene: Topo-Sortierung, Zyklenbruch, Band-Schnitt — mit synthetischen Atomen."""
"""Struktur-Ebene: Band-Schnitt, Judge-Ordnung mit Quellpositions-Prior,
Kapitel — mit synthetischen Atomen."""
import db
import inventar
import fake_agents
import llm
import struktur
from conftest import run_anlegen, topic_anlegen
@@ -10,7 +11,7 @@ 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, braucht=db.j([]))
soll_id=soll_id, ziel_id=ziel_id)
def _baustein_groessen(topic):
@@ -29,7 +30,7 @@ async def test_band_split_43_atome():
_atom(topic, f"A{i}", ziel, soll)
ctx = llm.Kontext(run, topic, "minimax")
ctx.ebene = "struktur"
struktur._bausteine_schneiden(ctx)
await struktur._bausteine_schneiden(ctx)
groessen = _baustein_groessen(topic)
assert sum(groessen) == 43
assert all(struktur.BAUSTEIN_MIN_ATOME <= n <= struktur.BAUSTEIN_MAX_ATOME
@@ -51,7 +52,7 @@ async def test_band_messen_konsistent_mit_schnitt():
_atom(topic, f"G{i}", z2, soll)
ctx = llm.Kontext(run, topic, "minimax")
ctx.ebene = "struktur"
struktur._bausteine_schneiden(ctx)
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)}
@@ -65,52 +66,10 @@ async def test_einsame_kleingruppe_kein_band():
_atom(topic, f"A{i}", ziel, soll)
ctx = llm.Kontext(run, topic, "minimax")
ctx.ebene = "struktur"
struktur._bausteine_schneiden(ctx)
await struktur._bausteine_schneiden(ctx)
assert "band" not in {b["art"] for b in struktur.messen(ctx)}
def test_topo_ordnung():
rang = {1: (0, 1), 2: (0, 2), 3: (0, 3)}
# 3 braucht 1, 2 braucht 3 → 1, 3, 2
assert struktur._topo([1, 2, 3], [(3, 1), (2, 3)], rang) == [1, 3, 2]
def test_topo_fremde_kanten_bleiben_draussen():
# Kante zu Atom 3 (nicht in der Gruppe) darf weder 3 hineinziehen
# noch Gruppenmitglieder verdrängen (aak-Bug: 22 Atome ohne Baustein)
rang = {1: (0, 1), 2: (0, 2), 3: (0, 3)}
out = struktur._topo([1, 2], [(1, 3), (2, 1)], rang)
assert out == [1, 2]
def test_zyklus_finden():
assert struktur._finde_zyklus([1, 2], [(1, 2), (2, 1)]) is not None
assert struktur._finde_zyklus([1, 2, 3], [(2, 1), (3, 2)]) is None
async def test_zyklen_brechen_und_schneiden():
topic = topic_anlegen()
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")
a = _atom(topic, "A", ziel, soll)
b = _atom(topic, "B", ziel, soll)
db.execute("INSERT INTO kanten(topic, von_atom, zu_atom, art) VALUES(?,?,?,'braucht')",
(topic, a, b))
db.execute("INSERT INTO kanten(topic, von_atom, zu_atom, art) VALUES(?,?,?,'braucht')",
(topic, b, a))
ctx = llm.Kontext(run, topic, "minimax")
ctx.ebene = "struktur"
await struktur._zyklen_brechen(ctx)
aktiv = db.query("SELECT * FROM kanten WHERE topic=? AND art='braucht' AND status='aktiv'",
(topic,))
assert len(aktiv) == 1 # genau eine Kante gebrochen
struktur._bausteine_schneiden(ctx)
atome = struktur._atome(topic)
assert all(x["baustein_id"] for x in atome)
async def test_band_split():
topic = topic_anlegen("band")
run = run_anlegen(topic)
@@ -119,34 +78,31 @@ async def test_band_split():
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")
struktur._bausteine_schneiden(ctx)
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", "zyklus")]
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 Kante);
# Kapitel entstehen danach als kontiguierliche Segmente
# 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")
a = _atom(topic, "A", z1, s1) # 1-Atom-Ziel, anderer Soll-Punkt als z2
b0 = _atom(topic, "B0", z2, s2)
for i in (1, 2, 3):
_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)
db.execute("INSERT INTO kanten(topic, von_atom, zu_atom, art) VALUES(?,?,?,'braucht')",
(topic, a, b0))
ctx = llm.Kontext(run, topic, "minimax")
ctx.ebene = "struktur"
struktur._bausteine_schneiden(ctx)
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)
@@ -155,8 +111,79 @@ async def test_merge_ohne_soll_schranke_und_kapitel():
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):
import fake_agents
topic = topic_anlegen("kapfall")
run = run_anlegen(topic)
ziel = db.insert("lernziele", topic=topic, text="Kann X", status="aktiv")
@@ -185,14 +212,13 @@ async def test_kapitel_retry_und_fallback(monkeypatch):
async def test_level_kalibrierung(monkeypatch):
import fake_agents
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, braucht=db.j([]))
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, braucht=db.j([]))
level="M", status="neu", ziel_id=ziel)
def judge(prompt):
return [{"atom": a1, "level": "E"}, {"atom": a2, "level": "S"},
@@ -205,50 +231,6 @@ async def test_level_kalibrierung(monkeypatch):
assert db.one("SELECT level FROM atome WHERE id=?", (a2,))["level"] == "S"
def test_level_konflikt_absenkung():
# v(E) braucht z(S), z braucht w(M) → Kaskade senkt z UND w auf E
topic = topic_anlegen("konflikt")
run = run_anlegen(topic)
ziel = db.insert("lernziele", topic=topic, text="Kann X", status="aktiv")
def atom(titel, level):
return db.insert("atome", topic=topic, titel=titel, typ="begriff",
definition="d", level=level, status="neu", ziel_id=ziel,
braucht=db.j([]))
v, z, w = atom("V", "E"), atom("Z", "S"), atom("W", "M")
for von, zu in ((v, z), (z, w)):
db.execute("INSERT INTO kanten(topic, von_atom, zu_atom, art)"
" VALUES(?,?,?,'braucht')", (topic, von, zu))
struktur._level_konflikte_loesen(topic)
assert db.one("SELECT level FROM atome WHERE id=?", (z,))["level"] == "E"
assert db.one("SELECT level FROM atome WHERE id=?", (w,))["level"] == "E"
ctx = llm.Kontext(run, topic, "minimax")
assert not [b for b in struktur.messen(ctx) if b["art"] == "level_konflikt"]
async def test_braucht_fallback_kompositum():
# braucht-Titel "Approximationsalgorithmus" (Kompositum) trifft weder norm noch
# titel_kern das Atom "Approximativer Algorithmus" (Phrase). Der Fallback-Judge
# legt die Kante; _level_konflikte_loesen senkt die Voraussetzung von M auf E.
topic = topic_anlegen("brfall")
run = run_anlegen(topic)
ziel = db.insert("lernziele", topic=topic, text="Kann X", status="aktiv")
a = db.insert("atome", topic=topic, titel="Absolute Güte", typ="begriff",
definition="d", level="E", status="neu", ziel_id=ziel,
braucht=db.j(["Approximationsalgorithmus"]))
b = db.insert("atome", topic=topic, titel="Approximativer Algorithmus", typ="begriff",
definition="d", level="M", status="neu", ziel_id=ziel, braucht=db.j([]))
ctx = llm.Kontext(run, topic, "minimax")
ctx.ebene = "struktur"
inventar._kanten_aufloesen(topic) # deterministisch: verfehlt das Kompositum
assert db.query("SELECT * FROM kanten WHERE topic=? AND von_atom=?", (topic, a)) == []
await inventar._braucht_fallback(ctx)
kante = db.one("SELECT * FROM kanten WHERE topic=? AND von_atom=? AND zu_atom=?",
(topic, a, b))
assert kante and kante["art"] == "braucht"
struktur._level_konflikte_loesen(topic)
assert db.one("SELECT level FROM atome WHERE id=?", (b,))["level"] == "E"
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
@@ -258,12 +240,12 @@ async def test_level_split_ordnung_kapitel():
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, braucht=db.j([]))
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, braucht=db.j([]))
level="M", status="neu", soll_id=soll, ziel_id=ziel)
ctx = llm.Kontext(run, topic, "minimax")
ctx.ebene = "struktur"
struktur._bausteine_schneiden(ctx)
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)
@@ -273,7 +255,7 @@ async def test_level_split_ordnung_kapitel():
assert not arten & {"level_mix", "kapitel_level", "band", "partition"}
def test_kein_merge_ueber_level():
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)
@@ -282,11 +264,11 @@ def test_kein_merge_ueber_level():
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, braucht=db.j([]))
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, braucht=db.j([]))
level="M", status="neu", soll_id=soll, ziel_id=z2)
ctx = llm.Kontext(run, topic, "minimax")
struktur._bausteine_schneiden(ctx)
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"]}