diff --git a/aria-brain/agent.py b/aria-brain/agent.py index f02ca6c..7204f9e 100644 --- a/aria-brain/agent.py +++ b/aria-brain/agent.py @@ -33,6 +33,7 @@ from proxy_client import ProxyClient, Message as ProxyMessage import router as router_mod import metrics from local_llm import local_llm_chat +import local_compact import skills as skills_mod import triggers as triggers_mod import watcher as watcher_mod @@ -2359,7 +2360,8 @@ class Agent: # /shared/config/local_llm.json (Default aus → alles Claude wie bisher). # Rueckgabe: fertige Antwort (str) wenn lokal erledigt, sonst None → Claude. - _LOCAL_WINDOW_TURNS = 8 # nur die letzten N Turns ans lokale Modell (Speed) + # Wie viele der letzten Turns ans lokale Modell gehen, regelt jetzt + # local_compact (nach Tokens + LOCAL_MAX_TURNS, aeltere als Notizzettel). _LOCAL_TOOL_ITERATIONS = 3 # max Tool-Runden lokal, sonst → Claude # Weiche Fehler-Marker im Tool-Ausgabetext. Viele Action-Skills geben einen @@ -2476,9 +2478,21 @@ class Agent: sys_prompt = router_mod.build_local_system_prompt(IDENTITY_ANCHOR, has_tools=bool(tools)) - # Nur die letzten paar Turns ans lokale Modell (Speed — volles Fenster - # wuerde das Prefill aufblaehen). - window = self.conversation.window(project_id=active_project_id)[-self._LOCAL_WINDOW_TURNS:] + # Compacting (local_compact): aeltere Nachrichten als Notizzettel im + # Systemprompt, dazu so viele der letzten Nachrichten, wie ins + # Kontextfenster der Karte passen (hoechstens LOCAL_MAX_TURNS — Tempo). + notes = local_compact.summary_block(active_project_id) + if notes: + sys_prompt += "\n\n" + notes + all_turns = self.conversation.window(project_id=active_project_id) + ctx = local_compact.model_ctx(local_model) + budget = int(ctx * 0.75) - 500 - local_compact.est_tokens(sys_prompt) + window = local_compact.fit_window(all_turns, max(budget, 256)) + try: + local_compact.maybe_compact(active_project_id, all_turns, window, + local_model, local_llm_chat) + except Exception as exc: + logger.debug("[compact] uebersprungen: %s", exc) messages = [{"role": "system", "content": sys_prompt}] messages += [{"role": t.role, "content": t.content} for t in window] diff --git a/aria-brain/local_compact.py b/aria-brain/local_compact.py new file mode 100644 index 0000000..cc25d72 --- /dev/null +++ b/aria-brain/local_compact.py @@ -0,0 +1,186 @@ +""" +Compacting fuer ARIAs lokalen Modus (Fast-Lane / localOnly). + +Ein lokales Modell hat ein kleines Kontextfenster (z.B. 32k auf einer +12-GB-Karte). Frueher bekam es starr die letzten 8 Nachrichten — alles davor +war fuer das lokale Modell einfach weg. Jetzt: + + * Fenster nach TOKENS statt fester Anzahl: so viele der letzten Nachrichten, + wie in ~75 % des Kontextfensters passen (Systemprompt + Antwortplatz + abgezogen), hoechstens LOCAL_MAX_TURNS (Tempo). + * Was aus dem Fenster faellt, fasst das lokale Modell im HINTERGRUND zu + einer laufenden Zusammenfassung zusammen ("Notizzettel"). Die liegt pro + Projekt in /data/local_summaries.json und geht bei jeder lokalen Anfrage + als "Bisheriger Verlauf" in den Systemprompt. + +Unabhaengig davon laeuft weiter das Destillat (agent.distill_old_turns): +dauerhaft Wichtiges wandert als Fakt ins Gedaechtnis (Qdrant). Compacting = +Notizzettel fuer DIESES Gespraech, Destillat = Karteikarten fuers Archiv. +""" + +from __future__ import annotations + +import json +import logging +import os +import threading +import time +import urllib.request +from pathlib import Path + +logger = logging.getLogger(__name__) + +BRIDGE_URL = os.environ.get("BRIDGE_URL", "http://aria-bridge:8090") +SUMMARY_FILE = Path(os.environ.get("LOCAL_SUMMARY_FILE", "/data/local_summaries.json")) +LOCAL_MAX_TURNS = int(os.environ.get("LOCAL_MAX_TURNS", "12")) +DEFAULT_CTX = 8192 +CHARS_PER_TOKEN = 2.8 # Deutsch/Code: eher vorsichtig schaetzen +COMPACT_MIN_TURNS = 4 # erst ab so vielen neu herausgefallenen Nachrichten +SUMMARY_MAX_TOKENS = 450 + +_lock = threading.Lock() +_running: set = set() # Projekte, fuer die gerade verdichtet wird +_ctx_cache: dict = {"ts": 0.0, "models": {}} + + +def est_tokens(text: str) -> int: + return int(len(text or "") / CHARS_PER_TOKEN) + 4 + + +def model_ctx(model: str) -> int: + """Kontextfenster des lokalen Modells laut Cluster (60 s gecacht).""" + now = time.time() + if now - _ctx_cache["ts"] > 60: + try: + with urllib.request.urlopen(f"{BRIDGE_URL}/internal/llm-models", timeout=3) as r: + data = json.loads(r.read().decode("utf-8", "ignore")) + _ctx_cache["models"] = {m.get("model"): int(m.get("ctx") or 0) + for m in data.get("models") or []} + _ctx_cache["ts"] = now + except Exception as exc: + logger.debug("[compact] Kontextfenster nicht abfragbar: %s", exc) + _ctx_cache["ts"] = now - 50 # in 10 s nochmal versuchen + return _ctx_cache["models"].get(model) or DEFAULT_CTX + + +def fit_window(turns: list, budget_tokens: int) -> list: + """Die neuesten Nachrichten, die ins Budget passen (mind. die letzte).""" + out, used = [], 0 + for t in reversed(turns[-LOCAL_MAX_TURNS:]): + need = est_tokens(t.content) + if out and used + need > budget_tokens: + break + out.append(t) + used += need + return list(reversed(out)) + + +# ── Laufende Zusammenfassung pro Projekt ───────────────────────────── + +def _load() -> dict: + try: + return json.loads(SUMMARY_FILE.read_text(encoding="utf-8")) + except Exception: + return {} + + +def get_summary(project_id: str) -> dict: + """{text, upto_ts} — upto_ts = Zeitstempel der juengsten verdichteten Nachricht.""" + s = _load().get(project_id or "", {}) + return {"text": s.get("text", ""), "upto_ts": s.get("upto_ts", "")} + + +def _save_summary(project_id: str, text: str, upto_ts: str) -> None: + with _lock: + data = _load() + data[project_id or ""] = {"text": text, "upto_ts": upto_ts, + "updated": time.strftime("%Y-%m-%dT%H:%M:%S")} + tmp = SUMMARY_FILE.with_suffix(".tmp") + tmp.write_text(json.dumps(data, ensure_ascii=False, indent=1), encoding="utf-8") + tmp.replace(SUMMARY_FILE) + + +def reset(project_id=None) -> None: + """Zusammenfassung(en) verwerfen (z.B. beim Zuruecksetzen des Gespraechs).""" + with _lock: + if project_id is None: + try: + SUMMARY_FILE.unlink() + except FileNotFoundError: + pass + return + data = _load() + data.pop(project_id or "", None) + SUMMARY_FILE.write_text(json.dumps(data, ensure_ascii=False, indent=1), encoding="utf-8") + + +def summary_block(project_id: str) -> str: + s = get_summary(project_id).get("text", "").strip() + if not s: + return "" + return ("## BISHERIGER VERLAUF (Zusammenfassung aelterer Nachrichten)\n" + "Das ist eine Zusammenfassung dessen, was vor den letzten Nachrichten " + "besprochen wurde. Nutze es als Hintergrund, zitiere es nicht.\n" + s) + + +def maybe_compact(project_id: str, all_turns: list, window: list, model: str, + chat_fn) -> None: + """Startet im Hintergrund eine Verdichtung, wenn genug Nachrichten aus dem + Fenster gefallen sind, die noch nicht im Notizzettel stehen. + chat_fn = local_llm_chat (wird nur im Hintergrund-Thread aufgerufen).""" + if not window: + return + first_kept = window[0].ts + upto = get_summary(project_id)["upto_ts"] + dropped = [t for t in all_turns if t.ts < first_kept and t.ts > upto] + if len(dropped) < COMPACT_MIN_TURNS: + return + key = project_id or "" + with _lock: + if key in _running: + return + _running.add(key) + threading.Thread(target=_compact, args=(key, dropped, model, chat_fn), + daemon=True, name=f"compact-{key or 'main'}").start() + + +def _compact(project_id: str, dropped: list, model: str, chat_fn) -> None: + try: + prev = get_summary(project_id)["text"] + ctx = model_ctx(model) + # Was in die Verdichtungs-Anfrage passt: ~60 % des Fensters + room = int(ctx * 0.6 * CHARS_PER_TOKEN) - len(prev) - 2000 + lines, used, last_ts = [], 0, "" + for t in dropped: # aelteste zuerst + who = "Stefan" if t.role == "user" else "ARIA" + line = f"{who}: {t.content.strip()}" + if used + len(line) > room: + line = line[: max(0, room - used)] + " …" + lines.append(line) + used += len(line) + last_ts = t.ts + if used >= room: + break # Rest beim naechsten Durchlauf + system = ("Du fuehrst einen laufenden Notizzettel ueber ein Gespraech zwischen " + "Stefan und seiner Assistentin ARIA. Schreibe auf Deutsch eine knappe " + "Zusammenfassung (hoechstens ~250 Woerter, Stichpunkte erlaubt): worum " + "es ging, was entschieden oder erledigt wurde, offene Punkte, wichtige " + "Namen/Zahlen. Kein Smalltalk, keine Einleitung, nur der Notizzettel.") + user = ((f"Bisheriger Notizzettel:\n{prev}\n\n" if prev else "") + + "Diese Nachrichten kommen neu dazu:\n" + "\n".join(lines) + + "\n\nSchreibe den aktualisierten Notizzettel.") + res = chat_fn([{"role": "system", "content": system}, {"role": "user", "content": user}], + max_tokens=SUMMARY_MAX_TOKENS, temperature=0.2, model=model, timeout=180) + text = (res.get("content") or "").strip() if res.get("ok") else "" + if not text: + logger.warning("[compact] Verdichtung fehlgeschlagen (%s) — Notizzettel bleibt", + res.get("error")) + return + _save_summary(project_id, text, last_ts) + logger.info("[compact] %d Nachrichten verdichtet (Projekt %r) → %d Zeichen Notizzettel", + len(dropped), project_id, len(text)) + except Exception as exc: + logger.warning("[compact] Fehler: %s", exc) + finally: + with _lock: + _running.discard(project_id or "") diff --git a/aria-brain/local_llm.py b/aria-brain/local_llm.py index 728b9f6..da2b743 100644 --- a/aria-brain/local_llm.py +++ b/aria-brain/local_llm.py @@ -30,7 +30,7 @@ LOCAL_LLM_HTTP_TIMEOUT_SEC = float(os.environ.get("LOCAL_LLM_HTTP_TIMEOUT_SEC", def local_llm_chat(messages: list, *, max_tokens: int = 512, temperature: float = 0.7, stop=None, tools=None, - model=None) -> dict: + model=None, timeout=None) -> dict: """Ein Chat-Call ans lokale LLM. messages = [{role, content}, ...]. model (B0.5): welches Modell llama-swap laden soll. tools (B1b): optionale OpenAI-Tool-Defs; das Ergebnis kann dann result['tool_calls'] enthalten. @@ -44,13 +44,15 @@ def local_llm_chat(messages: list, *, max_tokens: int = 512, req["tools"] = tools if model: req["model"] = model + if timeout: + req["timeout"] = timeout # Bridge-seitige Frist (sonst 30 s) try: body = json.dumps(req).encode("utf-8") http_req = urllib.request.Request( f"{BRIDGE_URL}/internal/local-llm", data=body, method="POST", headers={"Content-Type": "application/json"}, ) - with urllib.request.urlopen(http_req, timeout=LOCAL_LLM_HTTP_TIMEOUT_SEC) as resp: + with urllib.request.urlopen(http_req, timeout=(timeout + 5) if timeout else LOCAL_LLM_HTTP_TIMEOUT_SEC) as resp: result = json.loads(resp.read().decode("utf-8", "ignore")) except urllib.error.HTTPError as exc: try: diff --git a/aria-brain/main.py b/aria-brain/main.py index 4cd24c4..616c700 100644 --- a/aria-brain/main.py +++ b/aria-brain/main.py @@ -1190,6 +1190,11 @@ def conversation_reset(): """Hardes Reset — der Rolling-Window-Verlauf wird komplett geleert. Destillierte facts bleiben in der DB.""" conversation().reset() + try: + import local_compact + local_compact.reset() # Notizzettel des lokalen Modus gehoert dazu + except Exception: + pass return {"ok": True, "turns": 0} diff --git a/bridge/aria_bridge.py b/bridge/aria_bridge.py index 4cb52ec..77ecc6a 100644 --- a/bridge/aria_bridge.py +++ b/bridge/aria_bridge.py @@ -4271,7 +4271,7 @@ class ARIABridge: async def _local_llm(self, messages: list, max_tokens: int = 512, temperature: float = 0.7, stop=None, tools=None, - model=None) -> dict: + model=None, timeout: Optional[float] = None) -> dict: """Schickt einen llm_request an den llm-adapter (AI-Box), wartet auf llm_response. tools (B1b) werden durchgereicht; tool_calls kommen zurueck. Rueckgabe: {ok, content, tool_calls, model, elapsedMs} oder {ok:False, error}.""" @@ -4286,7 +4286,7 @@ class ARIABridge: res = await self._llm_cluster.route({ "model": model or "", "messages": messages, "max_tokens": max_tokens, "temperature": temperature, "stop": stop, "tools": tools, - }, timeout=self._LLM_TIMEOUT_S) + }, timeout=timeout or self._LLM_TIMEOUT_S) if not res.get("ok"): err = res.get("error") or "llm-adapter Fehler" if res.get("errorCode") == "model_unavailable": @@ -4773,6 +4773,10 @@ class ARIABridge: elif method == "POST" and path == "/internal/satellite-list": # Brain fragt: welche Satelliten/Netze sind online + Capabilities. await _send_response(writer, 200, {"ok": True, "satellites": self._satellite_list()}) + elif method in ("GET", "POST") and path == "/internal/llm-models": + # Brain fragt: welche lokalen Modelle sind geladen + ihr + # Kontextfenster (fuer ARIAs Compacting im lokalen Modus). + await _send_response(writer, 200, {"ok": True, "models": self._llm_cluster.models()}) elif method in ("GET", "POST") and path == "/internal/worker-list": # Diagnostic/Brain fragt: welche Compute-Worker sind online (Flotte). await _send_response(writer, 200, {"ok": True, "workers": self._worker_list()}) @@ -4934,10 +4938,16 @@ class ARIABridge: temperature = 0.7 _tools = data.get("tools") if isinstance(data.get("tools"), list) else None _model = data.get("model") if isinstance(data.get("model"), str) else None + # Hintergrund-Arbeit (Compacting) darf laenger dauern als die + # 30 s fuer schnelle Antworten — gedeckelt auf 5 min. + try: + _timeout = min(300.0, float(data["timeout"])) if data.get("timeout") else None + except (TypeError, ValueError): + _timeout = None result = await self._local_llm( messages=messages, max_tokens=max_tokens, temperature=temperature, stop=data.get("stop"), - tools=_tools, model=_model, + tools=_tools, model=_model, timeout=_timeout, ) status = 200 if result.get("ok") else 502 await _send_response(writer, status, result) diff --git a/diagnostic/index.html b/diagnostic/index.html index 0f2bb75..289ac05 100644 --- a/diagnostic/index.html +++ b/diagnostic/index.html @@ -237,6 +237,13 @@
🗂️ Sitzung = der Aktenordner beim Programm: beliebig dick, aber pro Anfrage passt nur ein Teil auf den Schreibtisch.
🗄️ ARIAs Langzeitgedächtnis = das Archiv: daraus wird bei Bedarf gezielt etwas auf den Schreibtisch gelegt.
+
+
Wenn der Schreibtisch voll ist — drei Wege
+
✂️ Kürzen: die ältesten Blätter fliegen weg. Einfach, aber Frühes ist komplett vergessen.
+
🗜️ Compacting (Verdichten): das Modell liest die alten Blätter und schreibt einen Notizzettel — worum es ging, was entschieden wurde, was offen ist. Der Zettel bleibt liegen, die Blätter gehen. Zusammenhänge bleiben, Einzelheiten (genaue Fehlermeldungen, Code-Stellen) gehen verloren. So macht es z.B. Claude Code in VS Code.
+
🗄️ Gedächtnis: dauerhaft Wichtiges wird als Fakt ins Archiv gelegt und bei Bedarf wieder hervorgeholt.
+
ARIA nutzt Gedächtnis immer (die ältesten Nachrichten werden zu Fakten destilliert) und im lokalen Modus zusätzlich Compacting: was aus ihrem Fenster fällt, fasst das lokale Modell im Hintergrund zum Notizzettel zusammen. Angebundene Programme (Software-Schmiede, OpenHands …) entscheiden selbst, welchen Weg sie gehen.
+

Mehr Kontext = das Modell überblickt in einer Anfrage mehr vom Verlauf, es muss seltener gekürzt werden. Es kostet aber Grafikspeicher (die GB-Angabe in der Auswahl ist eine Schätzung für Modell + Kontext). Passt es nicht auf die Karte, setzt die Box sie von selbst auf den alten Wert zurück.

Höchstens so viel, wie das Modell gelernt hat (steht in der Modelldatei, z.B. Gemma 4 12B: 256k). Angebundene Programme bekommen den eingestellten Wert automatisch mitgeteilt.