diff --git a/backend/migrations/090_ai_prompt_stage_transition_states.sql b/backend/migrations/090_ai_prompt_stage_transition_states.sql new file mode 100644 index 0000000..1a04b80 --- /dev/null +++ b/backend/migrations/090_ai_prompt_stage_transition_states.sql @@ -0,0 +1,43 @@ +-- Migration 090: Stufenspecs — start_state / target_state pro Major Step (Soll-Verkettung) + +UPDATE ai_prompts SET + description = 'Phase C: Stufenspezifikation inkl. Soll-Start und Stufen-Ziel je Major Step.', + template = $t$Du bist Assistent für Kampfsport-Trainer und spezifizierst didaktische Stufen eines Progressionsgraphen. + +Anfrage: {{goal_query}} +Zielanalyse: {{goal_analysis_json}} +Major Steps: {{major_steps_json}} +Planungs-Intent (Pfadweite Regeln): {{intent_context_json}} +Semantic Brief: {{semantic_brief_json}} + +Jede Stufe ist ein Übergang im Gesamtpfad: +- start_state: Soll-Zustand zu Beginn (= Ziel der vorherigen Stufe; Stufe 0 = Pfad-Start) +- target_state: Zielzustand nach dieser Stufe (= Soll für die nächste Stufe) +- learning_goal: messbares Lernziel der Übungssuche (was die Übung bringen soll) + +Felder je Major Step: +- load_profile, exercise_type, success_criteria, anti_patterns (wie bisher) + +Regeln: +1. start_state/target_state aus Zielanalyse und Major Steps ableiten — konsistente Kette. +2. explicit_exclusions aus intent_context in anti_patterns jeder Stufe. +3. success_criteria: prüfbar an Kurzbeschreibung + Übungsziel. +4. Keine erfundenen Ausschlüsse. + +Antworte NUR mit JSON: +{ + "stage_specs": [ + { + "major_step_index": 0, + "start_state": "…", + "target_state": "…", + "learning_goal": "…", + "load_profile": ["koordination"], + "exercise_type": "kihon_einzel", + "success_criteria": ["…"], + "anti_patterns": ["…"] + } + ] +}$t$, + default_template = template +WHERE slug = 'planning_progression_stage_spec'; diff --git a/backend/planning_exercise_path_builder.py b/backend/planning_exercise_path_builder.py index 44a1228..9a56cf5 100644 --- a/backend/planning_exercise_path_builder.py +++ b/backend/planning_exercise_path_builder.py @@ -13,6 +13,8 @@ from pydantic import BaseModel, Field from tenant_context import TenantContext, library_content_visibility_sql from planning_exercise_profiles import PlanningTargetProfile +from planning_path_qa_pipeline import run_multistage_path_qa +from planning_stage_context import build_contextualized_stage_goal, resolve_path_start_target from planning_exercise_path_qa import ( apply_llm_path_reorder, build_path_qa_summary, @@ -191,6 +193,8 @@ def _pick_best_path_hit( stage_anti_patterns: Optional[List[str]] = None, roadmap_stage_match: bool = False, stage_match_brief: Optional[PlanningSemanticBrief] = None, + path_primary_topic: Optional[str] = None, + path_technique_excludes: Optional[List[str]] = None, ) -> Optional[Dict[str, Any]]: return pick_best_path_hit( hits, @@ -200,6 +204,8 @@ def _pick_best_path_hit( stage_anti_patterns=stage_anti_patterns, roadmap_stage_match=roadmap_stage_match, stage_match_brief=stage_match_brief, + path_primary_topic=path_primary_topic, + path_technique_excludes=path_technique_excludes, ) @@ -304,6 +310,8 @@ def _run_path_step_retrieval( stage_success_criteria: Optional[List[str]] = None, stage_load_profile: Optional[List[str]] = None, path_context_note: Optional[str] = None, + path_primary_topic: Optional[str] = None, + path_technique_excludes: Optional[List[str]] = None, ) -> Tuple[List[Dict[str, Any]], PlanningTargetProfile, Dict[str, Any], str]: step_query = step_query_override or step_retrieval_query( semantic_brief, goal_query, step_index, max_steps @@ -346,6 +354,8 @@ def _run_path_step_retrieval( "stage_success_criteria": list(stage_success_criteria or []), "stage_load_profile": list(stage_load_profile or []), "path_context_note": (path_context_note or "").strip() or None, + "path_primary_topic": (path_primary_topic or "").strip() or None, + "path_technique_excludes": list(path_technique_excludes or []), } pack = apply_progression_context_to_pack( cur, @@ -514,6 +524,10 @@ def _annotate_roadmap_step( anti = list(anti_patterns_override or stage_spec.anti_patterns or []) if anti: step["roadmap_anti_patterns"] = anti + if (stage_spec.start_state or "").strip(): + step["roadmap_start_state"] = stage_spec.start_state.strip() + if (stage_spec.target_state or "").strip(): + step["roadmap_target_state"] = stage_spec.target_state.strip() step["roadmap_match_source"] = "stage_spec" if skill_expectations: step["skill_expectations"] = skill_expectations @@ -562,6 +576,11 @@ def _build_steps_roadmap_first( if roadmap_ctx.resolved_structured else None ) + path_start, path_target = resolve_path_start_target( + structured=roadmap_ctx.resolved_structured, + goal_analysis=roadmap_ctx.goal_analysis, + ) + stage_count = len(stage_specs) brief_summary = ( roadmap_ctx.semantic_brief if roadmap_ctx.semantic_brief @@ -593,6 +612,17 @@ def _build_steps_roadmap_first( ) step_kind = resolve_step_exercise_kind_filter(stage_spec, body.exercise_kind_any) stage_goal = (stage_spec.learning_goal or "").strip() + stage_start = (stage_spec.start_state or "").strip() + stage_target = (stage_spec.target_state or "").strip() + contextual_goal = build_contextualized_stage_goal( + learning_goal=stage_goal, + start_state=stage_start, + target_state=stage_target, + path_target_state=path_target, + path_start_state=path_start, + stage_index=step_index, + stage_count=stage_count, + ) path_context_note = None if rs_dump: ctx_parts = [ @@ -607,6 +637,14 @@ def _build_steps_roadmap_first( extra_context=path_context_note, ) stage_anti = list(dict.fromkeys([*(stage_spec.anti_patterns or []), *path_anti])) + path_primary = (semantic_brief.primary_topic or "").strip() + path_tech_excludes = list(semantic_brief.exclude_phrases or []) + if semantic_brief.topic_type == "technique" and path_primary: + from planning_exercise_semantics import technique_sibling_excludes + + for item in technique_sibling_excludes(path_primary): + if item not in path_tech_excludes: + path_tech_excludes.append(item) stage_match_brief = build_stage_match_brief( learning_goal=stage_goal, anti_patterns=stage_anti, @@ -615,6 +653,12 @@ def _build_steps_roadmap_first( phase=major.phase if major else None, path_context_note=path_context_note, path_anti_patterns=path_anti, + path_primary_topic=path_primary or None, + path_technique_excludes=path_tech_excludes or None, + stage_start_state=stage_start or None, + stage_target_state=stage_target or None, + path_target_state=path_target or None, + contextualized_learning_goal=contextual_goal or None, ) hits, _, _, _ = _run_path_step_retrieval( @@ -641,6 +685,8 @@ def _build_steps_roadmap_first( stage_success_criteria=list(stage_spec.success_criteria or []), stage_load_profile=list(stage_spec.load_profile or []), path_context_note=path_context_note, + path_primary_topic=path_primary or None, + path_technique_excludes=path_tech_excludes or None, ) hit = _pick_best_path_hit( @@ -651,6 +697,8 @@ def _build_steps_roadmap_first( stage_anti_patterns=stage_anti or None, roadmap_stage_match=True, stage_match_brief=stage_match_brief, + path_primary_topic=path_primary or None, + path_technique_excludes=path_tech_excludes or None, ) if not hit: @@ -878,6 +926,13 @@ def _run_evaluate_only_path_qa( roadmap_snapshot=path_roadmap_snapshot, ) + multistage_qa = run_multistage_path_qa( + off_topic_steps=off_topic_steps, + stripped_off_topic=stripped_off_topic, + gaps=gaps, + llm_qa=llm_qa, + llm_applied=llm_qa_applied, + ) path_qa = build_path_qa_summary( gaps=gaps, bridge_inserts=bridge_inserts, @@ -890,6 +945,7 @@ def _run_evaluate_only_path_qa( reorder_applied=False, reorder_notes=[], roadmap_qa_mode=roadmap_qa_mode, + multistage_qa=multistage_qa, ) return { "path_qa": path_qa, @@ -1318,6 +1374,14 @@ def suggest_progression_path( if offer.get("offer_id") not in seen_offer_ids: gap_fill_offers.append(offer) + multistage_qa = run_multistage_path_qa( + off_topic_steps=off_topic_steps, + stripped_off_topic=stripped_off_topic, + gaps=gaps, + llm_qa=llm_qa, + llm_applied=llm_qa_applied, + roadmap_unfilled=roadmap_unfilled if roadmap_first else None, + ) path_qa = build_path_qa_summary( gaps=gaps, bridge_inserts=bridge_inserts, @@ -1330,6 +1394,7 @@ def suggest_progression_path( reorder_applied=reorder_applied, reorder_notes=reorder_notes, roadmap_qa_mode=roadmap_qa_mode, + multistage_qa=multistage_qa, ) target_profile_summary = path_target_profile.to_summary_dict(cur) diff --git a/backend/planning_exercise_path_qa.py b/backend/planning_exercise_path_qa.py index de7a896..4ce77ba 100644 --- a/backend/planning_exercise_path_qa.py +++ b/backend/planning_exercise_path_qa.py @@ -23,10 +23,12 @@ from planning_exercise_semantics import ( brief_to_summary_dict, exercise_passes_path_semantic_gate, exercise_passes_stage_learning_goal_gate, + exercise_passes_technique_path_scope, resolve_path_anti_patterns, score_exercise_semantic_relevance, semantic_brief_for_stage, step_phase_for_index, + technique_sibling_excludes, ) _logger = logging.getLogger("shinkan.planning_exercise_path_qa") @@ -434,6 +436,31 @@ def detect_off_topic_steps( } ) continue + primary = (brief.primary_topic or "").strip() + if brief.topic_type == "technique" and primary: + siblings = technique_sibling_excludes(primary) + stage_goal_pre = (step.get("roadmap_learning_goal") or "").strip() + if not exercise_passes_technique_path_scope( + primary_topic=primary, + title=bundle["title"], + summary=bundle["summary"], + goal=bundle["goal"], + learning_goal=stage_goal_pre, + sibling_excludes=siblings, + relaxed=False, + ): + off_topic.append( + { + "step_index": idx, + "exercise_id": int(step["exercise_id"]), + "title": step.get("title") or bundle["title"], + "semantic_score": 0.0, + "expected_phase": (step.get("roadmap_phase") or "").strip().lower() or None, + "issue": "technique_scope", + "reasons": [f"Passt nicht zur Haupttechnik „{primary}“"], + } + ) + continue stage_goal = (step.get("roadmap_learning_goal") or "").strip() phase = (step.get("roadmap_phase") or "").strip().lower() or step_phase_for_index( brief, idx, total @@ -599,6 +626,7 @@ def build_path_qa_summary( reorder_applied: bool = False, reorder_notes: Optional[Sequence[str]] = None, roadmap_qa_mode: Optional[str] = None, + multistage_qa: Optional[Mapping[str, Any]] = None, ) -> Dict[str, Any]: offers = list(gap_fill_offers or []) off_topic = list(off_topic_steps or []) @@ -619,6 +647,10 @@ def build_path_qa_summary( "reorder_notes": list(reorder_notes or []), "roadmap_qa_mode": roadmap_qa_mode, } + if multistage_qa: + summary["qa_tiers"] = list(multistage_qa.get("qa_tiers") or []) + summary["optimization_hints"] = list(multistage_qa.get("optimization_hints") or []) + summary["optimization_hint_count"] = int(multistage_qa.get("optimization_hint_count") or 0) if llm_qa: summary["overall_ok"] = bool(llm_qa.get("overall_ok", True)) summary["quality_score"] = llm_qa.get("quality_score") diff --git a/backend/planning_exercise_retrieval.py b/backend/planning_exercise_retrieval.py index 3051518..6b748b1 100644 --- a/backend/planning_exercise_retrieval.py +++ b/backend/planning_exercise_retrieval.py @@ -351,6 +351,8 @@ def rank_visible_library_hits( stage_semantic_score=stage_semantic_score, anti_patterns=pack.get("stage_anti_patterns"), step_phase=step_phase, + path_primary_topic=pack.get("path_primary_topic"), + path_technique_excludes=pack.get("path_technique_excludes"), ): score_penalty = max(0.0, score_penalty - 0.10) stage_match_reason = "Passt zum Stufen-Lernziel" diff --git a/backend/planning_exercise_semantics.py b/backend/planning_exercise_semantics.py index 8c29c06..38798e2 100644 --- a/backend/planning_exercise_semantics.py +++ b/backend/planning_exercise_semantics.py @@ -180,6 +180,51 @@ def _find_technique_in_text(q_lower: str) -> Optional[Tuple[str, Tuple[str, ...] return None +def technique_sibling_excludes(primary_topic: str) -> List[str]: + """Andere Techniken derselben Familie (z. B. Mae/Yoko bei Mawashi) — aus Katalog.""" + topic = _normalize_phrase(primary_topic) + if not topic: + return [] + hit = _find_technique_in_text(topic) + if not hit: + return [] + out: List[str] = [] + for raw in hit[1]: + for expanded in _expand_stage_exclude_phrase(raw): + if expanded and expanded not in out: + out.append(expanded) + return out[:16] + + +def exercise_passes_technique_path_scope( + *, + primary_topic: str, + title: str, + summary: str = "", + goal: str = "", + learning_goal: str = "", + sibling_excludes: Optional[Sequence[str]] = None, + relaxed: bool = False, +) -> bool: + """ + Technik-Pfad: keine Geschwister-Technik; Haupttechnik im Übungstext oder Stufen-Lernziel. + """ + primary = _normalize_phrase(primary_topic) + if not primary: + return True + + blob = _blob_from_fields(title, summary, goal, []) + excludes = list(sibling_excludes or technique_sibling_excludes(primary)) + if excludes and _blob_matches_stage_excludes(blob, excludes): + return False + + in_exercise = _phrase_in_blob(primary, blob) + in_stage = _phrase_in_blob(primary, _blob_from_fields("", "", learning_goal, [])) + if in_exercise or in_stage: + return True + return relaxed + + def _detect_development_arc(q_lower: str) -> List[str]: found: List[str] = [] for phase, markers in _ARC_PHASES: @@ -850,13 +895,19 @@ def build_stage_match_brief( phase: Optional[str] = None, path_context_note: Optional[str] = None, path_anti_patterns: Optional[Sequence[str]] = None, + path_primary_topic: Optional[str] = None, + path_technique_excludes: Optional[Sequence[str]] = None, + stage_start_state: Optional[str] = None, + stage_target_state: Optional[str] = None, + path_target_state: Optional[str] = None, + contextualized_learning_goal: Optional[str] = None, ) -> PlanningSemanticBrief: """ Stufen-zentrierter Semantik-Brief — unabhängig vom Gesamt-Pfad-Thema. Primär für Roadmap-Match: Bewertung gegen Titel + Kurzbeschreibung + Übungsziel. """ - lg = (learning_goal or "").strip() + lg = (contextualized_learning_goal or learning_goal or "").strip() if len(lg) < 3: return PlanningSemanticBrief(semantic_strength=0.0) @@ -865,9 +916,20 @@ def build_stage_match_brief( s = str(raw or "").strip() if s and s not in merged_anti: merged_anti.append(s) + primary_path = _normalize_phrase(path_primary_topic or "") + if primary_path: + for item in technique_sibling_excludes(primary_path): + if item not in merged_anti: + merged_anti.append(item) + for raw in path_technique_excludes or []: + for expanded in _expand_stage_exclude_phrase(str(raw or "")): + if expanded and expanded not in merged_anti: + merged_anti.append(expanded) constraints = parse_stage_goal_constraints(lg, merged_anti) must: List[str] = [] norm_lg = _normalize_phrase(lg) + if primary_path and primary_path not in must: + must.insert(0, primary_path[:120]) for token in constraints.positive_tokens: if token not in must: must.append(token) @@ -883,6 +945,10 @@ def build_stage_match_brief( must.append(s[:60]) retrieval_parts = [norm_lg] + for raw in (stage_start_state, stage_target_state, path_target_state): + s = _normalize_phrase(str(raw or ""))[:200] + if s and s not in retrieval_parts: + retrieval_parts.append(s) if path_context_note: note = _normalize_phrase(path_context_note)[:200] if note: @@ -950,12 +1016,14 @@ def exercise_passes_stage_fit( stage_semantic_score: Optional[float] = None, anti_patterns: Optional[Sequence[str]] = None, step_phase: Optional[str] = None, + path_primary_topic: Optional[str] = None, + path_technique_excludes: Optional[Sequence[str]] = None, min_stage_semantic: float = _MIN_STAGE_FIT_SEMANTIC, relaxed: bool = False, ) -> bool: """Allgemeines Stufen-Fit-Gate: voller Übungstext vs. Stufen-Brief.""" lg = (learning_goal or "").strip() - if len(lg) < 3: + if len(lg) < 3 and not (path_primary_topic or "").strip(): return True blob = _blob_from_fields(title, summary, goal, []) @@ -963,6 +1031,18 @@ def exercise_passes_stage_fit( if constraints.exclude_phrases and _blob_matches_stage_excludes(blob, constraints.exclude_phrases): return False + primary_path = (path_primary_topic or "").strip() + if primary_path and not exercise_passes_technique_path_scope( + primary_topic=primary_path, + title=title, + summary=summary, + goal=goal, + learning_goal=lg, + sibling_excludes=path_technique_excludes, + relaxed=relaxed, + ): + return False + brief = stage_brief or build_stage_match_brief( learning_goal=lg, anti_patterns=anti_patterns, @@ -1102,6 +1182,8 @@ def pick_best_path_hit( stage_anti_patterns: Optional[Sequence[str]] = None, roadmap_stage_match: bool = False, stage_match_brief: Optional[PlanningSemanticBrief] = None, + path_primary_topic: Optional[str] = None, + path_technique_excludes: Optional[Sequence[str]] = None, ) -> Optional[Dict[str, Any]]: """Gestufte Auswahl: strikt → relaxed → optional Notfall-Fallback.""" if not hits: @@ -1138,6 +1220,8 @@ def pick_best_path_hit( stage_brief=stage_brief, stage_semantic_score=stage_sem, anti_patterns=stage_anti_patterns, + path_primary_topic=path_primary_topic, + path_technique_excludes=path_technique_excludes, relaxed=not strict, ): continue @@ -1209,8 +1293,10 @@ __all__ = [ "merge_semantic_brief_llm", "parse_stage_goal_constraints", "pick_best_path_hit", + "exercise_passes_technique_path_scope", "score_exercise_stage_fit", "semantic_brief_for_stage", + "technique_sibling_excludes", "resolve_semantic_skill_weights", "score_exercise_semantic_relevance", "semantic_core_phrases", diff --git a/backend/planning_intent_context.py b/backend/planning_intent_context.py index 89dd5e2..ce6e774 100644 --- a/backend/planning_intent_context.py +++ b/backend/planning_intent_context.py @@ -14,6 +14,7 @@ from typing import Any, Dict, List, Mapping, Optional, Sequence from planning_exercise_semantics import ( PlanningSemanticBrief, resolve_path_anti_patterns, + technique_sibling_excludes, ) _NEGATION_CLAUSE_RE = re.compile( @@ -46,14 +47,18 @@ class PlanningIntentContext: path_success_criteria: List[str] = field(default_factory=list) explicit_exclusions: List[str] = field(default_factory=list) context_notes: str = "" + topic_type: str = "general" + technique_sibling_excludes: List[str] = field(default_factory=list) def to_api_dict(self) -> Dict[str, Any]: return { "source_query": self.source_query, "primary_topic": self.primary_topic, + "topic_type": self.topic_type, "path_anti_patterns": self.path_anti_patterns[:16], "path_success_criteria": self.path_success_criteria[:10], "explicit_exclusions": self.explicit_exclusions[:10], + "technique_sibling_excludes": self.technique_sibling_excludes[:16], "context_notes": self.context_notes[:1200] or None, } @@ -101,15 +106,32 @@ def build_planning_intent_context( path_success.append(line) topic = (primary_topic or ga.get("primary_topic") or "").strip() - if semantic_brief and not topic: - topic = (semantic_brief.primary_topic or "").strip() + topic_type = "general" + siblings: List[str] = [] + if semantic_brief: + if not topic: + topic = (semantic_brief.primary_topic or "").strip() + topic_type = (semantic_brief.topic_type or "general").strip().lower() + if topic_type == "technique" and topic: + siblings = technique_sibling_excludes(topic) + for raw in semantic_brief.exclude_phrases or []: + s = str(raw or "").strip() + if s and s.lower() not in {x.lower() for x in siblings}: + siblings.append(s[:120]) + + if topic_type == "technique" and topic: + line = f"Haupttechnik {topic} in Kurzbeschreibung oder Übungsziel erkennbar" + if line not in path_success: + path_success.insert(0, line) return PlanningIntentContext( source_query=(goal_query or "").strip(), primary_topic=topic, + topic_type=topic_type, path_anti_patterns=path_anti, path_success_criteria=path_success, explicit_exclusions=explicit, + technique_sibling_excludes=siblings[:16], context_notes=combined_notes[:1200], ) @@ -148,13 +170,23 @@ def finalize_stage_spec_artifact( *(spec.anti_patterns or []), *intent.explicit_exclusions, *intent.path_anti_patterns, + *intent.technique_sibling_excludes, + ( + f"andere Technik als {intent.primary_topic}" + if intent.topic_type == "technique" and intent.primary_topic + else "" + ), ], limit=14, ) + stage_start = (spec.start_state or "").strip() + stage_target = (spec.target_state or "").strip() success = _dedupe_preserve( [ *(spec.success_criteria or []), *intent.path_success_criteria, + (f"Soll-Start der Stufe erreichbar: {stage_start[:180]}" if stage_start else ""), + (f"Stufen-Ziel erreichbar: {stage_target[:180]}" if stage_target else ""), ( f"Übung liefert messbar: {learning_goal[:160]}" if learning_goal diff --git a/backend/planning_path_qa_pipeline.py b/backend/planning_path_qa_pipeline.py new file mode 100644 index 0000000..2073880 --- /dev/null +++ b/backend/planning_path_qa_pipeline.py @@ -0,0 +1,176 @@ +""" +Mehrstufige Pfad-QS — Findings pro Stufe, daraus Optimierungspotenziale ableiten. + +Stufen (allgemein, domänenneutral): + 1. deterministische Gates (Technik-Scope, Ausschlüsse, Stufen-Fit) + 2. Übergangs-Kohärenz (Lücken zwischen Schritten) + 3. LLM-Ganzpfad-Bewertung (Empfehlungen, keine Auto-Patches) +""" +from __future__ import annotations + +from typing import Any, Dict, List, Mapping, Optional, Sequence + + +_ACTION_BY_ISSUE: Dict[str, str] = { + "technique_scope": "rematch_slot", + "path_exclude": "rematch_slot", + "stage_mismatch": "refine_stage_spec", + "off_topic": "rematch_slot", + "gap": "bridge_or_gap_fill", + "large_gap": "bridge_or_gap_fill", + "roadmap_unfilled": "rematch_slot", +} + + +def _action_for_finding(finding: Mapping[str, Any]) -> str: + issue = str(finding.get("issue") or finding.get("type") or "").strip().lower() + if finding.get("is_large_gap"): + return "bridge_or_gap_fill" + return _ACTION_BY_ISSUE.get(issue, "review") + + +def _hint_from_finding(finding: Mapping[str, Any], *, tier: str) -> Dict[str, Any]: + step_index = finding.get("step_index") + if step_index is None: + step_index = finding.get("major_step_index") + issue = str(finding.get("issue") or finding.get("type") or tier) + action = _action_for_finding(finding) + title = str(finding.get("title") or finding.get("removed_title") or "").strip() + reasons = finding.get("reasons") or [] + reason = reasons[0] if reasons else str(finding.get("rationale") or finding.get("detail") or "") + + hint: Dict[str, Any] = { + "tier": tier, + "action": action, + "issue": issue, + "step_index": step_index, + "title": title or None, + "reason": (reason or "")[:400] or None, + } + if finding.get("roadmap_learning_goal"): + hint["roadmap_learning_goal"] = finding.get("roadmap_learning_goal") + if finding.get("roadmap_major_step_index") is not None: + hint["roadmap_major_step_index"] = finding.get("roadmap_major_step_index") + return {k: v for k, v in hint.items() if v is not None and v != ""} + + +def derive_optimization_hints( + tiers: Sequence[Mapping[str, Any]], +) -> List[Dict[str, Any]]: + """Aus QS-Stufen konkrete Optimierungsaktionen (ohne anfrage-spezifische Heuristiken).""" + hints: List[Dict[str, Any]] = [] + seen: set[tuple] = set() + for tier in tiers: + tier_id = str(tier.get("id") or "") + for finding in tier.get("findings") or []: + if not isinstance(finding, dict): + continue + hint = _hint_from_finding(finding, tier=tier_id) + key = ( + hint.get("tier"), + hint.get("action"), + hint.get("step_index"), + hint.get("issue"), + ) + if key in seen: + continue + seen.add(key) + hints.append(hint) + return hints[:24] + + +def run_multistage_path_qa( + *, + off_topic_steps: Sequence[Mapping[str, Any]], + stripped_off_topic: Sequence[Mapping[str, Any]], + gaps: Sequence[Mapping[str, Any]], + llm_qa: Optional[Mapping[str, Any]] = None, + llm_applied: bool = False, + roadmap_unfilled: Optional[Sequence[Mapping[str, Any]]] = None, +) -> Dict[str, Any]: + """Orchestriert QS-Stufen und leitet Optimierungspotenziale ab.""" + tier1_findings: List[Dict[str, Any]] = [] + for item in stripped_off_topic or off_topic_steps or []: + if isinstance(item, dict): + tier1_findings.append(dict(item)) + + tier2_findings: List[Dict[str, Any]] = [dict(g) for g in gaps if isinstance(g, dict)] + + tier3_findings: List[Dict[str, Any]] = [] + llm_recommendations: List[str] = [] + if llm_applied and llm_qa: + q_score = llm_qa.get("quality_score") + tier3_findings.append( + { + "issue": "llm_assessment", + "quality_score": q_score, + "overall_ok": llm_qa.get("overall_ok"), + "detail": llm_qa.get("summary") or llm_qa.get("assessment"), + } + ) + for raw in llm_qa.get("recommendations") or llm_qa.get("suggestions") or []: + s = str(raw or "").strip() + if s: + llm_recommendations.append(s[:500]) + + unfilled = list(roadmap_unfilled or []) + if unfilled: + for item in unfilled: + if isinstance(item, (list, tuple)) and len(item) >= 2: + idx, spec = item[0], item[1] + tier1_findings.append( + { + "issue": "roadmap_unfilled", + "step_index": int(idx), + "roadmap_major_step_index": getattr(spec, "major_step_index", idx), + "roadmap_learning_goal": getattr(spec, "learning_goal", None), + "reasons": ["Keine passende Übung für Roadmap-Stufe"], + } + ) + elif isinstance(item, dict): + tier1_findings.append({**item, "issue": item.get("issue") or "roadmap_unfilled"}) + + tiers: List[Dict[str, Any]] = [ + { + "id": "tier1_deterministic", + "label": "Deterministische Gates", + "finding_count": len(tier1_findings), + "findings": tier1_findings[:16], + }, + { + "id": "tier2_transitions", + "label": "Übergangs-Kohärenz", + "finding_count": len(tier2_findings), + "findings": tier2_findings[:12], + }, + { + "id": "tier3_llm_holistic", + "label": "LLM-Ganzpfad", + "finding_count": len(tier3_findings), + "findings": tier3_findings, + "recommendations": llm_recommendations[:8], + "applied": bool(llm_applied), + }, + ] + optimization_hints = derive_optimization_hints(tiers) + for rec in llm_recommendations[:5]: + optimization_hints.append( + { + "tier": "tier3_llm_holistic", + "action": "review_roadmap", + "issue": "llm_recommendation", + "reason": rec, + } + ) + + return { + "qa_tiers": tiers, + "optimization_hints": optimization_hints[:28], + "optimization_hint_count": len(optimization_hints), + } + + +__all__ = [ + "derive_optimization_hints", + "run_multistage_path_qa", +] diff --git a/backend/planning_progression_roadmap.py b/backend/planning_progression_roadmap.py index 7881b55..5d95e7f 100644 --- a/backend/planning_progression_roadmap.py +++ b/backend/planning_progression_roadmap.py @@ -104,6 +104,10 @@ class RoadmapArtifact(BaseModel): class StageSpecArtifact(BaseModel): major_step_index: int = Field(ge=0) learning_goal: str = "" + """Soll-Start dieser Stufe (= Zielzustand der vorherigen Stufe / Pfad-Start).""" + start_state: str = "" + """Zielzustand dieser Stufe (= Soll für den nächsten Schritt).""" + target_state: str = "" load_profile: List[str] = Field(default_factory=list) exercise_type: str = "" success_criteria: List[str] = Field(default_factory=list) @@ -941,6 +945,8 @@ def roadmap_context_from_override( StageSpecArtifact( major_step_index=i, learning_goal=(spec.learning_goal or majors[i].learning_goal).strip(), + start_state=(spec.start_state or "").strip(), + target_state=(spec.target_state or "").strip(), load_profile=list(spec.load_profile or []), exercise_type=(spec.exercise_type or "").strip(), success_criteria=list(spec.success_criteria or []), @@ -1004,6 +1010,19 @@ def roadmap_context_from_override( semantic_brief=enriched_brief, ), ) + from planning_stage_context import derive_stage_specs_transition_states, resolve_path_start_target + + path_start, path_target = resolve_path_start_target( + structured=structured, + goal_analysis=goal_analysis, + ) + stage_specs = derive_stage_specs_transition_states( + stage_specs, + majors, + path_start=path_start, + path_target=path_target, + goal_analysis=goal_analysis, + ) return ProgressionRoadmapContext( goal_query=goal_query.strip(), @@ -1225,6 +1244,19 @@ def run_progression_roadmap_pipeline( intent=intent, fallback_specs=heuristic_specs, ) + from planning_stage_context import derive_stage_specs_transition_states, resolve_path_start_target + + path_start, path_target = resolve_path_start_target( + structured=resolved, + goal_analysis=goal_analysis, + ) + ctx.stage_specs = derive_stage_specs_transition_states( + ctx.stage_specs, + roadmap.major_steps, + path_start=path_start, + path_target=path_target, + goal_analysis=goal_analysis, + ) if ctx.llm_goal_analysis_applied or ctx.llm_roadmap_applied or ctx.llm_stage_spec_applied: ctx.pipeline_phase = "roadmap_v1_llm" diff --git a/backend/planning_stage_context.py b/backend/planning_stage_context.py new file mode 100644 index 0000000..0ed2aa8 --- /dev/null +++ b/backend/planning_stage_context.py @@ -0,0 +1,140 @@ +""" +Stufen-Kontext im Gesamtziel — Start/Ziel pro Roadmap-Stufe für Matching und QS. + +Übertragbar auf Trainingsplanung: Abschnitt-Soll (= Ende Vorabschnitt), Abschnitt-Ziel. +""" +from __future__ import annotations + +from typing import List, Optional, Sequence + +from planning_progression_roadmap import ( + GoalAnalysisArtifact, + MajorStep, + RoadmapStructuredInput, + StageSpecArtifact, +) + + +def build_contextualized_stage_goal( + *, + learning_goal: str, + start_state: str = "", + target_state: str = "", + path_target_state: str = "", + path_start_state: str = "", + stage_index: int = 0, + stage_count: int = 1, +) -> str: + """Stufen-Lernziel eingebettet in Übergang und Gesamtziel (für Brief/Retrieval).""" + lg = (learning_goal or "").strip() + if not lg: + return "" + + parts: List[str] = [] + start = (start_state or "").strip() + target = (target_state or "").strip() + path_end = (path_target_state or "").strip() + path_begin = (path_start_state or "").strip() + + if start: + parts.append(f"Soll-Start: {start[:220]}") + elif path_begin and stage_index == 0: + parts.append(f"Pfad-Start: {path_begin[:220]}") + if target: + parts.append(f"Stufen-Ziel: {target[:220]}") + parts.append(f"Lernziel: {lg[:280]}") + if path_end: + if stage_index >= max(0, stage_count - 1): + parts.append(f"Gesamtziel: {path_end[:220]}") + else: + parts.append(f"Gesamtziel (Kontext): {path_end[:180]}") + + return " | ".join(parts)[:900] + + +def derive_stage_specs_transition_states( + stage_specs: Sequence[StageSpecArtifact], + major_steps: Sequence[MajorStep], + *, + path_start: str = "", + path_target: str = "", + goal_analysis: Optional[GoalAnalysisArtifact] = None, +) -> List[StageSpecArtifact]: + """ + Verkettete Soll-/Zielzustände je Stufe. + + - Stufe 0 start = Pfad-Start + - Stufe n start = Zielzustand Stufe n-1 (Ziel des vorherigen Schritts) + - Letzte Stufe target = Pfad-Gesamtziel (falls gesetzt) + """ + start_path = (path_start or "").strip() + end_path = (path_target or "").strip() + if goal_analysis: + if not start_path: + start_path = (goal_analysis.start_assumption or "").strip() + if not end_path: + end_path = (goal_analysis.target_state or "").strip() + + by_idx = {int(s.major_step_index): s for s in stage_specs} + majors = sorted(major_steps, key=lambda m: m.index) + if not majors: + return list(stage_specs) + + out: List[StageSpecArtifact] = [] + prev_target = start_path + last_idx = majors[-1].index + + for major in majors: + spec = by_idx.get(major.index) + if spec is None: + spec = StageSpecArtifact( + major_step_index=major.index, + learning_goal=major.learning_goal, + ) + + explicit_start = (spec.start_state or "").strip() + explicit_target = (spec.target_state or "").strip() + stage_start = explicit_start or prev_target or start_path + if explicit_target: + stage_target = explicit_target + elif major.index == last_idx and end_path: + stage_target = end_path + else: + stage_target = (major.learning_goal or spec.learning_goal or "").strip() + + prev_target = stage_target + out.append( + spec.model_copy( + update={ + "start_state": (stage_start or "")[:500], + "target_state": (stage_target or "")[:500], + } + ) + ) + return out + + +def resolve_path_start_target( + *, + structured: Optional[RoadmapStructuredInput] = None, + goal_analysis: Optional[GoalAnalysisArtifact] = None, +) -> tuple[str, str]: + """Pfadweiter Start- und Zielzustand für Stufen-Verkettung.""" + start = "" + target = "" + if structured: + start = (structured.start_situation or "").strip() + target = (structured.target_state or "").strip() + if goal_analysis: + if not start: + start = (goal_analysis.start_assumption or "").strip() + if not target: + target = (goal_analysis.target_state or "").strip() + return start, target + + +__all__ = [ + "build_contextualized_stage_goal", + "derive_stage_specs_transition_states", + "resolve_path_start_target", +] diff --git a/backend/tests/test_planning_roadmap_stage_match.py b/backend/tests/test_planning_roadmap_stage_match.py index 9077640..0a9f97d 100644 --- a/backend/tests/test_planning_roadmap_stage_match.py +++ b/backend/tests/test_planning_roadmap_stage_match.py @@ -5,10 +5,12 @@ from planning_exercise_semantics import ( enrich_brief_with_path_constraints, exercise_passes_stage_learning_goal_gate, exercise_passes_stage_fit, + exercise_passes_technique_path_scope, pick_best_path_hit, resolve_path_anti_patterns, score_exercise_stage_fit, semantic_brief_for_stage, + technique_sibling_excludes, ) from planning_exercise_path_qa import strip_off_topic_steps_from_path @@ -247,6 +249,47 @@ def test_pick_best_skips_kumite_for_mawashi_athletic_path(): assert int(chosen["id"]) == 2 +def test_technique_scope_rejects_sibling_geri_for_mawashi_path(): + siblings = technique_sibling_excludes("mawashi geri") + assert any("mae" in s for s in siblings) + assert not exercise_passes_technique_path_scope( + primary_topic="mawashi geri", + title="Mae Geri Grundtechnik", + summary="Front kick", + goal="Präzision Mae Geri", + learning_goal="Sprungvorbereitung für Mawashi Geri", + sibling_excludes=siblings, + ) + assert exercise_passes_technique_path_scope( + primary_topic="mawashi geri", + title="Sprungkraft Plyometrie", + summary="Absprung", + goal="Vorbereitung gesprungener Mawashi Geri", + learning_goal="Sprungvorbereitung für Mawashi Geri", + sibling_excludes=siblings, + ) + + +def test_stage_fit_rejects_yoko_geri_on_mawashi_roadmap_stage(): + brief = build_semantic_brief("gesprungener Mawashi Geri Sprungphase") + primary = brief.primary_topic or "mawashi geri" + stage_goal = "Koordination Sprungphase Mawashi Geri" + stage_brief = build_stage_match_brief( + learning_goal=stage_goal, + path_primary_topic=primary, + path_technique_excludes=technique_sibling_excludes(primary), + ) + assert not exercise_passes_stage_fit( + learning_goal=stage_goal, + title="Yoko Geri seitlicher Tritt", + summary="Seitwärtskick", + goal="Yoko Geri Technik", + stage_brief=stage_brief, + path_primary_topic=primary, + path_technique_excludes=technique_sibling_excludes(primary), + ) + + def test_strip_off_topic_removes_partial_when_most_steps_bad(): steps = [{"exercise_id": i, "title": f"E{i}"} for i in range(1, 8)] off_topic = [{"step_index": i, "issue": "path_exclude"} for i in range(5)] diff --git a/backend/tests/test_planning_stage_context.py b/backend/tests/test_planning_stage_context.py new file mode 100644 index 0000000..dae3fa7 --- /dev/null +++ b/backend/tests/test_planning_stage_context.py @@ -0,0 +1,65 @@ +"""Tests Stufen-Kontext (Start/Ziel-Verkettung) und mehrstufige QS.""" +from planning_path_qa_pipeline import derive_optimization_hints, run_multistage_path_qa +from planning_progression_roadmap import MajorStep, StageSpecArtifact +from planning_stage_context import ( + build_contextualized_stage_goal, + derive_stage_specs_transition_states, +) + + +def test_derive_stage_transition_chain(): + majors = [ + MajorStep(index=0, phase="grundlage", learning_goal="Stand-Mawashi", consolidates=["m1"]), + MajorStep(index=1, phase="vertiefung", learning_goal="Sprungvorbereitung", consolidates=["m2"]), + MajorStep(index=2, phase="perfektion", learning_goal="Gesprungener Mawashi", consolidates=["m3"]), + ] + specs = [ + StageSpecArtifact(major_step_index=0, learning_goal=majors[0].learning_goal), + StageSpecArtifact(major_step_index=1, learning_goal=majors[1].learning_goal), + StageSpecArtifact(major_step_index=2, learning_goal=majors[2].learning_goal), + ] + out = derive_stage_specs_transition_states( + specs, + majors, + path_start="Anfänger mit Grundstellung", + path_target="Sauberer gesprungener Mawashi Geri", + ) + assert out[0].start_state == "Anfänger mit Grundstellung" + assert out[1].start_state == out[0].target_state + assert out[2].target_state == "Sauberer gesprungener Mawashi Geri" + + +def test_contextualized_stage_goal_includes_path_target(): + text = build_contextualized_stage_goal( + learning_goal="Sprungkoordination", + start_state="Stand-Mawashi sicher", + target_state="Explosiver Absprung", + path_target_state="Gesprungener Mawashi Geri", + stage_index=1, + stage_count=3, + ) + assert "Sprungkoordination" in text + assert "Gesamtziel" in text + assert "Soll-Start" in text + + +def test_multistage_qa_emits_optimization_hints(): + result = run_multistage_path_qa( + off_topic_steps=[], + stripped_off_topic=[ + { + "step_index": 2, + "issue": "technique_scope", + "title": "Yoko Geri", + "reasons": ["Passt nicht zur Haupttechnik"], + } + ], + gaps=[{"from_title": "A", "to_title": "B", "gap_score": 0.6, "is_large_gap": True}], + llm_qa={"quality_score": 0.25, "recommendations": ["Athletisches Training ergänzen"]}, + llm_applied=True, + ) + assert len(result["qa_tiers"]) == 3 + hints = result["optimization_hints"] + assert any(h.get("action") == "rematch_slot" for h in hints) + assert any(h.get("action") == "bridge_or_gap_fill" for h in hints) + assert derive_optimization_hints(result["qa_tiers"]) diff --git a/backend/version.py b/backend/version.py index 7d5fe5e..8ec0719 100644 --- a/backend/version.py +++ b/backend/version.py @@ -1,8 +1,8 @@ # Shinkan Jinkendo Version Information -APP_VERSION = "0.8.223" +APP_VERSION = "0.8.225" BUILD_DATE = "2026-06-07" -DB_SCHEMA_VERSION = "20260607089" +DB_SCHEMA_VERSION = "20260607090" MODULE_VERSIONS = { "legal_documents": "1.4.0", # Admin: Live-Vorschau pro Abschnitt + modale Vollvorschau (Editor + Dokumentenliste) @@ -53,6 +53,22 @@ MODULE_VERSIONS = { } CHANGELOG = [ + { + "version": "0.8.225", + "date": "2026-06-07", + "changes": [ + "Stufen start_state/target_state (Soll-Verkettung) + kontextualisiertes Matching.", + "Mehrstufige Pfad-QS (tier1–3) mit optimization_hints; Migration 090 Prompt.", + ], + }, + { + "version": "0.8.224", + "date": "2026-06-07", + "changes": [ + "Technik-Pfad-Scope: Geschwister-Techniken (Mae/Yoko bei Mawashi) als hartes Gate in Match/QS.", + "path_primary_topic in build_stage_match_brief; Intent technique_sibling_excludes.", + ], + }, { "version": "0.8.223", "date": "2026-06-07", diff --git a/docs/architecture/PLANNING_PROGRESSION_GRAPH_KI.md b/docs/architecture/PLANNING_PROGRESSION_GRAPH_KI.md index f456456..5b2df2d 100644 --- a/docs/architecture/PLANNING_PROGRESSION_GRAPH_KI.md +++ b/docs/architecture/PLANNING_PROGRESSION_GRAPH_KI.md @@ -241,6 +241,28 @@ LLM: Migration **089** — Prompts `planning_progression_goal_analysis` + `plann Matching: `anti_patterns` + `success_criteria` → `build_stage_match_brief` → Retrieval-Gate (Titel + Summary + Goal). +### Stufen im Gesamtziel (`planning_stage_context.py`) + +| Feld | Bedeutung | +|------|-----------| +| `start_state` | Soll-Start der Stufe (= `target_state` der Vorstufe / Pfad-Start) | +| `target_state` | Ziel nach dieser Stufe (= Soll für den nächsten Schritt) | +| `build_contextualized_stage_goal()` | Lernziel + Start + Stufen-Ziel + Gesamtziel → Brief/Retrieval | + +Deterministisch: `derive_stage_specs_transition_states()` nach Roadmap-Pipeline; LLM kann Felder überschreiben (Prompt **090**). + +### Mehrstufige Pfad-QS (`planning_path_qa_pipeline.py`) + +| Stufe | Inhalt | Ableitung | +|-------|--------|-----------| +| **tier1** | Deterministische Gates (Technik-Scope, Ausschlüsse, unfilled) | `optimization_hints` → `rematch_slot`, `refine_stage_spec` | +| **tier2** | Übergangs-Lücken zwischen Schritten | `bridge_or_gap_fill` | +| **tier3** | LLM-Ganzpfad + Empfehlungen | `review_roadmap` | + +API: `path_qa.qa_tiers`, `path_qa.optimization_hints` — **kein** anfrage-spezifischer Patch, sondern strukturierte Rückkopplung. Auto-Rematch-Schleife: Backlog (QS → Aktion → erneutes Match). + +Phase G: gleiche Tiers für Abschnitts-QS nach Einheits-Match. + ## 9. Fähigkeiten-Scoring-Anbindung Modul: `planning_skill_expectations.py`