Lars
f074a8bef0
All checks were successful
Deploy Development / deploy (push) Successful in 43s
Test Suite / pytest-backend (push) Successful in 47s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 14s
Test Suite / k6 /health Baseline (push) Successful in 34s
Test Suite / playwright-tests (push) Successful in 1m13s
- Added support for editable major steps in the roadmap, allowing users to modify phase, learning goals, and order before exercise matching. - Introduced a new `roadmap_override` feature to facilitate customized retrieval without re-invoking the roadmap AI. - Updated the `ExerciseProgressionPathBuilder` component to incorporate these new features, enhancing user interaction and flexibility. - Incremented application version to 0.8.207 to reflect these changes.
754 lines
26 KiB
Python
754 lines
26 KiB
Python
"""
|
|
Planungs-KI Phase F: Roadmap-first Progressionsgraph-Pipeline (Workflow-lite).
|
|
|
|
Ziel → Roadmap (micro → major) → Stufenspezifikation → danach Retrieval/KI (Phase D/E).
|
|
|
|
Kein Gruppenkontext — siehe PLANNING_PROGRESSION_ROADMAP_SPEC.md.
|
|
|
|
Prompt-Texte ausschließlich in ``ai_prompts`` (Admin konfigurierbar). Dieses Modul referenziert
|
|
nur Slugs — siehe ``PROMPT_SLUG_*`` und Migrationen 078/079.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple
|
|
|
|
from pydantic import BaseModel, Field, ValidationError
|
|
|
|
from ai_prompt_runtime import AiPromptUnavailableError, load_and_render_ai_prompt
|
|
from openrouter_chat import (
|
|
effective_openrouter_model_for_prompt_row,
|
|
normalize_openrouter_env,
|
|
openrouter_chat_completion,
|
|
)
|
|
from planning_exercise_semantics import (
|
|
PlanningSemanticBrief,
|
|
brief_to_summary_dict,
|
|
build_semantic_brief,
|
|
)
|
|
|
|
_logger = logging.getLogger("shinkan.planning_progression_roadmap")
|
|
|
|
# Nur Slugs — Templates in DB (ai_prompts), bearbeitbar im Admin.
|
|
PROMPT_SLUG_GOAL_ANALYSIS = "planning_progression_goal_analysis"
|
|
PROMPT_SLUG_ROADMAP = "planning_progression_roadmap"
|
|
PROMPT_SLUG_STAGE_SPEC = "planning_progression_stage_spec"
|
|
|
|
_PHASE_ORDER = {
|
|
"einstieg": 0,
|
|
"grundlage": 1,
|
|
"vertiefung": 2,
|
|
"anwendung": 3,
|
|
"perfektion": 4,
|
|
}
|
|
|
|
_DEFAULT_MICRO_TEMPLATES: Sequence[tuple[str, str, float]] = (
|
|
("einstieg", "Einstieg und Orientierung zum Thema", 0.75),
|
|
("grundlage", "Grundstellung und Basisbewegung", 0.9),
|
|
("vertiefung", "Koordination und Präzision vertiefen", 0.85),
|
|
("vertiefung", "Kraft und Geschwindigkeit mit Technikbezug", 0.8),
|
|
("anwendung", "Anwendung und Kombination", 0.85),
|
|
("perfektion", "Perfektion und Qualitätssicherung", 0.7),
|
|
)
|
|
|
|
_EXERCISE_TYPE_BY_PHASE = {
|
|
"einstieg": "kihon_einzel",
|
|
"grundlage": "kihon_einzel",
|
|
"vertiefung": "kihon_einzel",
|
|
"anwendung": "partner_drill",
|
|
"perfektion": "kombination",
|
|
}
|
|
|
|
_LOAD_BY_PHASE = {
|
|
"einstieg": ["koordination"],
|
|
"grundlage": ["koordination", "gleichgewicht"],
|
|
"vertiefung": ["präzision", "kraft", "geschwindigkeit"],
|
|
"anwendung": ["timing", "distanz"],
|
|
"perfektion": ["präzision", "kime"],
|
|
}
|
|
|
|
|
|
class GoalAnalysisArtifact(BaseModel):
|
|
primary_topic: str = ""
|
|
start_assumption: str = ""
|
|
target_state: str = ""
|
|
success_criteria: List[str] = Field(default_factory=list)
|
|
constraints: Dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class MicroObjective(BaseModel):
|
|
id: str
|
|
phase: str
|
|
title: str
|
|
weight: float = Field(ge=0.0, le=1.0, default=0.8)
|
|
depends_on: List[str] = Field(default_factory=list)
|
|
|
|
|
|
class MajorStep(BaseModel):
|
|
index: int = Field(ge=0)
|
|
phase: str
|
|
learning_goal: str
|
|
consolidates: List[str] = Field(default_factory=list)
|
|
rationale: str = ""
|
|
|
|
|
|
class RoadmapArtifact(BaseModel):
|
|
micro_objectives: List[MicroObjective] = Field(default_factory=list)
|
|
major_steps: List[MajorStep] = Field(default_factory=list)
|
|
consolidation_notes: List[str] = Field(default_factory=list)
|
|
|
|
|
|
class StageSpecArtifact(BaseModel):
|
|
major_step_index: int = Field(ge=0)
|
|
learning_goal: str = ""
|
|
load_profile: List[str] = Field(default_factory=list)
|
|
exercise_type: str = ""
|
|
success_criteria: List[str] = Field(default_factory=list)
|
|
anti_patterns: List[str] = Field(default_factory=list)
|
|
|
|
|
|
class RoadmapOverridePayload(BaseModel):
|
|
"""Vom Trainer bearbeitete Major Steps — steuert Retrieval ohne erneute Roadmap-KI."""
|
|
|
|
major_steps: List[MajorStep] = Field(..., min_length=2, max_length=10)
|
|
stage_specs: Optional[List[StageSpecArtifact]] = None
|
|
|
|
|
|
class ProgressionRoadmapContext(BaseModel):
|
|
goal_query: str
|
|
max_steps: int = Field(ge=2, le=10, default=5)
|
|
semantic_brief: Optional[Dict[str, Any]] = None
|
|
goal_analysis: Optional[GoalAnalysisArtifact] = None
|
|
roadmap: Optional[RoadmapArtifact] = None
|
|
stage_specs: List[StageSpecArtifact] = Field(default_factory=list)
|
|
pipeline_phase: str = "roadmap_v1"
|
|
llm_goal_analysis_applied: bool = False
|
|
llm_roadmap_applied: bool = False
|
|
llm_stage_spec_applied: bool = False
|
|
prompt_slugs: List[str] = Field(default_factory=list)
|
|
|
|
|
|
def _extract_json_object(text: str) -> Dict[str, Any]:
|
|
s = (text or "").strip()
|
|
if s.startswith("```"):
|
|
s = re.sub(r"^```[a-zA-Z0-9]*\s*", "", s)
|
|
if s.endswith("```"):
|
|
s = s[:-3].strip()
|
|
start = s.find("{")
|
|
end = s.rfind("}")
|
|
if start < 0 or end <= start:
|
|
raise ValueError("Kein JSON-Objekt in LLM-Antwort")
|
|
obj = json.loads(s[start : end + 1])
|
|
if not isinstance(obj, dict):
|
|
raise ValueError("LLM-Antwort ist kein JSON-Objekt")
|
|
return obj
|
|
|
|
|
|
def _run_prompt_json(
|
|
cur,
|
|
slug: str,
|
|
variables: Dict[str, str],
|
|
) -> Optional[Dict[str, Any]]:
|
|
api_key, _ = normalize_openrouter_env()
|
|
if not api_key or cur is None:
|
|
return None
|
|
try:
|
|
prow, rendered = load_and_render_ai_prompt(cur, slug, variables)
|
|
model = effective_openrouter_model_for_prompt_row(prow)
|
|
raw = openrouter_chat_completion(api_key=api_key, model=model, user_content=rendered.text)
|
|
return _extract_json_object(raw)
|
|
except AiPromptUnavailableError:
|
|
return None
|
|
except Exception as exc:
|
|
_logger.warning("Roadmap-Prompt %s fehlgeschlagen: %s", slug, exc)
|
|
return None
|
|
|
|
|
|
def try_llm_goal_analysis(
|
|
cur,
|
|
*,
|
|
goal_query: str,
|
|
brief: PlanningSemanticBrief,
|
|
) -> Tuple[Optional[GoalAnalysisArtifact], bool]:
|
|
obj = _run_prompt_json(
|
|
cur,
|
|
PROMPT_SLUG_GOAL_ANALYSIS,
|
|
{
|
|
"goal_query": goal_query or "",
|
|
"semantic_brief_json": json.dumps(brief_to_summary_dict(brief), ensure_ascii=False),
|
|
},
|
|
)
|
|
if not obj:
|
|
return None, False
|
|
try:
|
|
return GoalAnalysisArtifact.model_validate(obj), True
|
|
except ValidationError as exc:
|
|
_logger.warning("Zielanalyse-JSON ungültig: %s", exc)
|
|
return None, False
|
|
|
|
|
|
def try_llm_roadmap(
|
|
cur,
|
|
*,
|
|
goal_query: str,
|
|
brief: PlanningSemanticBrief,
|
|
goal_analysis: GoalAnalysisArtifact,
|
|
max_steps: int,
|
|
) -> Tuple[Optional[RoadmapArtifact], bool]:
|
|
obj = _run_prompt_json(
|
|
cur,
|
|
PROMPT_SLUG_ROADMAP,
|
|
{
|
|
"goal_query": goal_query or "",
|
|
"semantic_brief_json": json.dumps(brief_to_summary_dict(brief), ensure_ascii=False),
|
|
"goal_analysis_json": json.dumps(goal_analysis.model_dump(), ensure_ascii=False),
|
|
"max_steps": str(int(max_steps)),
|
|
},
|
|
)
|
|
if not obj:
|
|
return None, False
|
|
try:
|
|
micro = [MicroObjective.model_validate(m) for m in (obj.get("micro_objectives") or [])]
|
|
majors_raw = obj.get("major_steps") or []
|
|
majors = [MajorStep.model_validate(m) for m in majors_raw]
|
|
if len(majors) != max_steps:
|
|
majors, notes = consolidate_micro_to_major(
|
|
micro or develop_micro_objectives(brief, goal_analysis=goal_analysis, min_count=max_steps + 1),
|
|
max_steps=max_steps,
|
|
)
|
|
obj["consolidation_notes"] = list(obj.get("consolidation_notes") or []) + notes
|
|
for i, m in enumerate(majors):
|
|
m.index = i
|
|
return RoadmapArtifact(
|
|
micro_objectives=micro,
|
|
major_steps=majors,
|
|
consolidation_notes=[str(n) for n in (obj.get("consolidation_notes") or []) if str(n).strip()],
|
|
), True
|
|
except ValidationError as exc:
|
|
_logger.warning("Roadmap-JSON ungültig: %s", exc)
|
|
return None, False
|
|
|
|
|
|
def try_llm_stage_specs(
|
|
cur,
|
|
*,
|
|
goal_query: str,
|
|
goal_analysis: GoalAnalysisArtifact,
|
|
major_steps: Sequence[MajorStep],
|
|
) -> Tuple[Optional[List[StageSpecArtifact]], bool]:
|
|
obj = _run_prompt_json(
|
|
cur,
|
|
PROMPT_SLUG_STAGE_SPEC,
|
|
{
|
|
"goal_query": goal_query or "",
|
|
"goal_analysis_json": json.dumps(goal_analysis.model_dump(), ensure_ascii=False),
|
|
"major_steps_json": json.dumps([m.model_dump() for m in major_steps], ensure_ascii=False),
|
|
},
|
|
)
|
|
if not obj:
|
|
return None, False
|
|
raw_specs = obj.get("stage_specs")
|
|
if not isinstance(raw_specs, list):
|
|
return None, False
|
|
try:
|
|
specs = [StageSpecArtifact.model_validate(s) for s in raw_specs]
|
|
return specs, True
|
|
except ValidationError as exc:
|
|
_logger.warning("Stufenspez-JSON ungültig: %s", exc)
|
|
return None, False
|
|
|
|
|
|
def _phase_sort_key(phase: str) -> int:
|
|
return _PHASE_ORDER.get((phase or "").strip().lower(), 2)
|
|
|
|
|
|
def _topic_label(brief: PlanningSemanticBrief) -> str:
|
|
return (brief.primary_topic or brief.retrieval_query or "Technik").strip()
|
|
|
|
|
|
def build_goal_analysis(
|
|
goal_query: str,
|
|
brief: PlanningSemanticBrief,
|
|
) -> GoalAnalysisArtifact:
|
|
"""Phase A — deterministisch aus Anfrage + Semantic Brief."""
|
|
topic = _topic_label(brief)
|
|
target = goal_query.strip() or f"Entwicklung {topic}"
|
|
arc = list(brief.development_arc or [])
|
|
start_phase = arc[0] if arc else "grundlage"
|
|
target_phase = arc[-1] if arc else "perfektion"
|
|
criteria: List[str] = []
|
|
if brief.must_phrases:
|
|
criteria.extend(brief.must_phrases[:3])
|
|
if topic:
|
|
criteria.append(f"klarer Bezug zu {topic}")
|
|
|
|
return GoalAnalysisArtifact(
|
|
primary_topic=topic,
|
|
start_assumption=(
|
|
f"Einstieg auf Niveau „{start_phase}“ — Voraussetzungen der Zielgruppe werden im "
|
|
"Progressionsgraphen nicht analysiert (erst Trainingsplanung)."
|
|
),
|
|
target_state=target,
|
|
success_criteria=criteria or [f"sichere Entwicklung Richtung {target_phase}"],
|
|
constraints={"partner_required": False, "group_analysis": False},
|
|
)
|
|
|
|
|
|
def _micro_title_for_phase(phase: str, topic: str) -> str:
|
|
p = (phase or "vertiefung").lower()
|
|
labels = {
|
|
"einstieg": f"Einstieg {topic}",
|
|
"grundlage": f"{topic} — Grundstellung und Basis",
|
|
"vertiefung": f"{topic} — Vertiefung",
|
|
"anwendung": f"{topic} — Anwendung und Kombination",
|
|
"perfektion": f"{topic} — Perfektion",
|
|
}
|
|
return labels.get(p, f"{topic} — {p}")
|
|
|
|
|
|
def develop_micro_objectives(
|
|
brief: PlanningSemanticBrief,
|
|
*,
|
|
goal_analysis: GoalAnalysisArtifact,
|
|
min_count: int = 6,
|
|
) -> List[MicroObjective]:
|
|
"""Phase B1 — Zwischenziele (heuristisch aus development_arc)."""
|
|
topic = goal_analysis.primary_topic or _topic_label(brief)
|
|
arc = [str(p).lower() for p in (brief.development_arc or []) if str(p).strip()]
|
|
seen_phases: set = set()
|
|
micro: List[MicroObjective] = []
|
|
|
|
for i, phase in enumerate(arc):
|
|
if phase in seen_phases:
|
|
continue
|
|
seen_phases.add(phase)
|
|
mid = f"m{len(micro) + 1}"
|
|
deps = [f"m{len(micro)}"] if micro else []
|
|
micro.append(
|
|
MicroObjective(
|
|
id=mid,
|
|
phase=phase,
|
|
title=_micro_title_for_phase(phase, topic),
|
|
weight=0.85 if phase in {"grundlage", "vertiefung", "anwendung"} else 0.75,
|
|
depends_on=deps,
|
|
)
|
|
)
|
|
|
|
for phase, title_tpl, weight in _DEFAULT_MICRO_TEMPLATES:
|
|
if len(micro) >= min_count:
|
|
break
|
|
if phase in seen_phases:
|
|
continue
|
|
seen_phases.add(phase)
|
|
mid = f"m{len(micro) + 1}"
|
|
deps = [micro[-1].id] if micro else []
|
|
title = title_tpl.replace("Thema", topic) if "Thema" in title_tpl else f"{topic} — {title_tpl}"
|
|
micro.append(
|
|
MicroObjective(id=mid, phase=phase, title=title, weight=weight, depends_on=deps)
|
|
)
|
|
|
|
supplement_labels = (
|
|
("vertiefung", "Präzision und Zielpunkt"),
|
|
("vertiefung", "Kraft und Schnelligkeit"),
|
|
("anwendung", "Kombination im Ablauf"),
|
|
)
|
|
si = 0
|
|
while len(micro) < min_count and si < len(supplement_labels) * 3:
|
|
phase, label = supplement_labels[si % len(supplement_labels)]
|
|
si += 1
|
|
deps = [micro[-1].id] if micro else []
|
|
micro.append(
|
|
MicroObjective(
|
|
id=f"m{len(micro) + 1}",
|
|
phase=phase,
|
|
title=f"{topic} — {label}",
|
|
weight=0.8,
|
|
depends_on=deps,
|
|
)
|
|
)
|
|
|
|
micro.sort(key=lambda m: _phase_sort_key(m.phase))
|
|
for i, m in enumerate(micro):
|
|
m.id = f"m{i + 1}"
|
|
m.depends_on = [f"m{i}"] if i > 0 else []
|
|
return micro
|
|
|
|
|
|
def consolidate_micro_to_major(
|
|
micro_objectives: Sequence[MicroObjective],
|
|
*,
|
|
max_steps: int,
|
|
) -> tuple[List[MajorStep], List[str]]:
|
|
"""Phase B2 — deterministische Konsolidierung auf N Major Steps."""
|
|
if not micro_objectives:
|
|
return [], ["Keine Zwischenziele — Fallback leer"]
|
|
|
|
n = max(2, min(10, int(max_steps)))
|
|
notes: List[str] = []
|
|
if len(micro_objectives) <= n:
|
|
majors = [
|
|
MajorStep(
|
|
index=i,
|
|
phase=m.phase,
|
|
learning_goal=m.title,
|
|
consolidates=[m.id],
|
|
rationale=f"1:1 aus Zwischenziel {m.id}",
|
|
)
|
|
for i, m in enumerate(micro_objectives)
|
|
]
|
|
return majors, notes
|
|
|
|
notes.append(
|
|
f"{len(micro_objectives)} Zwischenziele auf {n} Major Steps reduziert (gleichmäßige Abdeckung des Bogens)."
|
|
)
|
|
count = len(micro_objectives)
|
|
majors: List[MajorStep] = []
|
|
for j in range(n):
|
|
start = (j * count) // n
|
|
end = ((j + 1) * count) // n
|
|
chunk = list(micro_objectives[start:end])
|
|
if not chunk:
|
|
continue
|
|
phase = chunk[len(chunk) // 2].phase
|
|
consolidates = [c.id for c in chunk]
|
|
goal = chunk[0].title if len(chunk) == 1 else f"{chunk[0].title} → {chunk[-1].title}"
|
|
majors.append(
|
|
MajorStep(
|
|
index=len(majors),
|
|
phase=phase,
|
|
learning_goal=goal,
|
|
consolidates=consolidates,
|
|
rationale=f"Konsolidiert {len(chunk)} Zwischenziele ({consolidates[0]}…{consolidates[-1]})",
|
|
)
|
|
)
|
|
for i, step in enumerate(majors):
|
|
step.index = i
|
|
return majors, notes
|
|
|
|
|
|
def _normalize_query(query: Optional[str]) -> str:
|
|
return re.sub(r"\s+", " ", (query or "").strip())
|
|
|
|
|
|
def stage_spec_retrieval_query(
|
|
*,
|
|
semantic_brief: PlanningSemanticBrief,
|
|
goal_query: str,
|
|
stage_spec: StageSpecArtifact,
|
|
major_step: Optional[MajorStep] = None,
|
|
) -> str:
|
|
"""Retrieval-Query für einen Roadmap-Major-Step (Phase F3)."""
|
|
parts: List[str] = []
|
|
topic = (semantic_brief.primary_topic or semantic_brief.retrieval_query or goal_query).strip()
|
|
if topic:
|
|
parts.append(topic)
|
|
learning_goal = (stage_spec.learning_goal or "").strip()
|
|
if learning_goal:
|
|
parts.append(learning_goal)
|
|
phase = (major_step.phase if major_step else "").strip().lower()
|
|
if phase:
|
|
parts.append(phase)
|
|
if stage_spec.load_profile:
|
|
parts.extend(str(x).strip() for x in stage_spec.load_profile[:2] if str(x).strip())
|
|
exercise_type = (stage_spec.exercise_type or "").strip().lower()
|
|
if exercise_type == "partner_drill":
|
|
parts.append("partner")
|
|
elif exercise_type == "kombination":
|
|
parts.append("kombination")
|
|
return _normalize_query(" ".join(parts)) or _normalize_query(goal_query)
|
|
|
|
|
|
def stage_spec_exercise_kind_filter(stage_spec: StageSpecArtifact) -> Optional[List[str]]:
|
|
"""Mappt didaktischen exercise_type auf DB exercise_kind (simple/combination)."""
|
|
et = (stage_spec.exercise_type or "").strip().lower()
|
|
if et == "kombination":
|
|
return ["combination"]
|
|
if et in ("kihon_einzel", "partner_drill", "grundtechnik"):
|
|
return ["simple"]
|
|
return None
|
|
|
|
|
|
def resolve_step_exercise_kind_filter(
|
|
stage_spec: StageSpecArtifact,
|
|
request_filter: Optional[Sequence[str]],
|
|
) -> Optional[List[str]]:
|
|
"""Schnittmenge aus Roadmap-Stufe und optionalem Request-Filter."""
|
|
stage_filter = stage_spec_exercise_kind_filter(stage_spec)
|
|
if not request_filter:
|
|
return stage_filter
|
|
req = [str(x).strip().lower() for x in request_filter if str(x).strip()]
|
|
if not stage_filter:
|
|
return req or None
|
|
merged = [k for k in stage_filter if k in req]
|
|
return merged or req
|
|
|
|
|
|
def build_roadmap_unfilled_gap_specs(
|
|
*,
|
|
unfilled_specs: Sequence[Tuple[int, StageSpecArtifact]],
|
|
major_steps_by_index: Mapping[int, MajorStep],
|
|
steps: Sequence[Mapping[str, Any]],
|
|
brief: PlanningSemanticBrief,
|
|
goal_query: str,
|
|
) -> List[Dict[str, Any]]:
|
|
"""Gap-Fill-Angebote für Roadmap-Stufen ohne Bibliothekstreffer."""
|
|
topic = (brief.primary_topic or "Technik").strip()
|
|
specs: List[Dict[str, Any]] = []
|
|
for roadmap_idx, stage_spec in unfilled_specs:
|
|
major = major_steps_by_index.get(stage_spec.major_step_index)
|
|
phase = (major.phase if major else "vertiefung").strip().lower()
|
|
insert_after = min(max(roadmap_idx - 1, -1), max(len(steps) - 1, -1))
|
|
title_hint = (stage_spec.learning_goal or f"{topic} — {phase}").strip()[:120]
|
|
sketch_parts = [
|
|
f"Planungsziel: {goal_query}",
|
|
f"Roadmap-Stufe {stage_spec.major_step_index + 1} ({phase}): {stage_spec.learning_goal}",
|
|
]
|
|
if stage_spec.success_criteria:
|
|
sketch_parts.append(f"Erfolgskriterien: {', '.join(stage_spec.success_criteria[:3])}")
|
|
specs.append(
|
|
{
|
|
"source": "roadmap_unfilled",
|
|
"insert_after_index": insert_after,
|
|
"gap": {
|
|
"expected_phase": phase,
|
|
"roadmap_major_step_index": stage_spec.major_step_index,
|
|
"learning_goal": stage_spec.learning_goal,
|
|
},
|
|
"phase": phase,
|
|
"title_hint": title_hint,
|
|
"sketch": "\n".join(sketch_parts),
|
|
"rationale": (
|
|
f"Keine passende Bibliotheks-Übung für Roadmap-Stufe "
|
|
f"{stage_spec.major_step_index + 1} ({phase})."
|
|
),
|
|
"roadmap_major_step_index": stage_spec.major_step_index,
|
|
}
|
|
)
|
|
return specs[:5]
|
|
|
|
|
|
def build_stage_specs(
|
|
major_steps: Sequence[MajorStep],
|
|
*,
|
|
goal_analysis: GoalAnalysisArtifact,
|
|
) -> List[StageSpecArtifact]:
|
|
"""Phase C — Stufenspezifikation je Major Step (heuristisch)."""
|
|
topic = goal_analysis.primary_topic or "Technik"
|
|
specs: List[StageSpecArtifact] = []
|
|
for step in major_steps:
|
|
phase = (step.phase or "vertiefung").lower()
|
|
specs.append(
|
|
StageSpecArtifact(
|
|
major_step_index=step.index,
|
|
learning_goal=step.learning_goal,
|
|
load_profile=list(_LOAD_BY_PHASE.get(phase, ["koordination"])),
|
|
exercise_type=_EXERCISE_TYPE_BY_PHASE.get(phase, "kihon_einzel"),
|
|
success_criteria=[
|
|
f"Bezug zu {topic}",
|
|
f"Phase {phase} erkennbar im Übungsziel",
|
|
],
|
|
anti_patterns=[
|
|
"reine Kraftübung ohne Technikbezug",
|
|
f"andere Technik als {topic}" if topic else "themenfremde Übung",
|
|
],
|
|
)
|
|
)
|
|
return specs
|
|
|
|
|
|
def normalize_major_steps_for_override(
|
|
major_steps: Sequence[MajorStep],
|
|
*,
|
|
max_steps: int,
|
|
) -> List[MajorStep]:
|
|
"""Indizes 0…n-1, mindestens 2, höchstens max_steps Major Steps."""
|
|
cleaned: List[MajorStep] = []
|
|
for raw in list(major_steps)[:max_steps]:
|
|
goal = (raw.learning_goal or "").strip()
|
|
phase = (raw.phase or "vertiefung").strip().lower()
|
|
if not goal:
|
|
continue
|
|
cleaned.append(
|
|
MajorStep(
|
|
index=len(cleaned),
|
|
phase=phase,
|
|
learning_goal=goal,
|
|
consolidates=list(raw.consolidates or []),
|
|
rationale=(raw.rationale or "").strip(),
|
|
)
|
|
)
|
|
if len(cleaned) < 2:
|
|
raise ValueError("Mindestens zwei Major Steps mit Lernziel nötig")
|
|
for i, step in enumerate(cleaned):
|
|
step.index = i
|
|
return cleaned
|
|
|
|
|
|
def roadmap_context_from_override(
|
|
goal_query: str,
|
|
*,
|
|
max_steps: int,
|
|
semantic_brief: PlanningSemanticBrief,
|
|
override: RoadmapOverridePayload,
|
|
) -> ProgressionRoadmapContext:
|
|
"""Phase F4: bearbeitete Roadmap → stage_specs → Retrieval (ohne LLM-Roadmap)."""
|
|
majors = normalize_major_steps_for_override(override.major_steps, max_steps=max_steps)
|
|
effective_max = len(majors)
|
|
goal_analysis = build_goal_analysis(goal_query, semantic_brief)
|
|
stage_specs: List[StageSpecArtifact]
|
|
if override.stage_specs and len(override.stage_specs) >= effective_max:
|
|
stage_specs = []
|
|
for i, spec in enumerate(override.stage_specs[:effective_max]):
|
|
stage_specs.append(
|
|
StageSpecArtifact(
|
|
major_step_index=i,
|
|
learning_goal=(spec.learning_goal or majors[i].learning_goal).strip(),
|
|
load_profile=list(spec.load_profile or []),
|
|
exercise_type=(spec.exercise_type or "").strip(),
|
|
success_criteria=list(spec.success_criteria or []),
|
|
anti_patterns=list(spec.anti_patterns or []),
|
|
)
|
|
)
|
|
if not all(s.exercise_type for s in stage_specs):
|
|
rebuilt = build_stage_specs(majors, goal_analysis=goal_analysis)
|
|
for i, spec in enumerate(stage_specs):
|
|
if not spec.exercise_type:
|
|
spec.exercise_type = rebuilt[i].exercise_type
|
|
if not spec.load_profile:
|
|
spec.load_profile = list(rebuilt[i].load_profile)
|
|
else:
|
|
stage_specs = build_stage_specs(majors, goal_analysis=goal_analysis)
|
|
|
|
return ProgressionRoadmapContext(
|
|
goal_query=goal_query.strip(),
|
|
max_steps=effective_max,
|
|
semantic_brief=brief_to_summary_dict(semantic_brief),
|
|
goal_analysis=goal_analysis,
|
|
roadmap=RoadmapArtifact(major_steps=majors),
|
|
stage_specs=stage_specs,
|
|
pipeline_phase="roadmap_v1_edited",
|
|
)
|
|
|
|
|
|
def run_progression_roadmap_pipeline(
|
|
goal_query: str,
|
|
*,
|
|
max_steps: int = 5,
|
|
semantic_brief: Optional[PlanningSemanticBrief] = None,
|
|
cur=None,
|
|
include_llm_roadmap: bool = False,
|
|
) -> ProgressionRoadmapContext:
|
|
"""Workflow-lite: Phase A → B → C. LLM über ai_prompts-Slugs; deterministischer Fallback."""
|
|
brief = semantic_brief or build_semantic_brief(goal_query)
|
|
ctx = ProgressionRoadmapContext(
|
|
goal_query=goal_query.strip(),
|
|
max_steps=max_steps,
|
|
semantic_brief=brief_to_summary_dict(brief),
|
|
)
|
|
|
|
goal_analysis = build_goal_analysis(goal_query, brief)
|
|
if include_llm_roadmap and cur is not None:
|
|
llm_ga, ga_ok = try_llm_goal_analysis(cur, goal_query=goal_query, brief=brief)
|
|
if ga_ok and llm_ga:
|
|
goal_analysis = llm_ga
|
|
ctx.llm_goal_analysis_applied = True
|
|
ctx.prompt_slugs.append(PROMPT_SLUG_GOAL_ANALYSIS)
|
|
ctx.goal_analysis = goal_analysis
|
|
|
|
roadmap: Optional[RoadmapArtifact] = None
|
|
if include_llm_roadmap and cur is not None:
|
|
llm_rm, rm_ok = try_llm_roadmap(
|
|
cur,
|
|
goal_query=goal_query,
|
|
brief=brief,
|
|
goal_analysis=goal_analysis,
|
|
max_steps=max_steps,
|
|
)
|
|
if rm_ok and llm_rm:
|
|
roadmap = llm_rm
|
|
ctx.llm_roadmap_applied = True
|
|
ctx.prompt_slugs.append(PROMPT_SLUG_ROADMAP)
|
|
|
|
if roadmap is None:
|
|
micro = develop_micro_objectives(
|
|
brief,
|
|
goal_analysis=goal_analysis,
|
|
min_count=max(max_steps + 1, 6),
|
|
)
|
|
majors, notes = consolidate_micro_to_major(micro, max_steps=max_steps)
|
|
roadmap = RoadmapArtifact(
|
|
micro_objectives=micro,
|
|
major_steps=majors,
|
|
consolidation_notes=notes,
|
|
)
|
|
ctx.roadmap = roadmap
|
|
|
|
stage_specs = build_stage_specs(roadmap.major_steps, goal_analysis=goal_analysis)
|
|
if include_llm_roadmap and cur is not None:
|
|
llm_specs, spec_ok = try_llm_stage_specs(
|
|
cur,
|
|
goal_query=goal_query,
|
|
goal_analysis=goal_analysis,
|
|
major_steps=roadmap.major_steps,
|
|
)
|
|
if spec_ok and llm_specs:
|
|
stage_specs = llm_specs
|
|
ctx.llm_stage_spec_applied = True
|
|
ctx.prompt_slugs.append(PROMPT_SLUG_STAGE_SPEC)
|
|
ctx.stage_specs = stage_specs
|
|
|
|
if ctx.llm_goal_analysis_applied or ctx.llm_roadmap_applied or ctx.llm_stage_spec_applied:
|
|
ctx.pipeline_phase = "roadmap_v1_llm"
|
|
return ctx
|
|
|
|
|
|
def progression_roadmap_to_api_dict(ctx: ProgressionRoadmapContext) -> Dict[str, Any]:
|
|
return {
|
|
"goal_analysis": ctx.goal_analysis.model_dump() if ctx.goal_analysis else None,
|
|
"roadmap": ctx.roadmap.model_dump() if ctx.roadmap else None,
|
|
"stage_specs": [s.model_dump() for s in ctx.stage_specs],
|
|
"pipeline_phase": ctx.pipeline_phase,
|
|
"major_step_count": len(ctx.roadmap.major_steps) if ctx.roadmap else 0,
|
|
"micro_objective_count": len(ctx.roadmap.micro_objectives) if ctx.roadmap else 0,
|
|
"llm_goal_analysis_applied": ctx.llm_goal_analysis_applied,
|
|
"llm_roadmap_applied": ctx.llm_roadmap_applied,
|
|
"llm_stage_spec_applied": ctx.llm_stage_spec_applied,
|
|
"prompt_slugs": list(ctx.prompt_slugs),
|
|
"prompt_slug_catalog": {
|
|
"goal_analysis": PROMPT_SLUG_GOAL_ANALYSIS,
|
|
"roadmap": PROMPT_SLUG_ROADMAP,
|
|
"stage_spec": PROMPT_SLUG_STAGE_SPEC,
|
|
},
|
|
}
|
|
|
|
|
|
__all__ = [
|
|
"PROMPT_SLUG_GOAL_ANALYSIS",
|
|
"PROMPT_SLUG_ROADMAP",
|
|
"PROMPT_SLUG_STAGE_SPEC",
|
|
"GoalAnalysisArtifact",
|
|
"MajorStep",
|
|
"MicroObjective",
|
|
"ProgressionRoadmapContext",
|
|
"RoadmapArtifact",
|
|
"RoadmapOverridePayload",
|
|
"normalize_major_steps_for_override",
|
|
"roadmap_context_from_override",
|
|
"StageSpecArtifact",
|
|
"build_goal_analysis",
|
|
"build_roadmap_unfilled_gap_specs",
|
|
"build_stage_specs",
|
|
"resolve_step_exercise_kind_filter",
|
|
"stage_spec_exercise_kind_filter",
|
|
"stage_spec_retrieval_query",
|
|
"consolidate_micro_to_major",
|
|
"develop_micro_objectives",
|
|
"progression_roadmap_to_api_dict",
|
|
"run_progression_roadmap_pipeline",
|
|
"try_llm_goal_analysis",
|
|
"try_llm_roadmap",
|
|
"try_llm_stage_specs",
|
|
]
|