refactor(brain): Fast-Path als Skill-Capability — fast_patterns im Manifest

Frueher: Spotify-spezifische Patterns hardcoded in agent.py — jeder neue
Steuer-Skill haette wieder Brain-Code-Aenderungen gebraucht.
Jetzt: jeder Skill deklariert seine eigenen Patterns im Manifest unter
fast_patterns: [{match, args, reply}]. Brain iteriert generisch, kein
Skill bekommt Sonderbehandlung.

- agent.py: _try_skill_fast_path liest aus skills.list_skills(), keine
  Spotify-Konstanten mehr. skill_create/skill_update Tool-Schema kennt
  fast_patterns (mit Beispiel + Wann-nutzen-Hinweis).
- skills.py: _normalize_fast_patterns validiert Regex + filtert kaputte
  Eintraege; create_skill/update_skill akzeptieren das Feld.
- main.py: einmalige Lifespan-Migration — wenn spotify-Skill existiert
  und kein fast_patterns hat, werden die alten Hardcoded-Patterns
  rueberkopiert. Idempotent, laeuft bei jedem Restart sicher mehrfach.
- seed_rules.py: neue Regel `seed/skill-rule/fast-patterns-for-control`
  erklaert ARIA wann sie das Feature nutzen soll (reines Steuern: ja,
  kreativer Output / Parametrisierung: nein) — mit Beispiel.

