Lars
3450a9296a
All checks were successful
Deploy Development / deploy (push) Successful in 45s
Test Suite / pytest-backend (push) Successful in 40s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 13s
Test Suite / k6 /health Baseline (push) Successful in 34s
Test Suite / playwright-tests (push) Successful in 1m13s
- Updated the AI gap filling logic to include structured offers for unfilled gaps, improving the user experience in the Exercise Progression Path Builder. - Introduced new functions for detecting off-topic steps and parsing LLM-suggested exercises, enhancing the contextual relevance of exercise suggestions. - Enhanced the frontend components to support new AI proposal features, including quick creation modals for newly suggested exercises. - Incremented version to 0.8.190 and updated changelog to reflect these improvements in planning AI functionality.
397 lines
12 KiB
Python
397 lines
12 KiB
Python
"""
|
|
Planungs-KI Phase E2/E3: KI-Neuanlage für Pfad-Lücken + strukturierte Angebote für die UI.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid
|
|
from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple
|
|
|
|
from ai_prompt_context import ExerciseFormAiPromptContext
|
|
from ai_prompt_job import run_exercise_form_ai_suggestion
|
|
from exercise_ai import strip_html_to_plain
|
|
|
|
from planning_exercise_path_qa import find_step_pair_index
|
|
from planning_exercise_semantics import PlanningSemanticBrief
|
|
|
|
_logger = logging.getLogger("shinkan.planning_exercise_path_ai_fill")
|
|
|
|
|
|
def _build_gap_ai_context(
|
|
*,
|
|
goal_query: str,
|
|
brief: PlanningSemanticBrief,
|
|
step_a: Mapping[str, Any],
|
|
step_b: Mapping[str, Any],
|
|
gap: Mapping[str, Any],
|
|
title_hint: Optional[str] = None,
|
|
sketch_hint: Optional[str] = None,
|
|
) -> ExerciseFormAiPromptContext:
|
|
topic = (brief.primary_topic or "Technik").strip()
|
|
phase = gap.get("expected_phase") or "vertiefung"
|
|
from_title = (step_a.get("title") or f"Übung #{step_a.get('exercise_id')}").strip()
|
|
to_title = (step_b.get("title") or f"Übung #{step_b.get('exercise_id')}").strip()
|
|
|
|
title = (title_hint or f"Brücke {topic} ({phase})").strip()[:280]
|
|
sketch = (sketch_hint or "").strip()
|
|
goal_parts = [
|
|
f"Planungsziel: {goal_query}",
|
|
"",
|
|
f"Didaktische Brücken-Übung zwischen „{from_title}“ und „{to_title}“.",
|
|
f"Phase: {phase}. Thema: {topic}.",
|
|
"Die Übung schließt die Lücke im Progressionspfad und bereitet sinnvoll auf den nächsten Schritt vor.",
|
|
]
|
|
if sketch:
|
|
goal_parts.extend(["", f"Hinweis: {sketch}"])
|
|
goal = "\n".join(goal_parts)
|
|
|
|
focus_hint = topic if brief.topic_type == "technique" else None
|
|
if brief.must_phrases:
|
|
focus_hint = ", ".join(brief.must_phrases[:2])
|
|
|
|
return ExerciseFormAiPromptContext(
|
|
title=title[:280],
|
|
goal=goal[:8000],
|
|
execution=None,
|
|
focus_hint=focus_hint,
|
|
)
|
|
|
|
|
|
def ai_proposal_to_path_step(
|
|
*,
|
|
ai_payload: Mapping[str, Any],
|
|
ctx_title: str,
|
|
gap: Mapping[str, Any],
|
|
step_a: Mapping[str, Any],
|
|
step_b: Mapping[str, Any],
|
|
) -> Dict[str, Any]:
|
|
summary_text = ""
|
|
summary_obj = ai_payload.get("summary")
|
|
if isinstance(summary_obj, dict):
|
|
summary_text = str(summary_obj.get("text") or "").strip()
|
|
elif isinstance(summary_obj, str):
|
|
summary_text = summary_obj.strip()
|
|
|
|
proposal_key = f"ai-{uuid.uuid4().hex[:10]}"
|
|
title = (ctx_title or "").strip() or "KI-Vorschlag (Brücke)"
|
|
reasons = ["KI-Neuanlage-Vorschlag — Lücke ohne passende Bibliotheks-Übung"]
|
|
|
|
return {
|
|
"exercise_id": None,
|
|
"proposal_key": proposal_key,
|
|
"variant_id": None,
|
|
"title": title,
|
|
"summary": summary_text or None,
|
|
"score": None,
|
|
"semantic_score": None,
|
|
"reasons": reasons,
|
|
"variants": [],
|
|
"is_bridge": True,
|
|
"is_ai_proposal": True,
|
|
"ai_suggestion": dict(ai_payload),
|
|
"bridge_for_gap": {
|
|
"from_exercise_id": step_a.get("exercise_id"),
|
|
"to_exercise_id": step_b.get("exercise_id"),
|
|
"gap_score": gap.get("gap_score"),
|
|
"expected_phase": gap.get("expected_phase"),
|
|
},
|
|
}
|
|
|
|
|
|
def try_suggest_ai_bridge_step(
|
|
cur,
|
|
*,
|
|
goal_query: str,
|
|
brief: PlanningSemanticBrief,
|
|
step_a: Mapping[str, Any],
|
|
step_b: Mapping[str, Any],
|
|
gap: Mapping[str, Any],
|
|
title_hint: Optional[str] = None,
|
|
sketch_hint: Optional[str] = None,
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""Ruft exercise AI suggest auf — kein Speichern in DB."""
|
|
ctx = _build_gap_ai_context(
|
|
goal_query=goal_query,
|
|
brief=brief,
|
|
step_a=step_a,
|
|
step_b=step_b,
|
|
gap=gap,
|
|
title_hint=title_hint,
|
|
sketch_hint=sketch_hint,
|
|
)
|
|
g_plain = strip_html_to_plain(ctx.goal)
|
|
if not g_plain.strip() and not (ctx.title or "").strip():
|
|
return None
|
|
try:
|
|
payload = run_exercise_form_ai_suggestion(
|
|
cur,
|
|
ctx,
|
|
want_summary=True,
|
|
want_skills=True,
|
|
want_instructions=False,
|
|
)
|
|
except Exception as exc:
|
|
_logger.warning("KI-Lückenfüller fehlgeschlagen: %s", exc)
|
|
return None
|
|
|
|
if not payload:
|
|
return None
|
|
return ai_proposal_to_path_step(
|
|
ai_payload=payload,
|
|
ctx_title=ctx.title or "",
|
|
gap=gap,
|
|
step_a=step_a,
|
|
step_b=step_b,
|
|
)
|
|
|
|
|
|
def _default_sketch(
|
|
*,
|
|
goal_query: str,
|
|
brief: PlanningSemanticBrief,
|
|
step_a: Optional[Mapping[str, Any]],
|
|
step_b: Optional[Mapping[str, Any]],
|
|
phase: str,
|
|
rationale: str = "",
|
|
) -> str:
|
|
topic = (brief.primary_topic or "Technik").strip()
|
|
from_t = (step_a or {}).get("title") or "vorherigem Schritt"
|
|
to_t = (step_b or {}).get("title") or "nächstem Schritt"
|
|
parts = [
|
|
f"Planungsziel: {goal_query}",
|
|
f"Zwischenschritt für {topic} ({phase}) zwischen „{from_t}“ und „{to_t}“.",
|
|
]
|
|
if rationale:
|
|
parts.append(rationale)
|
|
return " ".join(parts)[:1200]
|
|
|
|
|
|
def _spec_dedupe_key(spec: Mapping[str, Any]) -> Tuple[Any, ...]:
|
|
return (
|
|
spec.get("source"),
|
|
int(spec.get("insert_after_index") or 0),
|
|
str(spec.get("title_hint") or "")[:48],
|
|
)
|
|
|
|
|
|
def collect_gap_fill_specs(
|
|
*,
|
|
steps: Sequence[Mapping[str, Any]],
|
|
unfilled_gaps: Sequence[Mapping[str, Any]],
|
|
off_topic_steps: Sequence[Mapping[str, Any]],
|
|
llm_specs: Sequence[Mapping[str, Any]],
|
|
brief: PlanningSemanticBrief,
|
|
goal_query: str,
|
|
) -> List[Dict[str, Any]]:
|
|
"""Sammelt alle Lücken, für die ein KI-Anlege-Angebot sinnvoll ist."""
|
|
topic = (brief.primary_topic or "Technik").strip()
|
|
specs: List[Dict[str, Any]] = []
|
|
seen: set = set()
|
|
|
|
def add(spec: Dict[str, Any]) -> None:
|
|
key = _spec_dedupe_key(spec)
|
|
if key in seen:
|
|
return
|
|
seen.add(key)
|
|
specs.append(spec)
|
|
|
|
for gap in unfilled_gaps:
|
|
idx = find_step_pair_index(
|
|
steps,
|
|
int(gap["from_exercise_id"]),
|
|
int(gap["to_exercise_id"]),
|
|
)
|
|
if idx is None:
|
|
continue
|
|
phase = gap.get("expected_phase") or "vertiefung"
|
|
add(
|
|
{
|
|
"source": "unfilled_gap",
|
|
"insert_after_index": idx,
|
|
"gap": dict(gap),
|
|
"phase": phase,
|
|
"title_hint": f"{topic} — {phase}",
|
|
"sketch": _default_sketch(
|
|
goal_query=goal_query,
|
|
brief=brief,
|
|
step_a=steps[idx],
|
|
step_b=steps[idx + 1],
|
|
phase=str(phase),
|
|
rationale="Bibliothek enthält keine passende Brücke.",
|
|
),
|
|
"rationale": "Lücke zwischen benachbarten Schritten — keine passende Bibliotheks-Übung.",
|
|
}
|
|
)
|
|
|
|
for ot in off_topic_steps:
|
|
idx = int(ot.get("step_index") or 0)
|
|
if idx <= 0 or idx >= len(steps) - 1:
|
|
continue
|
|
phase = ot.get("expected_phase") or "vertiefung"
|
|
add(
|
|
{
|
|
"source": "off_topic",
|
|
"insert_after_index": idx - 1,
|
|
"replace_step_index": idx,
|
|
"gap": {
|
|
"expected_phase": phase,
|
|
"off_topic_title": ot.get("title"),
|
|
"off_topic_exercise_id": ot.get("exercise_id"),
|
|
},
|
|
"phase": phase,
|
|
"title_hint": f"{topic} — {phase} (Ersatz für themenfremden Schritt)",
|
|
"sketch": _default_sketch(
|
|
goal_query=goal_query,
|
|
brief=brief,
|
|
step_a=steps[idx - 1],
|
|
step_b=steps[idx + 1],
|
|
phase=str(phase),
|
|
rationale=f"Ersetzt themenfremden Schritt „{ot.get('title')}“.",
|
|
),
|
|
"rationale": f"Schritt „{ot.get('title')}“ passt nicht zum Pfad-Thema.",
|
|
}
|
|
)
|
|
|
|
for spec in llm_specs:
|
|
add(dict(spec))
|
|
|
|
return specs[:5]
|
|
|
|
|
|
def build_gap_fill_offer(
|
|
*,
|
|
spec: Mapping[str, Any],
|
|
steps: Sequence[Mapping[str, Any]],
|
|
proposal: Optional[Mapping[str, Any]] = None,
|
|
) -> Dict[str, Any]:
|
|
idx = int(spec.get("insert_after_index") or 0)
|
|
offer_id = f"{spec.get('source')}-{idx}-{uuid.uuid4().hex[:8]}"
|
|
offer: Dict[str, Any] = {
|
|
"offer_id": offer_id,
|
|
"source": spec.get("source"),
|
|
"insert_after_index": idx,
|
|
"replace_step_index": spec.get("replace_step_index"),
|
|
"title_hint": spec.get("title_hint"),
|
|
"sketch": spec.get("sketch"),
|
|
"phase": spec.get("phase"),
|
|
"rationale": spec.get("rationale"),
|
|
"has_ai_payload": False,
|
|
"from_title": (steps[idx].get("title") if idx < len(steps) else None),
|
|
"to_title": (steps[idx + 1].get("title") if idx + 1 < len(steps) else None),
|
|
}
|
|
if proposal:
|
|
offer["has_ai_payload"] = True
|
|
offer["proposal_key"] = proposal.get("proposal_key")
|
|
offer["ai_suggestion"] = proposal.get("ai_suggestion")
|
|
offer["proposal_title"] = proposal.get("title")
|
|
offer["proposal_summary"] = proposal.get("summary")
|
|
return offer
|
|
|
|
|
|
def apply_gap_fill_after_qa(
|
|
cur,
|
|
steps: List[Dict[str, Any]],
|
|
specs: Sequence[Mapping[str, Any]],
|
|
*,
|
|
goal_query: str,
|
|
brief: PlanningSemanticBrief,
|
|
include_ai_calls: bool = True,
|
|
max_ai_proposals: int = 3,
|
|
auto_insert_proposals: bool = False,
|
|
) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], List[Dict[str, Any]]]:
|
|
"""
|
|
Erzeugt gap_fill_offers für die UI; optional KI-Vorschläge einfügen.
|
|
Returns: (steps, ai_proposals, gap_fill_offers)
|
|
"""
|
|
if not specs:
|
|
return steps, [], []
|
|
|
|
out = list(steps)
|
|
proposals: List[Dict[str, Any]] = []
|
|
offers: List[Dict[str, Any]] = []
|
|
|
|
for spec in specs:
|
|
idx = int(spec.get("insert_after_index") or 0)
|
|
if idx < 0 or idx >= len(out) - 1:
|
|
continue
|
|
step_a = out[idx]
|
|
step_b = out[idx + 1]
|
|
if step_a.get("is_ai_proposal") or step_b.get("is_ai_proposal"):
|
|
offer = build_gap_fill_offer(spec=spec, steps=out, proposal=None)
|
|
offers.append(offer)
|
|
continue
|
|
|
|
gap = dict(spec.get("gap") or {})
|
|
if not gap.get("expected_phase"):
|
|
gap["expected_phase"] = spec.get("phase") or "vertiefung"
|
|
|
|
proposal: Optional[Dict[str, Any]] = None
|
|
if include_ai_calls and len(proposals) < max_ai_proposals:
|
|
proposal = try_suggest_ai_bridge_step(
|
|
cur,
|
|
goal_query=goal_query,
|
|
brief=brief,
|
|
step_a=step_a,
|
|
step_b=step_b,
|
|
gap=gap,
|
|
title_hint=str(spec.get("title_hint") or ""),
|
|
sketch_hint=str(spec.get("sketch") or ""),
|
|
)
|
|
|
|
offer = build_gap_fill_offer(spec=spec, steps=out, proposal=proposal)
|
|
offers.append(offer)
|
|
|
|
if proposal and auto_insert_proposals:
|
|
out.insert(idx + 1, proposal)
|
|
proposals.append(
|
|
{
|
|
"inserted_after_index": idx,
|
|
"proposal_key": proposal.get("proposal_key"),
|
|
"proposal_title": proposal.get("title"),
|
|
"gap": gap,
|
|
"offer_id": offer.get("offer_id"),
|
|
}
|
|
)
|
|
|
|
return out, proposals, offers
|
|
|
|
|
|
def insert_ai_proposals_for_gaps(
|
|
cur,
|
|
steps: list,
|
|
unfilled_gaps: list,
|
|
*,
|
|
goal_query: str,
|
|
brief: PlanningSemanticBrief,
|
|
max_proposals: int = 2,
|
|
) -> tuple[list, list]:
|
|
"""Legacy: Fügt KI-Vorschläge für Lücken ein, wenn Bibliotheks-Brücke fehlte."""
|
|
specs = collect_gap_fill_specs(
|
|
steps=steps,
|
|
unfilled_gaps=unfilled_gaps,
|
|
off_topic_steps=[],
|
|
llm_specs=[],
|
|
brief=brief,
|
|
goal_query=goal_query,
|
|
)
|
|
out, proposals, _offers = apply_gap_fill_after_qa(
|
|
cur,
|
|
steps,
|
|
specs,
|
|
goal_query=goal_query,
|
|
brief=brief,
|
|
include_ai_calls=True,
|
|
max_ai_proposals=max_proposals,
|
|
auto_insert_proposals=True,
|
|
)
|
|
return out, proposals
|
|
|
|
|
|
__all__ = [
|
|
"apply_gap_fill_after_qa",
|
|
"build_gap_fill_offer",
|
|
"collect_gap_fill_specs",
|
|
"insert_ai_proposals_for_gaps",
|
|
"try_suggest_ai_bridge_step",
|
|
]
|