init
This commit is contained in:
Team3
214
backend/pipeline.py
Normal file
214
backend/pipeline.py
Normal file
@@ -0,0 +1,214 @@
|
||||
"""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, 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]:
|
||||
t = state.thema(name) or {}
|
||||
return t.get("titel") or name, t.get("hinweise", "")
|
||||
|
||||
|
||||
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 = _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),
|
||||
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 = _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)) + 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, _ = _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"])
|
||||
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, _ = _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)
|
||||
aufgaben.append((f"pruefung-{k['n']:02d}", f"Kapitel {k['n']} prüfen", p))
|
||||
await _parallel(name, "pruefung", aufgaben)
|
||||
|
||||
|
||||
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"Querverweise 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, "konsistenz": st_konsistenz, "montage": st_montage}
|
||||
Reference in New Issue
Block a user