shinkan-jinkendo/backend/planning_path_qa_pipeline.py

"""
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",
]