""" Planungs-KI Phase C3/E/F: Pfad-Vorschläge für Progressionsgraphen. Legacy: retrieval-first. Phase F: optional Roadmap-Preview (A→B→C) parallel — siehe planning_progression_roadmap.py und PLANNING_PROGRESSION_ROADMAP_SPEC.md. """ from __future__ import annotations from typing import Any, Callable, Dict, List, Mapping, 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_off_topic_steps, detect_path_gaps, insert_bridge_exercises, parse_llm_suggested_new_exercises, strip_off_topic_steps_from_path, try_llm_qa_progression_path, ) from planning_exercise_path_ai_fill import ( apply_gap_fill_after_qa, build_gap_fill_offer, collect_gap_fill_specs, ) 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, pick_best_path_hit, 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 planning_exercise_form_context import build_progression_gap_snapshot from planning_progression_roadmap import ( MajorStep, ProgressionRoadmapContext, RoadmapOverridePayload, RoadmapStructuredInput, StageSpecArtifact, build_roadmap_unfilled_gap_specs, progression_roadmap_to_api_dict, resolve_step_exercise_kind_filter, roadmap_context_from_override, run_progression_roadmap_pipeline, run_start_target_resolve_only, stage_spec_retrieval_query, ) 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 include_roadmap_preview: bool = False include_llm_roadmap: bool = True include_llm_start_target: bool = True roadmap_first: bool = False roadmap_only: bool = False start_target_only: bool = False roadmap_override: Optional[RoadmapOverridePayload] = None start_situation: Optional[str] = Field(default=None, max_length=2000) target_state: Optional[str] = Field(default=None, max_length=2000) roadmap_notes: Optional[str] = Field(default=None, max_length=2000) progression_graph_id: Optional[int] = Field(default=None, ge=1) exercise_kind_any: Optional[List[str]] = None def _roadmap_gap_snapshot_for_spec( roadmap_ctx: Optional[ProgressionRoadmapContext], spec: Mapping[str, Any], *, semantic_brief: PlanningSemanticBrief, ) -> Dict[str, Any]: """Roadmap-Kontext für KI-Lücken-Übung (Start, Ziel, Stufenspec).""" major_idx = spec.get("roadmap_major_step_index") stage_spec_dict: Optional[Dict[str, Any]] = None if roadmap_ctx and major_idx is not None: for s in roadmap_ctx.stage_specs or []: if int(s.major_step_index) == int(major_idx): stage_spec_dict = s.model_dump() if roadmap_ctx.roadmap: for m in roadmap_ctx.roadmap.major_steps: if m.index == int(major_idx): stage_spec_dict["phase"] = m.phase break break ga = roadmap_ctx.goal_analysis.model_dump() if roadmap_ctx and roadmap_ctx.goal_analysis else None rs = ( roadmap_ctx.resolved_structured.model_dump() if roadmap_ctx and roadmap_ctx.resolved_structured else None ) brief_summary = ( roadmap_ctx.semantic_brief if roadmap_ctx and roadmap_ctx.semantic_brief else brief_to_summary_dict(semantic_brief) ) return build_progression_gap_snapshot( goal_analysis=ga, resolved_structured=rs, stage_spec=stage_spec_dict, semantic_brief=brief_summary, ) def _roadmap_structured_from_body(body: ProgressionPathSuggestRequest) -> Optional[RoadmapStructuredInput]: start = (body.start_situation or "").strip() or None target = (body.target_state or "").strip() or None notes = (body.roadmap_notes or "").strip() or None if not any([start, target, notes]): return None return RoadmapStructuredInput( start_situation=start, target_state=target, roadmap_notes=notes, ) def _pick_best_path_hit( hits: List[Dict[str, Any]], used_exercise_ids: Set[int], *, semantic_brief: Optional[PlanningSemanticBrief] = None, ) -> Optional[Dict[str, Any]]: return pick_best_path_hit(hits, used_exercise_ids, semantic_brief=semantic_brief) 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, step_query_override: Optional[str] = None, step_phase_override: Optional[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 ) 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_override or 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 ""), summary=str(h.get("summary") or ""), brief=semantic_brief, strict=False, ) ] return gated or hits[:12] return _bridge_search def _annotate_roadmap_step( step: Dict[str, Any], *, stage_spec: StageSpecArtifact, major_step: Optional[MajorStep], ) -> Dict[str, Any]: reasons = list(step.get("reasons") or []) learning_goal = (stage_spec.learning_goal or "").strip() if learning_goal: roadmap_reason = f"Roadmap: {learning_goal[:120]}" if roadmap_reason not in reasons: reasons.insert(0, roadmap_reason) step["reasons"] = reasons[:4] step["roadmap_major_step_index"] = stage_spec.major_step_index step["roadmap_phase"] = major_step.phase if major_step else None step["roadmap_learning_goal"] = learning_goal or None step["roadmap_match_source"] = "stage_spec" return step def _build_steps_roadmap_first( cur, *, tenant: TenantContext, body: ProgressionPathSuggestRequest, goal_query: str, max_steps: int, semantic_brief: PlanningSemanticBrief, path_target_profile: PlanningTargetProfile, path_intent: str, roadmap_ctx: ProgressionRoadmapContext, ) -> Tuple[List[Dict[str, Any]], List[Tuple[int, StageSpecArtifact]]]: """Retrieval pro stage_spec statt iterativem Pfad-Bau (Phase F3).""" stage_specs = list(roadmap_ctx.stage_specs or [])[:max_steps] if not stage_specs and roadmap_ctx.roadmap: stage_specs = [ StageSpecArtifact( major_step_index=m.index, learning_goal=m.learning_goal, ) for m in roadmap_ctx.roadmap.major_steps[:max_steps] ] major_by_index: Dict[int, MajorStep] = {} if roadmap_ctx.roadmap: major_by_index = {m.index: m for m in roadmap_ctx.roadmap.major_steps} used: Set[int] = set() steps: List[Dict[str, Any]] = [] planned_ids: List[int] = [] anchor_id: Optional[int] = None anchor_variant_id: Optional[int] = None unfilled: List[Tuple[int, StageSpecArtifact]] = [] for step_index, stage_spec in enumerate(stage_specs): major = major_by_index.get(stage_spec.major_step_index) step_query = stage_spec_retrieval_query( semantic_brief=semantic_brief, goal_query=goal_query, stage_spec=stage_spec, major_step=major, ) step_kind = resolve_step_exercise_kind_filter(stage_spec, body.exercise_kind_any) hits, _, _, _ = _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 and step_index == 0, exercise_kind_any=step_kind, semantic_brief=semantic_brief, path_target_profile=path_target_profile, path_intent=path_intent, step_query_override=step_query, step_phase_override=major.phase if major else None, ) hit = _pick_best_path_hit(hits, used, semantic_brief=semantic_brief) if not hit and step_query != goal_query: hits, _, _, _ = _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=False, exercise_kind_any=step_kind, semantic_brief=semantic_brief, path_target_profile=path_target_profile, path_intent=path_intent, step_query_override=goal_query, step_phase_override=major.phase if major else None, ) hit = _pick_best_path_hit(hits, used, semantic_brief=semantic_brief) if not hit: unfilled.append((step_index, stage_spec)) continue step = _annotate_roadmap_step( _hit_to_path_step(hit), stage_spec=stage_spec, major_step=major, ) 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") return steps, unfilled 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 ) roadmap_first = bool(body.roadmap_first) roadmap_only = bool(body.roadmap_only) start_target_only = bool(body.start_target_only) include_roadmap = ( roadmap_first or body.include_roadmap_preview or roadmap_only or start_target_only ) progression_roadmap: Optional[Dict[str, Any]] = None roadmap_ctx: Optional[ProgressionRoadmapContext] = None roadmap_edited = False roadmap_structured = _roadmap_structured_from_body(body) if body.roadmap_override is not None: try: roadmap_ctx = roadmap_context_from_override( goal_query, max_steps=max_steps, semantic_brief=semantic_brief, override=body.roadmap_override, structured=roadmap_structured, ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc progression_roadmap = progression_roadmap_to_api_dict(roadmap_ctx) progression_roadmap["roadmap_edited"] = True roadmap_edited = True max_steps = int(roadmap_ctx.max_steps) roadmap_first = True elif start_target_only: roadmap_ctx = run_start_target_resolve_only( goal_query, semantic_brief=semantic_brief, cur=cur, include_llm_start_target=body.include_llm_start_target, structured=roadmap_structured, ) progression_roadmap = progression_roadmap_to_api_dict(roadmap_ctx) elif include_roadmap: roadmap_ctx = run_progression_roadmap_pipeline( goal_query, max_steps=max_steps, semantic_brief=semantic_brief, cur=cur, include_llm_roadmap=body.include_llm_roadmap, include_llm_start_target=body.include_llm_start_target, structured=roadmap_structured, ) progression_roadmap = progression_roadmap_to_api_dict(roadmap_ctx) if start_target_only: return { "goal_query": goal_query, "max_steps_requested": max_steps, "steps": [], "step_count": 0, "target_profile_summary": None, "semantic_brief_summary": brief_to_summary_dict(semantic_brief), "semantic_llm_applied": semantic_llm_applied, "query_intent_summary": {}, "progression_graph_id": body.progression_graph_id, "path_qa": None, "gap_fill_offers": [], "progression_roadmap": progression_roadmap, "roadmap_first": False, "roadmap_only": False, "start_target_only": True, "roadmap_edited": False, "roadmap_unfilled_count": 0, "retrieval_phase": "start_target_only", } if roadmap_only: return { "goal_query": goal_query, "max_steps_requested": max_steps, "steps": [], "step_count": 0, "target_profile_summary": None, "semantic_brief_summary": brief_to_summary_dict(semantic_brief), "semantic_llm_applied": semantic_llm_applied, "query_intent_summary": {}, "progression_graph_id": body.progression_graph_id, "path_qa": None, "gap_fill_offers": [], "progression_roadmap": progression_roadmap, "roadmap_first": False, "roadmap_only": True, "roadmap_edited": roadmap_edited, "roadmap_unfilled_count": 0, "retrieval_phase": "roadmap_only", } 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, ) roadmap_unfilled: List[Tuple[int, StageSpecArtifact]] = [] roadmap_gap_offers: List[Dict[str, Any]] = [] used: Set[int] = set() steps: List[Dict[str, Any]] = [] planned_ids: List[int] = [] anchor_id: Optional[int] = None anchor_variant_id: Optional[int] = None if roadmap_first and roadmap_ctx is not None: steps, roadmap_unfilled = _build_steps_roadmap_first( cur, tenant=tenant, body=body, goal_query=goal_query, max_steps=max_steps, semantic_brief=semantic_brief, path_target_profile=path_target_profile, path_intent=path_intent, roadmap_ctx=roadmap_ctx, ) planned_ids = [int(s["exercise_id"]) for s in steps if s.get("exercise_id") is not None] if planned_ids: anchor_id = planned_ids[-1] anchor_variant_id = steps[-1].get("variant_id") if body.include_ai_gap_fill and roadmap_unfilled: major_by_index = ( {m.index: m for m in roadmap_ctx.roadmap.major_steps} if roadmap_ctx.roadmap else {} ) roadmap_gap_specs = build_roadmap_unfilled_gap_specs( unfilled_specs=roadmap_unfilled, major_steps_by_index=major_by_index, steps=steps, brief=semantic_brief, goal_query=goal_query, goal_analysis=roadmap_ctx.goal_analysis if roadmap_ctx else None, resolved_structured=roadmap_ctx.resolved_structured if roadmap_ctx else None, ) for spec in roadmap_gap_specs: roadmap_gap_offers.append( build_gap_fill_offer( spec=spec, steps=steps, goal_query=goal_query, brief=semantic_brief, proposal=None, roadmap_snapshot=_roadmap_gap_snapshot_for_spec( roadmap_ctx, spec, semantic_brief=semantic_brief ), ) ) else: 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]] = [] gap_fill_offers: List[Dict[str, Any]] = [] off_topic_steps: List[Dict[str, Any]] = [] stripped_off_topic: List[Dict[str, Any]] = [] llm_qa: Optional[Dict[str, Any]] = None llm_qa_applied = False reorder_applied = False reorder_notes: List[str] = [] roadmap_qa_mode: Optional[str] = None if body.include_path_qa: if roadmap_first: roadmap_qa_mode = "roadmap_first_lite" gaps = detect_path_gaps( cur, steps, brief=semantic_brief, roadmap_first=roadmap_first, ) unfilled_gaps: List[Dict[str, Any]] = [] if gaps and not roadmap_first: 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, ) elif gaps and roadmap_first: unfilled_gaps = list(gaps) 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 not roadmap_first 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) off_topic_steps = detect_off_topic_steps(cur, steps, brief=semantic_brief) steps, stripped_off_topic = strip_off_topic_steps_from_path(steps, off_topic_steps) if stripped_off_topic: off_topic_steps = [] gaps = detect_path_gaps( cur, steps, brief=semantic_brief, roadmap_first=roadmap_first, ) llm_gap_specs = parse_llm_suggested_new_exercises( llm_qa, brief=semantic_brief, step_count=len(steps), ) if body.include_ai_gap_fill: fresh_large_gaps = [g for g in gaps if g.get("is_large_gap")] gap_specs = collect_gap_fill_specs( steps=steps, unfilled_gaps=fresh_large_gaps or unfilled_gaps, off_topic_steps=off_topic_steps, llm_specs=llm_gap_specs, brief=semantic_brief, goal_query=goal_query, ) path_roadmap_snapshot = None if roadmap_ctx: path_roadmap_snapshot = build_progression_gap_snapshot( goal_analysis=( roadmap_ctx.goal_analysis.model_dump() if roadmap_ctx.goal_analysis else None ), resolved_structured=( roadmap_ctx.resolved_structured.model_dump() if roadmap_ctx.resolved_structured else None ), semantic_brief=roadmap_ctx.semantic_brief or brief_to_summary_dict(semantic_brief), ) steps, ai_proposals, gap_fill_offers = apply_gap_fill_after_qa( cur, steps, gap_specs, goal_query=goal_query, brief=semantic_brief, include_ai_calls=False, max_ai_proposals=0, auto_insert_proposals=False, roadmap_snapshot=path_roadmap_snapshot, ) if roadmap_gap_offers: seen_offer_ids = {o.get("offer_id") for o in gap_fill_offers} for offer in roadmap_gap_offers: if offer.get("offer_id") not in seen_offer_ids: gap_fill_offers.append(offer) path_qa = build_path_qa_summary( gaps=gaps, bridge_inserts=bridge_inserts, ai_proposals=ai_proposals, gap_fill_offers=gap_fill_offers, off_topic_steps=off_topic_steps, stripped_off_topic=stripped_off_topic, llm_qa=llm_qa, llm_applied=llm_qa_applied, reorder_applied=reorder_applied, reorder_notes=reorder_notes, roadmap_qa_mode=roadmap_qa_mode, ) target_profile_summary = path_target_profile.to_summary_dict(cur) retrieval_parts = ["profile_v1", "full_library", "path_builder", "semantics"] if roadmap_first: retrieval_parts.append("roadmap_first") if roadmap_qa_mode: retrieval_parts.append(roadmap_qa_mode) 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") if gap_fill_offers: retrieval_parts.append("gap_fill_offers") if include_roadmap: retrieval_parts.append("roadmap_preview") if roadmap_edited: retrieval_parts.append("roadmap_edited") if roadmap_unfilled: retrieval_parts.append("roadmap_unfilled") 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, "gap_fill_offers": gap_fill_offers, "progression_roadmap": progression_roadmap, "roadmap_first": roadmap_first, "roadmap_only": False, "roadmap_edited": roadmap_edited, "roadmap_unfilled_count": len(roadmap_unfilled), "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