""" Hermes Agent — lokaler KI-Assistent fuer das Agentic OS. Empfaengt Aufgaben per Text (und Voice), nutzt llama-swap als LLM-Backend, kann per Tools auf das Bosgame und spaeter per SSH auf den Windows-PC zugreifen. """ import asyncio import json import subprocess from pathlib import Path from typing import Callable, Awaitable import httpx from config import LLAMA_SWAP_URL, HERMES_SIMPLE_MODEL, HERMES_COMPLEX_MODEL # --------------------------------------------------------------------------- # Tool-Definitionen (OpenAI Function Calling Format) # --------------------------------------------------------------------------- TOOL_DEFS = [ { "type": "function", "function": { "name": "read_file", "description": "Liest eine Textdatei auf dem Bosgame.", "parameters": { "type": "object", "properties": { "path": {"type": "string", "description": "Absoluter Pfad zur Datei"} }, "required": ["path"], }, }, }, { "type": "function", "function": { "name": "list_directory", "description": "Listet Dateien und Ordner eines Verzeichnisses auf dem Bosgame.", "parameters": { "type": "object", "properties": { "path": {"type": "string", "description": "Absoluter Pfad zum Verzeichnis"} }, "required": ["path"], }, }, }, { "type": "function", "function": { "name": "run_command", "description": ( "Fuehrt einen Lese-Befehl auf dem Bosgame aus (kein sudo, kein rm). " "Geeignet fuer: ls, cat, ps, df, free, journalctl, systemctl status, ip, curl, etc." ), "parameters": { "type": "object", "properties": { "command": {"type": "string", "description": "Shell-Befehl"} }, "required": ["command"], }, }, }, { "type": "function", "function": { "name": "get_system_status", "description": "Gibt aktuellen System-Status zurueck: CPU, RAM, GPU, laufende Modelle.", "parameters": {"type": "object", "properties": {}}, }, }, { "type": "function", "function": { "name": "get_memories", "description": "Laedt alle gespeicherten Fakten und Entscheidungen.", "parameters": {"type": "object", "properties": {}}, }, }, { "type": "function", "function": { "name": "add_memory", "description": "Speichert eine neue Information dauerhaft ins Gedaechtnis.", "parameters": { "type": "object", "properties": { "content": {"type": "string", "description": "Der zu speichernde Fakt"}, "category": { "type": "string", "enum": ["user", "instruction", "stable", "versioned", "ephemeral"], "description": ( "user=Fakten ueber den User (Wer ist er? Wie arbeitet er?), " "instruction=Verhaltensregeln fuer alle KI-Tools, " "stable=Projektfakten, " "versioned=Tech-Versionen (bei Updates ueberschreiben), " "ephemeral=temporaer (7 Tage)" ), }, }, "required": ["content"], }, }, }, { "type": "function", "function": { "name": "web_search", "description": "Sucht im Internet nach aktuellen Informationen.", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Suchbegriff"} }, "required": ["query"], }, }, }, ] # --------------------------------------------------------------------------- # Tool-Implementierungen # --------------------------------------------------------------------------- _BLOCKED_CMDS = { "rm", "rmdir", "mv", "cp", "dd", "mkfs", "fdisk", "parted", "shutdown", "reboot", "halt", "poweroff", "chmod", "chown", "passwd", "userdel", "useradd", } def _exec_read_file(path: str) -> str: try: p = Path(path) if not p.exists(): return f"Datei nicht gefunden: {path}" content = p.read_text(errors="replace") if len(content) > 8000: return content[:8000] + f"\n\n... (gekuerzt, gesamt {len(content):,} Zeichen)" return content except Exception as e: return f"Fehler beim Lesen: {e}" def _exec_list_directory(path: str) -> str: try: p = Path(path) if not p.exists(): return f"Pfad nicht gefunden: {path}" items = sorted(p.iterdir(), key=lambda x: (x.is_file(), x.name.lower())) lines = [] for item in items[:80]: if item.is_dir(): lines.append(f"📁 {item.name}/") else: size = item.stat().st_size sz = f"{size/1024/1024:.1f} MB" if size > 1024*1024 else f"{size/1024:.1f} KB" if size > 1024 else f"{size} B" lines.append(f"📄 {item.name} ({sz})") total = sum(1 for _ in p.iterdir()) if total > 80: lines.append(f"... und {total - 80} weitere") return "\n".join(lines) if lines else "(leer)" except Exception as e: return f"Fehler: {e}" def _exec_run_command(command: str) -> str: first = command.strip().split()[0] if command.strip() else "" if first in _BLOCKED_CMDS or "sudo" in command: return f"Befehl '{first}' blockiert. Nutze Mission Control fuer Systemoperationen." try: result = subprocess.run( command, shell=True, capture_output=True, text=True, timeout=30 ) out = (result.stdout + result.stderr).strip() if len(out) > 4000: out = out[:4000] + "\n... (Ausgabe gekuerzt)" return out or "(kein Output)" except subprocess.TimeoutExpired: return "Timeout nach 30 Sekunden." except Exception as e: return f"Fehler: {e}" def _exec_get_system_status() -> str: base = "http://127.0.0.1:9000" try: s = httpx.get(f"{base}/api/status", timeout=5).json() models = s.get("models", []) running = [m for m in models if m.get("state") in ("running", "ready", "loading")] model_str = ", ".join(m["name"] for m in running) if running else "keines" except Exception: model_str = "unbekannt" try: sys = httpx.get(f"{base}/api/system/status", timeout=5).json() return ( f"Aktives Modell: {model_str}\n" f"CPU: {sys.get('cpu_pct', '?')}% | " f"RAM: {sys.get('ram_used_gb', '?')} / {sys.get('ram_total_gb', '?')} GB | " f"GPU: {sys.get('gpu_mem_used_gb', '?')} / {sys.get('gpu_mem_total_gb', '?')} GB GTT\n" f"Temp: CPU {sys.get('cpu_temp_c', '?')}°C" ) except Exception as e: return f"Aktives Modell: {model_str}\nSystem-Metriken nicht verfuegbar: {e}" def _exec_get_memories() -> str: try: items = httpx.get("http://127.0.0.1:9000/api/memory", timeout=5).json() if not items: return "Kein Gedaechtnis vorhanden." return "\n".join(f"[{m['category']}] {m['content']}" for m in items) except Exception as e: return f"Fehler: {e}" def _exec_add_memory(content: str, category: str = "stable") -> str: try: r = httpx.post( "http://127.0.0.1:9000/api/memory", json={"content": content, "category": category, "source": "hermes"}, timeout=5, ) r.raise_for_status() return f"Gespeichert: {content}" except Exception as e: return f"Fehler: {e}" def _exec_web_search(query: str) -> str: try: r = httpx.get( "https://api.duckduckgo.com/", params={"q": query, "format": "json", "no_html": 1, "skip_disambig": 1}, timeout=10, headers={"User-Agent": "HermesAgent/1.0"}, follow_redirects=True, ) data = r.json() results = [] if data.get("AbstractText"): results.append(data["AbstractText"]) for rt in data.get("RelatedTopics", [])[:4]: if isinstance(rt, dict) and rt.get("Text"): results.append(rt["Text"]) return "\n\n".join(results) if results else "Keine direkten Ergebnisse. Versuche eine genauere Suchanfrage." except Exception as e: return f"Websuche fehlgeschlagen: {e}" def execute_tool(name: str, args: dict) -> str: if name == "read_file": return _exec_read_file(args.get("path", "")) if name == "list_directory": return _exec_list_directory(args.get("path", "")) if name == "run_command": return _exec_run_command(args.get("command", "")) if name == "get_system_status": return _exec_get_system_status() if name == "get_memories": return _exec_get_memories() if name == "add_memory": return _exec_add_memory(args.get("content", ""), args.get("category", "stable")) if name == "web_search": return _exec_web_search(args.get("query", "")) return f"Unbekanntes Tool: {name}" # --------------------------------------------------------------------------- # Modell-Auswahl # --------------------------------------------------------------------------- _COMPLEX_KW = { "schreib", "erstell", "baue", "plan", "refactor", "debug", "analysier", "implementier", "entwickl", "code", "programm", "erklaer ausfuehrlich", "erstelle", "generier", } def choose_model(message: str) -> str: lower = message.lower() if any(kw in lower for kw in _COMPLEX_KW): return HERMES_COMPLEX_MODEL return HERMES_SIMPLE_MODEL # --------------------------------------------------------------------------- # System-Prompt # --------------------------------------------------------------------------- SYSTEM_PROMPT = """\ Du bist Hermes, ein lokaler KI-Assistent der dauerhaft auf dem Bosgame M5 laeuft. Dein Zuhause (Bosgame M5): - AMD Strix Halo, ~124 GB GTT-Speicher, Ubuntu 26.04, Kernel 7.0 - LAN-IP: 192.168.178.151 - Dienste: llama-swap :8080 (Inference), Mission Control :9000 (Dashboard) - Modelle: coder (Qwen3-30B-A3B), scout (Qwen3-8B), vision (Qwen3-VL) - Du laeuft als Teil von Mission Control Gespeichertes Wissen (nach Kategorie): {memories} Verhaltensregeln (fest): - Antworte praegnant auf Deutsch (max 2-3 Saetze wenn nicht ausdruecklich mehr gewuenscht) - Kuendige an was du tust bevor du es tust - Frag nach bevor du Dateien ueberschreibst oder Dienste neustartest - Nutze get_system_status() wenn du nicht sicher bist was gerade laeuft - Bei einfachen Fragen: keine Tools, direkt antworten Gedaechtnis-Kategorien (fuer add_memory): - user → Fakten ueber den User: Wer er ist, wie er arbeitet, was er bevorzugt - instruction → Verhaltensregeln die fuer ALLE KI-Tools gelten sollen - stable → Technische Projektfakten (IPs, Ports, Architektur) - versioned → Tech-Versionen (bei Updates ueberschreiben, nicht anhaengen) - ephemeral → Temporaerer Kontext (7 Tage, dann weg) Speichere wichtige Erkenntnisse aktiv per add_memory() — besonders wenn der User etwas ueber sich selbst erwaehnt (→ user) oder Verhaltensregeln nennt (→ instruction). """ # --------------------------------------------------------------------------- # LLM-Aufruf (synchron, in Executor ausfuehren) # --------------------------------------------------------------------------- def _call_llm(model: str, messages: list) -> dict: r = httpx.post( f"{LLAMA_SWAP_URL}/v1/chat/completions", json={ "model": model, "messages": messages, "tools": TOOL_DEFS, "tool_choice": "auto", "temperature": 0.3, }, timeout=120, ) r.raise_for_status() return r.json() # --------------------------------------------------------------------------- # Haupt-Agent-Loop # --------------------------------------------------------------------------- SendFn = Callable[[dict], Awaitable[None]] async def run_agent(message: str, send: SendFn) -> None: """ Fuehrt den Agenten-Loop aus und streamt Ergebnisse via send()-Callback. Nachrichten-Typen: {"type": "info", "content": str} — Status-Meldung {"type": "thinking"} — LLM denkt nach {"type": "token", "content": str} — Wort der Antwort {"type": "tool_call", "name": str, "args": dict} {"type": "tool_result", "name": str, "content": str} {"type": "error", "content": str} {"type": "done"} """ memories = _exec_get_memories() model = choose_model(message) await send({"type": "info", "content": f"Modell: {model}"}) messages = [ {"role": "system", "content": SYSTEM_PROMPT.format(memories=memories)}, {"role": "user", "content": message}, ] for _round in range(8): await send({"type": "thinking"}) try: result = await asyncio.get_event_loop().run_in_executor( None, lambda: _call_llm(model, messages) ) except Exception as exc: await send({"type": "error", "content": f"LLM nicht erreichbar: {exc}"}) return choice = result["choices"][0] msg = choice["message"] finish = choice.get("finish_reason", "") tool_calls = msg.get("tool_calls") or [] # Text-Antwort: Wort fuer Wort streamen if msg.get("content"): words = msg["content"].split(" ") for i, word in enumerate(words): token = word + (" " if i < len(words) - 1 else "") await send({"type": "token", "content": token}) await asyncio.sleep(0.012) if not tool_calls or finish == "stop": break # Tool-Calls ausfuehren messages.append(msg) for tc in tool_calls: fn = tc["function"] name = fn["name"] try: args = json.loads(fn.get("arguments", "{}")) except Exception: args = {} await send({"type": "tool_call", "name": name, "args": args}) tool_out = await asyncio.get_event_loop().run_in_executor( None, lambda n=name, a=args: execute_tool(n, a) ) preview = tool_out[:300] + "..." if len(tool_out) > 300 else tool_out await send({"type": "tool_result", "name": name, "content": preview}) messages.append({ "role": "tool", "tool_call_id": tc["id"], "content": tool_out, }) await send({"type": "done"})