odoo-at-payroll/pv-agent/agent/query_planner.py

"""Query-Planer für komplexe Fragen (Stufe 1, M6).

Ein kleiner LLM-Call zerlegt komplexe Fragen in 1-3 unabhängige Sub-Queries
(+ optional Geltungsjahr). Ein Heuristik-Gate entscheidet, ob der Planer-Call
überhaupt läuft — einfache Fragen bleiben deterministischer Single-Shot
(keine Zusatz-Latenz). Parse-/Call-Fehler fallen auf die Original-Frage
zurück; die Grounding-Regeln werden dadurch nie berührt.
"""
from __future__ import annotations

import json
import re
from dataclasses import dataclass
from datetime import date

from .normalize import normalize_text

YEAR_RE = re.compile(r"\b(?:19|20)\d{2}\b")
JSON_RE = re.compile(r"\{.*\}", re.DOTALL)

AGGREGATION_RE = re.compile(
    r"neuerungen|übersicht|zusammenfassung|alle\s|übersicht", re.I
)
COMPARISON_RE = re.compile(r"unterschied|vergleic|\bbzw\.|\bsowie\b", re.I)

# Muster gelten auf normalize_text(question) (NFKD-Folding). Neben dem
# gefoldeten "gunstig/losung" decken die Alternativen auch die ue-Schreibweise
# ab ("guenstigste loesung"), die Nutzer häufig ohne Umlaute eingeben.
DECISION_SUPPORT_RE = re.compile(
    r"gu(e)?nstig(e|er|ste)?\s+lo(e)?sung"
    r"|abgabenbegu(e)?nstigt"
    r"|steuerfrei.*(?:auszahl|pra(e)?m)"
    r"|(?:zusa(e)?tzl?ich|extra).*(?:auszahl|pra(e)?m)"
    r"|\beinmalig(e|n|r|s)?\b.{0,80}?\bauszah"
    r"|\bbar\s+auszah|\bbarauszah"
    r"|\b(?:wie\s*viel|wieviel)\b.{0,60}?\bkostet\b"
    r"|\bkostet\s+mich\b"
    r"|\bmitarbeiterpra(e)?m",
    re.IGNORECASE,
)
BROAD_RE = re.compile(
    r"gu(e)?nstig|m(?:o|e)?glichkeiten|\bstatt\b|alternativ", re.IGNORECASE
)
COST_INTENT_RE = re.compile(
    r"\b(?:wie\s*viel|wieviel)\b.{0,60}?\bkostet\b"
    r"|\bkostet\s+mich\b"
    r"|\beinmalig(e|n|r|s)?\b.{0,80}?\bauszah"
    r"|\bbar\s+auszah|\bbarauszah",
    re.IGNORECASE,
)


PLANNER_PROMPT = """Du planst Suchanfragen für eine Wissensbasis zur österreichischen
Personalverrechnung (kuratierte Briefings, Kollektivverträge je Branche und
Geltungsjahr, Gesetzesparagraphen).

Frage: "{question}"

Zerlege die Frage in 1-3 unabhängige Suchanfragen, die zusammen die Frage
beantworten. Regeln:
- Besteht die Frage aus einem Suchthema, gib genau eine Suchanfrage zurück
  (dann die Frage unverändert, leicht stichwortartig gekürzt).
- Suchbegriffe statt Sätze (ohne Frageformulierung, ohne "Kollektivvertrag"-
  Wortballast).
- Fragt die Frage nach einem bestimmten Jahr (Geltung/Stand), setze
  "stand_year" auf dieses vierstellige Jahr, sonst null.
- Setze "scope" je Suchanfrage: "gesetz", wenn nach der gesetzlichen/
  allgemeinen Grundlage gefragt ist (nur Gesetze und Fachbriefings, ohne
  Branchen-Kollektivverträge); "kv", wenn ausdrücklich nach kollektivvertrag-
  lichen Branchenregelungen gefragt ist; null für alles anderes.
- Setze "type" auf "survey", wenn die Frage eine Übersicht über viele
  Dokumente verlangt (Neuerungen, Entwicklungen, alle …); sonst "specific".
- Antworte ausschließlich mit JSON, ohne Erklärung:
{{"type": "specific", "queries": [{{"text": "...", "stand_year": null, "scope": null}}]}}"""


