Files
buchwerk/backend/pipeline.py
team3 4c989bc93c Spezialisierung gehört zum Thema, Hinweise zur Version
Thema „Game Design“ plus Spezialisierung „digitale Sammelkartenspiele“
wird zur Bezeichnung „Game Design – digitale Sammelkartenspiele“
(Ordner, Liste). Die Prompts bekommen das Hauptthema und die
Spezialisierung getrennt. „Neue Version“ ändert nur noch die Hinweise.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-12 15:42:44 +02:00

244 lines
9.9 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Lauf-Orchestrierung: Stufen seriell, innerhalb einer Stufe parallele Agenten; jede Stufe resümierbar über
Dateimarker (Bericht existiert = Agent fertig)."""
import asyncio
import logging
from . import agenten, config, gliederung, lektorat, montage, prompts, state
log = logging.getLogger("buchwerk")
_laeufe: dict[str, asyncio.Task] = {}
class Abbruch(Exception):
pass
def laeuft(name: str) -> bool:
t = _laeufe.get(name)
return bool(t and not t.done())
def starten(name: str) -> None:
if laeuft(name):
return
st = state.zustand(name)
st.update(status="laeuft", grund="")
state.speichern(name)
_laeufe[name] = asyncio.create_task(_lauf(name))
async def pausieren(name: str) -> None:
t = _laeufe.get(name)
if t and not t.done():
t.cancel()
try:
await t
except (asyncio.CancelledError, Exception):
pass
async def _lauf(name: str) -> None:
st = state.zustand(name)
try:
for sid, _ in config.STUFEN:
s = state.stufe(name, sid)
if s["status"] == "fertig":
continue
s.update(status="laeuft", gestartet=s["gestartet"] or state.jetzt(), beendet="")
st["stufe"] = sid
state.speichern(name)
await STUFEN_FN[sid](name)
s.update(status="fertig", beendet=state.jetzt())
state.speichern(name)
st.update(status="fertig", grund="", stufe="")
except asyncio.CancelledError:
st.update(status="pausiert", grund="Pausiert")
_offen(name)
state.speichern(name)
raise
except Exception as e:
log.exception("Lauf %s", name)
st.update(status="fehler", grund=f"{type(e).__name__}: {e}"[:500])
for s in st["stufen"]:
if s["status"] == "laeuft":
s["status"] = "fehler"
finally:
state.speichern(name)
def _offen(name: str) -> None:
for s in state.zustand(name)["stufen"]:
if s["status"] == "laeuft":
s["status"] = "offen"
def _thema(name: str) -> tuple[str, str, str]:
"""Hauptthema (ohne Spezialisierung), Hinweise der Version, Spezialisierung des Themas."""
t = state.thema(name) or {}
return t.get("thema") or state.thema_von(name), state.hinweise_von(name), state.spezialisierung_von(name)
def _kapitel(name: str) -> list[dict]:
p = state.verzeichnis(name) / "buch" / "GLIEDERUNG.md"
g = gliederung.parsen(p.read_text())
if g["fehler"]:
raise Abbruch("Gliederung unlesbar: " + "; ".join(g["fehler"][:3]))
return g["kapitel"]
async def _parallel(name: str, sid: str, aufgaben: list) -> None:
"""aufgaben: [(agent_id, titel, prompt)], schon fertige sind vorher entfernt."""
sem = asyncio.Semaphore(config.PARALLEL)
s = state.stufe(name, sid)
fehler = []
async def eins(agent_id, titel, prompt):
async with sem:
r = await agenten.laufen(name, agent_id, sid, titel, prompt)
if r["status"] == "fertig":
s["fertig"] += 1
else:
fehler.append(f"{titel}: {r['bericht'][:120]}")
state.speichern(name)
await asyncio.gather(*(eins(*a) for a in aufgaben))
if fehler:
raise Abbruch(f"{len(fehler)} Agent(en) gescheitert – „Fortsetzen“ wiederholt nur die offenen: " + " | ".join(fehler[:3]))
# ---- Stufen -------------------------------------------------------------------
async def st_recherche(name: str) -> None:
thema, hinweise, spez = _thema(name)
s = state.stufe(name, "recherche")
s.update(gesamt=1, fertig=0)
v = state.verzeichnis(name)
if not state.bericht_lesen(name, "recherche"):
p = prompts.laden("recherche", thema=thema, hinweise_block=prompts.hinweise_block(hinweise),
spezialisierung_block=prompts.spezialisierung_block(spez),
stil_basis=(config.PROMPTS / "stil_basis.md").read_text())
r = await agenten.laufen(name, "recherche", "recherche", "Thema deuten und Quellen beschaffen", p)
if r["status"] != "fertig":
raise Abbruch("Recherche gescheitert: " + r["bericht"][:200])
fehlend = [f for f in ("deutung.json", "buch/STIL.md", "quellen/QUELLEN.md") if not (v / f).exists()]
if fehlend:
(v / "berichte" / "recherche.md").unlink(missing_ok=True)
raise Abbruch("Recherche unvollständig, es fehlt: " + ", ".join(fehlend))
d = state.deutung(name)
state.zustand(name)["deutung"] = d
if d and d.get("titel"):
t = state.thema(name)
t["titel"] = d["titel"]
state._schreiben(v / "thema.json", t)
s["fertig"] = 1
async def st_gliederung(name: str) -> None:
thema, hinweise, spez = _thema(name)
s = state.stufe(name, "gliederung")
s.update(gesamt=1, fertig=0)
p = state.verzeichnis(name) / "buch" / "GLIEDERUNG.md"
hinweis = ""
for versuch in (1, 2):
if p.exists() and not gliederung.parsen(p.read_text())["fehler"] and state.bericht_lesen(name, "gliederung"):
break
pr = prompts.laden("gliederung", thema=thema, hinweise_block=prompts.hinweise_block(hinweise),
spezialisierung_block=prompts.spezialisierung_block(spez)) + hinweis
r = await agenten.laufen(name, "gliederung", "gliederung", "Gliederung entwerfen", pr)
if r["status"] != "fertig":
raise Abbruch("Gliederung gescheitert: " + r["bericht"][:200])
g = gliederung.parsen(p.read_text()) if p.exists() else {"fehler": ["Datei fehlt"]}
if not g["fehler"]:
break
hinweis = ("\n\nFORMATFEHLER im vorigen Versuch – bitte buch/GLIEDERUNG.md exakt im vorgegebenen Format neu schreiben:\n- "
+ "\n- ".join(g["fehler"][:10]))
state.ereignis(name, "gliederung", "fehler", "Format ungültig, zweiter Versuch")
else:
raise Abbruch("Gliederung auch im zweiten Versuch nicht parsebar")
s["fertig"] = 1
async def st_schreiben(name: str) -> None:
thema, _, spez = _thema(name)
kap = _kapitel(name)
s = state.stufe(name, "schreiben")
offen = [k for k in kap if not state.bericht_lesen(name, f"kapitel-{k['n']:02d}")]
s.update(gesamt=len(kap), fertig=len(kap) - len(offen))
aufgaben = []
for k in offen:
p = prompts.laden("kapitel", thema=thema, n=str(k["n"]), kapitel_titel=k["titel"], datei=k["datei"],
spezialisierung_block=prompts.spezialisierung_block(spez))
if (state.verzeichnis(name) / "buch" / k["datei"]).exists():
p += "\n\nHinweis: `buch/" + k["datei"] + "` existiert bereits aus einem abgebrochenen Lauf – prüfe sie und vervollständige oder ersetze sie."
aufgaben.append((f"kapitel-{k['n']:02d}", f"Kapitel {k['n']} schreiben", p))
await _parallel(name, "schreiben", aufgaben)
async def st_pruefung(name: str) -> None:
thema, _, spez = _thema(name)
kap = _kapitel(name)
s = state.stufe(name, "pruefung")
offen = [k for k in kap if not state.bericht_lesen(name, f"pruefung-{k['n']:02d}")]
s.update(gesamt=len(kap), fertig=len(kap) - len(offen))
aufgaben = []
for k in offen:
autor = state.bericht_lesen(name, f"kapitel-{k['n']:02d}") or "(kein Autorenbericht vorhanden)"
p = prompts.laden("pruefung", thema=thema, n=str(k["n"]), kapitel_titel=k["titel"], datei=k["datei"], autor_bericht=autor,
spezialisierung_block=prompts.spezialisierung_block(spez))
aufgaben.append((f"pruefung-{k['n']:02d}", f"Kapitel {k['n']} prüfen", p))
await _parallel(name, "pruefung", aufgaben)
async def st_lektorat(name: str) -> None:
"""Geprüfte Kapitel sprachlich glätten; Marker: berichte/lektorat-NN.md."""
thema, _, _ = _thema(name)
kap = [k for k in _kapitel(name) if (state.verzeichnis(name) / "buch" / k["datei"]).exists()]
s = state.stufe(name, "lektorat")
offen = [k for k in kap if state.bericht_lesen(name, f"pruefung-{k['n']:02d}") and not state.bericht_lesen(name, f"lektorat-{k['n']:02d}")]
s.update(gesamt=len(kap), fertig=len(kap) - len(offen))
sem = asyncio.Semaphore(config.PARALLEL)
fehler = []
async def eins(k):
async with sem:
r = await lektorat.kapitel(name, k, thema)
if r["status"] == "fertig":
s["fertig"] += 1
else:
fehler.append(f"Kapitel {k['n']}")
state.speichern(name)
await asyncio.gather(*(eins(k) for k in offen))
if fehler:
raise Abbruch("Lektorat gescheitert bei " + ", ".join(fehler[:5]) + " – „Fortsetzen“ wiederholt nur die offenen")
async def st_konsistenz(name: str) -> None:
thema, _, _ = _thema(name)
kap = _kapitel(name)
gruppen = [kap[i:i + config.KONSISTENZ_GRUPPE] for i in range(0, len(kap), config.KONSISTENZ_GRUPPE)]
s = state.stufe(name, "konsistenz")
aufgaben = []
for gi, g in enumerate(gruppen, 1):
if state.bericht_lesen(name, f"konsistenz-{gi}"):
continue
dateien = ", ".join(f"`buch/{k['datei']}` (Kapitel {k['n']})" for k in g)
aufgaben.append((f"konsistenz-{gi}", f"Konsistenz Kapitel {g[0]['n']}–{g[-1]['n']}",
prompts.laden("konsistenz", thema=thema, dateien=dateien, gruppe=str(gi))))
if not state.bericht_lesen(name, "vorwort"):
aufgaben.append(("vorwort", "Vorwort schreiben", prompts.laden("vorwort", thema=thema)))
s.update(gesamt=len(gruppen) + 1, fertig=len(gruppen) + 1 - len(aufgaben))
await _parallel(name, "konsistenz", aufgaben)
async def st_montage(name: str) -> None:
s = state.stufe(name, "montage")
s.update(gesamt=1, fertig=0)
ok = await asyncio.to_thread(montage.bauen, name)
state.ereignis(name, "montage", "text" if ok else "fehler", "README, buch.md und PDF erzeugt" if ok else "README und buch.md erzeugt, PDF fehlgeschlagen")
s["fertig"] = 1
STUFEN_FN = {"recherche": st_recherche, "gliederung": st_gliederung, "schreiben": st_schreiben,
"pruefung": st_pruefung, "lektorat": st_lektorat, "konsistenz": st_konsistenz, "montage": st_montage}