329 lines
16 KiB
Python
329 lines
16 KiB
Python
"""Deterministischer Agenten-Ersatz für E2E-Tests: beantwortet run_agent-Aufrufe ohne LLM.
|
||
|
||
Eine `Welt` beschreibt das Thema (Blöcke → Subs → Facts …); `respond()` routet per
|
||
agent_key-Muster und liefert (rc, stdout, stderr) wie ein echter Agent — files-Agenten
|
||
schreiben die out_path-Datei aus dem Prompt, none-Agenten antworten als Text (der
|
||
Engine-Sink parst/persistiert). Damit laufen ALLE echten Schichten (_race, Quorum,
|
||
Retry, Panels, Producer, QA-Gate) in Sekunden.
|
||
|
||
Störungen sind deterministisch: `Welt.stoerungen` matcht agent_keys per Regex und
|
||
liefert n-mal einen Fehler / kaputtes JSON / eine feste Antwort (z. B. Dissens).
|
||
|
||
Aktivierung: pytest-Fixture `fake_welt` (tests/conftest.py) oder ENV CREATOR_FAKE_AGENTS=1
|
||
(echter Server, Sekunden-Smoke im Frontend).
|
||
"""
|
||
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
|
||
_PATH_RE = re.compile(r"(/\S+\.(?:json|md))")
|
||
_NUM_RE = re.compile(r"^\s*(\d+)[.)]\s+(.*\S)", re.MULTILINE)
|
||
_SUBLIST_RE = re.compile(r"^- (?:\[(\w+)\] )?(.+\S)\s*$", re.MULTILINE)
|
||
_ZIEL_RE = re.compile(r"\(([a-z]\d+)\)")
|
||
_PAIR_RE = re.compile(r"^(\d+)\.\s*\nA: \[Block: (.*?)\] (.*?)\n", re.MULTILINE)
|
||
|
||
|
||
def _norm(s: str) -> str:
|
||
return " ".join((s or "").casefold().split())
|
||
|
||
|
||
class Welt:
|
||
"""Deterministisches Themen-Modell. bloecke: {titel: {"beschreibung": str,
|
||
"subs": [titel]}}; optionale Regeln steuern Konsolidierung/Cross-Block."""
|
||
|
||
def __init__(self, bloecke: dict | None = None, *, gruppen: list | None = None,
|
||
kataloge: list | None = None, stoerungen: list | None = None):
|
||
self.bloecke = bloecke if bloecke is not None else standard_bloecke()
|
||
self.gruppen = gruppen or [] # [(haupt_titel, [weitere_titel])] → In-Block-Fold
|
||
self.kataloge = kataloge or [] # [(katalog_titel, [mitglieder_titel])]
|
||
self.stoerungen = [dict(s, rest=int(s.get("mal", 1))) for s in (stoerungen or [])]
|
||
self.calls: list[str] = [] # Auditspur: jeder agent_key in Reihenfolge
|
||
|
||
# ── Nachschlagen ────────────────────────────────────────────────────────────────
|
||
def _alle_subs(self) -> dict[str, str]:
|
||
"""sub_norm → sub_titel über alle Blöcke (inkl. Katalog-Titel)."""
|
||
out = {}
|
||
for b in self.bloecke.values():
|
||
for s in b["subs"]:
|
||
out[_norm(s)] = s
|
||
for kt, _m in self.kataloge:
|
||
out[_norm(kt)] = kt
|
||
return out
|
||
|
||
def _bloecke_im_prompt(self, prompt: str) -> list[str]:
|
||
return [t for t in self.bloecke if t in prompt]
|
||
|
||
def _subs_im_prompt(self, prompt: str) -> list[str]:
|
||
gefunden = [s for s in self._alle_subs().values() if s in prompt]
|
||
return gefunden
|
||
|
||
# ── Störungen ───────────────────────────────────────────────────────────────────
|
||
def _stoerung(self, agent_key: str):
|
||
for s in self.stoerungen:
|
||
if s["rest"] > 0 and re.search(s["muster"], agent_key):
|
||
s["rest"] -= 1
|
||
return s
|
||
return None
|
||
|
||
# ── Haupteinstieg ──────────────────────────────────────────────────────────────
|
||
def respond(self, agent_key: str, prompt: str, capabilities: str) -> tuple[int, str, str]:
|
||
self.calls.append(agent_key)
|
||
if (s := self._stoerung(agent_key)):
|
||
if s["modus"] == "fehler":
|
||
return 1, "", "fake-stoerung"
|
||
if s["modus"] == "garbage":
|
||
return self._liefern(prompt, '{"kaputt": ')
|
||
if s["modus"] == "antwort":
|
||
return self._liefern(prompt, s["antwort"])
|
||
text = self._antwort(agent_key, prompt)
|
||
if text is None:
|
||
return 1, "", f"fake: kein Handler für {agent_key}"
|
||
return self._liefern(prompt, text)
|
||
|
||
@staticmethod
|
||
def _liefern(prompt: str, text: str) -> tuple[int, str, str]:
|
||
"""files-Agenten schreiben die out_path-Datei aus dem Prompt; none-Agenten
|
||
antworten als Text. Wir tun einfach BEIDES — steht ein Pfad im Prompt, wird
|
||
er geschrieben (dann liest das payload die Datei), und stdout trägt den Text
|
||
(dann parst ihn der Sink). Ein Weg von beiden greift immer."""
|
||
if (m := _PATH_RE.search(prompt)):
|
||
p = Path(m.group(1))
|
||
try:
|
||
p.parent.mkdir(parents=True, exist_ok=True)
|
||
p.write_text(text, encoding="utf-8")
|
||
except OSError:
|
||
pass
|
||
return 0, text, ""
|
||
|
||
# ── Antwort-Generatoren je Key-Muster ──────────────────────────────────────────
|
||
def _antwort(self, key: str, prompt: str) -> str | None: # noqa: C901 — bewusst ein Router
|
||
j = json.dumps
|
||
# Board 1 / Inventar
|
||
if "-research-" in key:
|
||
zeilen = []
|
||
n = 1
|
||
for t, b in self.bloecke.items():
|
||
zeilen.append(f"{n}. {t} — {b['beschreibung']}")
|
||
n += 1
|
||
return "\n".join(zeilen)
|
||
if "-pair-" in key:
|
||
n = prompt.count("\nA: ") or 1
|
||
return j({"pairs": {str(i): "ja" for i in range(1, n + 1)}})
|
||
if "-dedup-" in key:
|
||
n = prompt.count("\nA: ") or 1
|
||
return j({"pairs": {str(i): "nein" for i in range(1, n + 1)}})
|
||
if "-clarify-" in key:
|
||
keep = [ln[2:] for ln in prompt.splitlines() if ln.startswith("- ")]
|
||
return j({"keep": keep, "rest": []})
|
||
if "-naming-" in key: # deckt auch naming_check (gleicher Key)
|
||
return j({"best": 1})
|
||
if "-sanierung-" in key: # Titel/Beschreibung unverändert zurück (Fake-Welt ist sauber)
|
||
t = re.search(r"^Title: (.+)$", prompt, re.M)
|
||
d = re.search(r"^Description: (.+)$", prompt, re.M)
|
||
beschr = (d.group(1).strip() if d else "")
|
||
if beschr == "(leer)":
|
||
beschr = "Beschreibung aus dem Material."
|
||
return j({"title": t.group(1).strip() if t else "", "description": beschr})
|
||
if "-filter-" in key: # auch filter-recheck
|
||
return j({"fragments": {}, "drop": []})
|
||
if "-gruppierung-completion-" in key:
|
||
return j({"additions": []})
|
||
if "-gruppierung-" in key:
|
||
return j({"umbrellas": []})
|
||
if "-supplement-beleg" in key or "-anker-beleg-" in key:
|
||
nums = {m.group(1) for m in _NUM_RE.finditer(prompt)}
|
||
return j({"relevant": {k: "ja" for k in sorted(nums, key=int)} or {"1": "ja"}})
|
||
if "-supplement" in key:
|
||
return j({"blocks": []})
|
||
if "-source-relevance-" in key:
|
||
nums = {m.group(1) for m in _NUM_RE.finditer(prompt)}
|
||
return j({"relevant": {k: "ja" for k in sorted(nums, key=int)} or {"1": "ja"}})
|
||
|
||
# Board 2 / Artefakte (verschmolzene Calls, block_calls.py)
|
||
if "-sb-gen-" in key:
|
||
subs = []
|
||
for t in self._bloecke_im_prompt(prompt):
|
||
for s in self.bloecke[t]["subs"]:
|
||
f = self._fakt(t, s)
|
||
subs.append({"title": s, "level": "beginner", "relevance": "relevant",
|
||
**{k: f[k] for k in ("key_points", "prerequisites", "hurdles",
|
||
"cited_facts", "example_idea")}})
|
||
return j({"subs": subs})
|
||
if "-sb-verify-" in key:
|
||
nummern = {_norm(m.group(2)): m.group(1) for m in _NUM_RE.finditer(prompt)}
|
||
gruppen = []
|
||
for haupt, weitere in self.gruppen:
|
||
h, w = nummern.get(_norm(haupt)), [nummern[_norm(x)] for x in weitere
|
||
if _norm(x) in nummern]
|
||
if h and w:
|
||
gruppen.append({"haupt": int(h), "weitere": [int(x) for x in w]})
|
||
kataloge = []
|
||
for kt, mitglieder in self.kataloge:
|
||
m = [int(nummern[_norm(x)]) for x in mitglieder if _norm(x) in nummern]
|
||
if len(m) >= 2:
|
||
kataloge.append({"titel": kt, "mitglieder": m})
|
||
unsicher = {}
|
||
if (u := prompt.find("UNSICHER")) != -1: # alle Unsicher-Nummern übernehmen
|
||
import re as _re
|
||
for m in _re.finditer(r"entries (\d+)–(\d+)", prompt[u:u + 200]):
|
||
unsicher = {str(k): "ja" for k in range(int(m.group(1)), int(m.group(2)) + 1)}
|
||
return j({"gruppen": gruppen, "kataloge": kataloge, "fremd": [], "luecken": [],
|
||
"uebernehmen": unsicher, "facts_probleme": [], "levels": {}, "relevanz": {}})
|
||
if "-sb-fix-" in key:
|
||
return j({"subs": []})
|
||
if "-sub-crossblock-" in key:
|
||
urteile = {}
|
||
for m in _PAIR_RE.finditer(prompt):
|
||
urteile[m.group(1)] = "a" # identischer Text (nur so wird gepaart) → A behält
|
||
return j({"pairs": urteile or {"1": "nein"}})
|
||
if "-art-gen-" in key:
|
||
subs = self._subs_im_prompt(prompt)
|
||
t = (self._bloecke_im_prompt(prompt) or ["?"])[0]
|
||
return j({"pattern": [{"block": t, "subblock": s, "question": f"Was ist {s}?"} for s in subs],
|
||
"cards": [{"block": t, "subblock": s, "question": f"F: {s}?", "answer": f"A: {s}"} for s in subs],
|
||
"examples": [{"block": t, "subblock": s, "problem": f"Aufgabe zu {s}",
|
||
"steps": ["Schritt 1", "Schritt 2"], "result": "Ergebnis"} for s in subs]})
|
||
if "-art-check-" in key:
|
||
return j({"ok": True})
|
||
if "-outline-prereqs" in key:
|
||
return j({"prereqs": {}})
|
||
if "-outline-review" in key:
|
||
return j({"moves": {}})
|
||
if "-outline-" in key or key.endswith("-outline-judge"):
|
||
nums = sorted({int(m.group(1)) for m in _NUM_RE.finditer(prompt)}) or [1]
|
||
return j({"chapters": [{"title": "Kapitel 1", "numbers": nums}]})
|
||
|
||
# Guide-Board
|
||
if "-ziele-" in key:
|
||
ziele = [{"id": f"z{i}", "text": f"Verstehen von {s}", "sub": s}
|
||
for i, s in enumerate(self._subs_im_prompt(prompt), 1)][:8]
|
||
return j({"ziele": ziele or [{"id": "z1", "text": "Grundlagen verstehen", "sub": ""}]})
|
||
if "-gfix-" in key:
|
||
return self._section_aus_prompt(prompt) or "<!-- section: X -->\nRepariert."
|
||
if "-pruef-" in key: # verschmolzener Prüfer: Fakten + Coverage + Lesbarkeit
|
||
ids = sorted(set(_ZIEL_RE.findall(prompt)))
|
||
return j({"claims": [], "ziele": {z: True for z in ids}, "luecken": [],
|
||
"ballast": [], "lese_probleme": []})
|
||
if "-w-" in key:
|
||
return self._writer_md(prompt)
|
||
|
||
# QA / Repair / Guide-QA (alle Text, _yesno_schema)
|
||
if key.startswith(("qa-guide-",)):
|
||
nums = {m.group(1) for m in _NUM_RE.finditer(prompt)}
|
||
return j({"relevant": {k: "nein" for k in sorted(nums, key=int)} or {"1": "nein"}})
|
||
if key.startswith(("qa-", "repair-")):
|
||
# Semantik je Template: Bausteine „ja" = echt; Dubletten/Lücken „nein" = kein Befund
|
||
wert = "ja" if ("bausteine" in key or "beleg" in key or "fremd" in key) else "nein"
|
||
nums = {m.group(1) for m in _NUM_RE.finditer(prompt)}
|
||
return j({"relevant": {k: wert for k in sorted(nums, key=int)} or {"1": wert}})
|
||
return None
|
||
|
||
# ── Bausteine der Antworten ─────────────────────────────────────────────────────
|
||
@staticmethod
|
||
def _fakt(block: str, sub: str) -> dict:
|
||
return {"block": block, "subblock": sub,
|
||
"key_points": [f"Kernaussage zu {sub}", f"Zweite Aussage zu {sub}"],
|
||
"prerequisites": "", "hurdles": "",
|
||
"cited_facts": [{"text": f"Beleg für {sub}", "source": "Fake-Quelle"}],
|
||
"example_idea": f"Beispiel zu {sub}"}
|
||
|
||
def _writer_md(self, prompt: str) -> str:
|
||
"""Section im Marker-Format; Subs aus der SUBBLOCKS-Liste des Prompts
|
||
(`- [label] titel`), Länge je Sub im 150–1080-Rahmen."""
|
||
block = next(iter(self._bloecke_im_prompt(prompt)), None) or "Abschnitt"
|
||
subs = [(lv or "beginner", t) for lv, t in _SUBLIST_RE.findall(prompt)
|
||
if _norm(t) in self._alle_subs()]
|
||
if not subs:
|
||
subs = [("beginner", s) for s in self.bloecke.get(block, {}).get("subs", ["Inhalt"])]
|
||
kompakt = "\n".join(f"<!-- sub: {lv} | {t} -->\n- Merksatz zu {t}" for lv, t in subs)
|
||
prosa = "\n".join(f"<!-- sub: {lv} | {t} -->\n" + (f"Lehrtext über {t}. " * 12)
|
||
for lv, t in subs)
|
||
return (f"<!-- kapitel: Kapitel 1 -->\n<!-- section: {block} -->\n"
|
||
f"<!-- compact -->\n{kompakt}\n<!-- ausführlich -->\n"
|
||
f"Einstieg in {block}.\n{prosa}")
|
||
|
||
@staticmethod
|
||
def _section_aus_prompt(prompt: str) -> str | None:
|
||
"""Fix-Agenten geben die Section unverändert zurück (minimal-invasiv)."""
|
||
m = re.search(r"(<!--\s*(?:kapitel|section):.*)", prompt, re.DOTALL)
|
||
return m.group(1).strip() if m else None
|
||
|
||
|
||
def standard_bloecke() -> dict:
|
||
"""3 Blöcke; „Gemeinsamer Grundbegriff" liegt in Alpha UND Beta (Cross-Block-Fall)."""
|
||
return {
|
||
"Alpha-Konzept": {"beschreibung": "Das erste Grundkonzept",
|
||
"subs": ["Definition Alpha", "Alpha Eigenschaften",
|
||
"Gemeinsamer Grundbegriff"]},
|
||
"Beta-Verfahren": {"beschreibung": "Das zentrale Verfahren",
|
||
"subs": ["Beta Ablauf", "Beta Grenzen", "Gemeinsamer Grundbegriff"]},
|
||
"Gamma-Anwendung": {"beschreibung": "Praktische Anwendung",
|
||
"subs": ["Gamma Praxisfall", "Gamma Werkzeuge"]},
|
||
}
|
||
|
||
|
||
_WELT: Welt | None = None # ENV-Modus (CREATOR_FAKE_AGENTS=1): eine Welt pro Prozess
|
||
|
||
|
||
async def respond(agent_key: str, prompt: str, capabilities: str) -> tuple[int, str, str]:
|
||
global _WELT
|
||
if _WELT is None:
|
||
_WELT = Welt()
|
||
return _WELT.respond(agent_key, prompt, capabilities)
|
||
|
||
|
||
def aktivieren(welt: Welt, setattr_fn=setattr) -> None:
|
||
"""Alle Patches für einen Fake-E2E-Lauf (run_agent überall, Tempo-Bremsen raus,
|
||
Text-Identitäts-Embedding). pytest übergibt monkeypatch.setattr (auto-Rollback);
|
||
train_f0 nutzt den Default — der Prozess stirbt nach dem Lauf sowieso."""
|
||
import asyncio
|
||
|
||
import agents
|
||
import blocks
|
||
import board_artefacts as ba
|
||
import board_inventory as bi
|
||
import guide
|
||
import guide_board
|
||
import kanban
|
||
import pipeline
|
||
import qa
|
||
|
||
async def fake_run_agent(agent_key, prompt, timeout, provider="", role="fast",
|
||
capabilities="none", lane="batch", scope=None, on_line=None, label=""):
|
||
return welt.respond(agent_key, prompt, capabilities)
|
||
|
||
for mod in (agents, pipeline, guide):
|
||
setattr_fn(mod, "run_agent", fake_run_agent)
|
||
setattr_fn(blocks, "CONSENSUS_GRACE", 0)
|
||
setattr_fn(bi, "_QA_GATE_POLL", 0.05)
|
||
setattr_fn(kanban, "RETRY_BACKOFF", 0.05)
|
||
setattr_fn(guide_board, "READABILITY_ACTIVE", False)
|
||
setattr_fn(bi, "_ingest_lock", asyncio.Lock())
|
||
|
||
class _FakeEmb: # identischer Text → cos 1.0, sonst 0.0 (deterministisch, ohne Modell)
|
||
@staticmethod
|
||
def available():
|
||
return True
|
||
|
||
@staticmethod
|
||
def embed(texts):
|
||
import numpy as np
|
||
uniq = {t: k for k, t in enumerate(dict.fromkeys(texts))}
|
||
arr = np.zeros((len(texts), max(len(uniq), 1)))
|
||
for r, t in enumerate(texts):
|
||
arr[r, uniq[t]] = 1.0
|
||
return arr
|
||
|
||
@staticmethod
|
||
def embed_sims(texts):
|
||
arr = _FakeEmb.embed(texts)
|
||
return arr @ arr.T
|
||
|
||
for mod in (blocks, ba, qa):
|
||
setattr_fn(mod, "embedding", _FakeEmb)
|
||
|
||
async def emb_ok(flow): # Board-1-Vektorpfade aus — Judge-Wellen reichen
|
||
return False
|
||
setattr_fn(bi, "_emb_ok", emb_ok)
|