Enhance Path QA and Stage Matching Logic
All checks were successful
Deploy Development / deploy (push) Successful in 43s
Test Suite / pytest-backend (push) Successful in 44s
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 1m14s

- Introduced multistage path quality assurance (QA) functionality to improve exercise relevance and feedback through structured tiers and optimization hints.
- Updated stage specifications to include `start_state` and `target_state` for better contextualization in roadmap matching.
- Enhanced semantic brief construction with technique sibling exclusions to refine exercise selection based on primary topics.
- Improved path retrieval logic to incorporate new parameters for nuanced matching against learning goals.
- Incremented application version to reflect these updates.
This commit is contained in:
Lars 2026-06-11 10:19:58 +02:00
parent 4ef3f00e6b
commit a152218c45
13 changed files with 760 additions and 6 deletions

View File

@ -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';

View File

@ -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)

View File

@ -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")

View File

@ -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"

View File

@ -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",

View File

@ -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

View File

@ -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",
]

View File

@ -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"

View File

@ -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",
]

View File

@ -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)]

View File

@ -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"])

View File

@ -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 (tier13) 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",

View File

@ -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`