diff --git a/.env.example b/.env.example index 8147a44..60d39c0 100644 --- a/.env.example +++ b/.env.example @@ -5,3 +5,9 @@ CLAUDE_CODE_OAUTH_TOKEN= # MiniMax-Provider: API-Key aus der MiniMax-Console (Coding-Plan). MINIMAX_API_KEY= + +# Agent-Parallelität (optional). Zwei verschachtelte Limits, Default je 10 = bisheriges Verhalten. +# Global gilt über ALLE Themen, der Thema-Wert je Thema. Lokal das globale Limit hochsetzen, +# um mehrere Themen parallel mit je 10 Agenten zu fahren (z.B. global 50, Thema 10). +# MAX_CONCURRENT_AGENTS=10 +# MAX_CONCURRENT_AGENTS_PER_TOPIC=10 diff --git a/backend/agents.py b/backend/agents.py index ee3bec3..d19eda2 100644 --- a/backend/agents.py +++ b/backend/agents.py @@ -13,9 +13,11 @@ import signal import tempfile import time import urllib.request +from contextlib import asynccontextmanager from pathlib import Path -from config import PROVIDERS, DEFAULT_PROVIDER, MAX_CONCURRENT_AGENTS, MAX_CONCURRENT_INTERACTIVE +from config import (PROVIDERS, DEFAULT_PROVIDER, MAX_CONCURRENT_AGENTS, + MAX_CONCURRENT_AGENTS_PER_TOPIC, MAX_CONCURRENT_INTERACTIVE) log = logging.getLogger("creator.agents") @@ -43,6 +45,24 @@ def _scope_cancelled(agent_key: str) -> bool: # does not count against the agent timeout. _batch_sem = asyncio.Semaphore(MAX_CONCURRENT_AGENTS) _interactive_sem = asyncio.Semaphore(MAX_CONCURRENT_INTERACTIVE) +# Per-topic caps (lazily created): each topic gets its own batch semaphore of size +# MAX_CONCURRENT_AGENTS_PER_TOPIC, nested INSIDE the global _batch_sem. +_topic_sems: dict[str, asyncio.Semaphore] = {} + + +@asynccontextmanager +async def _batch_gate(scope: str | None): + """Acquire a batch slot: per-topic semaphore FIRST, then the global one. The order matters — + a waiter holds only its (per-topic) slot while queueing for the global cap, so a saturated topic + never blocks other topics on the global semaphore. scope=None → global cap only.""" + topic_sem = _topic_sems.setdefault(scope, asyncio.Semaphore(MAX_CONCURRENT_AGENTS_PER_TOPIC)) if scope else None + if topic_sem is None: + async with _batch_sem: + yield + else: + async with topic_sem: + async with _batch_sem: + yield # Serialize OpenCode starts: processes starting simultaneously collide on the # internal session DB ("database is locked", exit after <1s). The short @@ -116,6 +136,7 @@ async def run_agent( role: str = "fast", capabilities: str = "none", lane: str = "batch", + scope: str | None = None, ) -> tuple[int, str, str]: if _scope_cancelled(agent_key): # before queueing: don't even enter the queue return 1, "", "cancelled" @@ -123,8 +144,8 @@ async def run_agent( return 1, "", f"Unknown provider: {provider}" if shutil.which(PROVIDERS[provider]["cli"]) is None: return 1, "", f"CLI '{PROVIDERS[provider]['cli']}' not installed (provider: {provider})" - sem = _interactive_sem if lane == "interactive" else _batch_sem - async with sem: + gate = _interactive_sem if lane == "interactive" else _batch_gate(scope) + async with gate: if _scope_cancelled(agent_key): # after the acquire: cancelled in the queue → no spawn return 1, "", "cancelled" if PROVIDERS[provider]["cli"] == "opencode": diff --git a/backend/blocks.py b/backend/blocks.py index 0418831..ec83337 100644 --- a/backend/blocks.py +++ b/backend/blocks.py @@ -23,7 +23,7 @@ from pathlib import Path import database as db import embedding from agents import kill_process, cancel_scope, clear_scope, run_agent -from config import CONSENSUS_GRACE, RESEARCH_GRACE, CONSENSUS_MAX_ROUNDS, DEFAULT_PROVIDER, CRAWL_KEEP_PATTERNS, CRAWL_NOISE_PATTERNS, CRAWL_MIN_CHARS, QUELLE_RELEVANZ_CHUNK, QUELLE_RELEVANZ_SNIPPET, EMBEDDING_AKTIV, EMBEDDING_SUB_DUP +from config import CONSENSUS_GRACE, RESEARCH_GRACE, CONSENSUS_MAX_ROUNDS, DEFAULT_PROVIDER, CRAWL_KEEP_PATTERNS, CRAWL_NOISE_PATTERNS, CRAWL_MIN_CHARS, QUELLE_RELEVANZ_CHUNK, QUELLE_RELEVANZ_SNIPPET, EMBEDDING_AKTIV, EMBEDDING_SUB_DUP, EMBEDDING_SUB_SAME from fsutil import atomic_write_text, atomic_write_json from jsonio import read_json_file as _json_file from paths import arbeit_dir, blocks_path, question_pattern_path, project_dir, subblocks_path, source_path, source_crawl_dir, safe_folder @@ -198,7 +198,7 @@ def _blocks_steps(topic: str) -> tuple: all packages run in parallel; the step remains until the last package is done. """ q = load_source(topic) - base = ("Research", "Consolidation", "Clarification", "Dedup", "Blocks-Filter") + base = ("Research", "Consolidation", "Clarification", "Blocks-Filter") rest = ( "Subblocks find", "Subblocks select", "Subblocks clarify", "Facts find", "Facts check", "Facts fix", @@ -228,7 +228,7 @@ def _report_p(set_p, topic: str, step: str): # Special steps (Source laden, Supplement) belong to the "Inventory" phase. PHASEN = ( ("Source", ("Source prep",)), - ("Inventory", ("Research", "Consolidation", "Clarification", "Dedup", "Blocks-Filter", "Supplement")), + ("Inventory", ("Research", "Consolidation", "Clarification", "Blocks-Filter", "Supplement")), ("Subblocks", ("Subblocks find", "Subblocks select", "Subblocks clarify")), ("Facts", ("Facts find", "Facts check", "Facts fix")), ("Levels", ("Levels find", "Levels select", "Levels clarify")), @@ -317,7 +317,7 @@ async def _resume_step(topic: str) -> int: files = _blocks_files(topic) steps_all = _blocks_steps(topic) if not files["final"].exists(): - for step in ("Source prep", "Research", "Consolidation", "Clarification", "Dedup", "Blocks-Filter"): + for step in ("Source prep", "Research", "Consolidation", "Clarification", "Blocks-Filter"): if step in steps_all and await db.get_step_status(topic, step) != "done": return _step_idx(topic, step) return _step_idx(topic, "Blocks-Filter") # statuses done but artefact gone → rewrite @@ -451,15 +451,19 @@ def _reset_from_phase(topic: str, label: str) -> None: files["final"].unlink(missing_ok=True) -async def _reset_from_step(topic: str, step_idx: int) -> None: - """Fine reset FROM a sub-step (0-based index in _blocks_steps). Resets - pipeline_state + artefacts + DB from here on; earlier steps stay. Inventory sub-steps - reconstruct the DB status from the artefacts (dedup/filter safe; clarification robustly falls back - to consolidation, because the clarification renames → a title mismatch would be fragile).""" +async def _reset_from_step(topic: str, step_idx: int, to_idx: int | None = None) -> None: + """Fine reset FROM a sub-step (0-based index in _blocks_steps). With `to_idx` set, ONLY the + span [step_idx, to_idx] is reset — later steps and the blocks.md aggregate stay (isolated + single-step / bounded-range regenerate). Without it the reset cascades to the end (full re-run). + Earlier steps always stay. Inventory sub-steps reconstruct the DB status from the artefacts + (filter safe; clarification robustly falls back to consolidation, because the clarification + renames → a title mismatch would be fragile).""" fine_steps = list(_blocks_steps(topic)) if not (0 <= step_idx < len(fine_steps)): return - affected = set(fine_steps[step_idx:]) + last = len(fine_steps) - 1 if to_idx is None else max(step_idx, min(to_idx, len(fine_steps) - 1)) + span = fine_steps[step_idx:last + 1] + affected = set(span) files = _blocks_files(topic) work_dir = files["arbeit"] @@ -468,7 +472,7 @@ async def _reset_from_step(topic: str, step_idx: int) -> None: for p in work_dir.glob(pat): p.unlink(missing_ok=True) - await db.delete_pipeline_state(topic, list(fine_steps[step_idx:])) + await db.delete_pipeline_state(topic, span) # Later artefacts/DB cumulatively from the affected step (back to front). if {"Examples", "Flashcards"} & affected: files["artefakte"].unlink(missing_ok=True); gd("artifact-*"); await db.delete_sub_artefakte(topic) @@ -501,21 +505,12 @@ async def _reset_from_step(topic: str, step_idx: int) -> None: if any(s.startswith("Subblock") for s in affected): files["sub_roh"].unlink(missing_ok=True); gd("subblock-*"); await db.delete_subblocks(topic) # --- Inventory (DB status cascades) --- - if "Blocks-Filter" in affected and not ({"Clarification", "Consolidation", "Research", "Dedup"} & affected): + if "Blocks-Filter" in affected and not ({"Clarification", "Consolidation", "Research"} & affected): # Only filter rebuilt: degraded blocks back to consensus. d = _json_file(work_dir / "inventar-filter.json") for f in (d.get("fragments", []) if isinstance(d, dict) else []): await db.set_block_status(topic, _norm_title(f.get("fragment", "")), "consensus") gd("inventar-filter*") - if "Dedup" in affected and not ({"Clarification", "Consolidation", "Research"} & affected): - # Dedup (+filter) rebuilt: all blocks discarded in dedup/filter back to consensus. - for kind in ("dedup-runde-1.json", "inventar-filter.json"): - d = _json_file(work_dir / kind) - title = ([t for g in d.get("groups", []) for t in g] if isinstance(d, dict) and "groups" in d - else [f.get("fragment", "") for f in d.get("fragments", [])] if isinstance(d, dict) else []) - for t in title: - await db.set_block_status(topic, _norm_title(t), "consensus") - gd("dedup-*"); gd("inventar-filter*") if {"Clarification", "Consolidation"} & affected and not ({"Research"} & affected): # Clarification/consolidation rebuilt: clear inventory DB (research readers stay). Clarification rollback # would be fragile due to renaming → cleanly rebuild from consolidation. @@ -527,8 +522,9 @@ async def _reset_from_step(topic: str, step_idx: int) -> None: await db.delete_blocks(topic) # blocks.md is the inventory aggregate — stale once any inventory sub-step is reset. Delete it so the # status/resume see the inventory as open from the reset step (the pipeline rewrites it; the DB step - # statuses of the kept earlier steps let those skip). - if {"Research", "Consolidation", "Clarification", "Dedup", "Blocks-Filter"} & affected: + # statuses of the kept earlier steps let those skip). On a BOUNDED reset (to_idx set) the later steps + # are kept on purpose → keep their aggregate too. + if to_idx is None and {"Research", "Consolidation", "Clarification", "Blocks-Filter"} & affected: files["final"].unlink(missing_ok=True) @@ -867,7 +863,7 @@ async def _subblocks_block(ctx: GenContext, set_p, files: dict, entries: dict, i chunk_idx = _title_index({num: title_by_num[num] for num in chunk}) paths = [work_dir / f"subblock-final-c{c}-j{j}.md" for j in range(1, SUBBLOCK_PANEL + 1)] - pending = [(j, p) for j, p in enumerate(paths, 1) if _parse_subblocks(_read(p)) is None] + pending = [(j, p) for j, p in enumerate(paths, 1) if not _parse_subblocks(_read(p))] # {} (empty/missing) also pending — _parse_subblocks never returns None for _, p in pending: p.unlink(missing_ok=True) if pending: @@ -890,12 +886,13 @@ async def _subblocks_block(ctx: GenContext, set_p, files: dict, entries: dict, i atomic_write_text(fp, text) return - # code majority per block/sub-norm: keep if a majority of judges list it (tie → keep). + # Semantic majority per block: cluster the judges' subblocks by meaning, keep a cluster a + # majority of judges contributed to. Exact-string voting discarded paraphrased core facts + # (each judge wording → 1 vote) while verbatim side-notes won. block_texts_out = [] for num in chunk: - votes: dict[str, int] = {} - form: dict[str, str] = {} - for d in outs: + cand: list[tuple[int, str]] = [] # (judge index, subblock) — exact-dedup per judge + for ji, d in enumerate(outs): seen = set() for marker, subs in d.items(): if _resolve_title(chunk_idx, marker) != num: @@ -905,9 +902,10 @@ async def _subblocks_block(ctx: GenContext, set_p, files: dict, entries: dict, i if not sn or sn in seen: continue seen.add(sn) - form.setdefault(sn, sub) - votes[sn] = votes.get(sn, 0) + 1 - kept = [form[sn] for sn in form if votes[sn] * 2 >= len(outs)] + cand.append((ji, sub)) + if not cand: + continue + kept = await _cluster_vote(cand, len(outs)) if kept: block_texts_out.append(f"\n" + "\n".join(f"- {s}" for s in kept)) atomic_write_text(fp, "\n\n".join(block_texts_out)) @@ -947,6 +945,42 @@ async def _subblocks_block(ctx: GenContext, set_p, files: dict, entries: dict, i return raw +async def _cluster_vote(cand: list[tuple[int, str]], n_judges: int) -> list[str]: + """Semantic majority over the judges' subblocks. Cluster by cosine ≥ EMBEDDING_SUB_SAME (same + point), keep a cluster a majority of distinct judges contributed to, pick the longest (most + informative) phrasing. Replaces exact-string voting, which split paraphrases of the same fact. + Model missing → norm-union fallback (keep everything ≥1 judge, exact-dedup — never lose content).""" + texts = [s for _, s in cand] + sims = (await asyncio.to_thread(embedding.embed_sims, texts) + if EMBEDDING_AKTIV and await asyncio.to_thread(embedding.available) else None) + if sims is None: + out, seen = [], set() + for _, s in cand: + sn = _norm_title(s) + if sn and sn not in seen: + seen.add(sn) + out.append(s) + return out + parent = list(range(len(texts))) + def find(x: int) -> int: + while parent[x] != x: + parent[x] = parent[parent[x]] + x = parent[x] + return x + for i in range(len(texts)): + for k in range(i + 1, len(texts)): + if float(sims[i][k]) >= EMBEDDING_SUB_SAME: + parent[find(i)] = find(k) + clusters: dict[int, list[int]] = {} + for i in range(len(texts)): + clusters.setdefault(find(i), []).append(i) + kept = [] + for idxs in clusters.values(): + if len({cand[i][0] for i in idxs}) * 2 >= n_judges: # majority of judges + kept.append(max((texts[i] for i in idxs), key=len)) # longest phrasing + return kept + + async def _dedup_subblocks(topic: str, raw: dict[str, list[str]]) -> None: """Deterministic near-duplicate filter per block: subblocks with cosine ≥ EMBEDDING_SUB_DUP are the same statement (reliable in the narrow block context — no LLM @@ -1324,7 +1358,7 @@ async def _facts_block(ctx, set_p, files: dict, raw: dict, q: dict, folder, inst await asyncio.gather(*[ run_agent(f"blocks-{topic}-facts-check-c{ci}-j{j}", _prompt("Facts-Check", topic=topic, source=source, facts=facts_text, out_path=chk_path(ci, j), extra=_extra(instructions)), - _timeout("content_check", len(per)), provider=provider, role="judge", capabilities=caps) + _timeout("content_check", len(per)), provider=provider, role="judge", capabilities=caps, scope=topic) for j in pending], return_exceptions=True) outs = [s for j in (1, 2, 3)[:FACTS_CHECK_PANEL] if (s := _facts_check_schema(_json_file(chk_path(ci, j)))) is not None] bvotes: dict[str, int] = {} @@ -1920,6 +1954,37 @@ async def _prepare_source(ctx: GenContext, set_p, files: dict, q: dict, folder, return True +async def _ingest_research(topic: str, reader_id: str, text: str) -> None: + """Parse one reader's research file (lines `title — desc — source`) into block candidates and + upsert them with the reader id (one reader = one vote per concept). Shared by the research step + and the candidate re-ingest below.""" + seen_set = set() + for record in _parse_selection(text).values(): + title = _title(record) + norm = _norm_title(title) + if not norm or norm in seen_set: + continue + seen_set.add(norm) # one reader = one vote per concept + split_parts = [t.strip() for t in record.split(" — ")] + desc = split_parts[1] if len(split_parts) >= 2 else "" + source = [split_parts[2]] if len(split_parts) >= 3 and split_parts[2] else [] + await db.upsert_block(topic, norm, title, desc, source, reader=reader_id) + + +async def _reingest_research_files(topic: str, work_dir: Path) -> None: + """Rebuild the research candidates in the blocks DB from the saved research-*.md reader files + (no agents, no web calls). Consolidation CONSUMES the candidates (overwrites them with + consensus/rest), so a consolidation re-run / reset needs them restored. Reader id = file suffix + (pure number N → "tN" for the web readers; "a…"/"b…" section/batch readers stay as-is).""" + for p in sorted(work_dir.glob("research-*.md")): + suffix = p.stem[len("research-"):] + if not suffix: + continue + rid = f"t{suffix}" if suffix.isdigit() else suffix + if text := _file_payload(p): + await _ingest_research(topic, rid, text) + + async def _research_batch(ctx: GenContext, set_p, files: dict, q: dict, folder, instructions: str) -> bool: """Fills DB table `blocks` with candidates (+ mention counter). FIXED file batches: each crawl page is assigned to exactly one batch and read by RESEARCH_READERS agents @@ -1933,17 +1998,7 @@ async def _research_batch(ctx: GenContext, set_p, files: dict, q: dict, folder, await db.set_step_status(topic, "Research", "running") async def _ingest(reader_id: str, text: str) -> None: - seen_set = set() - for record in _parse_selection(text).values(): - title = _title(record) - norm = _norm_title(title) - if not norm or norm in seen_set: - continue - seen_set.add(norm) # one reader = one vote per concept - split_parts = [t.strip() for t in record.split(" — ")] - desc = split_parts[1] if len(split_parts) >= 2 else "" - source = [split_parts[2]] if len(split_parts) >= 3 and split_parts[2] else [] - await db.upsert_block(topic, norm, title, desc, source, reader=reader_id) + await _ingest_research(topic, reader_id, text) pages = await db.list_content(topic) # pages marked as content by the triage if not pages and folder: @@ -2068,31 +2123,6 @@ async def _research_batch(ctx: GenContext, set_p, files: dict, q: dict, folder, return True -def _grp_schema(data, ids: set[int]): - """{"groups": [[1,3],[2], …]} → partition of `ids` as a list of index groups. - Tolerant: ignores foreign/duplicate numbers; forgotten candidates are added standalone - (singleton group). None only on structurally broken JSON.""" - if not isinstance(data, dict) or not isinstance(data.get("groups"), list): - return None - groups, seen_set = [], set() - for g in data["groups"]: - if not isinstance(g, list): - return None - grp = [] - for x in g: - try: - num = int(x) - except (ValueError, TypeError): - continue - if num in ids and num not in seen_set: - seen_set.add(num) - grp.append(num) - if grp: - groups.append(grp) - groups += [[r] for r in sorted(ids - seen_set)] # forgotten candidates stay standalone - return groups or None - - _ASPECT_MARKER = ("∈ np", "∈np", " in np", "np-schwer", "np-vollständig", "verifizierer", "zertifikat", "ndtm", "nicht-determ", "lower bound", "untere schranke", "bzgl", "als sprache") @@ -2146,47 +2176,58 @@ def _canonical(candidates: list[dict], idxs: list[int], seen_norm: set[str]) -> return {"title": title, "description": candidates[k]["description"]} -async def _group_blocks(ctx: GenContext, set_p, work_dir: Path, candidates: list[dict], - blocks: list[list[int]], prefix: str = "consolidation", - step: str = "Consolidation") -> list[list[int]]: - """Per similarity block, a judge groups the titles into the real blocks (merge - paraphrases, split over-merges). Singletons directly. Error/timeout → conservatively each - candidate alone (avoids false over-merging). → final groups (global indices). - `praefix`/`step` separate consolidation and dedup (artefacts, race key, progress).""" +async def _pairwise_groups(ctx: GenContext, set_p, work_dir: Path, candidates: list[dict], + blocks: list[list[int]], sims) -> list[list[int]] | None: + """Verify candidate PAIRS individually (ja/nein) inside each similarity block, then form + COMPLETE-LINK cliques — same entity-resolution mechanism as the dedup pass: no chaining + (A=B + B=C without A=C does NOT merge), no aspect over-merging like the old N→groups judge. + Only block-internal pairs with cosine ≥ DEDUP_PAIR_FLOOR are checked; members without a + confirmed edge stay singletons. → final groups (global candidate indices) · None on cancel.""" topic, is_cancelled = ctx.topic, ctx.is_cancelled - multi = [(bi, b) for bi, b in enumerate(blocks) if len(b) > 1] - outcome: list[list[int]] = [list(b) for b in blocks if len(b) == 1] # singletons directly + n = len(candidates) + pairs: list[tuple[int, int]] = [] # block-internal candidate pairs above the pair floor + for b in blocks: + for x in range(len(b)): + for y in range(x + 1, len(b)): + i, j = b[x], b[y] + if float(sims[i][j]) >= DEDUP_PAIR_FLOOR: + pairs.append((i, j)) + if not pairs: + return [[i] for i in range(n)] + packages = [pairs[k:k + DEDUP_PAIRS_CHUNK] for k in range(0, len(pairs), DEDUP_PAIRS_CHUNK)] - def _line(k: int, g: int) -> str: - b = candidates[g] - return f"{k}. {b['title']}" + (f" — {b['description']}" if b["description"] else "") + def pair_path(pi): return work_dir / f"consolidation-paar-c{pi}.json" - async def _grp(bi: int, block: list[int]) -> None: - ids = set(range(1, len(block) + 1)) - p = work_dir / f"{prefix}-block-c{bi}.json" - part = _grp_schema(_json_file(p), ids) - if part is None: # resume: don't recompute a valid file - p.unlink(missing_ok=True) - if is_cancelled(): - return - lines = [_line(k, block[k - 1]) for k in range(1, len(block) + 1)] - status, part = await run_single_slot( - ctx, f"Block grouping {bi}", - key=f"blocks-{topic}-{prefix}-block-c{bi}", - prompt=_prompt("Blocks-Block-Grouping", topic=topic, entries="\n".join(lines), out_path=p), - role="judge", capabilities="files", - payload=(lambda result, p=p, ids=ids: _grp_schema(_json_file(p), ids)), - timeout=_timeout("research_mapping", len(block)), - ) - part = part if status == OK else None - if part is None: # judge failed → individually (no over-merge) - outcome.extend([idx] for idx in block) - else: # local numbers → global candidate indices - outcome.extend([block[k - 1] for k in g] for g in part) + async def _filt(pi, paare): + fp = pair_path(pi) + if _pairs_schema(_json_file(fp)): + return # resume + lines = "\n\n".join( + f"{j + 1}.\nA: {candidates[a]['title']} — {candidates[a]['description']}" + f"\nB: {candidates[b]['title']} — {candidates[b]['description']}" + for j, (a, b) in enumerate(paare)) + await run_single_slot( + ctx, f"Consolidation pairs {pi}", + key=f"blocks-{topic}-consolidation-paar-c{pi}", + prompt=_prompt("Blocks-Paar-Filter", topic=topic, pairs=lines, out_path=fp), + role="judge", capabilities="files", + payload=lambda result, p=fp: _pairs_schema(_json_file(p)), + timeout=_timeout("selection_mapping", len(paare)), + ) - await _gather_progress([_grp(bi, b) for bi, b in multi], - len(multi), _report_p(set_p, topic, step)) - return outcome + await _gather_progress([_filt(pi, p) for pi, p in enumerate(packages)], + len(packages), _report_p(set_p, topic, "Consolidation")) + if is_cancelled(): + return None + edge_list: list[tuple[int, int]] = [] + for pi, paare in enumerate(packages): + verdict = _pairs_schema(_json_file(pair_path(pi))) or {} + for j, (a, b) in enumerate(paare): + if verdict.get(j + 1): + edge_list.append((a, b)) + cliques = _cliques(n, edge_list) + covered = {i for g in cliques for i in g} + return cliques + [[i] for i in range(n) if i not in covered] async def _consolidate_embedding(ctx: GenContext, set_p, files: dict, candidates: list[dict]) -> bool: @@ -2198,11 +2239,12 @@ async def _consolidate_embedding(ctx: GenContext, set_p, files: dict, candidates sims = await asyncio.to_thread(embedding.embed_sims, texts) if sims is None: # model not available after all → fallback return await _consolidate_llm(ctx, set_p, files, candidates) - # Level 1: coarse similarity blocks (capped, no giant component). + # Level 1: coarse similarity blocks (capped, no giant component) — pure blocking for recall. blocks = await asyncio.to_thread(embedding.capped_blocks, sims, None, None) - # Level 2: one judge groups EACH multi-block into the real blocks. - groups = await _group_blocks(ctx, set_p, work_dir, candidates, blocks) - if is_cancelled(): + # Level 2: verify candidate PAIRS individually + complete-link cliques (no chaining, no aspect + # over-merging) instead of an N→groups judge that fused whole topics into one block. + groups = await _pairwise_groups(ctx, set_p, work_dir, candidates, blocks, sims) + if groups is None or is_cancelled(): return False def _min_cos(idxs): # internal coherence as a check (chains would be ~0.3) @@ -2230,7 +2272,7 @@ async def _consolidate_embedding(ctx: GenContext, set_p, files: dict, candidates "mitglieder": [candidates[k]["title"] for k in idxs]}) atomic_write_json(work_dir / "consolidation-cluster.json", debug, indent=1) multi_blocks = sum(1 for b in blocks if len(b) > 1) - _log(topic, f"Consolidation (embedding): {len(blocks)} blocks ({multi_blocks} grouped via LLM) " + _log(topic, f"Consolidation (pairwise): {len(blocks)} blocks ({multi_blocks} multi) " f"→ {len(groups)} clusters from {len(candidates)} candidates " f"→ {len(consensus)} consensus / {len(rest)} rest") @@ -2251,6 +2293,11 @@ async def _consolidate(ctx: GenContext, set_p, files: dict) -> bool: return True set_p("Consolidating research…", step=_step_idx(topic, "Consolidation")) candidates = await db.list_blocks(topic) + if not candidates: + # Candidates were consumed by an earlier consolidation (overwritten with consensus/rest) or + # wiped by a reset → rebuild them from the saved research files so this step can re-run. + await _reingest_research_files(topic, files["arbeit"]) + candidates = await db.list_blocks(topic) if not candidates: _blocks_errors[topic] = "Consolidation: no candidates" return False @@ -2434,7 +2481,7 @@ async def _clarify_inventory(ctx: GenContext, set_p, files: dict) -> bool: if new: renames.setdefault(_norm_title(str(old)), {}).setdefault(new, 0) renames[_norm_title(str(old))][new] += 1 - seen_norm = {b["title_norm"] for b in await db.list_blocks(topic, status="consensus")} + seen_norm = {b["title_norm"] for b in await db.list_blocks(topic)} # all stati: UNIQUE(topic,title_norm) spans every status, not just consensus for b in check_rows: accept = votes.get(b["title_norm"], 0) * 2 >= len(outs) if not accept: @@ -2494,83 +2541,6 @@ def _cliques(n: int, edge_list: list[tuple[int, int]]) -> list[list[int]]: return groups -async def _dedup_inventory(ctx: GenContext, set_p, files: dict) -> bool: - """Final dedup pass over the finished consensus list: pairwise verification (entity - resolution). Embedding yields candidate PAIRS (cosine ≥ DEDUP_PAAR_FLOOR), a judge - confirms EACH pair individually (ja = the same duplicate). ONLY confirmed pairs become - merge edges (union-find) — no chaining, no aspect over-merging like the block mixer. - Per group ONE representative (main concept) stays, the rest is discarded.""" - topic, is_cancelled = ctx.topic, ctx.is_cancelled - if await db.get_step_status(topic, "Dedup") == "done": - return True - if not (EMBEDDING_AKTIV and await asyncio.to_thread(embedding.available)): - await db.set_step_status(topic, "Dedup", "done") # without a model: silently skip - return True - set_p("Dedup…", step=_step_idx(topic, "Dedup")) - work_dir = files["arbeit"] - consensus = await db.list_blocks(topic, status="consensus") - if len(consensus) >= 2: - import numpy as np - texts = [f"{b['title']} — {b['description']}" if b["description"] else b["title"] for b in consensus] - sims = await asyncio.to_thread(embedding.embed_sims, texts) - if sims is not None: - n = len(consensus) - iu = np.triu_indices(n, k=1) - cands = [(int(iu[0][m]), int(iu[1][m])) for m in np.where(sims[iu] >= DEDUP_PAIR_FLOOR)[0]] - _log(topic, f"Dedup: {len(cands)} candidate pairs (cosine ≥ {DEDUP_PAIR_FLOOR}) → pairwise filter") - packages = [cands[i:i + DEDUP_PAIRS_CHUNK] for i in range(0, len(cands), DEDUP_PAIRS_CHUNK)] - - def pair_path(pi): return work_dir / f"dedup-paar-c{pi}.json" - - async def _filt(pi, paare): - fp = pair_path(pi) - if _pairs_schema(_json_file(fp)): - return # resume - lines = "\n\n".join( - f"{j + 1}.\nA: {consensus[a]['title']} — {consensus[a]['description']}" - f"\nB: {consensus[b]['title']} — {consensus[b]['description']}" - for j, (a, b) in enumerate(paare)) - await run_single_slot( - ctx, f"Dedup pairs {pi}", - key=f"blocks-{topic}-dedup-paar-c{pi}", - prompt=_prompt("Blocks-Paar-Filter", topic=topic, pairs=lines, out_path=fp), - role="judge", capabilities="files", - payload=lambda result, p=fp: _pairs_schema(_json_file(p)), - timeout=_timeout("selection_mapping", len(paare)), - ) - - await _gather_progress([_filt(pi, p) for pi, p in enumerate(packages)], - len(packages), _report_p(set_p, topic, "Dedup")) - if is_cancelled(): - return False - # Collect confirmed "ja" edges, then COMPLETE-LINK (greedy cliques) instead of single-link - # union-find — prevents chaining (A=B + B=C does NOT merge A,C without a direct A=C). - edge_list, ja = [], 0 - for pi, paare in enumerate(packages): - verdict = _pairs_schema(_json_file(pair_path(pi))) or {} - for j, (a, b) in enumerate(paare): - if verdict.get(j + 1): - edge_list.append((a, b)) - ja += 1 - groups = _cliques(n, edge_list) - removed = 0 - for idxs in groups: - # representative = main concept (fewest property markers), then shortest title. - rep = min(idxs, key=lambda k: (_aspect_marker(consensus[k]["title"]), len(consensus[k]["title"]), k)) - for k in idxs: - if k != rep: - await db.set_block_status(topic, consensus[k]["title_norm"], "discarded") - removed += 1 - from collections import Counter - atomic_write_json(work_dir / "dedup-runde-1.json", - {"vorher": n, "entfernt": removed, "paare_geprueft": len(cands), "paare_ja": ja, - "clique_groessen": dict(sorted(Counter(len(g) for g in groups).items())), - "groups": [[consensus[k]["title"] for k in g] for g in groups]}, indent=1) - _log(topic, f"Dedup (pairwise): {n} → {n - removed} (−{removed}); {ja}/{len(cands)} pairs confirmed") - await db.set_step_status(topic, "Dedup", "done") - return True - - def _filter_schema(data) -> dict[int, int] | None: """{"fragments": {"3": 7, "12": 8}} → {block_nr: parent_nr} · None on invalid structure. Empty dict = valid (nothing to degrade). Parent ≠ itself.""" @@ -2963,7 +2933,7 @@ async def _artefacts_block(ctx: GenContext, set_p, files: dict, sidecar: dict, i await asyncio.gather(*[ run_agent(f"blocks-{topic}-artifact-example-check-c{ci}-j{j}", _prompt("Artifact-Example-Check", topic=topic, facts=block_text(idxs), examples=examples_txt, out_path=cpath(j), extra=_extra(instructions)), - _timeout("content_check", len(items)), provider=provider, role="judge", capabilities=caps) + _timeout("content_check", len(items)), provider=provider, role="judge", capabilities=caps, scope=topic) for j in pending], return_exceptions=True) outs = [s for j in (1, 2, 3)[:FACTS_CHECK_PANEL] if (s := _example_check_schema(_json_file(cpath(j)))) is not None] if not outs: @@ -3073,7 +3043,7 @@ async def _reset_db_from_phase(topic: str, label: str) -> None: await db.delete_subblocks(topic) if idx <= 1: # Inventory: inventory + research steps — triage stays await db.delete_blocks(topic) - await db.delete_pipeline_state(topic, ["Research", "Consolidation", "Clarification", "Dedup", "Blocks-Filter"]) + await db.delete_pipeline_state(topic, ["Research", "Consolidation", "Clarification", "Blocks-Filter"]) if idx <= 0: # Source: redo triage (coverage/content + step) await db.delete_coverage(topic) await db.delete_pipeline_state(topic, ["Source prep"]) @@ -3114,7 +3084,7 @@ async def generate_blocks(topic: str, instructions: str = "", provider: str = DE # Re-run from the chosen phase: delete artefacts from there; the fresh-start block # below is skipped (with a preserved sidecar it would otherwise wipe everything). if ab_step is not None: # fine sub-step re-run (takes precedence over ab_phase) - await _reset_from_step(topic, ab_step) + await _reset_from_step(topic, ab_step, to_step) elif ab_phase is not None: phasen = _phases(topic) label = phasen[ab_phase - 1][0] if 1 <= ab_phase <= len(phasen) else "Inventory" @@ -3156,9 +3126,6 @@ async def generate_blocks(topic: str, instructions: str = "", provider: str = DE if _past_limit("Clarification"): return if not await _stage(_clarify_inventory(ctx, set_p, files)): return - if _past_limit("Dedup"): return - if not await _stage(_dedup_inventory(ctx, set_p, files)): - return if _past_limit("Blocks-Filter"): return if not await _stage(_filter_inventory(ctx, set_p, files)): return diff --git a/backend/config.py b/backend/config.py index 54959f4..992ed39 100644 --- a/backend/config.py +++ b/backend/config.py @@ -1,3 +1,4 @@ +import os from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parent.parent @@ -41,11 +42,17 @@ EMBEDDING_BLOCK_CAP = 25 # max. titles per block (keep the LLM list short/s # block context — from this cosine on two are the same statement (checked on aak: ≥0.88 are # without exception true duplicates). Conservative 0.90 so different aspects (∈NP ≠ NP-hard) stay separate. EMBEDDING_SUB_DUP = 0.90 +# Cosine: two judge-subblocks state the SAME point → one cluster in the clarify majority vote. +# Lower than _DUP because it must merge paraphrases (not just typo-variants). Empirically 0.80 keeps +# distinct aspects (∈NP vs NP-hard) apart while clustering re-wordings of the same fact. +EMBEDDING_SUB_SAME = 0.80 -# Cap for concurrent CLI agent processes (across all generations). -# Own lane for interactive calls (chat, elements) so they don't hang behind -# running writers in the queue. -MAX_CONCURRENT_AGENTS = 10 +# Caps for concurrent CLI agent processes (env-overridable). Two nested limits, both always active: +# a per-topic cap and a global cap across all topics. Defaults 10/10 = previous behavior (global +# dominates). Locally raise the global cap to actually parallelize across topics (per-topic stays 10). +# Own lane for interactive calls (chat, elements) so they don't hang behind running writers. +MAX_CONCURRENT_AGENTS = int(os.getenv("MAX_CONCURRENT_AGENTS", "10")) # global, all topics +MAX_CONCURRENT_AGENTS_PER_TOPIC = int(os.getenv("MAX_CONCURRENT_AGENTS_PER_TOPIC", "10")) # per topic MAX_CONCURRENT_INTERACTIVE = 8 # Grace window of the consensus races (blocks, guide, OnePager): after the first diff --git a/backend/guide.py b/backend/guide.py index f653512..00db2e4 100644 --- a/backend/guide.py +++ b/backend/guide.py @@ -469,7 +469,7 @@ async def _generate_sections( topic=topic, assignment=assignments[i], facts=facts, out_path=content_paths[i], extra=_extra(instructions), ), - _timeout("content", chunk_sizes[i]), provider=provider, role="guide", capabilities="full", + _timeout("content", chunk_sizes[i]), provider=provider, role="guide", capabilities="full", scope=topic, ) for i in pending ], writer_count, report, start=writer_count - len(pending)) @@ -509,7 +509,7 @@ async def _generate_sections( topic=topic, assignment=_assignment_subs(followup_chunks[k], entries, subs_by_title), facts=facts, out_path=followup_paths[k], extra=_extra(instructions), ), - _timeout("content", len(followup_chunks[k][0]["nums"])), provider=provider, role="guide", capabilities="full", + _timeout("content", len(followup_chunks[k][0]["nums"])), provider=provider, role="guide", capabilities="full", scope=topic, ) for k in followup_pending ], len(followup_chunks), report_n, start=len(followup_chunks) - len(followup_pending)) @@ -583,7 +583,7 @@ async def _generate_sections( ), out_path=fix_paths[i], extra=_extra(instructions), ), - _timeout("content", len(fix_chunks[i])), provider=provider, role="guide", capabilities="full", + _timeout("content", len(fix_chunks[i])), provider=provider, role="guide", capabilities="full", scope=topic, ) for i in fix_pending ], return_exceptions=True) @@ -620,7 +620,7 @@ async def _generate_sections( contents=content_text(w_chunks[i]), spec=spec, out_path=paths[i], extra=_extra(instructions), ), - _timeout("writer", 1), provider=provider, role="guide", capabilities="files", + _timeout("writer", 1), provider=provider, role="guide", capabilities="files", scope=topic, ) for i in pending ], len(w_chunks), report, start=len(w_chunks) - len(pending)) @@ -668,7 +668,7 @@ async def _generate_sections( topic=topic, format_name=format_name, assignment=_assignment_subs(nw_chunks[k], entries, subs_by_title), contents=content_text(nw_chunks[k]), spec=spec, out_path=nw_paths[k], extra=_extra(instructions), ), - _timeout("writer", 1), provider=provider, role="guide", capabilities="files", + _timeout("writer", 1), provider=provider, role="guide", capabilities="files", scope=topic, ) for k in nw_pending ], len(nw_chunks), report_nw, start=len(nw_chunks) - len(nw_pending)) @@ -770,7 +770,7 @@ async def _generate_sections( tasks=tasks_text(fix_chunks[i], problems_by_num), out_path=fix_paths[i], extra=_extra(instructions), ), - _timeout("writer", len(fix_chunks[i])), provider=provider, role="guide", capabilities="full", + _timeout("writer", len(fix_chunks[i])), provider=provider, role="guide", capabilities="full", scope=topic, ) for i in fix_pending ], return_exceptions=True) diff --git a/backend/pipeline.py b/backend/pipeline.py index ecfdea6..5a2be44 100644 --- a/backend/pipeline.py +++ b/backend/pipeline.py @@ -191,6 +191,7 @@ async def _race(topic: str, label: str, slots: list[dict], quorum: int, timeout: task = asyncio.create_task(run_agent( slot["key"], slot["prompt"], timeout, provider=provider, role=slot["role"], capabilities=slot["capabilities"], + scope=topic, )) tasks[task] = i diff --git a/frontend/src/components/BlocksOverview.vue b/frontend/src/components/BlocksOverview.vue index 2fc824f..8b084cb 100644 --- a/frontend/src/components/BlocksOverview.vue +++ b/frontend/src/components/BlocksOverview.vue @@ -47,7 +47,11 @@ function arm(action, fn) { if (confirm.value === action) { confirm.value = null; fn() } else confirm.value = action } -function regenerateFromHere() { const from = startSel.value, to = endSel.value; clearSel(); emit('restartFrom', { from, to }) } +// regenerate: re-run ONLY the picked step(s). No end → bound to the start itself (single step); +// with an end → the whole [start, end] range. Later steps stay intact (bounded reset). +function regenerateFromHere() { const from = startSel.value, to = endSel.value ?? startSel.value; clearSel(); emit('restartFrom', { from, to }) } +// continue: run from the start point straight through to the end of the pipeline (full cascade). +function continueFromHere() { const from = startSel.value; clearSel(); emit('restartFrom', { from, to: null }) } function deleteFromHere() { const from = startSel.value; clearSel(); emit('resetFrom', from) } const items = ref([]) @@ -139,7 +143,8 @@ const subTotal = computed(() => items.value.reduce((n, b) => n + (b.subblocks?.l
From «{{ startLabel }}» to «{{ endLabel }}»: - + +
diff --git a/templates/Prompt/Blocks-Paar-Filter.md b/templates/Prompt/Blocks-Paar-Filter.md index ce6e421..4c40aeb 100644 --- a/templates/Prompt/Blocks-Paar-Filter.md +++ b/templates/Prompt/Blocks-Paar-Filter.md @@ -1,18 +1,23 @@ -Two research passes have noted blocks for the topic "{topic}". For EACH pair, decide whether A and B denote the SAME block (the same concept, just worded differently) → **ja**, or whether they are TWO DIFFERENT blocks → **nein**. +Research has noted block candidates for the topic "{topic}". For EACH pair, decide whether A and B denote the SAME block (the same concept, just worded differently) → **ja**, or whether they are TWO DIFFERENT blocks → **nein**. PAIRS: {pairs} Rules: -- **Watch the CORE ENTITY first** (the problem/object in question): Clique, Vertex Cover, Independent Set, Dominating Set, Set Cover, FVS, Knapsack … If the entities are DIFFERENT → **nein**, no matter how identical the phrasing. -- Identical phrasing is deceptive. These pairs are **nein** (different entity despite nearly identical wording): - - "Lower Bound **Clique** bzgl. Knoten" ↔ "Lower Bound **Vertex Cover** bzgl. Knoten" - - "Lower Bound Clique bzgl. **Knoten**" ↔ "Lower Bound Clique bzgl. **Kanten**" - - "Verifizierer für **FVS**" ↔ "Verifizierer für **Knapsack**" - - "**Cliquenproblem**" ↔ "**Vertex-Cover-Problem**" -- **ja** only on genuine semantic equivalence: same solution to the same problem, the same entity, just different wording/naming (e.g. "SET COVER" ↔ "Mengenüberdeckungsproblem", "Cliquenproblem" ↔ "k-CLIQUE", "List Scheduling" ↔ "LPT-Algorithmus"). -- **nein** also for different aspects of the same problem: "Set Cover (Problem)" ↔ "Set Cover ETH-Schranke"; a problem ↔ its reduction to another; a problem ↔ its verifier. -- When in doubt **nein** — better two separate blocks than wrongly merging two concepts. +- **Watch the CORE ENTITY first** (the object/concept/operation in question). If the entities are DIFFERENT → **nein**, no matter how identical the phrasing. Examples of **nein** despite near-identical wording: + - "Plugin-**Lebenszyklus**" ↔ "App-**Lebenszyklus**" (different object: Plugin vs App) + - "**install()**-Methode" ↔ "**uninstall()**-Methode" (different operation) + - "Lower Bound **Clique**" ↔ "Lower Bound **Vertex Cover**" (different problem) + - "Present **Perfect**" ↔ "Past **Perfect**" (different tense) +- **ja** ONLY on genuine semantic equivalence: the same concept/operation, just different wording, naming, language or abbreviation. Examples: + - "EntityRepository" ↔ "Repository für CRUD-Operationen am DAL" + - "Set Cover" ↔ "Mengenüberdeckungsproblem" + - "List Scheduling" ↔ "LPT-Algorithmus" +- **nein** for different ASPECTS, properties, parts or methods of the same thing — they are their own blocks (a later step folds true fragments back in): + - a concept ↔ one of its properties/details ("DAL" ↔ "DAL-Versionierung") + - a concept ↔ a single method/step of it ("Plugin-Lebenszyklus" ↔ "install()-Methode") + - a thing ↔ its sub-component, its verifier, its reduction to another thing +- When in doubt → **nein**. Better two separate blocks than wrongly merging two concepts. Write ONLY the JSON file to: {out_path}