Lars
c2c736dafc
All checks were successful
Deploy Development / deploy (push) Successful in 40s
Test Suite / pytest-backend (push) Successful in 51s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 13s
Test Suite / k6 /health Baseline (push) Successful in 34s
Test Suite / playwright-tests (push) Successful in 1m20s
- Introduced path reordering functionality using LLM with `ordered_step_indices`, allowing for dynamic adjustment of exercise progression paths. - Added AI gap filling capabilities, enabling the system to propose new exercises when unbridgeable gaps are detected. - Updated the backend to support new request parameters for path reordering and AI gap filling. - Enhanced frontend components to reflect these new features, including alerts for AI proposals and adjustments in exercise display. - Incremented version to 0.8.187 and updated changelog to document these significant enhancements in planning AI functionality.
428 lines
14 KiB
Python
428 lines
14 KiB
Python
"""
|
|
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_dynamic_retrieval_weights,
|
|
brief_to_summary_dict,
|
|
build_semantic_brief,
|
|
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 (
|
|
INTENT_SUGGEST_NEXT,
|
|
_enrich_planning_hits_with_variant_meta,
|
|
_intent_weights,
|
|
_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],
|
|
) -> 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)
|
|
score = float(hit.get("score") or 0.0)
|
|
key = (sem, score)
|
|
if key > best_key:
|
|
best_key = key
|
|
best = hit
|
|
return best
|
|
|
|
|
|
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,
|
|
) -> 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:
|
|
step_query = f"{semantic_brief.retrieval_query or goal_query} brücke zwischen schritten"
|
|
|
|
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 = INTENT_SUGGEST_NEXT
|
|
|
|
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",
|
|
}
|
|
|
|
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_dynamic_retrieval_weights(
|
|
_intent_weights(intent),
|
|
semantic_brief,
|
|
scenario="free_search" if step_index == 0 and not bridge_mode else "progression",
|
|
has_planning_reference=has_plan_ref,
|
|
)
|
|
|
|
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],
|
|
) -> 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,
|
|
)
|
|
return hits
|
|
|
|
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
|
|
)
|
|
|
|
used: Set[int] = set()
|
|
steps: List[Dict[str, Any]] = []
|
|
planned_ids: List[int] = []
|
|
anchor_id: Optional[int] = None
|
|
anchor_variant_id: Optional[int] = None
|
|
target_profile: Optional[PlanningTargetProfile] = None
|
|
first_intent_summary: Dict[str, Any] = {}
|
|
|
|
for step_index in range(max_steps):
|
|
hits, target_profile, query_intent_summary, _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,
|
|
)
|
|
if step_index == 0:
|
|
first_intent_summary = query_intent_summary
|
|
|
|
hit = _pick_best_path_hit(hits, used)
|
|
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,
|
|
)
|
|
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:
|
|
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 = target_profile.to_summary_dict(cur) if target_profile else None
|
|
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
|