""" Planungs-KI Phase C3/E: Pfad-Vorschläge für Progressionsgraphen. Ziel-Freitext → semantisch gewichtete Schritte → Lücken/Brücken → optional LLM-QA. """ from __future__ import annotations from typing import Any, Callable, Dict, List, Optional, Set, Tuple from fastapi import HTTPException from pydantic import BaseModel, Field from tenant_context import TenantContext, library_content_visibility_sql from planning_exercise_profiles import PlanningTargetProfile from planning_exercise_path_qa import ( apply_llm_path_reorder, build_path_qa_summary, detect_path_gaps, insert_bridge_exercises, try_llm_qa_progression_path, ) from planning_exercise_path_ai_fill import insert_ai_proposals_for_gaps from planning_exercise_retrieval import run_multistage_planning_retrieval from planning_exercise_semantics import ( PlanningSemanticBrief, apply_path_retrieval_weights, brief_to_summary_dict, build_semantic_brief, enrich_target_with_semantic_expectations, exercise_passes_path_semantic_gate, resolve_semantic_skill_weights, step_phase_for_index, step_retrieval_query, try_enrich_semantic_brief_with_llm, ) from planning_exercise_target_pipeline import build_planning_target_with_query_pipeline from planning_exercise_progression import apply_progression_context_to_pack from planning_exercise_suggest import ( _enrich_planning_hits_with_variant_meta, _load_skill_ids_for_exercise, _normalize_query, resolve_planning_exercise_intent, ) from routers.training_planning import _has_planning_role class ProgressionPathSuggestRequest(BaseModel): query: str = Field(..., min_length=3, max_length=2000) max_steps: int = Field(default=5, ge=2, le=10) include_llm_intent: bool = True include_path_qa: bool = True include_llm_path_qa: bool = True include_path_reorder: bool = True include_ai_gap_fill: bool = True progression_graph_id: Optional[int] = Field(default=None, ge=1) exercise_kind_any: Optional[List[str]] = None def _pick_best_path_hit( hits: List[Dict[str, Any]], used_exercise_ids: Set[int], *, semantic_brief: Optional[PlanningSemanticBrief] = None, ) -> Optional[Dict[str, Any]]: best: Optional[Dict[str, Any]] = None best_key: Tuple[float, float] = (-1.0, -1.0) for hit in hits: eid = int(hit["id"]) if eid in used_exercise_ids: continue sem = float(hit.get("semantic_score") or 0.0) if semantic_brief and not exercise_passes_path_semantic_gate( semantic_score=sem, title=str(hit.get("title") or ""), brief=semantic_brief, ): continue score = float(hit.get("score") or 0.0) key = (sem, score) if key > best_key: best_key = key best = hit return best def _build_path_target_profile( cur, *, goal_query: str, semantic_brief: PlanningSemanticBrief, include_llm_intent: bool, ) -> Tuple[PlanningTargetProfile, Dict[str, Any], str]: """Einmaliges Erwartungsprofil für den gesamten Pfad (Query + Semantik + Skills).""" empty_unit = { "id": None, "framework_slot_id": None, "origin_framework_slot_id": None, } pipeline_context = { "unit_title": None, "group_name": None, "section_title": None, "section_guidance_notes": goal_query, "section_exercise_count": 0, "planned_count": 0, "anchor_title": None, "anchor_exercise_id": None, "last_section_exercise_title": None, "progression_graph_id": None, "unit_skill_profile": None, "section_skill_profile": None, "has_planning_reference": False, "expectation_mode": "query_only", } target, intent, _scenario, query_intent_summary = build_planning_target_with_query_pipeline( cur, unit=empty_unit, planned_exercise_ids=[], section_planned_exercise_ids=[], anchor_exercise_id=None, query=goal_query, heuristic_intent=resolve_planning_exercise_intent(goal_query, "free_search"), include_llm_intent=include_llm_intent, context_summary=pipeline_context, has_planning_reference=False, ) skill_weights = resolve_semantic_skill_weights(cur, semantic_brief) target = enrich_target_with_semantic_expectations(target, skill_weights=skill_weights) return target, query_intent_summary, intent def _hit_to_path_step(hit: Dict[str, Any], *, is_bridge: bool = False) -> Dict[str, Any]: raw_vid = hit.get("suggested_variant_id") variant_id: Optional[int] = None if raw_vid is not None: try: vid = int(raw_vid) if vid > 0: variant_id = vid except (TypeError, ValueError): variant_id = None step = { "exercise_id": int(hit["id"]), "variant_id": variant_id, "title": hit.get("title"), "summary": hit.get("summary"), "score": hit.get("score"), "semantic_score": hit.get("semantic_score"), "reasons": list(hit.get("reasons") or []), "variants": hit.get("variants") or [], "suggested_variant_id": hit.get("suggested_variant_id"), "suggested_variant_name": hit.get("suggested_variant_name"), } if is_bridge: step["is_bridge"] = True return step def _run_path_step_retrieval( cur, *, tenant: TenantContext, goal_query: str, step_index: int, max_steps: int, planned_ids: List[int], anchor_id: Optional[int], anchor_variant_id: Optional[int], progression_graph_id: Optional[int], include_llm_intent: bool, exercise_kind_any: Optional[List[str]], semantic_brief: PlanningSemanticBrief, bridge_mode: bool = False, step_a: Optional[Dict[str, Any]] = None, step_b: Optional[Dict[str, Any]] = None, path_target_profile: Optional[PlanningTargetProfile] = None, path_intent: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], PlanningTargetProfile, Dict[str, Any], str]: step_query = step_retrieval_query(semantic_brief, goal_query, step_index, max_steps) if bridge_mode and step_a and step_b: phase = step_phase_for_index(semantic_brief, step_index, max_steps) parts = [semantic_brief.primary_topic or semantic_brief.retrieval_query or goal_query] if phase: parts.append(phase) step_query = _normalize_query(" ".join(p for p in parts if p) + " brücke") pack: Dict[str, Any] = { "unit_id": None, "unit": { "id": None, "framework_slot_id": None, "origin_framework_slot_id": None, }, "unit_title": None, "group_id": None, "group_name": None, "section_order_index": None, "section_title": None, "section_guidance_notes": goal_query if step_index == 0 and not bridge_mode else step_query, "planned_exercise_ids": list(planned_ids), "anchor_exercise_id": anchor_id, "anchor_title": None, "anchor_skill_ids": sorted(_load_skill_ids_for_exercise(cur, anchor_id)), "group_recent_exercise_ids": [], "context_mode": "progression_path", "has_planning_reference": bool(planned_ids or anchor_id or bridge_mode), "semantic_brief": semantic_brief, "retrieval_query": step_query, "path_step_phase": step_phase_for_index(semantic_brief, step_index, max_steps), } pack = apply_progression_context_to_pack( cur, tenant, pack, explicit_graph_id=progression_graph_id, anchor_variant_id=anchor_variant_id, ) if step_index == 0 and not bridge_mode: heuristic_intent = resolve_planning_exercise_intent(goal_query, "free_search") else: heuristic_intent = path_intent or resolve_planning_exercise_intent(goal_query, "free_search") has_plan_ref = bool(pack.get("has_planning_reference")) pipeline_context = { "unit_title": None, "group_name": None, "section_title": pack.get("section_title"), "section_guidance_notes": pack.get("section_guidance_notes"), "section_exercise_count": len(planned_ids), "planned_count": len(planned_ids), "anchor_title": pack.get("anchor_title"), "anchor_exercise_id": pack.get("anchor_exercise_id"), "last_section_exercise_title": None, "progression_graph_id": pack.get("progression_graph_id"), "unit_skill_profile": None, "section_skill_profile": None, "has_planning_reference": has_plan_ref, "expectation_mode": "query_only" if step_index == 0 and not planned_ids else "planning_hybrid", } if path_target_profile is not None: target_profile = path_target_profile intent = path_intent or resolve_planning_exercise_intent(goal_query, "free_search") query_intent_summary = {} else: target_profile, intent, _scenario, query_intent_summary = build_planning_target_with_query_pipeline( cur, unit=pack["unit"], planned_exercise_ids=pack["planned_exercise_ids"], section_planned_exercise_ids=[], anchor_exercise_id=pack.get("anchor_exercise_id"), query=goal_query if step_index == 0 and not bridge_mode else step_query, heuristic_intent=heuristic_intent, include_llm_intent=include_llm_intent and step_index == 0 and not bridge_mode, context_summary=pipeline_context, has_planning_reference=has_plan_ref, ) weights = apply_path_retrieval_weights(semantic_brief) profile_id = tenant.profile_id role = tenant.global_role vis_sql, vis_params = library_content_visibility_sql( alias="e", profile_id=profile_id, role=role, effective_club_id=tenant.effective_club_id, ) hits, _skills_by_ex, _full_lib = run_multistage_planning_retrieval( cur, vis_sql=vis_sql, vis_params=vis_params, query=step_query, exercise_kind_any=exercise_kind_any, target=target_profile, intent=intent, intent_weights=weights, pack=pack, ) hits = _enrich_planning_hits_with_variant_meta(cur, hits[:32]) return hits, target_profile, query_intent_summary, intent def _make_bridge_search_fn( cur, *, tenant: TenantContext, goal_query: str, max_steps: int, progression_graph_id: Optional[int], include_llm_intent: bool, exercise_kind_any: Optional[List[str]], semantic_brief: PlanningSemanticBrief, planned_ids: List[int], path_target_profile: PlanningTargetProfile, path_intent: str, ) -> Callable[..., List[Dict[str, Any]]]: def _bridge_search( step_a: Dict[str, Any], step_b: Dict[str, Any], _gap: Dict[str, Any], ) -> List[Dict[str, Any]]: hits, _, _, _ = _run_path_step_retrieval( cur, tenant=tenant, goal_query=goal_query, step_index=1, max_steps=max_steps, planned_ids=list(planned_ids) + [int(step_a["exercise_id"])], anchor_id=int(step_a["exercise_id"]), anchor_variant_id=step_a.get("variant_id"), progression_graph_id=progression_graph_id, include_llm_intent=include_llm_intent, exercise_kind_any=exercise_kind_any, semantic_brief=semantic_brief, bridge_mode=True, step_a=step_a, step_b=step_b, path_target_profile=path_target_profile, path_intent=path_intent, ) gated = [ h for h in hits if exercise_passes_path_semantic_gate( semantic_score=float(h.get("semantic_score") or 0.0), title=str(h.get("title") or ""), brief=semantic_brief, ) ] return gated or hits[:8] return _bridge_search def suggest_progression_path( cur, *, tenant: TenantContext, body: ProgressionPathSuggestRequest, ) -> Dict[str, Any]: role = tenant.global_role if not _has_planning_role(role): raise HTTPException(status_code=403, detail="Nur Trainer dürfen Pfad-Vorschläge abrufen") goal_query = _normalize_query(body.query) if len(goal_query) < 3: raise HTTPException(status_code=400, detail="Ziel-Anfrage: mindestens 3 Zeichen") max_steps = int(body.max_steps) semantic_brief = build_semantic_brief(goal_query) semantic_llm_applied = False if body.include_llm_intent and semantic_brief.semantic_strength >= 0.35: semantic_brief, semantic_llm_applied = try_enrich_semantic_brief_with_llm( cur, goal_query, semantic_brief ) path_target_profile, first_intent_summary, path_intent = _build_path_target_profile( cur, goal_query=goal_query, semantic_brief=semantic_brief, include_llm_intent=body.include_llm_intent, ) used: Set[int] = set() steps: List[Dict[str, Any]] = [] planned_ids: List[int] = [] anchor_id: Optional[int] = None anchor_variant_id: Optional[int] = None for step_index in range(max_steps): hits, _tp, _qis, _intent = _run_path_step_retrieval( cur, tenant=tenant, goal_query=goal_query, step_index=step_index, max_steps=max_steps, planned_ids=planned_ids, anchor_id=anchor_id, anchor_variant_id=anchor_variant_id, progression_graph_id=body.progression_graph_id, include_llm_intent=body.include_llm_intent, exercise_kind_any=body.exercise_kind_any, semantic_brief=semantic_brief, path_target_profile=path_target_profile, path_intent=path_intent, ) hit = _pick_best_path_hit(hits, used, semantic_brief=semantic_brief) if not hit: break step = _hit_to_path_step(hit) steps.append(step) eid = int(step["exercise_id"]) used.add(eid) planned_ids.append(eid) anchor_id = eid anchor_variant_id = step.get("variant_id") if len(steps) < 2: raise HTTPException( status_code=422, detail="Zu wenig passende Übungen für einen Pfad (mindestens 2 Schritte). Ziel präzisieren oder max_steps senken.", ) gaps: List[Dict[str, Any]] = [] bridge_inserts: List[Dict[str, Any]] = [] ai_proposals: List[Dict[str, Any]] = [] llm_qa: Optional[Dict[str, Any]] = None llm_qa_applied = False reorder_applied = False reorder_notes: List[str] = [] if body.include_path_qa: gaps = detect_path_gaps(cur, steps, brief=semantic_brief) unfilled_gaps: List[Dict[str, Any]] = [] if gaps: bridge_fn = _make_bridge_search_fn( cur, tenant=tenant, goal_query=goal_query, max_steps=max_steps, progression_graph_id=body.progression_graph_id, include_llm_intent=body.include_llm_intent, exercise_kind_any=body.exercise_kind_any, semantic_brief=semantic_brief, planned_ids=planned_ids, path_target_profile=path_target_profile, path_intent=path_intent, ) steps, bridge_inserts, unfilled_gaps = insert_bridge_exercises( cur, steps, gaps, brief=semantic_brief, bridge_search_fn=bridge_fn, ) if body.include_ai_gap_fill and unfilled_gaps: steps, ai_proposals = insert_ai_proposals_for_gaps( cur, steps, unfilled_gaps, goal_query=goal_query, brief=semantic_brief, ) if body.include_llm_path_qa: llm_qa, llm_qa_applied = try_llm_qa_progression_path( cur, goal_query=goal_query, brief=semantic_brief, steps=steps, gaps=gaps, bridge_inserts=bridge_inserts, ) if body.include_path_reorder and llm_qa_applied and llm_qa: q_score = llm_qa.get("quality_score") try: q_val = float(q_score) if q_score is not None else None except (TypeError, ValueError): q_val = None if llm_qa.get("overall_ok") or (q_val is not None and q_val >= 0.45): steps, reorder_applied, reorder_notes = apply_llm_path_reorder(steps, llm_qa) path_qa = build_path_qa_summary( gaps=gaps, bridge_inserts=bridge_inserts, ai_proposals=ai_proposals, llm_qa=llm_qa, llm_applied=llm_qa_applied, reorder_applied=reorder_applied, reorder_notes=reorder_notes, ) target_profile_summary = path_target_profile.to_summary_dict(cur) retrieval_parts = ["profile_v1", "full_library", "path_builder", "semantics"] if body.include_path_qa: retrieval_parts.append("path_qa") if llm_qa_applied: retrieval_parts.append("llm_path_qa") if reorder_applied: retrieval_parts.append("path_reorder") if ai_proposals: retrieval_parts.append("ai_gap_fill") return { "goal_query": goal_query, "max_steps_requested": max_steps, "steps": steps, "step_count": len(steps), "target_profile_summary": target_profile_summary, "semantic_brief_summary": brief_to_summary_dict(semantic_brief), "semantic_llm_applied": semantic_llm_applied, "query_intent_summary": first_intent_summary, "progression_graph_id": body.progression_graph_id, "path_qa": path_qa, "retrieval_phase": "+".join(retrieval_parts), } __all__ = [ "ProgressionPathSuggestRequest", "suggest_progression_path", "_pick_best_path_hit", "_pick_next_path_hit", ] # Legacy-Alias für Tests _pick_next_path_hit = _pick_best_path_hit