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", "localLlmModel": "qwen3-8b"}


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",
            # Welches lokale Modell llama-swap laden soll (B0.5). Muss zu einem
            # Key in xtts/llama-swap/config.yaml passen.
            "localLlmModel": (data.get("localLlmModel") or "qwen3-8b").strip(),
        }
    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.

# CLAUDE-ONLY-Themen → nicht lokal. Seit B1b hat das lokale Tier Werkzeuge
# (web_search, memory_search, trigger_timer, Spotify), daher gehen Wetter, News,
# Fakten, Timer, Musik, Gedaechtnis-Suche jetzt LOKAL. Nur was das lokale Tier
# nicht kann bleibt hier: Bilder, Skills, Projekte, OAuth, Smart-Home (keine
# Anbindung), Kalender/Mail (kein Tool).
_TOOL_HINTS = re.compile(
    r"\b(bild|generier|male?\b|malen|zeichne|foto|"
    r"skill|projekt|oauth|"
    r"licht|lampe|steckdose|rollade|heizung|"
    r"kalender|termin|"
    r"maild?|e-?mail|nachricht schreiben)\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).

# Was das lokale Tier NICHT selbst kann → dafuer eskaliert es an Claude.
_AWARENESS = (
    "Nur das grosse Modell (nicht du im Schnell-Modus) kann: Bilder generieren, "
    "Skills bauen/aendern, Projekte & OAuth verwalten, ins Gedaechtnis SCHREIBEN, "
    "sowie tiefe/technische Analysen und langen Code."
)


def build_local_system_prompt(identity_anchor: str, has_tools: bool = False,
                              pinned_persona: str = "") -> str:
    """Schlanker System-Prompt fuers lokale LLM. identity_anchor = derselbe
    Anker wie bei Claude (Rolle haelt)."""
    parts = [
        identity_anchor.strip(),
        "",
        "## SCHNELL-MODUS",
        "Du laeufst gerade als schnelles lokales Modell fuer Alltags-Konversation "
        "und einfache Aufgaben. Antworte knapp, freundlich, auf Deutsch, als ARIA.",
        "WICHTIG zur Ausgabe: Antworte in ganz NORMALEM Text. Verwende KEINE "
        "`<voice>`-Tags, kein `[FILE:]`, kein `<tool_call>`, kein HTML/Markup — "
        "nur ein oder zwei natuerliche Saetze. (Der Text wird direkt angezeigt "
        "UND vorgelesen.)",
    ]
    if has_tools:
        parts += [
            "",
            "## DEINE WERKZEUGE — nur nutzen wenn die Frage es WIRKLICH braucht",
            "- `web_search`: aktuelle Infos aus dem Netz — Wetter, News, Fakten, "
            "Preise, Oeffnungszeiten. Bei Wetter: nimm Stefans Ort aus dem "
            "GPS-Hinweis in der Nachricht.",
            "- `memory_search`: in ARIAs Gedaechtnis nachsehen (lesen).",
            "- `trigger_timer`: Timer/Erinnerung setzen ('in 10 Minuten…').",
            "- `run_spotify`: Musik steuern (naechster Titel, Pause, weiter).",
            "WICHTIG: Bei reinem Smalltalk ('wie gehts', Begruessung, Meinung) "
            "KEIN Werkzeug — einfach direkt antworten. Werkzeuge nur bei echtem "
            "Bedarf; erfinde keine.",
        ]
    parts += [
        "",
        "## WAS DU HIER NICHT KANNST",
        _AWARENESS,
        "Dafuer — und bei tiefen/technischen Fragen, langem Code, oder wenn du "
        f"unsicher bist — antworte AUSSCHLIESSLICH mit exakt `{ESCALATE_MARKER}` "
        "(nichts sonst). Dann uebernimmt das grosse Modell mit vollem Zugriff. "
        "Lieber einmal eskalieren als falsch raten oder ein Werkzeug erfinden.",
    ]
    if pinned_persona.strip():
        parts += ["", "## PERSONA", pinned_persona.strip()]
    return "\n".join(parts)