@dataclass
class SubQuery:
    text: str
    stand_year: str | None = None  # YYYY, wenn nach einem Geltungsjahr gefragt
    scope: str | None = None        # "gesetz" (ohne Branchen-KV) | "kv" | None


def is_decision_support(question: str) -> bool:
    """Erkennt Gestaltungsfragen, die den deterministischen Plan benötigen.

    Die Frage wird vorher normalisiert, damit auch Schreibweisen ohne Umlaute
    ("guenstigste loesung") erkannt werden.
    """
    return bool(DECISION_SUPPORT_RE.search(normalize_text(question)))


def should_plan(question: str) -> bool:
    """Heuristik-Gate: nur komplexe Fragen bekommen einen Planer-Call.

    Signale: Jahreszahl, Vergleichs-/Aggregationsmarker, langer Text,
    Mehrfach-Konjunktion. Einfache Fragen bleiben Single-Shot (Latenz)."""
    q = question.strip()
    if YEAR_RE.search(q):
        return True
    words = q.split()
    if len(words) >= 14:
        return True
    if COMPARISON_RE.search(q):
        return True
    if is_decision_support(q):
        return True
    if AGGREGATION_RE.search(q):
        return True
    if " und " in q.casefold() and len(words) >= 10:
        return True
    return False


def decision_support_plan(
    question: str, default_year: str | None = None
) -> tuple[list[SubQuery], str] | None:
    """Deterministische Zerlegung für Gestaltungsfragen zu zusätzlichen
    Arbeitnehmerleistungen. Sie verhindert, dass die vielen Branchen-KV-
    Treffer allgemeine Abgabenregeln verdrängen.

    Ohne genanntes Jahr wird der neueste Indexstand verwendet; die Antwort
    muss fehlenden Einzelfallkontext weiterhin über die Prompt-Regel erfragen.
    """
    if not is_decision_support(question):
        return None
    folded = normalize_text(question)
    explicit_year = YEAR_RE.search(question)
    year = explicit_year.group(0) if explicit_year else default_year
    year_text = f" {year}" if year else ""
    direct = SubQuery(
        text=(
            f"Mitarbeiterprämie{year_text} WIKU Personal aktuell Nr 12 FAQ "
            "Höchstbetrag 500 Lohnsteuer SV BV Beitragsfreie Bezugsbestandteile "
            "§ 49 Abs 3 Z 30"
        ),
        stand_year=year,
        scope="gesetz",
    )
    cash_and_benefits = SubQuery(
        text=(
            "Zukunftssicherungsmaßnahmen 300 Kalenderjahr Risikoversicherung "
            "Gruppenerfordernis Bezugsumwandlung Lohnsteuer SV BV DB DZ KommSt"
        ),
        stand_year=year,
        scope="gesetz",
    )
    purpose_bound = SubQuery(
        text=(
            "Sachzuwendungen Geschenke Barzahlungen stets steuerpflichtig Freibetrag "
            "Beitragsfreie Bezugsbestandteile Z 1 bis Z 16a Mahlzeiten "
            "Kinderbetreuungszuschuss"
        ),
        stand_year=year,
        scope="gesetz",
    )
    # Kostenaufstellung: einmalige Barzahlung -> Lohnsteuer-Behandlung der
    # Bezüge plus Arbeitgeberbelastung (SV-Beitragssätze, DB/DZ, KommSt).
    cost_tax = SubQuery(
        text=(
            "einmalige Bezüge Sonderzahlungen Lohnsteuer Jahressechstel "
            "6 Prozent steuerfrei 620 Freibetrag Zuwendung EStG"
        ),
        stand_year=year,
        scope="gesetz",
    )
    cost_lnk = SubQuery(
        text=(
            "Beitragssätze des jeweiligen Jahres für Dienstnehmer "
            "Dienstgeberanteil Sozialversicherungsbeiträge DB DZ Zuschlag "
            "zum Dienstgeberbeitrag Kommunalsteuer Lohnnebenkosten Zuwendung"
        ),
        stand_year=year,
        scope="gesetz",
    )
    broad = bool(BROAD_RE.search(folded))
    cost_intent = bool(COST_INTENT_RE.search(folded))
    queries = [direct]
    if broad and not cost_intent:
        # Ein Gestaltungsvergleich ist kein Korpus-Survey: Die gezielten
        # Queries passen in den normalen Antwortpfad und vermeiden unnötiges
        # Map-Reduce samt zusätzlicher Latenz/Fehlverweigerungsrisiko.
        queries += [cash_and_benefits, purpose_bound]
    elif broad:
        # Explizite Kostenaufstellung: Die Zukunftssicherungs-Query verdrängt
        # bei Slots für vier Queries die Beitrags-/Lohnsteuerquellen; die
        # zweckgebundenen Alternativen bleiben über purpose_bound abgedeckt.
        queries.append(purpose_bound)
    if cost_intent:
        queries += [cost_tax, cost_lnk]
    return queries[:4], "specific"


