""" Mehrstufige Pfad-QS — Findings pro Stufe, daraus Optimierungspotenziale ableiten. Stufen (allgemein, domänenneutral): 1. deterministische Gates (Technik-Scope, Ausschlüsse, Stufen-Fit) 2. Übergangs-Kohärenz (Lücken zwischen Schritten) 3. LLM-Ganzpfad-Bewertung (Empfehlungen, keine Auto-Patches) """ from __future__ import annotations from typing import Any, Dict, List, Mapping, Optional, Sequence _ACTION_BY_ISSUE: Dict[str, str] = { "technique_scope": "rematch_slot", "path_exclude": "rematch_slot", "stage_mismatch": "refine_stage_spec", "off_topic": "rematch_slot", "gap": "bridge_or_gap_fill", "large_gap": "bridge_or_gap_fill", "roadmap_unfilled": "rematch_slot", } def _action_for_finding(finding: Mapping[str, Any]) -> str: issue = str(finding.get("issue") or finding.get("type") or "").strip().lower() if finding.get("is_large_gap"): return "bridge_or_gap_fill" return _ACTION_BY_ISSUE.get(issue, "review") def _hint_from_finding(finding: Mapping[str, Any], *, tier: str) -> Dict[str, Any]: step_index = finding.get("step_index") if step_index is None: step_index = finding.get("major_step_index") issue = str(finding.get("issue") or finding.get("type") or tier) action = _action_for_finding(finding) title = str(finding.get("title") or finding.get("removed_title") or "").strip() reasons = finding.get("reasons") or [] reason = reasons[0] if reasons else str(finding.get("rationale") or finding.get("detail") or "") hint: Dict[str, Any] = { "tier": tier, "action": action, "issue": issue, "step_index": step_index, "title": title or None, "reason": (reason or "")[:400] or None, } if finding.get("roadmap_learning_goal"): hint["roadmap_learning_goal"] = finding.get("roadmap_learning_goal") if finding.get("roadmap_major_step_index") is not None: hint["roadmap_major_step_index"] = finding.get("roadmap_major_step_index") return {k: v for k, v in hint.items() if v is not None and v != ""} def derive_optimization_hints( tiers: Sequence[Mapping[str, Any]], ) -> List[Dict[str, Any]]: """Aus QS-Stufen konkrete Optimierungsaktionen (ohne anfrage-spezifische Heuristiken).""" hints: List[Dict[str, Any]] = [] seen: set[tuple] = set() for tier in tiers: tier_id = str(tier.get("id") or "") for finding in tier.get("findings") or []: if not isinstance(finding, dict): continue hint = _hint_from_finding(finding, tier=tier_id) key = ( hint.get("tier"), hint.get("action"), hint.get("step_index"), hint.get("issue"), ) if key in seen: continue seen.add(key) hints.append(hint) return hints[:24] def run_multistage_path_qa( *, off_topic_steps: Sequence[Mapping[str, Any]], stripped_off_topic: Sequence[Mapping[str, Any]], gaps: Sequence[Mapping[str, Any]], llm_qa: Optional[Mapping[str, Any]] = None, llm_applied: bool = False, roadmap_unfilled: Optional[Sequence[Mapping[str, Any]]] = None, ) -> Dict[str, Any]: """Orchestriert QS-Stufen und leitet Optimierungspotenziale ab.""" tier1_findings: List[Dict[str, Any]] = [] for item in stripped_off_topic or off_topic_steps or []: if isinstance(item, dict): tier1_findings.append(dict(item)) tier2_findings: List[Dict[str, Any]] = [dict(g) for g in gaps if isinstance(g, dict)] tier3_findings: List[Dict[str, Any]] = [] llm_recommendations: List[str] = [] if llm_applied and llm_qa: q_score = llm_qa.get("quality_score") tier3_findings.append( { "issue": "llm_assessment", "quality_score": q_score, "overall_ok": llm_qa.get("overall_ok"), "detail": llm_qa.get("summary") or llm_qa.get("assessment"), } ) for raw in llm_qa.get("recommendations") or llm_qa.get("suggestions") or []: s = str(raw or "").strip() if s: llm_recommendations.append(s[:500]) unfilled = list(roadmap_unfilled or []) if unfilled: for item in unfilled: if isinstance(item, (list, tuple)) and len(item) >= 2: idx, spec = item[0], item[1] tier1_findings.append( { "issue": "roadmap_unfilled", "step_index": int(idx), "roadmap_major_step_index": getattr(spec, "major_step_index", idx), "roadmap_learning_goal": getattr(spec, "learning_goal", None), "reasons": ["Keine passende Übung für Roadmap-Stufe"], } ) elif isinstance(item, dict): tier1_findings.append({**item, "issue": item.get("issue") or "roadmap_unfilled"}) tiers: List[Dict[str, Any]] = [ { "id": "tier1_deterministic", "label": "Deterministische Gates", "finding_count": len(tier1_findings), "findings": tier1_findings[:16], }, { "id": "tier2_transitions", "label": "Übergangs-Kohärenz", "finding_count": len(tier2_findings), "findings": tier2_findings[:12], }, { "id": "tier3_llm_holistic", "label": "LLM-Ganzpfad", "finding_count": len(tier3_findings), "findings": tier3_findings, "recommendations": llm_recommendations[:8], "applied": bool(llm_applied), }, ] optimization_hints = derive_optimization_hints(tiers) for rec in llm_recommendations[:5]: optimization_hints.append( { "tier": "tier3_llm_holistic", "action": "review_roadmap", "issue": "llm_recommendation", "reason": rec, } ) return { "qa_tiers": tiers, "optimization_hints": optimization_hints[:28], "optimization_hint_count": len(optimization_hints), } __all__ = [ "derive_optimization_hints", "run_multistage_path_qa", ]