""" System-Prompt-Bau aus Memory-Punkten. Strategie: 1. Alle pinned Punkte (Hot Memory) — gruppiert nach Type — in den System-Prompt schreiben. IMMER drin. 2. Top-K semantisch aehnliche Punkte (Cold Memory) zur aktuellen User-Nachricht — als "Moeglicherweise relevant" eingehaengt. 3. Aktive Skills als kompakte Liste (nur Name + Description) — damit ARIA weiss was sie hat. Phase B Punkt 1: nur Hot-Memory-Bau, Skills + Cold-Search kommen mit dem Conversation-Loop in spaeteren Phasen. """ from __future__ import annotations from datetime import datetime, timezone, timedelta from typing import List from memory import MemoryPoint def build_time_section() -> str: """Aktueller Zeitstempel — damit ARIA Timer korrekt anlegen kann und Watcher-Conditions mit hour_of_day etc. einordenbar bleiben.""" now_utc = datetime.now(timezone.utc) # Europa/Berlin: Sommerzeit CEST = UTC+2, Winterzeit CET = UTC+1. # Wir nehmen den simplen Fall (kein zoneinfo-Import noetig im Brain-Image): # Stefans VM laeuft auf UTC, die Bridge in der Wohnung — Anzeige reicht. local_offset_h = 2 if 3 <= now_utc.month <= 10 else 1 local = now_utc + timedelta(hours=local_offset_h) lines = [ "## Aktuelle Zeit", f"- UTC: {now_utc.isoformat(timespec='seconds')}", f"- Lokal (Europa/Berlin, UTC+{local_offset_h}): " f"{local.strftime('%Y-%m-%d %H:%M:%S')} ({local.strftime('%A')})", "", "Nutze das fuer Trigger-Timestamps und um Watcher-Conditions wie " "`hour_of_day == 8` einzuordnen. Fuer relative Angaben " "('in 10min', 'in 2 Stunden') nutze beim `trigger_timer` den " "`in_seconds`-Parameter — Server rechnet dann selbst.", ] return "\n".join(lines) TYPE_HEADINGS = { "identity": "## Wer du bist", "rule": "## Sicherheitsregeln & Prinzipien", "preference": "## Benutzer-Praeferenzen", "tool": "## Tool-Freigaben", "skill": "## Deine Skills", } def build_hot_memory_section(pinned: List[MemoryPoint]) -> str: """Baue den 'IMMER-im-Prompt'-Block aus pinned Punkten.""" grouped: dict[str, List[MemoryPoint]] = {} for p in pinned: grouped.setdefault(p.type, []).append(p) parts: List[str] = [] # Sortier-Reihenfolge: identity → rule → preference → tool → skill → Rest order = ["identity", "rule", "preference", "tool", "skill"] for t in order: items = grouped.pop(t, []) if not items: continue parts.append(TYPE_HEADINGS.get(t, f"## {t}")) for p in items: parts.append(f"### {p.title}") parts.append(p.content.strip()) parts.append("") # uebrige Types (falls jemand was anderes als pinned markiert) for t, items in grouped.items(): parts.append(f"## {t}") for p in items: parts.append(f"### {p.title}") parts.append(p.content.strip()) parts.append("") return "\n".join(parts).strip() def build_cold_memory_section(matches: List[MemoryPoint]) -> str: """Baue 'Moeglicherweise relevant'-Block aus Search-Treffern.""" if not matches: return "" lines = ["## Moeglicherweise relevant (aus Gedaechtnis)"] for p in matches: score = f" [score={p.score:.2f}]" if p.score is not None else "" lines.append(f"- **{p.title}**{score}") lines.append(f" {p.content.strip()}") return "\n".join(lines) def build_skills_section(skills: List[dict]) -> str: """Listet alle Skills (aktiv + deaktiviert) damit ARIA weiss was es gibt und keine doppelt baut. Plus klare Schwelle wann ein Skill sich lohnt.""" lines = ["## Deine Skills"] if skills: for s in skills: active = s.get("active", True) marker = "" if active else " [DEAKTIVIERT — kann nicht aufgerufen werden]" lines.append(f"- **{s.get('name', '?')}**{marker} — {s.get('description', '(ohne Beschreibung)')}") lines.append("") lines.append("Wenn ein vorhandener Skill zur Aufgabe passt: nutze ihn via Tool-Call.") else: lines.append("(noch keine Skills vorhanden)") lines.append("") lines.append("### Wann lohnt sich ein neuer Skill?") lines.append("") lines.append("**Skills sind IMMER Python** — eigene venv pro Skill mit den noetigen " "pip-Paketen. Kein apt im Skill, kein systemweiter Install. Python deckt " "in der Regel alles ab (yt-dlp, requests, pypdf, pillow, openpyxl, " "static-ffmpeg, beautifulsoup4, …). Falls etwas WIRKLICH nur via apt geht: " "Stefan fragen ob es ins Brain-Dockerfile soll.") lines.append("") lines.append("**Harte Regel — IMMER Skill anlegen wenn:** die Loesung erfordert eine " "pip-Library. Begruendung: Brain-Container hat keinen persistenten State " "ausser /data/skills/. Ohne Skill wuerde der Install bei jedem " "Container-Restart wiederholt.") lines.append("") lines.append("**Sonst — Skill nur wenn alle vier zutreffen:**") lines.append("") lines.append("1. **Wiederkehrend** — die Aufgabe wird realistisch nochmal gestellt. " "Einmal-Faelle (\"wie spaet ist es jetzt\") kein Skill.") lines.append("2. **Nicht-trivial** — mehrere Schritte. Ein einzelner Shell-Befehl " "(`date`, `hostname`, `ls`) ist KEIN Skill — das macht Bash direkt.") lines.append("3. **Parametrisierbar** — der Skill nimmt Eingaben (URL, Datei, Suchbegriff) " "und gibt ein nuetzliches Ergebnis zurueck.") lines.append("4. **Wiederverwendbar als ganzes** — Stefan wuerde es zukuenftig per Name " "ansprechen (\"mach mir den YouTube zu MP3\") statt jedes Mal zu erklaeren.") lines.append("") lines.append("Wenn nichts installiert werden muss UND nicht alle vier zutreffen: einfach " "die Aufgabe loesen ohne Skill anzulegen. Stefan kann jederzeit sagen " "'bau daraus einen Skill'.") return "\n".join(lines) def build_triggers_section( triggers: List[dict], condition_vars: List[dict], condition_funcs: List[dict] | None = None, ) -> str: """Triggers (passive Aufweck-Quellen) + verfuegbare Condition-Variablen + Funktionen.""" lines = ["## Trigger (passive Aufweck-Quellen)"] lines.append("") lines.append("Trigger sind ANDERS als Skills: das System ruft DICH wenn ein Event passiert. " "Du legst sie an wenn Stefan sagt 'erinner mich an X' oder 'sag bescheid wenn Y'.") lines.append("") if triggers: lines.append("### Aktuelle Trigger") for t in triggers: active = t.get("active", True) mark = "" if active else " [INAKTIV]" if t["type"] == "timer": lines.append(f"- **{t['name']}**{mark} (timer) feuert {t.get('fires_at')}: \"{t.get('message','')[:80]}\"") elif t["type"] == "watcher": lines.append(f"- **{t['name']}**{mark} (watcher) cond=`{t.get('condition')}`: \"{t.get('message','')[:80]}\"") lines.append("") lines.append("### Verfuegbare Condition-Variablen (fuer Watcher)") for v in condition_vars: lines.append(f"- `{v['name']}` ({v['type']}) — {v['desc']}") if condition_funcs: lines.append("") lines.append("### Verfuegbare Funktionen in Conditions") for fn in condition_funcs: lines.append(f"- `{fn['signature']}` — {fn['desc']}") lines.append("") lines.append("Operatoren in Conditions: `<` `>` `<=` `>=` `==` `!=` `and` `or` `not`. " "Beispiele: `disk_free_gb < 5 and hour_of_day >= 8`, " "`day_of_week == \"mon\"`, `near(53.123, 7.456, 500)`. " "Funktionen nur mit Konstanten als Argumenten (keine Variablen, " "keine geschachtelten Funktionen).") lines.append("") lines.append("### Wann welcher Typ?") lines.append("- **Timer** fuer einmalige Erinnerungen mit konkreter Zeit ('in 10min', 'um 14:30').") lines.append("- **Watcher** fuer 'wenn X passiert' (Disk voll, bestimmte Tageszeit, GPS-Naehe).") lines.append("- ARIA legt Trigger NUR auf Stefan-Wunsch an, nicht eigenmaechtig.") lines.append("") lines.append("### GPS-Watcher mit near()") lines.append( "Wenn du einen Watcher mit `near()` anlegst: die App sendet GPS-Position " "nur kontinuierlich wenn Tracking AN ist (Default: AUS, Akku-Schutz). " "Rufe dafuer `request_location_tracking(on=true, reason=\"...\")` auf " "bevor oder gleich nach dem trigger_watcher. Sonst hat current_lat/lon " "veraltete Werte und der Watcher feuert nie. " "Beim Loeschen des letzten GPS-Watchers (trigger_cancel) wieder " "`request_location_tracking(on=false)` aufrufen.") return "\n".join(lines) def build_system_prompt( pinned: List[MemoryPoint], cold: List[MemoryPoint] | None = None, skills: List[dict] | None = None, triggers: List[dict] | None = None, condition_vars: List[dict] | None = None, condition_funcs: List[dict] | None = None, ) -> str: """Kompletter System-Prompt: Hot + Cold + Skills + Triggers.""" parts = [build_hot_memory_section(pinned), "", build_time_section()] if skills: parts.append("") parts.append(build_skills_section(skills)) if condition_vars: parts.append("") parts.append(build_triggers_section(triggers or [], condition_vars, condition_funcs)) if cold: parts.append("") parts.append(build_cold_memory_section(cold)) return "\n".join(parts).strip()