128 lines
5.2 KiB
Python
128 lines
5.2 KiB
Python
"""Roadmap-Generierung: ein read-only Übersichts-Graph pro Thema (Advance Organizer).
|
|
|
|
Quelle = vorhandene Bausteine (keine Web-Recherche). Zwei Schritte:
|
|
Bauen (1 Agent synthetisiert aus der Baustein-Übersicht einen gerichteten Graph aus
|
|
Bereichen) und Prüfung (1 Judge fixt Struktur/Knappheit). Schritt-Dateien bleiben liegen →
|
|
Abbruch erhält Fortschritt, ▶ setzt am offenen Schritt fort.
|
|
"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from bausteine import lade_uebersicht
|
|
from config import KONSENS_MAX_RUNDEN
|
|
from jsonio import read_json_file as _json_datei
|
|
from pipeline import (
|
|
CANCELLED, FAILED, GenContext, _extra, _fail, _log,
|
|
_prompt, _set_step, _timeout, is_guide_cancelled, run_single_slot,
|
|
)
|
|
|
|
ROADMAP_STEPS = ("Bauen", "Prüfung")
|
|
|
|
_PRIO = ("grund", "kern", "vertiefung")
|
|
|
|
|
|
def _roadmap_schema(data) -> dict | None:
|
|
"""{"richtung", "knoten":[{id,title,text,prioritaet}], "kanten":[{von,nach,typ}]} →
|
|
bereinigtes Graph-Dict · sonst None. IDs eindeutig, Kanten referenzieren echte Knoten."""
|
|
if not isinstance(data, dict) or not isinstance(data.get("knoten"), list):
|
|
return None
|
|
knoten, ids = [], set()
|
|
for k in data["knoten"]:
|
|
if not isinstance(k, dict):
|
|
return None
|
|
kid = str(k.get("id", "")).strip()
|
|
title = str(k.get("title", "")).strip()
|
|
text = str(k.get("text", "")).strip()
|
|
if not kid or not title or kid in ids:
|
|
continue
|
|
ids.add(kid)
|
|
prio = k.get("prioritaet") if k.get("prioritaet") in _PRIO else "kern"
|
|
knoten.append({"id": kid, "title": title, "text": text, "prioritaet": prio})
|
|
if not knoten:
|
|
return None
|
|
kanten = []
|
|
for e in (data.get("kanten") or []):
|
|
if not isinstance(e, dict):
|
|
continue
|
|
von, nach = str(e.get("von", "")).strip(), str(e.get("nach", "")).strip()
|
|
if von not in ids or nach not in ids or von == nach:
|
|
continue # hängende/Selbst-Kante verwerfen
|
|
typ = e.get("typ") if e.get("typ") in ("prereq", "optional") else "prereq"
|
|
kanten.append({"von": von, "nach": nach, "typ": typ})
|
|
richtung = data.get("richtung") if data.get("richtung") in ("TB", "LR") else "TB"
|
|
return {"richtung": richtung, "knoten": knoten, "kanten": kanten}
|
|
|
|
|
|
def _uebersicht_block(topic: str) -> str:
|
|
"""Baustein-Übersicht (nur Bausteine mit ≥1 relevantem Subbaustein) als Prompt-Text."""
|
|
zeilen = []
|
|
for b in lade_uebersicht(topic):
|
|
rel = [s["titel"] for s in b["subbausteine"] if s.get("relevanz") != "rand"]
|
|
if not rel:
|
|
continue
|
|
kopf = f"{b['num']}. {b['titel']}"
|
|
if b.get("beschreibung"):
|
|
kopf += f" — {b['beschreibung']}"
|
|
zeilen.append(kopf)
|
|
zeilen += [f" - {t}" for t in rel]
|
|
return "\n".join(zeilen)
|
|
|
|
|
|
async def _generate_roadmap(
|
|
guide_id: str, topic: str, instructions: str, provider: str, content_path: Path,
|
|
) -> dict | None:
|
|
is_cancelled = lambda: is_guide_cancelled(guide_id)
|
|
ctx = GenContext(topic=topic, provider=provider, is_cancelled=is_cancelled, guide_id=guide_id)
|
|
|
|
uebersicht = _uebersicht_block(topic)
|
|
if not uebersicht.strip():
|
|
await _fail(guide_id, "Keine relevanten Bausteine für die Roadmap")
|
|
return None
|
|
|
|
d, stem = content_path.parent, content_path.stem
|
|
bau_path = d / f"{stem}.roadmap-bau.json"
|
|
check_path = d / f"{stem}.roadmap-check.json"
|
|
|
|
# Schritt 0: Bauen — 1 Agent synthetisiert den Graph. Gültige Datei überspringt.
|
|
graph = _roadmap_schema(_json_datei(bau_path))
|
|
if graph is None:
|
|
await _set_step(guide_id, 0, "Baue Roadmap…")
|
|
status, _ = await run_single_slot(
|
|
ctx, "Roadmap-Bauen",
|
|
key=f"{guide_id}-roadmap-bau",
|
|
prompt=_prompt("Roadmap-Bauen", topic=topic, uebersicht=uebersicht,
|
|
out_path=bau_path, extra=_extra(instructions)),
|
|
role="guide", capabilities="files",
|
|
payload=lambda result: _roadmap_schema(_json_datei(bau_path)),
|
|
timeout=_timeout("roadmap"),
|
|
)
|
|
if status == CANCELLED:
|
|
return None
|
|
if status == FAILED:
|
|
await _fail(guide_id, "Roadmap-Bau fehlgeschlagen")
|
|
return None
|
|
graph = _roadmap_schema(_json_datei(bau_path))
|
|
|
|
# Schritt 1: Prüfung — Judge fixt Struktur/Knappheit; Fallback = Bau-Graph.
|
|
geprueft = _roadmap_schema(_json_datei(check_path))
|
|
if geprueft is None:
|
|
await _set_step(guide_id, 1, "Prüfe Roadmap…")
|
|
status, _ = await run_single_slot(
|
|
ctx, "Roadmap-Prüfung",
|
|
key=f"{guide_id}-roadmap-check",
|
|
prompt=_prompt("Roadmap-Check", topic=topic, uebersicht=uebersicht,
|
|
graph=json.dumps(graph, ensure_ascii=False, indent=1),
|
|
out_path=check_path, extra=_extra(instructions)),
|
|
role="judge", capabilities="files",
|
|
payload=lambda result: _roadmap_schema(_json_datei(check_path)),
|
|
timeout=_timeout("roadmap_check"),
|
|
)
|
|
if status == CANCELLED:
|
|
return None
|
|
if status == FAILED:
|
|
_log(topic, "Roadmap-Prüfung fehlgeschlagen — Bau-Graph bleibt")
|
|
geprueft = _roadmap_schema(_json_datei(check_path))
|
|
|
|
return geprueft or graph
|