259 lines
11 KiB
Python
259 lines
11 KiB
Python
"""Elemente (persönliche Zusammenfassung) und Tutor-Chat zum Guide."""
|
|
|
|
import json
|
|
import logging
|
|
import uuid
|
|
|
|
from agents import run_agent
|
|
from config import DEFAULT_PROVIDER
|
|
from jsonio import parse_json_text as _parse_json_text, read_json_file as _json_datei
|
|
from paths import bausteine_path, guide_content_path
|
|
from pipeline import _prompt
|
|
|
|
log = logging.getLogger("creator.elements")
|
|
|
|
|
|
# --- Tutor-Chat ---
|
|
|
|
def _build_guide_chat_prompt(topic: str, format_name: str, section: str, outline: str, messages: list[dict]) -> str:
|
|
transcript = "\n".join(
|
|
f"{'Nutzer' if m.get('role') == 'user' else 'Assistent'}: {m.get('content', '')}"
|
|
for m in messages
|
|
)
|
|
return _prompt(
|
|
"Chat",
|
|
topic=topic, format_name=format_name,
|
|
outline_block=outline.strip() or "(keine)",
|
|
section_block=section.strip() or "(kein Abschnitt erkannt)",
|
|
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 "Entschuldigung, das hat nicht geklappt. Bitte versuche es erneut."
|
|
reply = stdout.strip()
|
|
return reply or "Entschuldigung, ich habe keine Antwort erhalten."
|
|
except Exception:
|
|
log.warning("[%s] Guide-Chat fehlgeschlagen", topic, exc_info=True)
|
|
return "Entschuldigung, das hat nicht geklappt. Bitte versuche es erneut."
|
|
|
|
|
|
# --- Elemente ---
|
|
|
|
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"],
|
|
"hints": listen["hints"],
|
|
}
|
|
|
|
|
|
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", lane="interactive"
|
|
)
|
|
if returncode != 0:
|
|
return fallback
|
|
return _element_fields(_parse_json_text(stdout)) or fallback
|
|
except Exception:
|
|
log.warning("[%s] Element-Erstellung fehlgeschlagen", topic, exc_info=True)
|
|
return fallback
|
|
|
|
|
|
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"):
|
|
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", lane="interactive"
|
|
)
|
|
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", lane="interactive"
|
|
)
|
|
if returncode != 0:
|
|
return None
|
|
return _parse_suggestions(stdout)
|
|
except Exception:
|
|
log.warning("[%s] Element-Prüfung fehlgeschlagen", element.get("topic", "?"), exc_info=True)
|
|
return None
|
|
|
|
|
|
def _element_json(element: dict) -> str:
|
|
return json.dumps(
|
|
{k: element[k] for k in ("title", "description", "examples", "hints")},
|
|
ensure_ascii=False, indent=1,
|
|
)
|
|
|
|
|
|
def _validate_change(c, element: dict) -> dict | None:
|
|
"""Validiert einen Änderungs-Vorschlag aus KI-Output gegen das Element."""
|
|
if not isinstance(c, dict):
|
|
return None
|
|
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"):
|
|
return None
|
|
if target not in ("title", "description", "examples", "hints"):
|
|
return None
|
|
if action in ("anpassen", "hinzufuegen") and not content:
|
|
return None
|
|
if action == "entfernen" and target not in ("examples", "hints"):
|
|
return None
|
|
# 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])):
|
|
return None
|
|
else:
|
|
index = None
|
|
return {"text": text, "action": action, "target": target, "index": index, "content": content}
|
|
|
|
|
|
async def chat_with_element(element: dict, messages: list[dict], provider: str = DEFAULT_PROVIDER) -> tuple[str, list[dict]]:
|
|
"""Chat zum Element. Gibt (Antwort, Änderungs-Vorschläge) zurück — ändert nichts direkt."""
|
|
fehler = "Entschuldigung, das hat nicht geklappt. Bitte versuche es erneut."
|
|
try:
|
|
transcript = "\n".join(
|
|
f"{'Nutzer' if m.get('role') == 'user' else 'Assistent'}: {m.get('content', '')}"
|
|
for m in messages
|
|
)
|
|
prompt = _prompt("Element-Chat", topic=element["topic"], element_json=_element_json(element), transcript=transcript)
|
|
returncode, stdout, _ = await run_agent(
|
|
"element-chat-" + str(uuid.uuid4()), prompt, 240, provider=provider, role="fast", capabilities="none", lane="interactive"
|
|
)
|
|
if returncode != 0:
|
|
return fehler, []
|
|
data = _parse_json_text(stdout)
|
|
if not isinstance(data, dict):
|
|
return fehler, []
|
|
changes = [v for c in data.get("changes", []) if (v := _validate_change(c, element))]
|
|
reply = str(data.get("reply", "")).strip() or ("Vorschläge erstellt." if changes else fehler)
|
|
return reply, changes
|
|
except Exception:
|
|
log.warning("[%s] Element-Chat fehlgeschlagen", element.get("topic", "?"), exc_info=True)
|
|
return fehler, []
|
|
|
|
|
|
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:
|
|
prompt = _prompt("Element-Stil", topic=element["topic"], element_json=_element_json(element))
|
|
returncode, stdout, _ = await run_agent(
|
|
"element-stil-" + str(uuid.uuid4()), prompt, 240, provider=provider, role="fast", capabilities="none", lane="interactive"
|
|
)
|
|
if returncode != 0:
|
|
return None
|
|
data = _parse_json_text(stdout)
|
|
if not isinstance(data, dict):
|
|
return None
|
|
return [v for c in data.get("changes", []) if (v := _validate_change(c, element))]
|
|
except Exception:
|
|
log.warning("[%s] Stil-Prüfung fehlgeschlagen", element.get("topic", "?"), exc_info=True)
|
|
return None
|
|
|
|
|
|
async def refine_suggestion(element: dict, suggestion: dict, instruction: str, provider: str = DEFAULT_PROVIDER) -> dict | None:
|
|
"""Überarbeitet einen einzelnen Vorschlag nach Nutzer-Anweisung. None bei Fehler."""
|
|
try:
|
|
prompt = _prompt(
|
|
"Element-Refine",
|
|
topic=element["topic"], element_json=_element_json(element),
|
|
suggestion_json=json.dumps(suggestion, ensure_ascii=False, indent=1),
|
|
instruction=instruction,
|
|
)
|
|
returncode, stdout, _ = await run_agent(
|
|
"element-refine-" + str(uuid.uuid4()), prompt, 240, provider=provider, role="fast", capabilities="none", lane="interactive"
|
|
)
|
|
if returncode != 0:
|
|
return None
|
|
data = _parse_json_text(stdout)
|
|
if not isinstance(data, dict):
|
|
return None
|
|
return _validate_change(data.get("change"), element)
|
|
except Exception:
|
|
log.warning("[%s] Vorschlags-Überarbeitung fehlgeschlagen", element.get("topic", "?"), exc_info=True)
|
|
return None
|