Lars
9dd44ce3ca
All checks were successful
Deploy Development / deploy (push) Successful in 45s
Test Suite / pytest-backend (push) Successful in 43s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 14s
Test Suite / k6 /health Baseline (push) Successful in 33s
Test Suite / playwright-tests (push) Successful in 1m14s
- Introduced `RoadmapStructuredInput` to encapsulate structured inputs for start situation, target state, and roadmap notes. - Updated `ProgressionPathSuggestRequest` to include new fields for structured roadmap inputs. - Implemented parsing logic for goal queries to extract start and target states, enhancing the goal analysis process. - Enhanced `build_goal_analysis` to utilize structured inputs, improving the clarity and relevance of generated goals. - Updated the `ExerciseProgressionPathBuilder` component to support new structured input fields, enhancing user experience. - Incremented application version to 0.8.210 to reflect these changes.
952 lines
34 KiB
Python
952 lines
34 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 RoadmapStructuredInput(BaseModel):
|
|
"""Optionale Strukturierung: Start, Ziel, Ergänzungen (Progressionsgraph, kein Gruppen-Tracking)."""
|
|
|
|
start_situation: Optional[str] = Field(default=None, max_length=2000)
|
|
target_state: Optional[str] = Field(default=None, max_length=2000)
|
|
roadmap_notes: Optional[str] = Field(default=None, max_length=2000)
|
|
|
|
|
|
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
|
|
|
|
|
|
_GENERIC_START_MARKER = "Voraussetzungen der Zielgruppe werden im Progressionsgraphen nicht analysiert"
|
|
|
|
|
|
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()
|
|
|
|
|
|
_PHASE_TOPIC_WORDS = frozenset(
|
|
{"einstieg", "grundlage", "vertiefung", "anwendung", "perfektion", "technik"}
|
|
)
|
|
|
|
|
|
def _extract_topic_from_goal_query(goal_query: str, brief: PlanningSemanticBrief) -> str:
|
|
q = (goal_query or "").strip()
|
|
m = re.match(r"^(.+?)\s+von\s+(?:der|die|dem|das|einer?|einem)\s+", q, flags=re.IGNORECASE)
|
|
if m:
|
|
cand = m.group(1).strip().rstrip(".,;")
|
|
if len(cand) >= 3:
|
|
return cand
|
|
topic = _topic_label(brief)
|
|
if topic and topic.lower() not in _PHASE_TOPIC_WORDS and len(topic) >= 4:
|
|
return topic
|
|
return topic or "Technik"
|
|
|
|
|
|
def parse_start_target_from_goal_query(goal_query: str) -> Tuple[Optional[str], Optional[str]]:
|
|
"""„von … bis …“ aus Freitext (z. B. Kumite Beinarbeit von X bis Y)."""
|
|
q = (goal_query or "").strip()
|
|
if not q:
|
|
return None, None
|
|
m = re.search(
|
|
r"\bvon\s+((?:(?:der|die|dem|das|einer?|einem)\s+)?.+?)\s+bis\s+"
|
|
r"(?:zur?|zum|zu der|zu einem)?\s*(.+?)\s*$",
|
|
q,
|
|
flags=re.IGNORECASE | re.DOTALL,
|
|
)
|
|
if not m:
|
|
return None, None
|
|
start = m.group(1).strip().rstrip(".,;")
|
|
target = m.group(2).strip().rstrip(".,;")
|
|
if len(start) < 4 or len(target) < 4:
|
|
return None, None
|
|
return start[:800], target[:800]
|
|
|
|
|
|
def _roadmap_llm_goal_block(
|
|
goal_query: str,
|
|
*,
|
|
structured: Optional[RoadmapStructuredInput] = None,
|
|
parsed_start: Optional[str] = None,
|
|
parsed_target: Optional[str] = None,
|
|
) -> str:
|
|
"""Reicher Kontext für Roadmap-LLM ohne zwingend neue Prompt-Migration."""
|
|
lines = [f"Gesamtanfrage: {(goal_query or '').strip()}"]
|
|
start = (structured.start_situation if structured else None) or parsed_start
|
|
target = (structured.target_state if structured else None) or parsed_target
|
|
notes = structured.roadmap_notes if structured else None
|
|
if start:
|
|
lines.append(f"Ausgangslage/Startpunkt: {start.strip()}")
|
|
if target:
|
|
lines.append(f"Zielzustand: {target.strip()}")
|
|
if notes and notes.strip():
|
|
lines.append(f"Ergänzungen (Fokus, Gruppe, Besonderheiten): {notes.strip()}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def build_goal_analysis(
|
|
goal_query: str,
|
|
brief: PlanningSemanticBrief,
|
|
*,
|
|
structured: Optional[RoadmapStructuredInput] = None,
|
|
) -> GoalAnalysisArtifact:
|
|
"""Phase A — aus Anfrage, optionalen Feldern und Semantic Brief."""
|
|
topic = _extract_topic_from_goal_query(goal_query, brief)
|
|
parsed_start, parsed_target = parse_start_target_from_goal_query(goal_query)
|
|
|
|
start = (structured.start_situation if structured else None) or parsed_start
|
|
target = (structured.target_state if structured else None) or parsed_target
|
|
notes = (structured.roadmap_notes if structured else None) or ""
|
|
|
|
if not target:
|
|
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"
|
|
|
|
if start:
|
|
start_assumption = start.strip()
|
|
else:
|
|
start_assumption = (
|
|
f"Einstieg auf Niveau „{start_phase}“ — {_GENERIC_START_MARKER} "
|
|
"(erst Trainingsplanung)."
|
|
)
|
|
|
|
criteria: List[str] = []
|
|
if brief.must_phrases:
|
|
criteria.extend(brief.must_phrases[:3])
|
|
if topic:
|
|
criteria.append(f"klarer Bezug zu {topic}")
|
|
if start and target:
|
|
criteria.append(f"nachvollziehbarer Übergang von „{start[:80]}“ zu „{target[:80]}“")
|
|
if notes.strip():
|
|
criteria.append(f"Berücksichtigung: {notes.strip()[:200]}")
|
|
|
|
constraints: Dict[str, Any] = {"partner_required": False, "group_analysis": False}
|
|
if notes.strip():
|
|
constraints["trainer_notes"] = notes.strip()[:500]
|
|
|
|
return GoalAnalysisArtifact(
|
|
primary_topic=topic,
|
|
start_assumption=start_assumption,
|
|
target_state=target.strip(),
|
|
success_criteria=criteria or [f"sichere Entwicklung Richtung {target_phase}"],
|
|
constraints=constraints,
|
|
)
|
|
|
|
|
|
def _has_specific_start_target(goal_analysis: GoalAnalysisArtifact) -> bool:
|
|
start = (goal_analysis.start_assumption or "").strip()
|
|
target = (goal_analysis.target_state or "").strip()
|
|
if _GENERIC_START_MARKER in start:
|
|
return False
|
|
return len(start) >= 6 and len(target) >= 6 and start != target
|
|
|
|
|
|
def _target_facets(target: str) -> List[str]:
|
|
parts = re.split(r"\s+und\s+|\s+mit\s+|,\s*", target, flags=re.IGNORECASE)
|
|
out: List[str] = []
|
|
for p in parts:
|
|
s = p.strip().rstrip(".,;")
|
|
if len(s) >= 5 and s.lower() not in {x.lower() for x in out}:
|
|
out.append(s[:200])
|
|
return out[:6]
|
|
|
|
|
|
def develop_micro_objectives_from_start_target(
|
|
goal_analysis: GoalAnalysisArtifact,
|
|
*,
|
|
min_count: int,
|
|
) -> List[MicroObjective]:
|
|
"""Zwischenziele entlang Start → Ziel (heuristisch, themenspezifisch)."""
|
|
topic = goal_analysis.primary_topic or "Technik"
|
|
start = goal_analysis.start_assumption.strip()
|
|
target = goal_analysis.target_state.strip()
|
|
facets = _target_facets(target)
|
|
phases = ["einstieg", "grundlage", "vertiefung", "anwendung", "perfektion"]
|
|
|
|
titles: List[str] = [f"{topic}: Ausgang — {start}"]
|
|
n_middle = max(0, min_count - 2)
|
|
for i in range(n_middle):
|
|
if facets and i < len(facets):
|
|
titles.append(f"{topic}: {facets[i]} — schrittweise einführen")
|
|
else:
|
|
titles.append(
|
|
f"{topic}: Übergangsschritt {i + 1} — Annäherung vom Ausgang zum Ziel"
|
|
)
|
|
titles.append(f"{topic}: Ziel — {target}")
|
|
|
|
while len(titles) < min_count:
|
|
titles.insert(max(1, len(titles) - 1), f"{topic}: Vertiefung vor Zielerreichung")
|
|
|
|
titles = titles[:max(min_count, 2)]
|
|
micro: List[MicroObjective] = []
|
|
for i, title in enumerate(titles):
|
|
phase = phases[min(i, len(phases) - 1)]
|
|
micro.append(
|
|
MicroObjective(
|
|
id=f"m{i + 1}",
|
|
phase=phase,
|
|
title=title,
|
|
weight=0.9 if i in (0, len(titles) - 1) else 0.85,
|
|
depends_on=[f"m{i}"] if i > 0 else [],
|
|
)
|
|
)
|
|
return micro
|
|
|
|
|
|
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 (Start→Ziel oder development_arc-Fallback)."""
|
|
if _has_specific_start_target(goal_analysis):
|
|
return develop_micro_objectives_from_start_target(goal_analysis, min_count=min_count)
|
|
|
|
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,
|
|
structured: Optional[RoadmapStructuredInput] = None,
|
|
) -> 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, structured=structured)
|
|
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 _merge_structured_into_goal_analysis(
|
|
llm_ga: GoalAnalysisArtifact,
|
|
*,
|
|
goal_query: str,
|
|
brief: PlanningSemanticBrief,
|
|
structured: Optional[RoadmapStructuredInput],
|
|
) -> GoalAnalysisArtifact:
|
|
ga_struct = build_goal_analysis(goal_query, brief, structured=structured)
|
|
if not _has_specific_start_target(ga_struct):
|
|
return llm_ga
|
|
merged_criteria = list(
|
|
dict.fromkeys((llm_ga.success_criteria or []) + (ga_struct.success_criteria or []))
|
|
)[:6]
|
|
merged_constraints = {**(llm_ga.constraints or {}), **(ga_struct.constraints or {})}
|
|
return llm_ga.model_copy(
|
|
update={
|
|
"primary_topic": ga_struct.primary_topic or llm_ga.primary_topic,
|
|
"start_assumption": ga_struct.start_assumption,
|
|
"target_state": ga_struct.target_state,
|
|
"success_criteria": merged_criteria,
|
|
"constraints": merged_constraints,
|
|
}
|
|
)
|
|
|
|
|
|
def run_progression_roadmap_pipeline(
|
|
goal_query: str,
|
|
*,
|
|
max_steps: int = 5,
|
|
semantic_brief: Optional[PlanningSemanticBrief] = None,
|
|
cur=None,
|
|
include_llm_roadmap: bool = False,
|
|
structured: Optional[RoadmapStructuredInput] = None,
|
|
) -> ProgressionRoadmapContext:
|
|
"""Workflow-lite: Phase A → B → C. LLM über ai_prompts-Slugs; deterministischer Fallback."""
|
|
brief = semantic_brief or build_semantic_brief(goal_query)
|
|
parsed_start, parsed_target = parse_start_target_from_goal_query(goal_query)
|
|
llm_goal_query = _roadmap_llm_goal_block(
|
|
goal_query,
|
|
structured=structured,
|
|
parsed_start=parsed_start,
|
|
parsed_target=parsed_target,
|
|
)
|
|
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, structured=structured)
|
|
if include_llm_roadmap and cur is not None:
|
|
llm_ga, ga_ok = try_llm_goal_analysis(cur, goal_query=llm_goal_query, brief=brief)
|
|
if ga_ok and llm_ga:
|
|
goal_analysis = _merge_structured_into_goal_analysis(
|
|
llm_ga,
|
|
goal_query=goal_query,
|
|
brief=brief,
|
|
structured=structured,
|
|
)
|
|
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=llm_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=llm_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",
|
|
"RoadmapStructuredInput",
|
|
"normalize_major_steps_for_override",
|
|
"parse_start_target_from_goal_query",
|
|
"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",
|
|
]
|