def parse_plan(raw: str, original: str) -> tuple[list[SubQuery], str]:
    """Robustes JSON-Parsing; jeder Fehler → [Originalfrage]. Liefert
    (Sub-Queries, Fragetyp 'survey' | 'specific')."""
    try:
        match = JSON_RE.search(raw)
        if not match:
            raise ValueError("kein JSON-Objekt")
        data = json.loads(match.group(0))
        items = data.get("queries")
        if not isinstance(items, list) or not items:
            raise ValueError("leeres Plan-Array")
        subs: list[SubQuery] = []
        for item in items[:3]:
            text = str(item.get("text", "")).strip()
            if not text:
                raise ValueError("leere Sub-Query")
            year = item.get("stand_year")
            year = str(year) if year and re.fullmatch(r"20\d{2}", str(year)) else None
            scope = item.get("scope")
            scope = scope if scope in ("gesetz", "kv") else None
            subs.append(SubQuery(text=text, stand_year=year, scope=scope))
        qtype = data.get("type") if data.get("type") in ("survey", "specific") else "specific"
        return subs, qtype
    except (ValueError, TypeError, KeyError, json.JSONDecodeError):
        return [SubQuery(text=original)], "specific"
        return [SubQuery(text=original)]


def plan_queries(
    question: str,
    client,
    cfg,
) -> tuple[list[SubQuery], bool, str]:
    """Liefert (Sub-Queries, geplant?, Fragetyp) — Call-/Parse-Fehler →
    Original als Einzel-Query, Typ 'specific'. Der Planer-Call ist klein
    (Frage ohne Kontext, kurzes num_predict); Temperature 0."""
    deterministic = decision_support_plan(question, default_year=str(date.today().year))
    if deterministic:
        subs, qtype = deterministic
        return subs, True, qtype
    if not should_plan(question):
        return [SubQuery(text=question)], False, "specific"
    prompt = PLANNER_PROMPT.format(question=question.strip())
    try:
        raw = client.chat(
            cfg.planner_model or cfg.answer_model,
            [
                {"role": "system", "content": "Du antwortest ausschließlich mit JSON."},
                {"role": "user", "content": prompt},
            ],
            temperature=0.0,
            num_ctx=cfg.num_ctx,
            num_predict=cfg.planner_num_predict,
            think=False,
        )
    except Exception:
        return [SubQuery(text=question)], False, "specific"
    subs, qtype = parse_plan(raw, question)
    return subs, True, qtype