Lars
a19ed02300
All checks were successful
Deploy Development / deploy (push) Successful in 44s
Test Suite / pytest-backend (push) Successful in 40s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 12s
Test Suite / k6 /health Baseline (push) Successful in 33s
Test Suite / playwright-tests (push) Successful in 1m13s
- Incremented version to 0.8.185, reflecting the implementation of Phase C3 features. - Introduced the `POST /api/planning/progression-path-suggest` endpoint for generating exercise progression paths. - Enhanced the ExerciseProgressionGraphPanel with a new ExerciseProgressionPathBuilder for reviewing and saving paths. - Updated changelog to document the new capabilities in planning AI functionality.
253 lines
8.5 KiB
Python
253 lines
8.5 KiB
Python
"""
|
|
Planungs-KI Phase C3: Pfad-Vorschläge für Progressionsgraphen.
|
|
|
|
Ziel-Freitext → iterative Hybrid-Suche (Schritt 1 mit optional LLM-Profil, Folgeschritte deterministisch).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, 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_retrieval import run_multistage_planning_retrieval
|
|
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
|
|
progression_graph_id: Optional[int] = Field(default=None, ge=1)
|
|
exercise_kind_any: Optional[List[str]] = None
|
|
|
|
|
|
def _pick_next_path_hit(
|
|
hits: List[Dict[str, Any]],
|
|
used_exercise_ids: Set[int],
|
|
) -> Optional[Dict[str, Any]]:
|
|
for hit in hits:
|
|
eid = int(hit["id"])
|
|
if eid in used_exercise_ids:
|
|
continue
|
|
return hit
|
|
return None
|
|
|
|
|
|
def _hit_to_path_step(hit: Dict[str, Any]) -> 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
|
|
return {
|
|
"exercise_id": int(hit["id"]),
|
|
"variant_id": variant_id,
|
|
"title": hit.get("title"),
|
|
"summary": hit.get("summary"),
|
|
"score": hit.get("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"),
|
|
}
|
|
|
|
|
|
def _run_path_step_retrieval(
|
|
cur,
|
|
*,
|
|
tenant: TenantContext,
|
|
goal_query: str,
|
|
step_index: 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]],
|
|
) -> Tuple[List[Dict[str, Any]], PlanningTargetProfile, Dict[str, Any], str]:
|
|
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 else None,
|
|
"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),
|
|
}
|
|
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:
|
|
heuristic_intent = resolve_planning_exercise_intent(goal_query, "free_search")
|
|
step_query = goal_query
|
|
else:
|
|
heuristic_intent = INTENT_SUGGEST_NEXT
|
|
step_query = "nächste sinnvolle übung im pfad"
|
|
|
|
has_plan_ref = bool(pack.get("has_planning_reference")) or step_index > 0
|
|
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 else step_query,
|
|
heuristic_intent=heuristic_intent,
|
|
include_llm_intent=include_llm_intent and step_index == 0,
|
|
context_summary=pipeline_context,
|
|
has_planning_reference=has_plan_ref,
|
|
)
|
|
|
|
weights = _intent_weights(intent)
|
|
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 if step_index > 0 else goal_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 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)
|
|
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,
|
|
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,
|
|
)
|
|
if step_index == 0:
|
|
first_intent_summary = query_intent_summary
|
|
|
|
hit = _pick_next_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.",
|
|
)
|
|
|
|
target_profile_summary = target_profile.to_summary_dict(cur) if target_profile else None
|
|
|
|
return {
|
|
"goal_query": goal_query,
|
|
"max_steps_requested": max_steps,
|
|
"steps": steps,
|
|
"step_count": len(steps),
|
|
"target_profile_summary": target_profile_summary,
|
|
"query_intent_summary": first_intent_summary,
|
|
"progression_graph_id": body.progression_graph_id,
|
|
"retrieval_phase": "profile_v1+full_library+path_builder",
|
|
}
|
|
|
|
|
|
__all__ = [
|
|
"ProgressionPathSuggestRequest",
|
|
"suggest_progression_path",
|
|
"_pick_next_path_hit",
|
|
]
|