Fast-Path-Plugin raus, offizielle quick_commands rein (funktioniert auch in der TUI)
Root Cause fuers "geht trotzdem ans LLM": unser fast-paths-Plugin war auf pre_gateway_dispatch registriert -- der Hook feuert laut Sourcecode nur fuer die Gateway-Plattformen (Telegram/Discord/...), nicht fuer die TUI (tui_gateway/server.py), die komplett am Gateway-Dispatcher vorbei direkt in den Agent-Loop laeuft. Ein Patch an tui_gateway/server.py waere noetig gewesen, aber Concurrency-kritischer Kern-Code -- zu riskant. Hermes hat dafuer schon einen eigenen Mechanismus: quick_commands (type: exec) bypassen den Agent-Loop nachweislich auf JEDER Plattform inkl. TUI, kein LLM-Call, keine Tokens, offiziell dokumentiert. Trade-off: fester Slash-Befehl statt Freitext (/next statt "naechstes Lied") -- gleiches Prinzip wie Alexas Intent-Slots, ohne Slot-Fuellung. scripts/spotify_quick.py buendelt next/previous/pause/play/volume_up/ volume_down/current in einem Skript (ein Argument pro Aktion), nutzt denselben SpotifyClient wie der echte Tool-Dispatch. docker-compose.yml mountet scripts/ jetzt read-only nach /opt/data/scripts statt fast-paths/ nach /opt/data/plugins/fast-paths. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
ARIA
Claude Sonnet 5
@@ -320,31 +320,79 @@ oben), nicht an Spotify selbst — dann `docker logs hermes-proxy` waehrend
|
||||
der Reproduktion pruefen. Gleiches Nutzungsmuster wie
|
||||
`scripts/spotify_manual_auth.py` (Login-Skript weiter oben).
|
||||
|
||||
## Fast-Paths (Steuerbefehle ohne LLM-Roundtrip)
|
||||
## Fast-Path ohne LLM (quick_commands)
|
||||
|
||||
Selbst mit allen Fixes oben braucht jede Nachricht mindestens einen vollen
|
||||
LLM-Turn (mehrere Sekunden) — fuer "pause" oder "naechstes Lied" unnoetig
|
||||
langsam. `fast-paths/` ist ein eigenes, universelles Plugin (ein File pro
|
||||
Skill, siehe `fast-paths/README.md`), das einfache Steuerbefehle per Regex
|
||||
VOR dem LLM-Aufruf abfaengt und direkt ausfuehrt — analog zu ARIAs eigenen
|
||||
`fast_patterns`. Aktuell drin: `patterns/spotify.py`
|
||||
(pause/weiter/next/previous/lauter/leiser/Lautstaerke/aktueller Titel, plus
|
||||
"spiel Playlist X auf Geraet Y ab" per Fuzzy-Match ohne LLM).
|
||||
langsam.
|
||||
|
||||
Einmalig aktivieren (Plugins sind bei Hermes Opt-in) — in
|
||||
`hermes-data/agent-home/config.yaml`:
|
||||
**Erster Anlauf war ein eigenes Plugin** (`fast-paths/`, Regex-Hook auf
|
||||
`pre_gateway_dispatch`) — das ist wieder raus. Root Cause (im echten
|
||||
Hermes-Sourcecode verifiziert, nicht geraten): `pre_gateway_dispatch` feuert
|
||||
NUR fuer die Gateway-Plattformen (Telegram/Discord/Slack/...), NICHT fuer die
|
||||
TUI (`tui_gateway/server.py`) — die laeuft komplett am Gateway-Dispatcher
|
||||
vorbei direkt in den Agent-Loop. Da wir hier fast ausschliesslich die TUI
|
||||
nutzen, griff das Plugin praktisch nie; die vier `Spotify Playback`-Calls in
|
||||
Folge, die Claude bei "nächstes" gemacht hat, kamen weil's doch beim LLM
|
||||
landete. Ein Patch direkt in `tui_gateway/server.py` waere der einzige Weg
|
||||
gewesen — aber Concurrency-kritischer Kern-Code (~700 Zeilen, History-Lock,
|
||||
Busy-Queue), zu riskant fuer einen Sed-Patch.
|
||||
|
||||
**Stattdessen: Hermes' eigener, offizieller Mechanismus — `quick_commands`.**
|
||||
Laut Sourcecode (`cli.py::process_command`, `tui_gateway/server.py`,
|
||||
`gateway/run.py`) bypassen `type: exec`-Quick-Commands den kompletten
|
||||
Agent-Loop — kein LLM-Call, keine Tokens — und funktionieren offiziell
|
||||
dokumentiert auf JEDER Plattform (CLI/TUI, Telegram, Discord, Slack,
|
||||
WhatsApp, Signal, Email, Home Assistant). Kein Patch an Hermes-Core-Dateien
|
||||
noetig, nur Config + ein Skript.
|
||||
|
||||
**Der Trade-off (ehrlich):** quick_commands sind Slash-Befehle mit festem
|
||||
Namen — kein Freitext, keine Argumente werden durchgereicht. `/next` statt
|
||||
"naechstes Lied". Fuer feste Steuerbefehle reicht das genauso wie Alexas
|
||||
Intent-Slots (siehe Chat-Verlauf), nur ohne Slot-Fuellung. Freitext-Faelle
|
||||
wie "spiel Playlist Fliegen auf dem Handy ab" bleiben bewusst beim LLM (das
|
||||
braucht ohnehin ein Modell, um den Playlist-Namen aus dem Satz zu holen).
|
||||
|
||||
Skript: `scripts/spotify_quick.py <next|previous|pause|play|volume_up|volume_down|current>`
|
||||
— nutzt denselben `SpotifyClient` wie der echte Tool-Dispatch, gleiche
|
||||
OAuth-Session, kein doppelter Token-Code. Wird read-only nach
|
||||
`/opt/data/scripts/` gemountet (siehe `docker-compose.yml`).
|
||||
|
||||
**Einmalig einrichten** — in `hermes-data/agent-home/config.yaml` ergaenzen
|
||||
(und falls noch vorhanden: den alten `plugins: enabled: [fast-paths]`-Block
|
||||
entfernen):
|
||||
```yaml
|
||||
plugins:
|
||||
enabled:
|
||||
- fast-paths
|
||||
quick_commands:
|
||||
next:
|
||||
type: exec
|
||||
command: python3 /opt/data/scripts/spotify_quick.py next
|
||||
prev:
|
||||
type: exec
|
||||
command: python3 /opt/data/scripts/spotify_quick.py previous
|
||||
pause:
|
||||
type: exec
|
||||
command: python3 /opt/data/scripts/spotify_quick.py pause
|
||||
play:
|
||||
type: exec
|
||||
command: python3 /opt/data/scripts/spotify_quick.py play
|
||||
louder:
|
||||
type: exec
|
||||
command: python3 /opt/data/scripts/spotify_quick.py volume_up
|
||||
quieter:
|
||||
type: exec
|
||||
command: python3 /opt/data/scripts/spotify_quick.py volume_down
|
||||
nowplaying:
|
||||
type: exec
|
||||
command: python3 /opt/data/scripts/spotify_quick.py current
|
||||
```
|
||||
Dann:
|
||||
```bash
|
||||
git pull
|
||||
docker-compose up -d --build hermes-agent
|
||||
```
|
||||
Volle Details, Sicherheitsmodell (Autorisierungspruefung VOR jedem
|
||||
Fast-Path-Treffer!) und Anleitung fuer neue Skills: `fast-paths/README.md`.
|
||||
Danach in der TUI `/next`, `/pause`, `/louder`, `/nowplaying` etc. testen —
|
||||
sollte in Millisekunden reagieren, kein Modell-Aufruf, keine Tool-Call-Schleife
|
||||
mehr moeglich (es gibt schlicht keinen LLM-Turn dafuer).
|
||||
|
||||
## Mobiler Client (Android)?
|
||||
|
||||
|
||||
+6
-11
@@ -220,17 +220,12 @@ services:
|
||||
# Start einmalig hierher kopiert (siehe README) und dann von Hermes
|
||||
# NICHT mehr angefasst (nur geseedet wenn's fehlt).
|
||||
- ./hermes-data/agent-home:/opt/data
|
||||
# Unser eigener Fast-Path-Layer (siehe fast-paths/README.md): faengt
|
||||
# einfache Steuerbefehle wie "pause"/"naechstes Lied" per Regex VOR dem
|
||||
# LLM-Aufruf ab. Wird als Hermes-USER-Plugin unter $HERMES_HOME/plugins/
|
||||
# erwartet ($HERMES_HOME=/opt/data, siehe ENV oben) -- ein verschachtelter
|
||||
# Bind-Mount UEBER dem agent-home-Mount oben, versioniert in unserem
|
||||
# eigenen Git-Repo statt im gitignorten hermes-data/. Read-only, damit
|
||||
# Hermes (oder ein kompromittierter Plugin-Code) den Code hier nie
|
||||
# veraendern kann. MUSS zusaetzlich in config.yaml unter
|
||||
# "plugins: enabled: [fast-paths]" eingetragen werden (siehe README) --
|
||||
# Plugins sind bei Hermes standardmaessig opt-in.
|
||||
- ./fast-paths:/opt/data/plugins/fast-paths:ro
|
||||
# scripts/ (Spotify-Quick-Steuerung etc.) READ-ONLY in den Container,
|
||||
# damit `quick_commands` in config.yaml (siehe README, Abschnitt
|
||||
# "Fast-Path ohne LLM") sie unter einem festen Pfad aufrufen koennen,
|
||||
# ohne jedes Mal manuell `docker cp` machen zu muessen. Versioniert in
|
||||
# unserem eigenen Git-Repo statt im gitignorten hermes-data/.
|
||||
- ./scripts:/opt/data/scripts:ro
|
||||
environment:
|
||||
- HERMES_UID=${HERMES_UID:-10000}
|
||||
- HERMES_GID=${HERMES_GID:-10000}
|
||||
|
||||
@@ -1,117 +0,0 @@
|
||||
# fast-paths — Steuerbefehle ohne LLM-Roundtrip
|
||||
|
||||
## Warum
|
||||
|
||||
Jede Nachricht an Hermes geht normalerweise durch's volle LLM: Text →
|
||||
System-Prompt bauen → Claude aufrufen (mehrere Sekunden, siehe
|
||||
Root-Cause-Historie im Haupt-README) → Tool-Call parsen → Tool ausfuehren →
|
||||
Antwort formulieren. Fuer einfache Steuerbefehle ("pause", "naechstes Lied",
|
||||
"lauter") ist das unnoetig langsam — Alexa & Co. machen sowas mit
|
||||
Intent-Klassifikation in Millisekunden, ganz ohne generatives Modell.
|
||||
|
||||
Dieses Plugin baut das fuer Hermes nach: ein Regex-Layer, der VOR dem LLM
|
||||
sitzt. Matcht der Text, wird die Aktion direkt ausgefuehrt und Claude nie
|
||||
gefragt. Matcht nichts (oder ist sich ein Handler nicht sicher), laeuft
|
||||
alles normal weiter — als waere das Plugin gar nicht da.
|
||||
|
||||
## Architektur
|
||||
|
||||
```
|
||||
fast-paths/
|
||||
plugin.yaml <- Hermes-Plugin-Manifest (kind: standalone)
|
||||
__init__.py <- Loader: sammelt patterns/*.py ein, ein Hook-Callback
|
||||
patterns/
|
||||
__init__.py
|
||||
_base.py <- FastPattern-Dataclass, pattern()-Helper, send_reply()
|
||||
spotify.py <- EIN Skill = EINE Datei
|
||||
<dein-neuer-skill>.py
|
||||
```
|
||||
|
||||
Ein Skill = eine Datei unter `patterns/`. Kein zentrales Registrieren
|
||||
noetig — `__init__.py` importiert automatisch jede `*.py`-Datei in
|
||||
`patterns/` (ausser Dateien mit fuehrendem `_`, das ist Infrastruktur) und
|
||||
sammelt deren `PATTERNS`-Liste ein.
|
||||
|
||||
## Neuen Fast-Path hinzufuegen
|
||||
|
||||
1. Neue Datei `patterns/<skill>.py` anlegen.
|
||||
2. Handler-Funktion(en) schreiben: `def handler(match, event, gateway) -> str | None`.
|
||||
- `event` ist ein `MessageEvent` (siehe `gateway/platforms/base.py` im
|
||||
Hermes-Sourcecode) — `event.text`, `event.source.chat_id`,
|
||||
`event.source.platform`, etc.
|
||||
- `gateway` ist der `GatewayRunner` — u.a. `gateway._adapter_for_source(source)`
|
||||
um den Platform-Adapter zu bekommen, `gateway._is_user_authorized(source)`
|
||||
fuer Auth-Checks (macht der Loader schon automatisch, siehe unten).
|
||||
3. Am Ende der Datei: `PATTERNS = [pattern("name", r"^regex$", handler), ...]`
|
||||
mit dem Helper aus `_base.py`.
|
||||
4. Fertig — kein Eintrag irgendwo sonst noetig. Container neu starten
|
||||
(`docker-compose up -d --build hermes-agent`), Loader findet die Datei
|
||||
automatisch beim naechsten Plugin-Discovery.
|
||||
|
||||
### Rueckgabewerte des Handlers (wichtig!)
|
||||
|
||||
| Rueckgabe | Bedeutung |
|
||||
|---|---|
|
||||
| `"Text"` | Wird 1:1 an den User geschickt, LLM wird uebersprungen. |
|
||||
| `""` (leer) | Aktion ausgefuehrt, aber bewusst stumm. LLM trotzdem uebersprungen. |
|
||||
| `None` | Handler ist sich nicht sicher genug (z.B. mehrdeutiger Name) — faellt zurueck auf die normale LLM-Verarbeitung, GENAU wie "kein Match". **Nie raten, im Zweifel `None`.** |
|
||||
| *(Exception)* | Wird geloggt, faellt ebenfalls automatisch aufs LLM zurueck. |
|
||||
|
||||
### Regex-Konventionen
|
||||
|
||||
- Immer mit `^` und `$` anchorn — sonst matcht `pause` auch mitten in
|
||||
"pause die Musik und erzaehl mir was" und reisst den Rest des Satzes weg.
|
||||
- Mehrere Formulierungen/Synonyme + Fuellwoerter ueber Alternativen und
|
||||
optionale Gruppen abdecken (`(?:mal|bitte)?`), nicht nur eine Formulierung.
|
||||
- Text wird vor dem Matchen normalisiert (lowercase, Satzzeichen am Ende
|
||||
weg, Mehrfach-Leerzeichen zusammengefasst) — Regexe koennen daher
|
||||
durchgehend lowercase geschrieben werden.
|
||||
- Lieber ein Pattern zu eng als zu weit — ein Fast-Path, der faelschlich
|
||||
matcht, kapert eine Nachricht die eigentlich ans LLM sollte. Ein Pattern,
|
||||
das gar nicht matcht, kostet nur die eingesparte Geschwindigkeit (die
|
||||
Nachricht geht stattdessen ganz normal ans LLM, kein Schaden).
|
||||
|
||||
## Sicherheit
|
||||
|
||||
`pre_gateway_dispatch` (der Hook, an dem dieses Plugin haengt) feuert VOR
|
||||
Hermes' eigener Autorisierungspruefung. Der Loader in `__init__.py` prueft
|
||||
deshalb bei JEDEM Pattern-Treffer explizit `gateway._is_user_authorized(...)`
|
||||
BEVOR der Handler ueberhaupt aufgerufen wird — ein nicht autorisierter
|
||||
Absender (z.B. auf einer oeffentlich erreichbaren Plattform wie Telegram)
|
||||
kann also nie per Fast-Path an Hermes vorbei etwas ausloesen. Diese Pruefung
|
||||
lebt zentral im Loader, nicht in den einzelnen Skill-Dateien — neue Skills
|
||||
muessen sich darum nicht kuemmern.
|
||||
|
||||
## Bekannter Trade-off
|
||||
|
||||
`pre_gateway_dispatch`-Callbacks laufen **synchron** (Hermes' PluginManager
|
||||
awaited sie nicht, siehe `hermes_cli/plugins.py`). Handler duerfen also
|
||||
ganz normal blockierende Netzwerk-Calls machen (z.B. `httpx.request` wie
|
||||
`SpotifyClient`), das blockiert den Event-Loop aber kurz (typischerweise
|
||||
100-500ms pro API-Call). Fuer eine Text-Antwort an den User (das *ist*
|
||||
async — `adapter.send()`) gibt's `send_reply()` in `_base.py`: schedult
|
||||
einen Fire-and-forget `asyncio.Task` auf dem laufenden Event-Loop. Beides
|
||||
zusammen ist immer noch um Groessenordnungen schneller als der
|
||||
LLM-Roundtrip (mehrere Sekunden bis >20s, siehe Haupt-README).
|
||||
|
||||
## Aktivieren
|
||||
|
||||
Plugins sind bei Hermes standardmaessig Opt-in. Einmalig in
|
||||
`hermes-data/agent-home/config.yaml`:
|
||||
|
||||
```yaml
|
||||
plugins:
|
||||
enabled:
|
||||
- fast-paths
|
||||
```
|
||||
|
||||
Danach `docker-compose up -d --build hermes-agent` (Volume-Mount siehe
|
||||
`docker-compose.yml`, Kommentar beim `hermes-agent`-Service).
|
||||
|
||||
## Bisherige Skills
|
||||
|
||||
- **`patterns/spotify.py`** — pause/weiter/next/previous/lauter/leiser/
|
||||
Lautstaerke setzen/aktueller Titel, plus "spiel Playlist X [auf Geraet Y]
|
||||
ab" per Fuzzy-Match (kein LLM fuer den Freitext-Namen noetig, siehe
|
||||
`difflib`-basiertes Matching dort — bricht bei Mehrdeutigkeit bewusst ab
|
||||
und gibt an das LLM ab statt zu raten).
|
||||
@@ -1,163 +0,0 @@
|
||||
"""fast-paths -- universeller Fast-Path-Layer fuer Hermes.
|
||||
|
||||
Faengt einfache Steuerbefehle per Regex VOR dem eigentlichen LLM-Aufruf ab
|
||||
und fuehrt sie direkt aus -- kein Claude-Proxy-Roundtrip noetig. Motivation
|
||||
und Architektur-Entscheidung: siehe README.md daneben.
|
||||
|
||||
Funktionsweise in Kurzform:
|
||||
|
||||
1. Hermes' Plugin-System laedt dieses Verzeichnis als kind="standalone"-
|
||||
Plugin (siehe plugin.yaml) und ruft register(ctx) einmal beim Start auf.
|
||||
2. register() sammelt alle Fast-Path-Regeln aus patterns/*.py ein (ein File
|
||||
pro Skill, z.B. patterns/spotify.py -- neue Skills: einfach eine neue
|
||||
Datei dort reinlegen, siehe patterns/_base.py fuer die Konventionen) und
|
||||
registriert EINEN Callback auf den "pre_gateway_dispatch"-Hook.
|
||||
3. Dieser Hook feuert bei JEDER eingehenden Nachricht, auf JEDER Plattform
|
||||
(TUI, Telegram, Discord, ...) -- noch VOR Auth/Pairing und noch VOR dem
|
||||
eigentlichen Agent-Dispatch (siehe gateway/run.py _handle_message). Genau
|
||||
der richtige Abfangpunkt: universell fuer alle Plattformen, aber weit
|
||||
genug vorne dass wir dem LLM wirklich zuvorkommen.
|
||||
4. Matcht der normalisierte Text gegen eine der gesammelten Regeln, wird der
|
||||
zugehoerige Handler synchron aufgerufen. Antwortet der Handler mit einem
|
||||
String, wird der per adapter.send() direkt an den User geschickt (siehe
|
||||
patterns/_base.py send_reply()) und die Nachricht per
|
||||
{"action": "skip"} aus der normalen Pipeline genommen -- das LLM sieht
|
||||
diese Nachricht nie.
|
||||
5. Matcht nichts, ODER lehnt ein Handler bewusst ab (Rueckgabe None, z.B.
|
||||
weil ein Playlist-Name mehrdeutig war), faellt alles ganz normal auf den
|
||||
LLM-Turn zurueck -- als waere dieses Plugin gar nicht da.
|
||||
|
||||
Sicherheitshinweis (wichtig, nicht optional): "pre_gateway_dispatch" feuert
|
||||
VOR Hermes' eigener Autorisierungspruefung. Ohne expliziten Check wuerde
|
||||
ein Fast-Path-Treffer also auch fuer NICHT autorisierte Absender ausgefuehrt
|
||||
-- z.B. koennte irgendwer, der Hermes' Telegram-Bot anschreibt, ohne
|
||||
Autorisierung "pause" tippen und tatsaechlich Spotify pausieren. Deshalb
|
||||
prueft der Hook-Callback unten explizit gateway._is_user_authorized(...)
|
||||
BEVOR irgendein Pattern ueberhaupt versucht wird.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import logging
|
||||
import re
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any, List, Optional
|
||||
|
||||
logger = logging.getLogger("hermes_plugins.fast_paths")
|
||||
|
||||
_PATTERNS: List[Any] = []
|
||||
|
||||
|
||||
def _load_pattern_modules() -> None:
|
||||
"""Importiert jede patterns/<skill>.py und sammelt ihre PATTERNS-Liste."""
|
||||
global _PATTERNS
|
||||
loaded: List[Any] = []
|
||||
patterns_dir = Path(__file__).resolve().parent / "patterns"
|
||||
for py_file in sorted(patterns_dir.glob("*.py")):
|
||||
mod_name = py_file.stem
|
||||
if mod_name.startswith("_") or mod_name == "__init__":
|
||||
continue # _base.py etc. -- Infrastruktur, kein Skill
|
||||
try:
|
||||
mod = importlib.import_module(f"{__package__}.patterns.{mod_name}")
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"fast_paths: Skill-Datei '%s.py' konnte nicht geladen werden -- uebersprungen",
|
||||
mod_name,
|
||||
)
|
||||
continue
|
||||
skill_patterns = getattr(mod, "PATTERNS", None)
|
||||
if not skill_patterns:
|
||||
logger.warning("fast_paths: '%s.py' hat keine PATTERNS-Liste -- uebersprungen", mod_name)
|
||||
continue
|
||||
for p in skill_patterns:
|
||||
p.skill = mod_name
|
||||
loaded.append(p)
|
||||
logger.info("fast_paths: %d Pattern(s) aus '%s.py' geladen", len(skill_patterns), mod_name)
|
||||
_PATTERNS = loaded
|
||||
skill_count = len({p.skill for p in _PATTERNS})
|
||||
logger.info("fast_paths: insgesamt %d Pattern(s) aus %d Skill-Datei(en) aktiv", len(_PATTERNS), skill_count)
|
||||
|
||||
|
||||
def _normalize(text: str) -> str:
|
||||
"""Wie ARIAs eigene fast_patterns-Konvention: lowercase, Satzzeichen am
|
||||
Ende weg, Mehrfach-Leerzeichen zusammenfassen -- damit die Regexe in den
|
||||
Skill-Dateien einfach bleiben."""
|
||||
t = (text or "").strip().lower()
|
||||
t = re.sub(r"[.!?]+$", "", t).strip()
|
||||
t = re.sub(r"\s+", " ", t)
|
||||
return t
|
||||
|
||||
|
||||
def _on_pre_gateway_dispatch(event=None, gateway=None, session_store=None, **_kw) -> Optional[dict]:
|
||||
if event is None or gateway is None:
|
||||
return None
|
||||
if getattr(event, "internal", False):
|
||||
return None # System-generierte Events (z.B. Background-Notifications) nie fast-pathen
|
||||
|
||||
text = _normalize(getattr(event, "text", "") or "")
|
||||
if not text:
|
||||
return None
|
||||
|
||||
# Erst pruefen ob ueberhaupt ein Pattern matcht -- die (minimal teurere)
|
||||
# Autorisierungspruefung erst danach, damit der ganz normale Nachrichten-
|
||||
# Strom (der zu 99% NICHT matcht) keinen zusaetzlichen Call pro Turn zahlt.
|
||||
for fp in _PATTERNS:
|
||||
m = fp.regex.match(text)
|
||||
if not m:
|
||||
continue
|
||||
|
||||
# SICHERHEITSKRITISCH: pre_gateway_dispatch feuert VOR Hermes' eigener
|
||||
# Auth-Pruefung. Ein Fast-Path-Treffer darf NIEMALS fuer einen nicht
|
||||
# autorisierten Absender ausgefuehrt werden -- sonst koennte z.B. ein
|
||||
# fremder Telegram-User ungefragt "pause" tippen und wirklich Spotify
|
||||
# steuern. Bei fehlender Autorisierung: exakt wie "kein Match"
|
||||
# behandeln, normale Pipeline (inkl. Pairing-Flow) uebernimmt.
|
||||
try:
|
||||
authorized = gateway._is_user_authorized(event.source)
|
||||
except Exception:
|
||||
logger.exception("fast_paths: Autorisierungspruefung fehlgeschlagen -- Pattern wird NICHT ausgefuehrt")
|
||||
return None
|
||||
if not authorized:
|
||||
logger.info(
|
||||
"fast_paths: Pattern '%s' haette gematcht, Absender aber nicht autorisiert -- normale Pipeline uebernimmt",
|
||||
fp.name,
|
||||
)
|
||||
return None
|
||||
|
||||
t0 = time.monotonic()
|
||||
try:
|
||||
reply = fp.handler(m, event, gateway)
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"fast_paths: Handler '%s' (skill=%s) hat eine Exception geworfen -- Fallback aufs LLM",
|
||||
fp.name, fp.skill,
|
||||
)
|
||||
return None
|
||||
elapsed_ms = int((time.monotonic() - t0) * 1000)
|
||||
|
||||
if reply is None:
|
||||
# Handler ist sich bewusst nicht sicher (z.B. mehrdeutiger
|
||||
# Playlist-Name) -- NICHT raten, normale LLM-Verarbeitung uebernimmt.
|
||||
logger.info(
|
||||
"fast_paths: '%s' (skill=%s) hat gematcht, Handler aber abgelehnt (%dms) -- Fallback aufs LLM",
|
||||
fp.name, fp.skill, elapsed_ms,
|
||||
)
|
||||
return None
|
||||
|
||||
logger.info(
|
||||
"fast_paths: '%s' (skill=%s) in %dms erledigt, LLM-Aufruf uebersprungen",
|
||||
fp.name, fp.skill, elapsed_ms,
|
||||
)
|
||||
if reply:
|
||||
from .patterns._base import send_reply
|
||||
send_reply(gateway, event, reply)
|
||||
return {"action": "skip", "reason": f"fast_path:{fp.skill}:{fp.name}"}
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def register(ctx) -> None:
|
||||
_load_pattern_modules()
|
||||
ctx.register_hook("pre_gateway_dispatch", _on_pre_gateway_dispatch)
|
||||
@@ -1,8 +0,0 @@
|
||||
# Absichtlich (fast) leer -- macht "patterns/" zu einem regulaeren
|
||||
# Python-Package, damit __init__.py (eine Ebene hoeher) die einzelnen
|
||||
# Skill-Dateien hier drin per "from .patterns import <modname>" laden kann.
|
||||
#
|
||||
# Neue Fast-Path-Skills: einfach eine neue Datei hier reinlegen (z.B.
|
||||
# lights.py, heating.py) -- KEIN Eintrag hier noetig, der Loader in
|
||||
# ../__init__.py findet jede *.py-Datei automatisch (ausser Dateien mit
|
||||
# fuehrendem Unterstrich wie _base.py, das ist Infrastruktur, kein Skill).
|
||||
@@ -1,110 +0,0 @@
|
||||
"""Gemeinsame Bausteine fuer alle Fast-Path-Skill-Dateien.
|
||||
|
||||
Jede Skill-Datei in diesem Ordner (spotify.py, kuenftig z.B. lights.py)
|
||||
exportiert eine Modul-Variable ``PATTERNS: list[FastPattern]``. Der Loader
|
||||
in ``../__init__.py`` sammelt die automatisch ein -- siehe README.md dort
|
||||
fuer die Anleitung, wie man einen neuen Fast-Path ergaenzt.
|
||||
|
||||
Wichtig fuer Handler-Autoren:
|
||||
|
||||
def handler(match: re.Match, event: MessageEvent, gateway) -> str | None:
|
||||
...
|
||||
|
||||
- Rueckgabe ``str`` (auch nicht-leer) -> wird 1:1 als Antwort an den
|
||||
User geschickt, das LLM wird fuer diese Nachricht komplett uebersprungen.
|
||||
- Rueckgabe ``""`` (leerer String) -> Aktion wurde ausgefuehrt, aber
|
||||
bewusst OHNE Text-Antwort. LLM wird trotzdem uebersprungen.
|
||||
- Rueckgabe ``None`` -> Pattern hat zwar auf den Text
|
||||
gematcht, der Handler ist sich aber nicht sicher genug (z.B. Playlist-
|
||||
Name mehrdeutig, Geraet nicht gefunden) -- NICHT raten, sondern exakt
|
||||
wie "kein Match" behandeln und die Nachricht normal ans LLM
|
||||
durchreichen. Das ist das Sicherheitsnetz, das Fast-Paths ueberhaupt
|
||||
erst gefahrlos macht.
|
||||
- Eine Exception im Handler wird vom Loader abgefangen, geloggt, und
|
||||
faellt ebenfalls zurueck auf die normale LLM-Verarbeitung -- ein Bug in
|
||||
einem Fast-Path darf NIE eine Nachricht komplett verschlucken.
|
||||
|
||||
Bekannter Trade-off (bewusst in Kauf genommen, siehe Root-Cause-Historie im
|
||||
Hermes-Projekt): ``pre_gateway_dispatch`` wird synchron aufgerufen (kein
|
||||
await moeglich, siehe hermes_cli/plugins.py PluginManager.invoke_hook).
|
||||
Handler duerfen daher ganz normal synchrone/blockierende Netzwerk-Calls
|
||||
machen (z.B. ueber httpx.request wie SpotifyClient das tut) -- das blockiert
|
||||
den Event-Loop kurz (typ. 100-500ms fuer einen API-Call), ist aber
|
||||
IMMER NOCH um Groessenordnungen schneller als der LLM-Roundtrip (mehrere
|
||||
Sekunden bis >20s, siehe Hermes-Projekt-Historie). Fuer eine Antwort an den
|
||||
User (adapter.send(), das IST async) siehe ``send_reply()`` unten.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Callable, Optional
|
||||
|
||||
logger = logging.getLogger("hermes_plugins.fast_paths")
|
||||
|
||||
|
||||
@dataclass
|
||||
class FastPattern:
|
||||
"""Eine Fast-Path-Regel: eine anchored Regex -> ein Handler."""
|
||||
|
||||
name: str
|
||||
regex: "re.Pattern[str]"
|
||||
handler: Callable[["re.Match[str]", object, object], Optional[str]]
|
||||
# Wird vom Loader in ../__init__.py automatisch gesetzt (Dateiname ohne
|
||||
# .py) -- nicht manuell befuellen, nur fuer Logging/Debugging gedacht.
|
||||
skill: str = field(default="", compare=False)
|
||||
|
||||
|
||||
def pattern(name: str, regex: str, handler: Callable) -> FastPattern:
|
||||
"""Komfort-Konstruktor: kompiliert die Regex case-insensitive.
|
||||
|
||||
Regeln fuers Regex-Schreiben (siehe ARIAs eigene fast_patterns-Konvention,
|
||||
1:1 uebernommen):
|
||||
- IMMER mit ^ und $ anchorn -- sonst matcht "pause" auch mitten in
|
||||
"pause die musik dann erzaehl mir einen witz" und reisst den Rest weg.
|
||||
- Mehrere Formulierungen/Synonyme + Fuellwoerter ueber Alternativen und
|
||||
optionale Gruppen abdecken, nicht nur die eine Formulierung.
|
||||
- Der eingehende Text wird VOR dem Matchen normalisiert (lowercase,
|
||||
Mehrfach-Leerzeichen zusammengefasst, Satzzeichen am Ende entfernt)
|
||||
-- siehe _normalize() in ../__init__.py. Regexe koennen daher
|
||||
lowercase geschrieben werden und muessen kein trailendes [.!?]* haben.
|
||||
"""
|
||||
return FastPattern(name=name, regex=re.compile(regex, re.IGNORECASE), handler=handler)
|
||||
|
||||
|
||||
def send_reply(gateway, event, text: str) -> None:
|
||||
"""Schickt eine Antwort direkt ueber den Platform-Adapter -- OHNE LLM.
|
||||
|
||||
Muss aus einem SYNCHRONEN pre_gateway_dispatch-Hook heraus funktionieren
|
||||
(PluginManager.invoke_hook awaited seine Callbacks nicht). Wir sind aber
|
||||
innerhalb eines laufenden Event-Loops (der Hook wird von der async
|
||||
GatewayRunner._handle_message() aufgerufen), also reicht ein
|
||||
Fire-and-forget create_task(), um adapter.send() (das IST async)
|
||||
trotzdem loszuschicken.
|
||||
"""
|
||||
if not text:
|
||||
return
|
||||
try:
|
||||
adapter = gateway._adapter_for_source(event.source)
|
||||
except Exception:
|
||||
logger.exception("fast_paths: _adapter_for_source() fehlgeschlagen")
|
||||
return
|
||||
if adapter is None:
|
||||
logger.warning("fast_paths: kein Adapter fuer diese Quelle gefunden -- Antwort verworfen: %r", text)
|
||||
return
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
logger.warning("fast_paths: kein laufender Event-Loop -- Antwort verworfen: %r", text)
|
||||
return
|
||||
|
||||
async def _send() -> None:
|
||||
try:
|
||||
await adapter.send(event.source.chat_id, text)
|
||||
except Exception:
|
||||
logger.exception("fast_paths: adapter.send() fehlgeschlagen")
|
||||
|
||||
loop.create_task(_send())
|
||||
@@ -1,288 +0,0 @@
|
||||
"""Fast-Path-Skill: Spotify.
|
||||
|
||||
Deckt reine Steuerbefehle (pause/weiter/next/previous/lauter/leiser/
|
||||
Lautstaerke setzen/aktueller Titel) OHNE LLM-Roundtrip ab, plus als
|
||||
Kuer "spiel Playlist X [auf Geraet Y] ab" per Fuzzy-Match (kein LLM
|
||||
noetig fuer den Freitext-Playlistnamen -- difflib reicht).
|
||||
|
||||
Nutzt bewusst NICHT die injizierten <tool_call>-Tools (die laufen ueber
|
||||
den Proxy -> Claude -> zurueck, genau der Roundtrip den wir hier
|
||||
umgehen wollen), sondern importiert Hermes' eigenen SpotifyClient direkt
|
||||
-- gleiche OAuth-Session, gleicher Auto-Refresh, kein Code doppelt
|
||||
gepflegt (siehe plugins/spotify/client.py im Hermes-Sourcecode).
|
||||
|
||||
Sicherheitsnetz: jeder Handler gibt None zurueck, sobald er sich nicht
|
||||
sicher genug ist (z.B. Playlist-Name mehrdeutig) -- dann uebernimmt ganz
|
||||
normal das LLM, es wird NIE geraten.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import difflib
|
||||
import logging
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from ._base import pattern
|
||||
|
||||
logger = logging.getLogger("hermes_plugins.fast_paths.spotify")
|
||||
|
||||
|
||||
def _client():
|
||||
# Lazy-Import: erst bei tatsaechlichem Bedarf, damit ein Import-Fehler
|
||||
# hier (z.B. Spotify-Plugin bei einem Hermes-Update umbenannt) nicht das
|
||||
# Laden ALLER Fast-Paths verhindert -- siehe ../__init__.py, der jede
|
||||
# Skill-Datei einzeln try/except importiert.
|
||||
from plugins.spotify.client import SpotifyClient
|
||||
return SpotifyClient()
|
||||
|
||||
|
||||
def _friendly_error(exc: Exception) -> str:
|
||||
# str(exc) ist bei SpotifyAuthRequiredError/SpotifyAPIError bereits die
|
||||
# aufbereitete, menschenlesbare Meldung (siehe _friendly_spotify_error_message
|
||||
# in plugins/spotify/client.py) -- 1:1 zitieren, nicht neu formulieren
|
||||
# oder raten (siehe ARIAs eigene Anti-Halluzinations-Regel).
|
||||
return f"⚠️ Spotify: {exc}"
|
||||
|
||||
|
||||
def _devices(payload: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||||
return list((payload or {}).get("devices") or [])
|
||||
|
||||
|
||||
def _active_device_id(payload: Dict[str, Any]) -> Optional[str]:
|
||||
for d in _devices(payload):
|
||||
if d.get("is_active"):
|
||||
return d.get("id")
|
||||
return None
|
||||
|
||||
|
||||
def _best_match(query: str, candidates: List[str], cutoff: float = 0.55) -> Optional[int]:
|
||||
"""Index des besten Fuzzy-Treffers, oder None wenn zu unsicher.
|
||||
|
||||
"Zu unsicher" heisst: kein Treffer ueber cutoff, ODER die besten zwei
|
||||
Treffer liegen zu nah beieinander (< 0.08 Differenz) -- dann lieber ans
|
||||
LLM abgeben statt eine von zwei aehnlich benannten Playlists zu raten.
|
||||
Exakt das Verhalten, das ARIA bei "mehrere Substring-Matches" auch
|
||||
einfordert: nachfragen statt raten.
|
||||
"""
|
||||
if not candidates:
|
||||
return None
|
||||
scored = sorted(
|
||||
(
|
||||
(i, difflib.SequenceMatcher(None, query.lower(), (c or "").lower()).ratio())
|
||||
for i, c in enumerate(candidates)
|
||||
),
|
||||
key=lambda pair: pair[1],
|
||||
reverse=True,
|
||||
)
|
||||
if not scored or scored[0][1] < cutoff:
|
||||
return None
|
||||
if len(scored) > 1 and (scored[0][1] - scored[1][1]) < 0.08:
|
||||
return None
|
||||
return scored[0][0]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Reine Steuerbefehle
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _handle_pause(match, event, gateway) -> Optional[str]:
|
||||
try:
|
||||
_client().pause_playback()
|
||||
except Exception as exc:
|
||||
return _friendly_error(exc)
|
||||
return "⏸ Pausiert."
|
||||
|
||||
|
||||
def _handle_resume(match, event, gateway) -> Optional[str]:
|
||||
try:
|
||||
_client().start_playback()
|
||||
except Exception as exc:
|
||||
return _friendly_error(exc)
|
||||
return "▶️ Weiter."
|
||||
|
||||
|
||||
def _handle_next(match, event, gateway) -> Optional[str]:
|
||||
try:
|
||||
_client().skip_next()
|
||||
except Exception as exc:
|
||||
return _friendly_error(exc)
|
||||
return "⏭ Nächster Titel."
|
||||
|
||||
|
||||
def _handle_previous(match, event, gateway) -> Optional[str]:
|
||||
try:
|
||||
_client().skip_previous()
|
||||
except Exception as exc:
|
||||
return _friendly_error(exc)
|
||||
return "⏮ Vorheriger Titel."
|
||||
|
||||
|
||||
def _handle_volume_relative(match, event, gateway) -> Optional[str]:
|
||||
direction = (match.group("dir") or "").lower()
|
||||
delta = 10 if direction.startswith("laut") else -10
|
||||
client = _client()
|
||||
try:
|
||||
devices_payload = client.get_devices()
|
||||
device_id = _active_device_id(devices_payload)
|
||||
current = 50
|
||||
for d in _devices(devices_payload):
|
||||
if d.get("id") == device_id:
|
||||
current = int(d.get("volume_percent") or 50)
|
||||
break
|
||||
new_volume = max(0, min(100, current + delta))
|
||||
client.set_volume(volume_percent=new_volume, device_id=device_id)
|
||||
except Exception as exc:
|
||||
return _friendly_error(exc)
|
||||
return f"🔊 Lautstärke: {new_volume}%"
|
||||
|
||||
|
||||
def _handle_volume_set(match, event, gateway) -> Optional[str]:
|
||||
try:
|
||||
percent = max(0, min(100, int(match.group("pct"))))
|
||||
except (TypeError, ValueError):
|
||||
return None # Zahl nicht sauber parsebar -> lieber ans LLM
|
||||
try:
|
||||
_client().set_volume(volume_percent=percent)
|
||||
except Exception as exc:
|
||||
return _friendly_error(exc)
|
||||
return f"🔊 Lautstärke: {percent}%"
|
||||
|
||||
|
||||
def _handle_current(match, event, gateway) -> Optional[str]:
|
||||
try:
|
||||
payload = _client().get_currently_playing()
|
||||
except Exception as exc:
|
||||
return _friendly_error(exc)
|
||||
if not payload or payload.get("empty") or not payload.get("item"):
|
||||
return "⏸ Aktuell läuft nichts auf Spotify."
|
||||
item = payload["item"]
|
||||
name = item.get("name", "?")
|
||||
artists = ", ".join(a.get("name", "?") for a in item.get("artists", []) or [])
|
||||
device = (payload.get("device") or {}).get("name")
|
||||
suffix = f" ({device})" if device else ""
|
||||
return f"🎵 Läuft gerade: {name} – {artists}{suffix}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# "spiel Playlist X [auf Geraet Y] ab" -- Freitext-Namen per Fuzzy-Match,
|
||||
# kein LLM noetig. Sicherheitsnetz: bei Unsicherheit -> None -> LLM uebernimmt.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _handle_play_playlist(match, event, gateway) -> Optional[str]:
|
||||
playlist_query = (match.group("playlist") or "").strip().strip("\"“”'")
|
||||
device_query = ""
|
||||
if "device" in match.groupdict():
|
||||
device_query = (match.group("device") or "").strip()
|
||||
if not playlist_query:
|
||||
return None
|
||||
|
||||
client = _client()
|
||||
|
||||
# ALLE Playlists einsammeln bevor gematcht wird -- nicht auf
|
||||
# Teilergebnissen raten (ARIAs eigene Pagination-Regel).
|
||||
try:
|
||||
items: List[Dict[str, Any]] = []
|
||||
offset = 0
|
||||
while len(items) < 300:
|
||||
page = client.get_my_playlists(limit=50, offset=offset)
|
||||
page_items = page.get("items") or []
|
||||
items.extend(page_items)
|
||||
if not page.get("next") or not page_items:
|
||||
break
|
||||
offset += len(page_items)
|
||||
except Exception as exc:
|
||||
return _friendly_error(exc)
|
||||
|
||||
names = [p.get("name", "") for p in items]
|
||||
idx = _best_match(playlist_query, names)
|
||||
if idx is None:
|
||||
logger.info(
|
||||
"fast_paths/spotify: Playlist '%s' nicht eindeutig unter %d Playlists -> LLM uebernimmt",
|
||||
playlist_query, len(names),
|
||||
)
|
||||
return None
|
||||
playlist = items[idx]
|
||||
|
||||
try:
|
||||
devices_payload = client.get_devices()
|
||||
except Exception as exc:
|
||||
return _friendly_error(exc)
|
||||
devices = _devices(devices_payload)
|
||||
if not devices:
|
||||
return "⚠️ Spotify: kein Gerät gefunden — öffne Spotify auf irgendeinem Gerät und versuch's nochmal."
|
||||
|
||||
device_id = None
|
||||
if device_query:
|
||||
device_names = [d.get("name", "") for d in devices]
|
||||
d_idx = _best_match(device_query, device_names, cutoff=0.4)
|
||||
if d_idx is None:
|
||||
logger.info(
|
||||
"fast_paths/spotify: Geraet '%s' nicht eindeutig unter %s -> LLM uebernimmt",
|
||||
device_query, device_names,
|
||||
)
|
||||
return None
|
||||
device_id = devices[d_idx].get("id")
|
||||
else:
|
||||
device_id = _active_device_id(devices_payload) or devices[0].get("id")
|
||||
|
||||
try:
|
||||
client.start_playback(device_id=device_id, context_uri=playlist.get("uri"))
|
||||
except Exception as exc:
|
||||
return _friendly_error(exc)
|
||||
|
||||
device_name = next((d.get("name") for d in devices if d.get("id") == device_id), "?")
|
||||
return f"▶️ Spiele Playlist „{playlist.get('name')}“ auf {device_name}."
|
||||
|
||||
|
||||
PATTERNS = [
|
||||
pattern(
|
||||
"pause",
|
||||
r"^(?:spotify\s+)?(?:pause|pausier(?:e|en)?|stop|stopp|halt|anhalten)"
|
||||
# * statt ? -- deckt auch mehrere aneinandergereihte Fuellwoerter ab
|
||||
# ("pause bitte spotify"), nicht nur genau ein einzelnes.
|
||||
r"(?:\s+(?:mal|bitte|spotify|die\s+musik))*$",
|
||||
_handle_pause,
|
||||
),
|
||||
pattern(
|
||||
"resume",
|
||||
r"^(?:spotify\s+)?(?:weiter|weiterspielen|fortsetzen|play|abspielen|los)"
|
||||
r"(?:\s+(?:mal|bitte|spotify|die\s+musik))*$",
|
||||
_handle_resume,
|
||||
),
|
||||
pattern(
|
||||
"next",
|
||||
r"^(?:spotify\s+)?(?:n(?:ä|ae)chste[rs]?(?:\s+(?:lied|song|titel))?"
|
||||
r"|weiter(?:er)?\s+(?:lied|song|titel)|skip|next)(?:\s+(?:mal|bitte|spotify))*$",
|
||||
_handle_next,
|
||||
),
|
||||
pattern(
|
||||
"previous",
|
||||
r"^(?:spotify\s+)?(?:vorherige[rs]?(?:\s+(?:lied|song|titel))?"
|
||||
r"|zur(?:ü|ue)ck(?:\s+zum\s+vorherigen)?|previous)(?:\s+(?:mal|bitte|spotify))*$",
|
||||
_handle_previous,
|
||||
),
|
||||
pattern(
|
||||
"volume_relative",
|
||||
r"^(?:spotify\s+)?(?:mach\s+(?:es\s+)?)?(?P<dir>lauter|leiser)(?:\s+(?:mal|bitte|spotify))*$",
|
||||
_handle_volume_relative,
|
||||
),
|
||||
pattern(
|
||||
"volume_set",
|
||||
r"^(?:spotify\s+)?(?:lautst(?:ä|ae)rke|volume)\s+(?:auf\s+)?(?P<pct>\d{1,3})\s*%?$",
|
||||
_handle_volume_set,
|
||||
),
|
||||
pattern(
|
||||
"current",
|
||||
r"^(?:spotify\s+)?(?:was\s+l(?:ä|ae)uft(?:\s+(?:gerade|jetzt))?"
|
||||
r"|welches\s+lied\s+l(?:ä|ae)uft|aktueller\s+(?:song|titel)|current(?:ly\s+playing)?)"
|
||||
r"(?:\s+(?:mal|bitte))*$",
|
||||
_handle_current,
|
||||
),
|
||||
pattern(
|
||||
"play_playlist",
|
||||
r"^spiel(?:e)?\s+(?:mal\s+|bitte\s+)?(?:meine\s+)?playlist\s+"
|
||||
r"[\"“]?(?P<playlist>.+?)[\"”]?(?:\s+auf\s+(?:mein(?:em|er)?\s+)?(?P<device>.+?))?"
|
||||
r"\s*(?:ab)?$",
|
||||
_handle_play_playlist,
|
||||
),
|
||||
]
|
||||
@@ -1,12 +0,0 @@
|
||||
name: fast-paths
|
||||
version: 1.0.0
|
||||
description: >
|
||||
Universeller Fast-Path-Layer: einfache Steuerbefehle (z.B. Spotify
|
||||
pause/next/volume) werden per Regex VOR dem eigentlichen LLM-Aufruf
|
||||
abgefangen und direkt ausgefuehrt -- kein Claude-Roundtrip noetig. Ein
|
||||
eigenes File pro Skill unter patterns/, siehe README.md fuer die
|
||||
Anleitung neue Fast-Paths zu ergaenzen.
|
||||
author: "Stefan / ARIA"
|
||||
kind: standalone
|
||||
provides_hooks:
|
||||
- pre_gateway_dispatch
|
||||
@@ -0,0 +1,131 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Zero-LLM Spotify-Transportsteuerung fuer Hermes' offizielle `quick_commands`.
|
||||
|
||||
Hintergrund: Wir hatten zuerst ein eigenes Fast-Path-Plugin gebaut (Regex vor
|
||||
dem LLM abfangen), das ist wieder raus -- der Hook (`pre_gateway_dispatch`),
|
||||
auf den es registriert war, feuert nur fuer die Gateway-Plattformen
|
||||
(Telegram/Discord/...), NICHT fuer die TUI (siehe `gateway/run.py` vs.
|
||||
`tui_gateway/server.py` im echten Hermes-Sourcecode). Ein eigener Patch an
|
||||
`tui_gateway/server.py` waere noetig gewesen -- riskant, weil die Funktion
|
||||
dort ~700 Zeilen Concurrency-kritischen Code enthaelt (History-Lock,
|
||||
Busy-Queue, Session-State).
|
||||
|
||||
Hermes hat dafuer aber bereits einen EIGENEN, offiziellen Mechanismus:
|
||||
`quick_commands` (type: exec) in config.yaml. Laut Sourcecode
|
||||
(cli.py:process_command, tui_gateway/server.py, gateway/run.py) bypassen die
|
||||
NACHWEISLICH den Agent-Loop komplett -- kein LLM-Call, keine Tokens -- und
|
||||
funktionieren auf ALLEN Plattformen (CLI/TUI, Telegram, Discord, Slack,
|
||||
WhatsApp, Signal, Email, Home Assistant), offiziell dokumentiert unter
|
||||
website/docs/user-guide/configuration.md#quick-commands.
|
||||
|
||||
Der Haken (ehrlich, nicht schoengeredet): quick_commands reichen KEINE
|
||||
Argumente durch -- jeder Befehl ist ein fester Slash-Befehl (z.B. "/next"),
|
||||
kein Freitext wie ARIAs Regex-`fast_patterns`. Fuer feste Steuerbefehle
|
||||
(naechstes Lied, Pause, lauter/leiser) reicht das aber exakt so gut wie
|
||||
Alexas Intent-Slots -- nur ohne Slot-Fuellung. Freitext-Faelle ("spiel
|
||||
Playlist X auf Geraet Y ab") bleiben bewusst beim LLM, weil das ohne Modell
|
||||
so oder so nicht geht (siehe README).
|
||||
|
||||
Ein Skript, ein Argument pro Aktion -- vermeidet zehn Mini-Skripte. Nutzt
|
||||
denselben `SpotifyClient` (plugins.spotify.client) wie der echte
|
||||
Tool-Dispatch: gleiche OAuth-Session, kein doppelter Token-Code.
|
||||
|
||||
Aufruf (aus quick_commands heraus, IM Container):
|
||||
python3 /opt/data/scripts/spotify_quick.py next
|
||||
python3 /opt/data/scripts/spotify_quick.py previous
|
||||
python3 /opt/data/scripts/spotify_quick.py pause
|
||||
python3 /opt/data/scripts/spotify_quick.py play
|
||||
python3 /opt/data/scripts/spotify_quick.py volume_up
|
||||
python3 /opt/data/scripts/spotify_quick.py volume_down
|
||||
python3 /opt/data/scripts/spotify_quick.py current
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
|
||||
try:
|
||||
from plugins.spotify.client import (
|
||||
SpotifyAPIError,
|
||||
SpotifyAuthRequiredError,
|
||||
SpotifyClient,
|
||||
SpotifyError,
|
||||
)
|
||||
except ImportError as exc: # pragma: no cover
|
||||
sys.exit(
|
||||
"Konnte plugins.spotify.client nicht importieren -- Skript muss "
|
||||
f"INNERHALB des hermes-agent-Containers laufen. Fehler: {exc}"
|
||||
)
|
||||
|
||||
VOLUME_STEP = 10
|
||||
|
||||
|
||||
def _current_volume(client: "SpotifyClient") -> int:
|
||||
state = client.get_playback_state() or {}
|
||||
device = state.get("device") or {}
|
||||
vol = device.get("volume_percent")
|
||||
return int(vol) if isinstance(vol, (int, float)) else 50
|
||||
|
||||
|
||||
def run(action: str) -> str:
|
||||
client = SpotifyClient()
|
||||
|
||||
if action == "next":
|
||||
client.skip_next()
|
||||
return "⏭ naechstes Lied"
|
||||
|
||||
if action == "previous":
|
||||
client.skip_previous()
|
||||
return "⏮ vorheriges Lied"
|
||||
|
||||
if action == "pause":
|
||||
client.pause_playback()
|
||||
return "⏸ pausiert"
|
||||
|
||||
if action == "play":
|
||||
client.start_playback()
|
||||
return "▶ weiter"
|
||||
|
||||
if action == "volume_up":
|
||||
new_vol = min(100, _current_volume(client) + VOLUME_STEP)
|
||||
client.set_volume(volume_percent=new_vol)
|
||||
return f"🔊 Lautstaerke {new_vol}%"
|
||||
|
||||
if action == "volume_down":
|
||||
new_vol = max(0, _current_volume(client) - VOLUME_STEP)
|
||||
client.set_volume(volume_percent=new_vol)
|
||||
return f"🔉 Lautstaerke {new_vol}%"
|
||||
|
||||
if action == "current":
|
||||
state = client.get_currently_playing() or {}
|
||||
item = state.get("item") or {}
|
||||
name = item.get("name")
|
||||
artists = ", ".join(a.get("name", "") for a in (item.get("artists") or []))
|
||||
if not name:
|
||||
return "Gerade laeuft nichts."
|
||||
return f"Läuft: {name} – {artists}" if artists else f"Läuft: {name}"
|
||||
|
||||
return (
|
||||
f"Unbekannte Aktion '{action}'. Erlaubt: next, previous, pause, play, "
|
||||
"volume_up, volume_down, current"
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
if len(sys.argv) < 2:
|
||||
sys.exit("Usage: spotify_quick.py <next|previous|pause|play|volume_up|volume_down|current>")
|
||||
|
||||
action = sys.argv[1].strip().lower()
|
||||
try:
|
||||
print(run(action))
|
||||
except SpotifyAuthRequiredError as exc:
|
||||
sys.exit(f"Spotify nicht eingeloggt: {exc}")
|
||||
except SpotifyAPIError as exc:
|
||||
# Echten error.reason zitieren statt zu raten (z.B. NO_ACTIVE_DEVICE,
|
||||
# ALREADY_PAUSED, PREMIUM_REQUIRED) -- gleiche Regel wie ueberall sonst.
|
||||
sys.exit(f"Spotify-Fehler (status={exc.status_code}): {exc}")
|
||||
except SpotifyError as exc:
|
||||
sys.exit(f"Spotify-Fehler: {exc}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user