Trade-off: Volume-Patterns (lauter/leiser) fallen aus dem Fast-Path raus,
weil die Multi-Step-Logik (GET state → compute → PUT) sich nicht
deklarativ ausdruecken laesst. Wer das zurueck will: Spotify-Skill um
einen action=volume_relative-Arg erweitern der die Mathe intern macht.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-06-06 08:56:15 +02:00
parent e04bbef361
commit 61c9183033
4 changed files with 231 additions and 111 deletions
+90 -111
View File
@@ -127,6 +127,25 @@ META_TOOLS = [
"items": {"type": "object"},
"description": "Argumente-Schema [{name, type, required, description}]",
},
"fast_patterns": {
"type": "array",
"items": {"type": "object"},
"description": (
"OPTIONAL — fuer 'reines Steuern'-Skills (Licht an/aus, Spotify "
"pause/next, Rollade hoch/runter etc.) eine Liste von "
"[{match, args, reply}] eintragen. Wenn ein User-Befehl gegen "
"match (anchored Regex, case-insensitive) matched, ruft das "
"Brain run_skill(name, args) DIREKT auf und gibt reply zurueck — "
"ohne Claude (~5s Latenz gespart). Match wird gegen den "
"normalisierten Text (lowercase, Endsatzzeichen weg) gemacht; "
"schreibe Patterns mit ^...$ damit nur exakte Befehle matchen "
"und nicht Teilstrings (z.B. ^pause$ statt pause). NICHT fuer "
"Skills mit kreativem Output / parametrisierter Logik — die "
"brauchen Claude. Beispiel: "
"[{\"match\":\"^pause$\",\"args\":{\"path\":\"/v1/me/player/pause\",\"method\":\"PUT\"},"
"\"reply\":\"Spotify: pausiert ⏸\"}]"
),
},
},
"required": ["name", "description", "entry_code"],
},
@@ -193,6 +212,16 @@ META_TOOLS = [
"Setzt Stefan in Diagnostic; Skill bekommt CFG_<NAME> ENV."
),
},
"fast_patterns": {
"type": "array",
"items": {"type": "object"},
"description": (
"Optional komplette Fast-Path-Patterns-Liste UEBERSCHREIBEN — "
"[{match, args, reply}]. Siehe skill_create-Beschreibung fuer "
"Format. Leere Liste = alle Fast-Paths entfernen (alles geht "
"wieder durch Claude). Wenn nicht angegeben: bleibt unberuehrt."
),
},
},
"required": ["name"],
},
@@ -782,61 +811,26 @@ META_TOOLS = [
]
# ── Spotify Fast-Path ──────────────────────────────────────────────────
# ── Fast-Path (Skill-deklariert) ───────────────────────────────────────
#
# Einfache Media-Commands (nächster Track, Pause, lauter, ...) gehen
# direkt aufs spotify-Skill statt durch die volle Claude-Reasoning-Pipeline.
# Latenz: ~1-1.5s statt 5-10s. Stefan-Bug 06/2026: "ARIA braucht ewig nur
# fuer 'nächster Track'". Wenn ein Pattern nicht matcht, faellt der Call
# wie bisher in die normale chat()-Loop und Claude entscheidet — keine
# Funktionalitaet geht verloren.
# Skills koennen in ihrem Manifest `fast_patterns` deklarieren — eine Liste
# von {match: regex, args: dict, reply: str}. Wenn ein User-Text gegen
# ein Pattern matcht, ruft das Brain direkt run_skill(name, args) auf und
# returnt `reply` an den User — Claude wird komplett uebersprungen. Spart
# 5-10s LLM-Latenz pro "reines Steuern"-Befehl.
#
# Patterns sind anchored (^...$) gegen normalisierten Text (lowercase,
# Endsatzzeichen weg, Whitespace gestrafft). Bewusst eng gefasst: lieber
# einmal in Claude fallen als ein Kontextsatz wie "ich war kurz zurueck"
# faelschlich als "previous track" interpretieren.
_SPOTIFY_FAST_PATTERNS: list[tuple[str, str, str, Optional[int]]] = [
# (regex, action, http-method, volume-delta)
# NEXT
(r"^(naechster|nächster|naechste|nächste) (track|song|titel|lied)$", "next", "POST", None),
(r"^(weiter|skip|ueberspringen|überspringen|ueberspring|überspring)$", "next", "POST", None),
# PREVIOUS
(r"^(vorheriger|vorheriges|letzter|letztes) (track|song|titel|lied)$", "previous", "POST", None),
(r"^(zurueck|zurück)$", "previous", "POST", None),
# PAUSE
(r"^(pause|pausiere|pausieren|stop|stopp|halt)$", "pause", "PUT", None),
(r"^(musik|spotify) (pause|aus|stop|stopp)$", "pause", "PUT", None),
# PLAY / RESUME
(r"^(play|weiterspielen|weiter spielen|fortsetzen|abspielen)$", "play", "PUT", None),
(r"^(musik|spotify) (an|wieder an|weiter|fortsetzen)$", "play", "PUT", None),
# VOLUME — Delta wird auf den aktuell ermittelten Volume-Wert aufaddiert
(r"^(lauter|musik lauter|spotify lauter|volume hoch|lautstärke hoch)$", "volume", "PUT", 10),
(r"^(leiser|musik leiser|spotify leiser|volume runter|lautstärke runter)$", "volume", "PUT", -10),
(r"^(viel lauter|deutlich lauter)$", "volume", "PUT", 20),
(r"^(viel leiser|deutlich leiser)$", "volume", "PUT", -20),
]
# Patterns sollten anchored (^...$) gegen den normalisierten Text (lower-
# case, Endsatzzeichen weg, Whitespace gestrafft) geschrieben sein. Lieber
# eng matchen als breit — false-positives sind teurer als ein Cache-Miss.
#
# Diese Logik ist generisch — ARIA deklariert die Patterns selbst beim
# skill_create / skill_update, das Brain orchestriert nur.
def _spotify_fast_match(text: str) -> Optional[tuple[str, str, Optional[int]]]:
"""Returns (action, method, volume_delta) wenn ein Pattern matcht — sonst None."""
def _normalize_for_fast_match(text: str) -> str:
norm = (text or "").strip().lower()
norm = re.sub(r"[.!?]+$", "", norm)
norm = re.sub(r"\s+", " ", norm)
if not norm:
return None
for rx, action, method, delta in _SPOTIFY_FAST_PATTERNS:
if re.match(rx, norm):
return action, method, delta
return None
def _run_spotify_call(path: str, method: str, body: Optional[dict] = None) -> dict:
"""Fuehrt einen Spotify-Skill-Call aus. Skill-Args: path, method, body (JSON-String).
Returns das run_skill-Ergebnis."""
args: dict = {"path": path, "method": method}
if body is not None:
args["body"] = json.dumps(body)
return skills_mod.run_skill("spotify", args, timeout_sec=15)
return norm
def _skill_to_tool(s: dict) -> dict:
@@ -906,71 +900,52 @@ class Agent:
self._pending_events = []
return events
def _try_spotify_fast_path(self, user_message: str) -> Optional[str]:
"""Wenn die Nachricht ein einfacher Media-Command ist, direkt aufs
spotify-Skill routen und ein kurzes Reply zurueckgeben — Claude wird
komplett uebersprungen. Returnt None wenn kein Pattern matcht oder das
spotify-Skill nicht installiert ist (dann faellt's normal in Claude)."""
m = _spotify_fast_match(user_message)
if m is None:
return None
action, method, delta = m
def _try_skill_fast_path(self, user_message: str) -> Optional[str]:
"""Iteriert ueber alle aktiven Skills und probiert deren fast_patterns
gegen den normalisierten User-Text. Erster Treffer gewinnt — Skill
wird direkt aufgerufen, Reply geht ohne Claude zurueck.
# Skill muss installiert + aktiv sein. Sonst Fall-Through zu Claude.
try:
manifest = skills_mod.read_manifest("spotify")
except Exception:
manifest = None
if not manifest or not manifest.get("active", True):
logger.info("[spotify-fast] skill nicht verfuegbar — fall through zu Claude")
Returnt None wenn kein Pattern matcht. Bei Skill-Ausfuehrungs-Fehler
(ok=False) wird eine ehrliche Fehler-Reply gegeben statt durch Claude
zu fallen — sonst kostet ein gescheiterter Fast-Path doppelt (~1s
Skill-Versuch + 5-10s Claude). Bei unerwarteter Exception fallen wir
durch zu Claude (Claude kann ggf. besser diagnostizieren)."""
norm = _normalize_for_fast_match(user_message)
if not norm:
return None
logger.info("[spotify-fast] match action=%s method=%s delta=%s msg=%r",
action, method, delta, user_message[:60])
def _err_reply(label: str, res: dict) -> str:
# ok=False kommt von 401 (nicht eingeloggt), 404 (kein aktives
# Gerät) etc. — Skill schreibt den Spotify-Error nach stderr.
tail = (res.get("stderr") or res.get("stdout") or "").strip().splitlines()
hint = (tail[-1] if tail else "")[:120]
return f"Spotify: {label} fehlgeschlagen — {hint or 'siehe Brain-Log'}"
try:
if action == "next":
res = _run_spotify_call("/v1/me/player/next", method)
return "Spotify: nächster Track ⏭" if res.get("ok") else _err_reply("Skip", res)
if action == "previous":
res = _run_spotify_call("/v1/me/player/previous", method)
return "Spotify: vorheriger Track ⏮" if res.get("ok") else _err_reply("Zurück", res)
if action == "pause":
res = _run_spotify_call("/v1/me/player/pause", method)
return "Spotify: pausiert ⏸" if res.get("ok") else _err_reply("Pause", res)
if action == "play":
res = _run_spotify_call("/v1/me/player/play", method)
return "Spotify: spielt ▶" if res.get("ok") else _err_reply("Play", res)
if action == "volume" and delta is not None:
state = _run_spotify_call("/v1/me/player", "GET")
if not state.get("ok"):
return _err_reply("Lautstärke-Status", state)
cur_vol = 50
active_skills = [s for s in skills_mod.list_skills(active_only=False)
if s.get("active", True)]
for skill in active_skills:
patterns = skill.get("fast_patterns") or []
if not patterns:
continue
skill_name = skill.get("name") or ""
for pat in patterns:
rx = pat.get("match") or ""
if not rx:
continue
try:
out = (state.get("stdout") or "").strip()
if out:
data = json.loads(out)
dev = data.get("device") or {}
cur_vol = int(dev.get("volume_percent", 50))
if not re.match(rx, norm, re.IGNORECASE):
continue
except re.error:
# Sollte durch _normalize_fast_patterns rausgefiltert sein.
continue
args = pat.get("args") or {}
reply = pat.get("reply") or f"{skill_name}: ok"
logger.info("[fast-path] match skill=%s pattern=%r msg=%r",
skill_name, rx, user_message[:60])
try:
res = skills_mod.run_skill(skill_name, dict(args), timeout_sec=15)
except Exception as exc:
logger.warning("[spotify-fast] volume-state parse: %s", exc)
new_vol = max(0, min(100, cur_vol + delta))
res = _run_spotify_call(f"/v1/me/player/volume?volume_percent={new_vol}", "PUT")
logger.warning("[fast-path] %s exception — fall through zu Claude: %s",
skill_name, exc)
return None
if not res.get("ok"):
return _err_reply("Lautstärke", res)
arrow = "🔊" if delta > 0 else "🔉"
return f"Spotify: Lautstärke {new_vol}% {arrow}"
except Exception as exc:
logger.warning("[spotify-fast] action=%s exception — fall through zu Claude: %s",
action, exc)
return None
tail = (res.get("stderr") or res.get("stdout") or "").strip().splitlines()
hint = (tail[-1] if tail else "")[:120]
return f"{skill_name}: {reply} — Fehler: {hint or 'siehe Brain-Log'}"
return reply
return None
# ── Hauptpfad: ein User-Turn → Tool-Loop → finaler Reply ──
@@ -985,9 +960,10 @@ class Agent:
# Events vom letzten Turn weglassen
self._pending_events = []
# Spotify Fast-Path: einfache Media-Commands ueberspringen Claude komplett.
# Spart 4-9s Latenz fuer 'naechster Track', 'Pause', 'lauter' etc.
fast_reply = self._try_spotify_fast_path(user_message)
# Fast-Path: einfache "reines Steuern"-Commands ueberspringen Claude komplett.
# Jeder Skill kann in seinem Manifest fast_patterns deklarieren — das Brain
# iteriert hier ueber alle aktiven Skills und matched. Spart 5-10s Latenz.
fast_reply = self._try_skill_fast_path(user_message)
if fast_reply is not None:
self.conversation.add("user", user_message, source=source)
self.conversation.add("assistant", fast_reply)
@@ -1133,6 +1109,7 @@ class Agent:
args=arguments.get("args", []),
pip_packages=arguments.get("pip_packages", []),
config_schema=arguments.get("config_schema") or None,
fast_patterns=arguments.get("fast_patterns") or None,
author="aria",
)
# Side-Channel-Event: Stefan soll sehen wenn ARIA was anlegt
@@ -1196,6 +1173,8 @@ class Agent:
patch["pip_packages"] = arguments["pip_packages"]
if "config_schema" in arguments and isinstance(arguments["config_schema"], list):
patch["config_schema"] = arguments["config_schema"]
if "fast_patterns" in arguments and isinstance(arguments["fast_patterns"], list):
patch["fast_patterns"] = arguments["fast_patterns"]
if not patch:
return "FEHLER: keine Felder zum Update angegeben."
try:
+57
View File
@@ -45,6 +45,54 @@ logger = logging.getLogger("aria-brain")
QDRANT_HOST = os.environ.get("QDRANT_HOST", "aria-qdrant")
QDRANT_PORT = int(os.environ.get("QDRANT_PORT", "6333"))
def _seed_spotify_fast_patterns() -> None:
"""One-shot Migration: schreibt Standard-Steuer-Patterns ins Spotify-Skill
wenn das Skill existiert + aktiv ist + noch keine fast_patterns hat.
Nach diesem Run kann ARIA die Patterns frei via skill_update aendern."""
manifest = skills_mod.read_manifest("spotify")
if not manifest:
logger.info("[migrate] spotify skill nicht vorhanden — nichts zu tun")
return
if manifest.get("fast_patterns"):
logger.info("[migrate] spotify hat schon fast_patterns (%d) — skip",
len(manifest["fast_patterns"]))
return
default_patterns = [
# NEXT
{"match": r"^(naechster|nächster|naechste|nächste) (track|song|titel|lied)$",
"args": {"path": "/v1/me/player/next", "method": "POST"},
"reply": "Spotify: nächster Track ⏭"},
{"match": r"^(weiter|skip|ueberspringen|überspringen|ueberspring|überspring)$",
"args": {"path": "/v1/me/player/next", "method": "POST"},
"reply": "Spotify: nächster Track ⏭"},
# PREVIOUS
{"match": r"^(vorheriger|vorheriges|letzter|letztes) (track|song|titel|lied)$",
"args": {"path": "/v1/me/player/previous", "method": "POST"},
"reply": "Spotify: vorheriger Track ⏮"},
{"match": r"^(zurueck|zurück)$",
"args": {"path": "/v1/me/player/previous", "method": "POST"},
"reply": "Spotify: vorheriger Track ⏮"},
# PAUSE
{"match": r"^(pause|pausiere|pausieren|stop|stopp|halt)$",
"args": {"path": "/v1/me/player/pause", "method": "PUT"},
"reply": "Spotify: pausiert ⏸"},
{"match": r"^(musik|spotify) (pause|aus|stop|stopp)$",
"args": {"path": "/v1/me/player/pause", "method": "PUT"},
"reply": "Spotify: pausiert ⏸"},
# PLAY
{"match": r"^(play|weiterspielen|weiter spielen|fortsetzen|abspielen)$",
"args": {"path": "/v1/me/player/play", "method": "PUT"},
"reply": "Spotify: spielt ▶"},
{"match": r"^(musik|spotify) (an|wieder an|weiter|fortsetzen)$",
"args": {"path": "/v1/me/player/play", "method": "PUT"},
"reply": "Spotify: spielt ▶"},
]
skills_mod.update_skill("spotify", {"fast_patterns": default_patterns})
logger.info("[migrate] spotify fast_patterns gesetzt (%d Eintraege)",
len(default_patterns))
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Beim Brain-Start: System-Seed-Regeln idempotent in DB schreiben,
@@ -54,6 +102,15 @@ async def lifespan(app: FastAPI):
logger.info("Lifespan: seed_rules angewendet (%s)", result)
except Exception as exc:
logger.exception("Lifespan: seed_rules fehlgeschlagen — Brain startet trotzdem (%s)", exc)
# Einmalige Migration: Spotify-Skill ohne fast_patterns kriegt die Standard-
# Patterns injiziert. Idempotent — wenn schon welche da sind, nichts tun.
# ARIA kann sie spaeter via skill_update beliebig erweitern/ersetzen.
try:
_seed_spotify_fast_patterns()
except Exception as exc:
logger.warning("Lifespan: spotify fast_patterns Migration: %s", exc)
task = asyncio.create_task(background_mod.run_loop(agent))
logger.info("Lifespan: Trigger-Loop gestartet")
try:
+48
View File
@@ -131,6 +131,54 @@ SEED_RULES: List[dict] = [
"Skill-Friedhof und Stefan muss aufraeumen."
),
},
{
"migration_key": "seed/skill-rule/fast-patterns-for-control",
"type": "rule",
"title": "Skill-Regel: fast_patterns fuer reines Steuern (spart 5-10s Latenz)",
"category": "skills",
"content": (
"Wenn Du einen Skill baust oder aktualisierst, der **reine Steuer-"
"Befehle** behandelt (Licht an/aus, Spotify pause/next, Rollade "
"hoch/runter, Heizung +1°), trag ins Manifest `fast_patterns` ein. "
"Format pro Eintrag: `{match: \"^regex$\", args: {...}, reply: \"Text\"}`.\n"
"\n"
"Wirkung: das Brain matched eingehende User-Texte BEVOR Claude gerufen "
"wird. Match → run_skill(name, args) direkt → reply zurueck → Claude "
"uebersprungen. Stefan spart 5-10 Sekunden pro Befehl. Praktisch "
"Pflicht im Auto, wo Latenz nervt.\n"
"\n"
"REGELN beim Patterns schreiben:\n"
" - Mit `^` und `$` anchorn — sonst matched `pause` mitten in `pause "
"die musik dann erzaehl mir nen witz` und zerschiesst den Befehl.\n"
" - Case-insensitive (Brain matched mit re.IGNORECASE), Endsatzzeichen "
"werden vorher entfernt — schreibe Lowercase ohne Punkt.\n"
" - Mehrere Varianten = mehrere Eintraege (`^pause$`, `^pausiere$`, "
"`^stop$`). Sprachlich wechselt Stefan zwischen synonymen Kurzformen.\n"
" - reply = kurze Bestaetigung in genau einem Satz, gerne mit Emoji.\n"
"\n"
"NIE fast_patterns fuer:\n"
" - Skills mit kreativem Output (zusammenfassen, generieren, raten).\n"
" - Skills mit Parametern die aus Freitext extrahiert werden muessten "
" ('spiele jazz' geht nicht — was ist 'jazz'? Lass Claude entscheiden).\n"
" - Skills mit Multi-Step-Logik (z.B. Volumen +10 = erst Status holen, "
" rechnen, setzen). Wenn unbedingt: in den Skill-Code packen und "
" dem Skill einen `action`-Arg geben.\n"
"\n"
"Beispiel komplett:\n"
"```\n"
"fast_patterns = [\n"
" {\"match\": \"^pause$\",\n"
" \"args\": {\"path\": \"/v1/me/player/pause\", \"method\": \"PUT\"},\n"
" \"reply\": \"Spotify: pausiert ⏸\"}\n"
"]\n"
"```\n"
"\n"
"Stefan-Hinweis 06/2026: das war frueher hardcoded in agent.py fuer "
"Spotify und musste fuer jeden neuen Steuer-Skill nachgepflegt werden. "
"Jetzt steckt's pro Skill im Manifest — dein Job, ARIA, das gleich "
"mitzudenken wenn der Use-Case passt."
),
},
{
"migration_key": "seed/skill-rule/no-hardcoded-credentials",
"type": "rule",
+36
View File
@@ -164,6 +164,7 @@ def create_skill(
pip_packages: Optional[list[str]] = None,
author: str = "aria",
config_schema: Optional[list] = None,
fast_patterns: Optional[list] = None,
) -> dict:
"""Legt einen neuen Skill an. Wirft ValueError bei ungueltigen Inputs.
@@ -213,6 +214,7 @@ def create_skill(
"version": "1.0",
"author": author,
"config_schema": _normalize_config_schema(config_schema),
"fast_patterns": _normalize_fast_patterns(fast_patterns),
"version_history": [],
}
write_manifest(name, manifest)
@@ -261,6 +263,38 @@ def _normalize_config_schema(schema: Optional[list]) -> list:
return out
def _normalize_fast_patterns(patterns: Optional[list]) -> list:
"""Filter + Normalisiert fast_patterns. Erwartet Liste von Dicts mit:
- match (str) : Regex, wird gegen normalisierten User-Text (lowercase,
Endsatzzeichen weg, Whitespace gestrafft) gematched.
Sollte mit ^...$ anchored sein damit keine Teilmatches
reinrutschen. re.IGNORECASE wird automatisch gesetzt.
- args (dict?): Args fuer run_skill — leerer Dict wenn weggelassen.
- reply (str) : Fixe Antwort die ohne Claude an den User geht.
Patterns mit kaputter Regex werden ausgefiltert + geloggt — sonst wuerde
der ganze Fast-Path-Pass jedes Mal crashen wenn ARIA mal ein Pattern
falsch baut."""
if not patterns:
return []
out = []
for p in patterns:
if not isinstance(p, dict):
continue
match = (p.get("match") or "").strip()
reply = (p.get("reply") or "").strip()
if not match or not reply:
continue
try:
re.compile(match)
except re.error as exc:
logger.warning("fast_patterns: Regex %r kaputt — geskippt: %s", match, exc)
continue
args = p.get("args") if isinstance(p.get("args"), dict) else {}
out.append({"match": match, "args": args, "reply": reply[:300]})
return out
def _setup_venv(skill_dir: Path, pip_packages: list[str]) -> None:
venv = skill_dir / "venv"
logger.info("venv erstellen: %s", venv)
@@ -307,6 +341,8 @@ def update_skill(name: str, patch: dict) -> dict:
manifest[k] = v
if "config_schema" in patch:
manifest["config_schema"] = _normalize_config_schema(patch["config_schema"])
if "fast_patterns" in patch:
manifest["fast_patterns"] = _normalize_fast_patterns(patch["fast_patterns"])
# Code austauschen
if "entry_code" in patch and patch["entry_code"]: