Implement Phase B Enhancements for Planning Exercise Profiles

- Added support for section guidance notes and titles in the planning target profile, enabling richer context for exercise suggestions.
- Introduced deterministic text-to-catalog signal mapping, allowing for improved integration of planning text signals into the exercise retrieval process.
- Implemented a partner-related filter in exercise retrieval, enhancing the relevance of suggested exercises based on user intent.
- Updated the retrieval phase to account for text signals, improving the accuracy of exercise recommendations.
- Incremented version to 0.8.181 and updated changelog to reflect these significant enhancements in planning AI capabilities.

backend/planning_exercise_profiles.py

@@ -8,6 +8,10 @@ from __future__ import annotations
 from dataclasses import dataclass, field
 from typing import Any, Dict, List, Optional, Sequence, Set, Tuple
 
+from planning_exercise_text_signals import (
+    load_framework_planning_text_parts,
+    resolve_planning_text_to_catalog_weights,
+)
 from skill_scoring import (
     ExerciseOccurrence,
     collect_unit_exercise_occurrences,
@@ -339,6 +343,8 @@ def build_planning_target_profile(
     section_planned_exercise_ids: Optional[Sequence[int]] = None,
     anchor_exercise_id: Optional[int],
     intent: str,
+    section_guidance_notes: Optional[str] = None,
+    section_title: Optional[str] = None,
 ) -> PlanningTargetProfile:
     sources: List[str] = []
     focus: Dict[int, float] = {}
@@ -414,6 +420,30 @@ def build_planning_target_profile(
                 tg = _merge_weight_maps(tg, ap.target_group_ids, scale=0.75)
                 sources.append("anchor_exercise")
 
+    text_parts: List[str] = []
+    if (section_title or "").strip():
+        text_parts.append(str(section_title).strip())
+    if (section_guidance_notes or "").strip():
+        text_parts.append(str(section_guidance_notes).strip())
+    if fw:
+        text_parts.extend(
+            load_framework_planning_text_parts(
+                cur,
+                int(fw["framework_program_id"]),
+                slot_id=int(fw["slot_id"]) if fw.get("slot_id") else None,
+            )
+        )
+    if text_parts:
+        blob = "\n".join(text_parts)
+        tf, ts, ttt, ttg, tsk = resolve_planning_text_to_catalog_weights(cur, blob)
+        if tf or ts or ttt or ttg or tsk:
+            focus = _merge_weight_maps(focus, tf, scale=0.88)
+            style = _merge_weight_maps(style, ts, scale=0.8)
+            tt = _merge_weight_maps(tt, ttt, scale=0.8)
+            tg = _merge_weight_maps(tg, ttg, scale=0.8)
+            skill_target = _merge_weight_maps(skill_target, tsk, scale=0.92)
+            sources.append("planning_text_signals")
+
     skill_target = _normalize_weight_map(skill_target)
     skill_plan_norm = _normalize_weight_map(skill_plan)
     skill_gap: Dict[int, float] = {}
@@ -470,6 +500,8 @@ def score_exercise_against_target(
         reasons.append("Deckt Skill-Lücke im bisherigen Plan")
     if "query_intent" in (target.sources or []):
         reasons.append("Passt zur KI-interpretierten Suchanfrage")
+    if "planning_text_signals" in (target.sources or []):
+        reasons.append("Passt zu Abschnitts- oder Rahmen-Zieltext")
 
     # Intent-gewichtete Dimensionen (Summe = 1.0)
     if intent == INTENT_FREE_SEARCH:

backend/planning_exercise_retrieval.py

@@ -17,6 +17,17 @@ from planning_exercise_profiles import (
 
 _MAX_LIBRARY_ROWS = 8000
 _PROFILE_LOAD_BATCH = 400
+_PARTNER_TEXT_MARKERS = ("partner", "paar", "paarweise", "zu zweit")
+
+
+def _exercise_looks_partner_related(row: Mapping[str, Any]) -> bool:
+    parts = [
+        str(row.get("method_archetype") or ""),
+        str(row.get("title") or ""),
+        str(row.get("summary") or ""),
+    ]
+    blob = " ".join(parts).lower()
+    return any(m in blob for m in _PARTNER_TEXT_MARKERS)
 
 
 def _skill_jaccard(a: Set[int], b: Set[int]) -> float:
@@ -72,7 +83,7 @@ def fetch_all_visible_exercise_rows(
         params.extend(ek_filtered)
 
     sql = f"""
-        SELECT e.id, e.title, e.summary,
+        SELECT e.id, e.title, e.summary, e.method_archetype,
                (
                  SELECT fa.name FROM exercise_focus_areas efa
                  JOIN focus_areas fa ON fa.id = efa.focus_area_id
@@ -139,6 +150,7 @@ def rank_visible_library_hits(
     anchor_skills = set(pack.get("anchor_skill_ids") or [])
     anchor_id = pack.get("anchor_exercise_id")
     progression_notes = pack.get("progression_edge_notes") or {}
+    requires_partner = pack.get("requires_partner")
 
     last_planned_skills: Set[int] = set()
     planned_ids = pack.get("planned_exercise_ids") or []
@@ -154,6 +166,10 @@ def rank_visible_library_hits(
         eid = int(row["id"])
         if anchor_id and eid == int(anchor_id):
             continue
+        if requires_partner is True and not _exercise_looks_partner_related(row):
+            continue
+        if requires_partner is False and _exercise_looks_partner_related(row):
+            continue
         cand_rows.append(row)
 
     cand_ids = [int(r["id"]) for r in cand_rows]

backend/planning_exercise_suggest.py

@@ -638,14 +638,20 @@ def suggest_planning_exercises(
         target=target_profile,
         intent=intent,
         intent_weights=weights,
-        pack=pack,
+        pack={
+            **pack,
+            "requires_partner": query_intent_summary.get("requires_partner"),
+        },
     )
 
+    text_signals_applied = "planning_text_signals" in (target_profile.sources or [])
+
     planned_set = set(pack["planned_exercise_ids"])
 
     llm_rank_applied = False
     retrieval_phase = compose_retrieval_phase(
         full_library=full_library_ranked,
+        text_signals=text_signals_applied,
         query_intent=query_intent_applied,
         llm_expectation=llm_expectation_applied,
         llm_rank=False,
@@ -681,6 +687,7 @@ def suggest_planning_exercises(
         if llm_rank_applied:
             retrieval_phase = compose_retrieval_phase(
                 full_library=full_library_ranked,
+                text_signals=text_signals_applied,
                 query_intent=query_intent_applied,
                 llm_expectation=llm_expectation_applied,
                 llm_rank=True,
@@ -719,6 +726,7 @@ def suggest_planning_exercises(
         "query_intent_summary": query_intent_summary,
         "retrieval_phase": retrieval_phase,
         "full_library_ranked": full_library_ranked,
+        "text_signals_applied": text_signals_applied,
         "profile_preselect_applied": False,
         "llm_rank_applied": llm_rank_applied,
         "llm_intent_applied": query_intent_applied,

backend/planning_exercise_target_pipeline.py

@@ -271,6 +271,8 @@ def build_planning_target_with_query_pipeline(
             section_planned_exercise_ids=section_planned_exercise_ids or [],
             anchor_exercise_id=anchor_exercise_id,
             intent=heuristic_intent,
+            section_guidance_notes=(context_summary.get("section_guidance_notes") or None),
+            section_title=(context_summary.get("section_title") or None),
         )
     else:
         base = PlanningTargetProfile(sources=["query_only"])
@@ -387,6 +389,7 @@ def compose_retrieval_phase(
     *,
     full_library: bool = False,
     profile_preselect: bool = False,
+    text_signals: bool = False,
     query_intent: bool = False,
     llm_expectation: bool = False,
     llm_rank: bool = False,
@@ -394,6 +397,8 @@ def compose_retrieval_phase(
     parts = ["profile_v1"]
     if full_library or profile_preselect:
         parts.append("full_library")
+    if text_signals:
+        parts.append("text_signals")
     if llm_expectation:
         parts.append("llm_expectation")
     elif query_intent:

backend/planning_exercise_text_signals.py

@@ -0,0 +1,201 @@
+"""
+Phase B: Deterministische Text→Katalog-Signale für PlanningTargetProfile.
+
+Mappt Abschnitts-guidance, Rahmen-Ziele/-Notizen und Programmbeschreibung
+auf Skill-/Katalog-Gewichte (ohne LLM).
+"""
+from __future__ import annotations
+
+import re
+from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple
+
+_MIN_SKILL_NAME_LEN = 3
+_MAX_SKILL_MATCHES = 12
+_MAX_CATALOG_MATCHES = 6
+
+
+def _normalize_text_blob(*parts: Optional[str]) -> str:
+    chunks: List[str] = []
+    for p in parts:
+        s = (p or "").strip()
+        if s:
+            chunks.append(s)
+    return "\n".join(chunks).lower()
+
+
+def _load_skills_for_text_match(cur) -> List[Tuple[int, str, int]]:
+    cur.execute(
+        """
+        SELECT id, name FROM skills
+        WHERE (status IS NULL OR status = 'active')
+          AND name IS NOT NULL AND TRIM(name) <> ''
+        ORDER BY LENGTH(name) DESC, name ASC
+        """
+    )
+    out: List[Tuple[int, str, int]] = []
+    for row in cur.fetchall():
+        name = str(row.get("name") or "").strip()
+        if len(name) < _MIN_SKILL_NAME_LEN:
+            continue
+        out.append((int(row["id"]), name.lower(), len(name)))
+    return out
+
+
+def _load_catalog_names(cur, table: str, id_col: str = "id", name_col: str = "name") -> List[Tuple[int, str, int]]:
+    cur.execute(
+        f"""
+        SELECT {id_col} AS id, {name_col} AS name
+        FROM {table}
+        WHERE {name_col} IS NOT NULL AND TRIM({name_col}) <> ''
+        ORDER BY LENGTH({name_col}) DESC, {name_col} ASC
+        """
+    )
+    out: List[Tuple[int, str, int]] = []
+    for row in cur.fetchall():
+        name = str(row.get("name") or "").strip()
+        if len(name) < 2:
+            continue
+        out.append((int(row["id"]), name.lower(), len(name)))
+    return out
+
+
+def _match_catalog_names_in_text(
+    text: str,
+    catalog_rows: Sequence[Tuple[int, str, int]],
+    *,
+    weight: float = 0.85,
+    limit: int = _MAX_CATALOG_MATCHES,
+) -> Dict[int, float]:
+    if not text or not catalog_rows:
+        return {}
+    out: Dict[int, float] = {}
+    for cid, name_lower, _ in catalog_rows:
+        if len(out) >= limit:
+            break
+        if len(name_lower) < 2:
+            continue
+        if name_lower in text:
+            out[cid] = max(out.get(cid, 0.0), weight)
+    return out
+
+
+def _match_skills_in_text(
+    text: str,
+    skill_rows: Sequence[Tuple[int, str, int]],
+    *,
+    limit: int = _MAX_SKILL_MATCHES,
+) -> Dict[int, float]:
+    if not text or not skill_rows:
+        return {}
+    out: Dict[int, float] = {}
+    for sid, name_lower, name_len in skill_rows:
+        if len(out) >= limit:
+            break
+        if name_len < _MIN_SKILL_NAME_LEN:
+            continue
+        if name_lower in text:
+            w = min(1.0, 0.72 + min(name_len, 20) * 0.012)
+            out[sid] = max(out.get(sid, 0.0), w)
+    return out
+
+
+def load_framework_planning_text_parts(
+    cur,
+    framework_program_id: int,
+    *,
+    slot_id: Optional[int] = None,
+) -> List[str]:
+    """Sammelt Rahmen-Texte für Text-Signal-Matching."""
+    parts: List[str] = []
+    cur.execute(
+        "SELECT description FROM training_framework_programs WHERE id = %s",
+        (int(framework_program_id),),
+    )
+    row = cur.fetchone()
+    if row and (row.get("description") or "").strip():
+        parts.append(str(row["description"]).strip())
+
+    cur.execute(
+        """
+        SELECT title, notes FROM training_framework_goals
+        WHERE framework_program_id = %s
+        ORDER BY sort_order ASC
+        """,
+        (int(framework_program_id),),
+    )
+    for g in cur.fetchall():
+        t = (g.get("title") or "").strip()
+        n = (g.get("notes") or "").strip()
+        if t:
+            parts.append(t)
+        if n:
+            parts.append(n)
+
+    if slot_id:
+        cur.execute(
+            "SELECT title, notes FROM training_framework_slots WHERE id = %s",
+            (int(slot_id),),
+        )
+        srow = cur.fetchone()
+        if srow:
+            st = (srow.get("title") or "").strip()
+            sn = (srow.get("notes") or "").strip()
+            if st:
+                parts.append(st)
+            if sn:
+                parts.append(sn)
+
+    return parts
+
+
+def resolve_planning_text_to_catalog_weights(
+    cur,
+    text_blob: str,
+) -> Tuple[Dict[int, float], Dict[int, float], Dict[int, float], Dict[int, float], Dict[int, float]]:
+    """
+    Returns: focus, style, training_type, target_group, skill weight maps.
+    """
+    text = _normalize_text_blob(text_blob)
+    if not text or len(text) < 3:
+        return {}, {}, {}, {}, {}
+
+    skill_rows = _load_skills_for_text_match(cur)
+    focus_rows = _load_catalog_names(cur, "focus_areas")
+    style_rows = _load_catalog_names(cur, "style_directions")
+    tt_rows = _load_catalog_names(cur, "training_types")
+    tg_rows = _load_catalog_names(cur, "target_groups")
+
+    skills = _match_skills_in_text(text, skill_rows)
+    focus = _match_catalog_names_in_text(text, focus_rows, weight=0.88)
+    style = _match_catalog_names_in_text(text, style_rows, weight=0.82)
+    tt = _match_catalog_names_in_text(text, tt_rows, weight=0.82)
+    tg = _match_catalog_names_in_text(text, tg_rows, weight=0.8)
+
+    if re.search(r"\bpartner\b|\bpaar\b|\bpaarweise\b|\bzu zweit\b", text):
+        for gid, name_lower, _ in tg_rows:
+            if "partner" in name_lower or "paar" in name_lower:
+                tg[gid] = max(tg.get(gid, 0.0), 0.9)
+                break
+
+    return focus, style, tt, tg, skills
+
+
+def merge_text_signal_summary(
+    summary: Mapping[str, Any],
+    *,
+    text_sources: Sequence[str],
+    matched_skills: Sequence[Mapping[str, Any]],
+) -> Dict[str, Any]:
+    out = dict(summary)
+    if text_sources:
+        out["text_signal_sources"] = list(text_sources)
+    if matched_skills:
+        out["text_signal_skills"] = list(matched_skills)[:8]
+    return out
+
+
+__all__ = [
+    "load_framework_planning_text_parts",
+    "merge_text_signal_summary",
+    "resolve_planning_text_to_catalog_weights",
+]

backend/tests/test_planning_exercise_retrieval_partner.py

@@ -0,0 +1,7 @@
+"""Tests Partner-Filter im Planungs-Retrieval."""
+from planning_exercise_retrieval import _exercise_looks_partner_related
+
+
+def test_exercise_partner_heuristic():
+    assert _exercise_looks_partner_related({"title": "Partner-Fangspiel", "summary": ""})
+    assert not _exercise_looks_partner_related({"title": "Kihon Solo", "summary": "Allein"})

backend/tests/test_planning_exercise_text_signals.py

@@ -0,0 +1,47 @@
+"""Tests Phase B: Text-Signale für PlanningTargetProfile."""
+from planning_exercise_text_signals import resolve_planning_text_to_catalog_weights
+
+
+def test_resolve_planning_text_matches_skill_and_partner_hint():
+    class _Cur:
+        def execute(self, sql, params=None):
+            self._sql = sql
+
+        def fetchall(self):
+            sql = getattr(self, "_sql", "")
+            if "FROM skills" in sql:
+                return [
+                    {"id": 5, "name": "Kime"},
+                    {"id": 8, "name": "Schnellkraft"},
+                    {"id": 2, "name": "Ab"},
+                ]
+            if "FROM focus_areas" in sql:
+                return [{"id": 10, "name": "Karate Technik"}]
+            if "FROM style_directions" in sql:
+                return []
+            if "FROM training_types" in sql:
+                return []
+            if "FROM target_groups" in sql:
+                return [{"id": 3, "name": "Partnerübung"}]
+            return []
+
+    focus, style, tt, tg, skills = resolve_planning_text_to_catalog_weights(
+        _Cur(),
+        "Abschnitt: Kime vertiefen mit Partnerübung",
+    )
+    assert skills.get(5, 0) > 0
+    assert 8 not in skills
+    assert tg.get(3, 0) > 0
+    assert not style and not tt
+
+
+def test_resolve_planning_text_empty():
+    class _Cur:
+        def execute(self, sql, params=None):
+            pass
+
+        def fetchall(self):
+            return []
+
+    focus, style, tt, tg, skills = resolve_planning_text_to_catalog_weights(_Cur(), "  ")
+    assert not focus and not skills

backend/version.py

@@ -1,6 +1,6 @@
 # Shinkan Jinkendo Version Information
 
-APP_VERSION = "0.8.180"
+APP_VERSION = "0.8.181"
 BUILD_DATE = "2026-05-23"
 DB_SCHEMA_VERSION = "20260531074"
 
@@ -29,7 +29,7 @@ MODULE_VERSIONS = {
     "skill_profiles": "1.0.0",  # Phase 3: gewichtetes Fähigkeiten-Profil + skill-discovery/suggestions
     "methods": "0.1.0",
     "exercises": "2.37.0",  # Planungs-KI P1: Szenario-Pipeline + Query-Intent-Overlay
-    "planning_exercise_suggest": "0.8.0",  # Phase A: Voll-Library-Ranking gegen Erwartungsprofil
+    "planning_exercise_suggest": "0.9.0",  # Phase B: Text-Signale guidance/Rahmen-Ziele; requires_partner-Filter
     "training_units": "0.4.0",  # POST .../publish-to-framework: Ablauf aus geplanter Einheit → Rahmen-Slot-Blueprint
     "training_programs": "0.1.0",
     "planning": "0.15.0",  # Vorlagen: Strukturvorschau, Bearbeiten inkl. Split-Sessions + Beschreibung
@@ -44,6 +44,14 @@ MODULE_VERSIONS = {
 }
 
 CHANGELOG = [
+    {
+        "version": "0.8.181",
+        "date": "2026-05-23",
+        "changes": [
+            "Planungs-KI Phase B: guidance_notes + Rahmen-Ziele/Notizen → Text-Signale im Erwartungsprofil (planning_text_signals).",
+            "requires_partner aus Intent filtert Übungen; retrieval_phase +text_signals; Grund „Passt zu Abschnitts-/Rahmen-Zieltext“.",
+        ],
+    },
     {
         "version": "0.8.180",
         "date": "2026-05-23",