ARIA-AGENT/aria-brain/router.py

"""
Router (Plan B, B1a) — entscheidet pro Turn: lokales schnelles LLM oder Claude.

Gestaffelt:
- B1a (hier): „nur reden" — einfache Plauder-Turns → lokales Qwen (schlanker
  Prompt, KEINE Tools). Antwortet es sauber → fertig in <1 s. Sagt es
  `<<ESCALATE>>`, braucht ein Tool oder faellt aus → Claude (bestehender Pfad).
- B1b (spaeter): kuratierte lokale Tools + lokale Tool-Loop.

Schalter kommen aus /shared/config/local_llm.json (Diagnostic schreibt, Brain
liest pro Request):
    {
      "enabled": false,        # Master: lokales Tier an/aus (aus = alles Claude)
      "localOnly": false,      # Eval: erzwinge lokal, KEIN Claude-Fallback
      "toolVariant": "slim"    # "slim" | "full"  (B1b; "full" braucht mehr VRAM)
    }
Default (Datei fehlt/kaputt): enabled=false → Verhalten wie bisher (alles Claude).
"""

from __future__ import annotations

import json
import logging
import os
import re

logger = logging.getLogger(__name__)

CONFIG_PATH = os.environ.get("LOCAL_LLM_CONFIG", "/shared/config/local_llm.json")

ESCALATE_MARKER = "<<ESCALATE>>"

DEFAULT_CONFIG = {"enabled": False, "localOnly": False, "toolVariant": "slim"}


def load_config() -> dict:
    """Liest die Schalter. Nie werfen — bei Fehler Defaults (= alles Claude)."""
    try:
        with open(CONFIG_PATH, encoding="utf-8") as f:
            data = json.load(f) or {}
        return {
            "enabled": bool(data.get("enabled", False)),
            "localOnly": bool(data.get("localOnly", False)),
            "toolVariant": data.get("toolVariant", "slim") or "slim",
        }
    except (FileNotFoundError, json.JSONDecodeError):
        return dict(DEFAULT_CONFIG)
    except Exception as exc:
        logger.debug("local_llm-Config lesen fehlgeschlagen: %s", exc)
        return dict(DEFAULT_CONFIG)


# ── Heuristik: ist dieser Turn „einfach genug" fuers lokale Tier (B1a)? ──
#
# B1a ist reden-ohne-Tools. Also: alles, was ein Tool/Aktion braucht oder tief/
# technisch ist, geht an Claude. Lieber konservativ (im Zweifel Claude) — das
# lokale Tier soll nur die klaren Plauder-Turns abgreifen; Fehlklassifikation
# faengt zusaetzlich das <<ESCALATE>> im Modell selbst ab.

# Wortstämme, die auf Tool-/Aktionsbedarf hindeuten → nicht lokal (in B1a).
_TOOL_HINTS = re.compile(
    r"\b(wetter|temperatur|regne|timer|erinner|weck|licht|lampe|steckdose|"
    r"spotify|musik|lied|song|playlist|lauter|leiser|"
    r"bild|generier|male?\b|zeichne|foto|"
    r"merk dir|memory|gedächtnis|erinnere dich|"
    r"skill|trigger|projekt|oauth|spotify|kalender|termin|"
    r"such(e|st)?\b|google|internet|news|nachrichten|aktie|kurs|"
    r"maild?|email|nachricht schreiben|sende)\b",
    re.IGNORECASE,
)

# Technik-/Tiefe-Marker → Claude (lokales 8B soll das nicht raten).
_HARD_HINTS = re.compile(
    r"```|"                                    # Codeblock
    r"\b(code|fehler|error|stacktrace|exception|bug|debug|pentest|exploit|"
    r"vuln|payload|regex|sql|python|javascript|docker|kubernetes|"
    r"analysier|erklär.*genau|schritt für schritt|refactor|implementier)\b",
    re.IGNORECASE,
)

_MAX_LEN_FOR_LOCAL = 220   # laengere Nachrichten = eher komplexe Aufgaben → Claude


def should_try_local(user_message: str, cfg: dict) -> bool:
    """True, wenn der Router diesen Turn (B1a, reden-only) lokal versuchen soll.
    localOnly überschreibt die Heuristik (dann IMMER lokal)."""
    if not cfg.get("enabled"):
        return False
    if cfg.get("localOnly"):
        return True
    msg = (user_message or "").strip()
    if not msg or len(msg) > _MAX_LEN_FOR_LOCAL:
        return False
    if _TOOL_HINTS.search(msg):
        return False
    if _HARD_HINTS.search(msg):
        return False
    return True


# ── Schlanker System-Prompt fuers lokale Tier ──
#
# Klein halten (Speed!). Persona-Kern + Identitaets-Anker + kurze Awareness-Liste
# (WAS ARIA kann, ohne volle Schemas) + Escalation-Regel. KEINE Tool-Schemas,
# kein volles Memory (B1a).

_AWARENESS = (
    "ARIA kann ausserdem (ueber ihr grosses Backend, nicht in diesem schnellen "
    "Modus): Wetter/Zeit/Websuche, Timer & Erinnerungen, Musik (Spotify) & Licht "
    "steuern, Bilder generieren, ins Gedaechtnis schreiben/suchen, Skills bauen, "
    "Projekte & OAuth verwalten."
)


def build_local_system_prompt(identity_anchor: str, pinned_persona: str = "") -> str:
    """Schlanker System-Prompt fuers lokale LLM. identity_anchor = derselbe
    Anker wie bei Claude (Rolle haelt); pinned_persona optional kurz."""
    parts = [
        identity_anchor.strip(),
        "",
        "## SCHNELL-MODUS",
        "Du laeufst gerade als schnelles lokales Modell fuer einfache, kurze "
        "Alltags-Konversation. Antworte knapp, freundlich, auf Deutsch, als ARIA.",
        "",
        "## WAS DU HIER NICHT TUST",
        _AWARENESS,
        "Fuer alles, was ein Werkzeug/eine Aktion braucht, tiefer/technischer wird, "
        "aktuelles Wissen oder Nachschlagen verlangt, oder wo du unsicher bist: "
        f"antworte AUSSCHLIESSLICH mit exakt `{ESCALATE_MARKER}` (nichts sonst). "
        "Dann uebernimmt automatisch das grosse Modell mit vollem Zugriff. "
        "Lieber einmal eskalieren als falsch raten.",
    ]
    if pinned_persona.strip():
        parts += ["", "## PERSONA", pinned_persona.strip()]
    return "\n".join(parts)