update
This commit is contained in:
team3
@@ -1350,3 +1350,232 @@ async def chat_with_guide(topic: str, format_name: str, section: str, outline: s
|
||||
return reply or "Entschuldigung, ich habe keine Antwort erhalten."
|
||||
except Exception:
|
||||
return "Entschuldigung, das hat nicht geklappt. Bitte versuche es erneut."
|
||||
|
||||
|
||||
# --- Elemente (persönliche Zusammenfassung) ---
|
||||
|
||||
def _parse_json_text(text: str):
|
||||
"""Parst JSON aus KI-Output (Code-Fences und Drumherum-Text tolerant).
|
||||
|
||||
Repariert unescapte Anführungszeichen in Strings (z. B. MiniMax: "Titel „p" geändert"):
|
||||
das letzte `"` vor der Fehlerstelle escapen und erneut parsen.
|
||||
"""
|
||||
text = re.sub(r"^```(?:json)?\s*|\s*```$", "", (text or "").strip())
|
||||
start, end = text.find("{"), text.rfind("}")
|
||||
if start == -1 or end <= start:
|
||||
return None
|
||||
candidate = text[start:end + 1]
|
||||
for _ in range(20):
|
||||
try:
|
||||
return json.loads(candidate)
|
||||
except json.JSONDecodeError as e:
|
||||
if not e.msg.startswith(("Expecting ',' delimiter", "Expecting ':' delimiter")):
|
||||
return None
|
||||
q = candidate.rfind('"', 0, e.pos)
|
||||
if q <= 0:
|
||||
return None
|
||||
candidate = candidate[:q] + '\\"' + candidate[q + 1:]
|
||||
except Exception:
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def _element_fields(data: dict) -> dict | None:
|
||||
"""Validiert KI-Element-JSON und normalisiert auf die DB-Felder."""
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
title = str(data.get("title", "")).strip()
|
||||
if not title:
|
||||
return None
|
||||
listen = {}
|
||||
for key in ("examples", "hints"):
|
||||
raw = data.get(key, [])
|
||||
listen[key] = [str(e).strip() for e in raw if str(e).strip()] if isinstance(raw, list) else []
|
||||
return {
|
||||
"title": title[:200],
|
||||
"description": str(data.get("description", "")).strip(),
|
||||
"examples": listen["examples"][:5],
|
||||
"hints": listen["hints"][:5],
|
||||
}
|
||||
|
||||
|
||||
def _topic_context(topic: str, limit: int = 12000) -> str:
|
||||
"""Bausteine + Guide-Inhalte des Themas als Kontext-Text (gekürzt)."""
|
||||
parts: list[str] = []
|
||||
bp = bausteine_path(topic)
|
||||
if bp.exists():
|
||||
parts.append(bp.read_text(encoding="utf-8"))
|
||||
for fmt in ("FullGuide", "Guide", "MiniGuide", "OnePager"):
|
||||
content = _json_datei(guide_content_path(topic, fmt))
|
||||
if content:
|
||||
for ch in content.get("chapters", []):
|
||||
for sec in ch.get("sections", []):
|
||||
parts.append(sec if isinstance(sec, str) else json.dumps(sec, ensure_ascii=False))
|
||||
break # bester verfügbarer Guide reicht
|
||||
text = "\n\n".join(parts).strip()
|
||||
return text[:limit] if text else "(kein Material vorhanden)"
|
||||
|
||||
|
||||
async def generate_element(topic: str, hint: str, provider: str = DEFAULT_PROVIDER) -> dict:
|
||||
"""Erstellt Element-Felder per KI. Fallback: nur Titel aus dem Stichwort."""
|
||||
fallback = {"title": hint.strip() or "Neues Element", "description": "", "examples": [], "hints": []}
|
||||
try:
|
||||
prompt = _prompt(
|
||||
"Element-Create",
|
||||
topic=topic, hint=hint.strip() or "(keins — wähle selbst ein Kernkonzept)",
|
||||
context=_topic_context(topic),
|
||||
)
|
||||
returncode, stdout, _ = await run_agent(
|
||||
"element-" + str(uuid.uuid4()), prompt, 240, provider=provider, role="fast", capabilities="none"
|
||||
)
|
||||
if returncode != 0:
|
||||
return fallback
|
||||
return _element_fields(_parse_json_text(stdout)) or fallback
|
||||
except Exception:
|
||||
return fallback
|
||||
|
||||
|
||||
def _fence(content: str) -> str:
|
||||
"""Beispiele müssen Codeblöcke sein — fehlende Fences nachrüsten."""
|
||||
if content.startswith("```"):
|
||||
return content
|
||||
return f"```\n{content}\n```"
|
||||
|
||||
|
||||
def _parse_suggestions(stdout: str) -> list[dict] | None:
|
||||
"""Validiert Vorschlags-JSON aus KI-Output. None bei ungültigem JSON."""
|
||||
data = _parse_json_text(stdout)
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
suggestions = []
|
||||
for s in data.get("suggestions", []):
|
||||
if not isinstance(s, dict):
|
||||
continue
|
||||
text = str(s.get("text", "")).strip()
|
||||
target = s.get("target")
|
||||
content = str(s.get("content", "")).strip()
|
||||
if text and content and target in ("description", "examples", "hints"):
|
||||
if target == "examples":
|
||||
content = _fence(content)
|
||||
suggestions.append({"text": text, "target": target, "content": content})
|
||||
return suggestions
|
||||
|
||||
|
||||
async def check_element(element: dict, provider: str = DEFAULT_PROVIDER) -> list[dict] | None:
|
||||
"""Zweischrittige Prüfung auf fehlende Infos: Recherche → Verifizieren. None bei Fehler."""
|
||||
try:
|
||||
element_json = json.dumps(
|
||||
{k: element[k] for k in ("title", "description", "examples", "hints")},
|
||||
ensure_ascii=False, indent=1,
|
||||
)
|
||||
context = _topic_context(element["topic"])
|
||||
|
||||
# Schritt 1: Recherche — breit Kandidaten sammeln
|
||||
prompt = _prompt("Element-Check", topic=element["topic"], element_json=element_json, context=context)
|
||||
returncode, stdout, _ = await run_agent(
|
||||
"element-check-" + str(uuid.uuid4()), prompt, 240, provider=provider, role="fast", capabilities="none"
|
||||
)
|
||||
if returncode != 0:
|
||||
return None
|
||||
candidates = _parse_suggestions(stdout)
|
||||
if candidates is None:
|
||||
return None
|
||||
if not candidates:
|
||||
return []
|
||||
|
||||
# Schritt 2: Verifizieren — nur Wichtiges, nicht Redundantes durchlassen
|
||||
prompt = _prompt(
|
||||
"Element-Verify",
|
||||
topic=element["topic"], element_json=element_json,
|
||||
candidates_json=json.dumps({"suggestions": candidates}, ensure_ascii=False, indent=1),
|
||||
context=context,
|
||||
)
|
||||
returncode, stdout, _ = await run_agent(
|
||||
"element-verify-" + str(uuid.uuid4()), prompt, 240, provider=provider, role="fast", capabilities="none"
|
||||
)
|
||||
if returncode != 0:
|
||||
return None
|
||||
return _parse_suggestions(stdout)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
async def _element_rewrite(template: str, element: dict, provider: str, **extra) -> tuple[str, dict | None]:
|
||||
"""Gemeinsames Muster: Element + Template → (Antwort, neue Felder|None)."""
|
||||
fehler = "Entschuldigung, das hat nicht geklappt. Bitte versuche es erneut."
|
||||
try:
|
||||
element_json = json.dumps(
|
||||
{k: element[k] for k in ("title", "description", "examples", "hints")},
|
||||
ensure_ascii=False, indent=1,
|
||||
)
|
||||
prompt = _prompt(template, topic=element["topic"], element_json=element_json, **extra)
|
||||
returncode, stdout, _ = await run_agent(
|
||||
f"element-{template.lower()}-" + str(uuid.uuid4()), prompt, 240,
|
||||
provider=provider, role="fast", capabilities="none",
|
||||
)
|
||||
if returncode != 0:
|
||||
return fehler, None
|
||||
data = _parse_json_text(stdout)
|
||||
if not isinstance(data, dict):
|
||||
return fehler, None
|
||||
fields = _element_fields(data.get("element"))
|
||||
reply = str(data.get("reply", "")).strip() or ("Erledigt." if fields else fehler)
|
||||
return reply, fields
|
||||
except Exception:
|
||||
return fehler, None
|
||||
|
||||
|
||||
async def chat_with_element(element: dict, messages: list[dict], provider: str = DEFAULT_PROVIDER) -> tuple[str, dict | None]:
|
||||
"""Passt ein Element per Chat an. Gibt (Antwort, neue Felder|None) zurück."""
|
||||
transcript = "\n".join(
|
||||
f"{'Nutzer' if m.get('role') == 'user' else 'Assistent'}: {m.get('content', '')}"
|
||||
for m in messages
|
||||
)
|
||||
return await _element_rewrite("Element-Chat", element, provider, transcript=transcript)
|
||||
|
||||
|
||||
async def style_element(element: dict, provider: str = DEFAULT_PROVIDER) -> list[dict] | None:
|
||||
"""Prüft ein Element auf die Stil-Regeln und schlägt Änderungen vor. None bei Fehler."""
|
||||
try:
|
||||
element_json = json.dumps(
|
||||
{k: element[k] for k in ("title", "description", "examples", "hints")},
|
||||
ensure_ascii=False, indent=1,
|
||||
)
|
||||
prompt = _prompt("Element-Stil", topic=element["topic"], element_json=element_json)
|
||||
returncode, stdout, _ = await run_agent(
|
||||
"element-stil-" + str(uuid.uuid4()), prompt, 240, provider=provider, role="fast", capabilities="none"
|
||||
)
|
||||
if returncode != 0:
|
||||
return None
|
||||
data = _parse_json_text(stdout)
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
changes = []
|
||||
for c in data.get("changes", []):
|
||||
if not isinstance(c, dict):
|
||||
continue
|
||||
text = str(c.get("text", "")).strip()
|
||||
action = c.get("action")
|
||||
target = c.get("target")
|
||||
index = c.get("index")
|
||||
content = str(c.get("content", "")).strip()
|
||||
if not text or action not in ("entfernen", "anpassen", "hinzufuegen"):
|
||||
continue
|
||||
if target not in ("title", "description", "examples", "hints"):
|
||||
continue
|
||||
if action in ("anpassen", "hinzufuegen") and not content:
|
||||
continue
|
||||
if action == "entfernen" and target not in ("examples", "hints"):
|
||||
continue
|
||||
# Index nur für anpassen/entfernen in Listen-Feldern; muss existieren
|
||||
if target in ("examples", "hints") and action in ("anpassen", "entfernen"):
|
||||
if not isinstance(index, int) or not (0 <= index < len(element[target])):
|
||||
continue
|
||||
else:
|
||||
index = None
|
||||
if target == "examples" and action in ("anpassen", "hinzufuegen"):
|
||||
content = _fence(content)
|
||||
changes.append({"text": text, "action": action, "target": target, "index": index, "content": content})
|
||||
return changes
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
Reference in New Issue
Block a user