"""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]: 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_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"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, "lektorat": st_lektorat, "konsistenz": st_konsistenz, "montage": st_montage}