433 lines
18 KiB
Python
433 lines
18 KiB
Python
"""Guide generation as a consensus pipeline.
|
||
|
||
Outline: select the blocks (deterministic per format) → 3 proposals
|
||
(Grace) that order the blocks into chapters by NUMBER → a judge merges the
|
||
proposals into one coherent order.
|
||
Writing: one writer per block. Reading exam: Check→Fix (one round),
|
||
follow-up rounds check only replaced sections; remaining complaints then stand.
|
||
Step files are kept → an abort preserves progress, ▶ resumes at the open step.
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import math
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
import uuid
|
||
|
||
from agents import run_agent
|
||
from blocks import _convert_pdfs, source_folder
|
||
from config import (
|
||
DEFAULT_PROVIDER, FORMAT_PURPOSE, CONSENSUS_GRACE,
|
||
READABILITY_ACTIVE, TEMPLATES_DIR,
|
||
)
|
||
import readability
|
||
from database import (list_guides, update_guide, list_blocks, list_subblocks, set_guide_content,
|
||
get_guide_content, get_outline, guide_stage_counts, delete_guide_board)
|
||
from fsutil import atomic_write_json, atomic_write_text
|
||
from jsonio import read_json_file as _json_file, parse_json_text as _parse_json_text
|
||
from paths import blocks_path, guide_content_path, project_dir, subblocks_path
|
||
from pipeline import (
|
||
CANCELLED, FAILED, GenContext, _claude_error, _extra,
|
||
_fail, _gather_error, _gather_progress, _log, _prompt, _race,
|
||
_semaphore, _set_progress, _set_step, _timeout, clear_guide_cancelled,
|
||
is_guide_cancelled, run_single_slot,
|
||
)
|
||
from textkit import (
|
||
_unique_title, _load_blocks, _norm_title, _parse_fragment, _split_chunks,
|
||
_title, _resolve_title, _title_index,
|
||
)
|
||
|
||
log = logging.getLogger("creator.guide")
|
||
|
||
|
||
# Content/Content-Check/Reading-Exam run in packets of ~GUIDE_CHUNK blocks per agent.
|
||
# Only the writer (Writing) stays at 1 agent per block (variable lengths, no trimming, no
|
||
# length alignment between blocks).
|
||
|
||
# Check steps as a panel: CHECK_PANEL judges per chunk, section flagged on a majority.
|
||
# A single judge is bias/sampling prone; a small panel is more stable.
|
||
|
||
# Reading exam: only ONE round (Check + Fix). Follow-up rounds added little value
|
||
# (1 agent per block checks finely anyway) but cost extra agents.
|
||
|
||
|
||
# Valid level values: new (learning path) + old (difficulty) backward-compatible.
|
||
_LEVELS_OK = ("beginner", "advanced", "expert", "easy", "medium", "hard")
|
||
|
||
|
||
async def _load_subblocks(topic: str) -> dict[str, list[dict]]:
|
||
"""Subblocks per block — DB-first ({title, level, relevance}), fallback sidecar file.
|
||
Both missing → {} (guide takes everything). A missing/invalid level defaults to
|
||
'advanced' instead of dropping the row: re-run resume left 25 consensus subs
|
||
level-less, the writer silently lost them while the guide QA still counted them."""
|
||
out: dict[str, list[dict]] = {}
|
||
for r in await list_subblocks(topic):
|
||
if r["status"] == "consensus" and r["sub_title"]:
|
||
try:
|
||
facts = json.loads(r["facts"]) if r.get("facts") else {}
|
||
except (ValueError, TypeError):
|
||
facts = {}
|
||
level = r["level"] if r["level"] in _LEVELS_OK else "advanced"
|
||
out.setdefault(r["block"], []).append(
|
||
{"title": r["sub_title"], "level": level, "relevance": r["relevance"], "facts": facts})
|
||
if out:
|
||
return out
|
||
data = _json_file(subblocks_path(topic))
|
||
if not isinstance(data, dict):
|
||
return {}
|
||
for title, subs in data.items():
|
||
if not isinstance(subs, list):
|
||
continue
|
||
good = [s for s in subs if isinstance(s, dict) and str(s.get("title", "")).strip()
|
||
and s.get("level") in _LEVELS_OK]
|
||
if good:
|
||
out[title] = good
|
||
return out
|
||
|
||
|
||
def _level_label(s: dict) -> str:
|
||
"""View level of a subblock: peripheral → 'peripheral' (level 4), otherwise the level (1–3)."""
|
||
return "peripheral" if s.get("relevance") == "peripheral" else (s.get("level") or "beginner")
|
||
|
||
|
||
|
||
|
||
|
||
|
||
def guide_slot_files(content_path: Path) -> list[Path]:
|
||
"""All step files of a guide (for a fresh start)."""
|
||
return [p for p in content_path.parent.glob(f"{content_path.stem}.*") if p != content_path]
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
# Slot-file globs per step (index = GUIDE_STEPS). Stem-anchored, collision-free.
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
def _fallback_outline(entries: dict[int, str]) -> list[dict]:
|
||
"""Deterministic outline when the agents deliver none: one chapter with
|
||
all selected blocks in order. Guarantees full coverage."""
|
||
return [{"title": "Contents", "nums": list(entries)}]
|
||
|
||
|
||
def _with_remainder(plan: list[dict], entries: dict[int, str]) -> list[dict]:
|
||
"""Ensures that EVERY selected block is in the plan — missing ones land in
|
||
an "Additional" chapter (against agents/judges that drop blocks)."""
|
||
present = {num for ch in plan for num in ch.get("nums", [])}
|
||
missing = [num for num in entries if num not in present]
|
||
return [*plan, {"title": "Additional", "nums": missing}] if missing else plan
|
||
|
||
|
||
def _facts_grounding(subs_raw: dict[str, list[dict]]) -> str:
|
||
"""Verified sub-facts (extract-once from the blocks phase) as a grounding block for the
|
||
content agent. Empty if no facts are stored (legacy data → fallback to a source hint)."""
|
||
blocks = []
|
||
for title, subs in subs_raw.items():
|
||
lines = []
|
||
for s in subs:
|
||
fk = s.get("facts") if isinstance(s.get("facts"), dict) else None
|
||
if not fk:
|
||
continue
|
||
parts = []
|
||
if fk.get("key_points"):
|
||
parts.append("Core: " + " · ".join(fk["key_points"]))
|
||
for bf in fk.get("cited_facts", []):
|
||
parts.append(f"FACT[{bf.get('source', '?')}]: {bf.get('text', '')}")
|
||
if fk.get("prerequisites"):
|
||
parts.append("Prerequisite: " + fk["prerequisites"])
|
||
if fk.get("hurdles"):
|
||
parts.append("Hurdle: " + fk["hurdles"])
|
||
if fk.get("example_idea"):
|
||
parts.append("Example: " + fk["example_idea"])
|
||
if parts:
|
||
lines.append(f"- {s['title']}: " + " | ".join(parts))
|
||
if lines:
|
||
blocks.append(f"BLOCK: {title}\n" + "\n".join(lines))
|
||
if not blocks:
|
||
return ""
|
||
return ("VERIFIED FACTS per subblock — binding basis. Quote cited facts (FACT[Source]) "
|
||
"VERBATIM, invent nothing extra, do NOT re-research. Use examples as examples, "
|
||
"never as fact.\n\n" + "\n\n".join(blocks))
|
||
|
||
|
||
async def _outline_from_db(topic: str, sel_entries: dict[int, str]) -> list[dict] | None:
|
||
"""Read the outline from the blocks artifact (DB) and map it onto the selected blocks.
|
||
Title-based (robust against number drift): blocks outside the selection are ignored
|
||
(format filter), missing ones are added later by _with_remainder. None → no artifact (legacy)."""
|
||
raw = await get_outline(topic)
|
||
if not raw:
|
||
return None
|
||
try:
|
||
data = json.loads(raw)
|
||
except (ValueError, TypeError):
|
||
return None
|
||
chapters = data.get("chapters") if isinstance(data, dict) else None
|
||
if not isinstance(chapters, list):
|
||
return None
|
||
norm_to_num = {_norm_title(_title(t)): num for num, t in sel_entries.items()}
|
||
plan, seen = [], set()
|
||
for ch in chapters:
|
||
if not isinstance(ch, dict):
|
||
continue
|
||
nums = []
|
||
for bt in ch.get("blocks", []):
|
||
num = norm_to_num.get(_norm_title(str(bt)))
|
||
if num is not None and num not in seen:
|
||
seen.add(num)
|
||
nums.append(num)
|
||
if nums:
|
||
plan.append({"title": str(ch.get("title", "")).strip() or "Chapter", "nums": nums})
|
||
return plan or None
|
||
|
||
|
||
|
||
|
||
_LEVEL_RANK = {"beginner": 1, "advanced": 2, "expert": 3, "peripheral": 4,
|
||
"easy": 1, "medium": 2, "hard": 3} # old values backward-compatible
|
||
|
||
|
||
def _section_for_level(sec: dict, level: int) -> dict:
|
||
"""Reconstruct md/compact of a section from subblocks up to the level (anchor stays)."""
|
||
subs = sec.get("subs") or []
|
||
if not subs:
|
||
return sec # no sub tags (legacy) → unchanged, visible
|
||
visible = [s for s in subs if _LEVEL_RANK.get(s.get("level"), 1) <= level]
|
||
md = "\n\n".join(t for t in [sec.get("anchor", ""), *(s.get("md", "") for s in visible)] if t).strip()
|
||
compact = "\n".join(t for t in [sec.get("anker_compact", ""), *(s.get("compact", "") for s in visible)] if t).strip()
|
||
return {**sec, "md": md, "compact": compact, "leer": not visible}
|
||
|
||
|
||
def content_fuer_level(content: dict, level: int) -> dict:
|
||
"""Filter guide content to a view level (1=A · 2=F · 3=E · 4=V). Level 4 = full version.
|
||
Sections without visible subs are hidden, empty chapters drop out."""
|
||
if not isinstance(content, dict) or level >= 4:
|
||
return content
|
||
chapters = []
|
||
for ch in content.get("chapters", []):
|
||
secs = [s for s in (_section_for_level(x, level) for x in ch.get("sections", [])) if not s.get("leer")]
|
||
if secs:
|
||
chapters.append({**ch, "sections": secs})
|
||
return {**content, "chapters": chapters}
|
||
|
||
|
||
async def reconcile_guides() -> None:
|
||
"""Reconcile DB↔filesystem: status=done without content file → error.
|
||
|
||
Runs at server start (after init_db) — catches crashes between
|
||
file write and status update.
|
||
"""
|
||
for g in await list_guides():
|
||
if g["status"] == "done" and not guide_content_path(g["topic"], g["format"]).exists():
|
||
log.warning("[%s] Guide %s: done without content file — set to error", g["topic"], g["id"])
|
||
now = datetime.now(timezone.utc).isoformat()
|
||
await update_guide(g["id"], status="error", error_msg="Content missing — regenerate", updated_at=now)
|
||
|
||
|
||
async def generate_guide(guide_id: str, topic: str, format_name: str, instructions: str = "", provider: str = DEFAULT_PROVIDER, ab_step: int | None = None) -> None:
|
||
async with _semaphore:
|
||
now = datetime.now(timezone.utc).isoformat()
|
||
await update_guide(guide_id, status="generating", progress="Starting…", updated_at=now)
|
||
|
||
content_path = guide_content_path(topic, format_name)
|
||
content_path.parent.mkdir(parents=True, exist_ok=True)
|
||
project = source_folder(topic) # folder source (project/uni/link) → path, else None
|
||
|
||
try:
|
||
if is_guide_cancelled(guide_id):
|
||
return
|
||
|
||
if project:
|
||
await asyncio.to_thread(_convert_pdfs, project)
|
||
|
||
import guide_board # lazy — guide_board imports helpers from this module
|
||
# Re-run from stage: cards from `ab_step` onward back to that column.
|
||
# A FINISHED guide without ab_step → complete fresh start (board + slots wiped).
|
||
# Otherwise cards are leftovers of an abort/error → resume at their stored stage.
|
||
if ab_step is not None:
|
||
await guide_board.reset_from_stage(topic, format_name, ab_step)
|
||
elif content_path.exists():
|
||
counts = await guide_stage_counts(topic, format_name)
|
||
if not counts or set(counts) == {"done"}:
|
||
await delete_guide_board(topic, format_name)
|
||
for p_alt in guide_slot_files(content_path):
|
||
p_alt.unlink(missing_ok=True)
|
||
|
||
bs = await list_blocks(topic, status="consensus")
|
||
if bs:
|
||
alle = {i: (f"{b['title']} — {b['description']}" if b["description"] else b["title"])
|
||
for i, b in enumerate(bs, 1)}
|
||
else: # fallback: blocks.md (legacy topics)
|
||
bp = blocks_path(topic)
|
||
alle = _load_blocks(bp.read_text(encoding="utf-8")) if bp.exists() else {}
|
||
if not alle:
|
||
await _fail(guide_id, "No blocks found")
|
||
return
|
||
entries = _unique_title(alle)
|
||
chapters = await guide_board.run_guide_board(
|
||
guide_id, topic, format_name, entries, instructions, provider, content_path,
|
||
)
|
||
if is_guide_cancelled(guide_id):
|
||
return
|
||
if chapters is None:
|
||
await _fail(guide_id, "No finished sections (see board — cards with errors)")
|
||
return
|
||
content = {"topic": topic, "format": format_name, "chapters": chapters}
|
||
|
||
atomic_write_json(content_path, content, indent=1) # bridge (resume/fallback)
|
||
await set_guide_content(topic, format_name, json.dumps(content, ensure_ascii=False))
|
||
|
||
now = datetime.now(timezone.utc).isoformat()
|
||
await update_guide(guide_id, status="done", progress=None, step=None, updated_at=now)
|
||
|
||
except asyncio.TimeoutError:
|
||
await _fail(guide_id, "Timeout during generation")
|
||
except FileNotFoundError:
|
||
await _fail(guide_id, "Blocks missing")
|
||
except Exception as e:
|
||
log.exception("[%s] Guide generation failed (%s)", topic, guide_id)
|
||
await _fail(guide_id, str(e)[:2000])
|
||
finally:
|
||
clear_guide_cancelled(guide_id)
|
||
|
||
|
||
# --- On-demand: check / fix / rewrite one section (focus, interactive) ---
|
||
|
||
SECTION_CHECK_TIMEOUT = 300
|
||
|
||
|
||
def _section_spec() -> str:
|
||
return (TEMPLATES_DIR / "Format" / "Section.md").read_text(encoding="utf-8")
|
||
|
||
|
||
def _section_facts(topic: str) -> str:
|
||
project = source_folder(topic)
|
||
return _prompt("Guide-Facts-Projekt", project=project) if project else _prompt("Guide-Facts-Thema")
|
||
|
||
|
||
def _hint_block(hint: str) -> str:
|
||
hint = (hint or "").strip()
|
||
return f"NOTE FROM THE USER (pay special attention):\n{hint}" if hint else ""
|
||
|
||
|
||
async def _subs_text(topic: str, block: str) -> str:
|
||
"""Relevant subblocks of a block with level — as a checklist for the agents."""
|
||
subs_raw = await _load_subblocks(topic)
|
||
subs = [s for s in subs_raw.get(_title(block), []) if s.get("relevance") != "peripheral"]
|
||
if not subs:
|
||
return "(no subblocks recorded — cover 3–7 concise points)"
|
||
return "\n".join(f"- [{s['level']}] {s['title']}" for s in subs)
|
||
|
||
|
||
async def _load_guide_content(topic: str, format_name: str) -> dict | None:
|
||
js = await get_guide_content(topic, format_name)
|
||
if not js:
|
||
return None
|
||
try:
|
||
return json.loads(js)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _find_section(content: dict, block: str) -> dict | None:
|
||
for ch in content.get("chapters", []):
|
||
for s in ch.get("sections", []):
|
||
if s.get("title") == block:
|
||
return s
|
||
return None
|
||
|
||
|
||
async def block_pruefen(topic: str, format_name: str, block: str, spot: str, snippet: str, hint: str = "", provider: str = DEFAULT_PROVIDER) -> str | None:
|
||
"""Check one section (Markdown block) against the guide rules → corrected
|
||
block version as Markdown. None = error/section missing."""
|
||
content = await _load_guide_content(topic, format_name)
|
||
sec = _find_section(content, block) if content else None
|
||
if sec is None:
|
||
return None
|
||
whole = sec.get("compact", "") if str(spot).startswith("compact") else sec.get("md", "")
|
||
prompt = _prompt(
|
||
"Block-Pruefen", topic=topic, spec=_section_spec(), facts=_section_facts(topic),
|
||
subblocks=await _subs_text(topic, block), context=whole, snippet=snippet, hint=_hint_block(hint),
|
||
)
|
||
rc, stdout, _ = await run_agent(
|
||
f"block-pruefen-{uuid.uuid4()}", prompt, SECTION_CHECK_TIMEOUT,
|
||
provider=provider, role="judge", capabilities="none", lane="interactive",
|
||
)
|
||
new = stdout.strip() if rc == 0 else ""
|
||
return new or None
|
||
|
||
|
||
async def block_adopt(topic: str, format_name: str, block: str, spot: str, old: str, new: str) -> dict | None:
|
||
"""Replace one block (old→new) in the compact/detailed field + persist.
|
||
→ {compact, md, found}; None = section missing."""
|
||
content = await _load_guide_content(topic, format_name)
|
||
sec = _find_section(content, block) if content else None
|
||
if sec is None:
|
||
return None
|
||
is_compact = str(spot).startswith("compact")
|
||
field = "compact" if is_compact else "md"
|
||
current = sec.get(field, "") or ""
|
||
found = old in current
|
||
if found:
|
||
sec[field] = current.replace(old, new, 1)
|
||
# Also replace in anchor + subs (sources of the filtered E/M/S view), otherwise the
|
||
# leveled view keeps showing the old block.
|
||
anchor_field = "anker_compact" if is_compact else "anchor"
|
||
if old in (sec.get(anchor_field) or ""):
|
||
sec[anchor_field] = sec[anchor_field].replace(old, new, 1)
|
||
for sub in sec.get("subs", []):
|
||
if old in (sub.get(field, "") or ""):
|
||
sub[field] = sub[field].replace(old, new, 1)
|
||
break
|
||
js = json.dumps(content, ensure_ascii=False)
|
||
await set_guide_content(topic, format_name, js)
|
||
atomic_write_json(guide_content_path(topic, format_name), content, indent=1)
|
||
return {"compact": sec.get("compact", ""), "md": sec.get("md", ""), "found": found}
|
||
|
||
|
||
# --- Tutor chat (moved from the removed elements module) ---
|
||
|
||
def _build_guide_chat_prompt(topic: str, format_name: str, section: str, outline: str, messages: list[dict]) -> str:
|
||
transcript = "\n".join(
|
||
f"{'User' if m.get('role') == 'user' else 'Assistant'}: {m.get('content', '')}"
|
||
for m in messages
|
||
)
|
||
return _prompt(
|
||
"Chat",
|
||
topic=topic, format_name=format_name,
|
||
outline_block=outline.strip() or "(none)",
|
||
section_block=section.strip() or "(no section detected)",
|
||
transcript=transcript,
|
||
)
|
||
|
||
|
||
async def chat_with_guide(topic: str, format_name: str, section: str, outline: str, messages: list[dict], provider: str = DEFAULT_PROVIDER) -> str:
|
||
try:
|
||
prompt = _build_guide_chat_prompt(topic, format_name, section, outline, messages)
|
||
returncode, stdout, stderr = await run_agent(
|
||
"chat-" + str(uuid.uuid4()), prompt, 240, provider=provider, role="fast", capabilities="none", lane="interactive"
|
||
)
|
||
if returncode != 0:
|
||
return "Sorry, that didn't work. Please try again."
|
||
reply = stdout.strip()
|
||
return reply or "Sorry, I didn't get a response."
|
||
except Exception:
|
||
log.warning("[%s] Guide chat failed", topic, exc_info=True)
|
||
return "Sorry, that didn't work. Please try again."
|