"""Die Agentenschleife: Modell fragen, Werkzeuge ausfuehren, wieder fragen. Ollama /api/chat mit `tools`. Antwortet das Modell mit tool_calls, werden sie (nach Freigabe) ausgefuehrt und als role=tool zurueckgegeben, bis es ohne Werkzeug antwortet. Kontext: Ollama schneidet zu lange Eingaben **stillschweigend** vorne ab - das Modell vergisst dann die Aufgabe, ohne dass es jemand merkt. Deshalb wird vor jedem Aufruf geschaetzt und notfalls gekuerzt: erst alte Werkzeugergebnisse, dann eine Zusammenfassung des Verlaufs. """ import json import os from collections import Counter import platform import re import time import uuid from dataclasses import dataclass, field from datetime import date from pathlib import Path from typing import Any, Callable, Protocol import httpx from . import freigabe, karte, katalog, konfig, protokoll, werkzeuge MAX_SCHRITTE = 60 ZEICHEN_JE_TOKEN = 3.2 # grob fuer Code und Deutsch; lieber zu vorsichtig KUERZEN_AB = 0.70 VERDICHTEN_AB = 0.85 RESERVE_ANTWORT = 4096 SYSTEM = """You are kode, an autonomous coding agent working in a terminal on a Linux server ({os}). You run as root. You can read, write and edit files and run shell commands with the provided tools. Working directory: {cwd} Date: {datum} How to work: - Explore before changing: use list_dir, glob, grep and read_file to understand the code. - Always read a file before editing it. Use edit_file for targeted changes, write_file only for new files or full rewrites. - Match the existing style, naming and comment language of the code you change. - After changes, verify them (run tests, linters, or the program) when that is possible. - Do not run destructive commands (rm -rf, dropping databases, force pushes) unless the user explicitly asked. - Be concise. Explain what you did in a few sentences at the end. No long preambles. - Call tools directly; do not describe tool calls in prose instead of making them. - Always answer the user in German. {projekt}""" _TEXT_AUFRUF = re.compile(r"\s*(\{.*?\})\s*", re.S) class Oberflaeche(Protocol): def text(self, stueck: str) -> None: ... def denken(self, stueck: str) -> None: ... def antwort_ende(self) -> None: ... def werkzeug(self, name: str, args: dict) -> None: ... def ergebnis(self, name: str, args: dict, text: str) -> None: ... def fragen(self, name: str, args: dict, diff: str | None) -> tuple[str, str]: ... def hinweis(self, text: str) -> None: ... @dataclass class Statistik: prompt_token: int = 0 antwort_token: int = 0 tok_s: float = 0.0 letzter_kontext: int = 0 @dataclass class Sitzung: modell_id: str cwd: Path modus: str = "fragen" verlauf: list[dict[str, Any]] = field(default_factory=list) stat: Statistik = field(default_factory=Statistik) kennung: str = field(default_factory=lambda: time.strftime("%Y%m%d-%H%M%S-") + uuid.uuid4().hex[:4]) def __post_init__(self) -> None: self.umgebung = werkzeuge.Umgebung(cwd=self.cwd, sitzung=self.kennung) self.projekt = self.cwd.resolve() self._an_kontext_anpassen() @property def modell(self) -> dict[str, Any]: return katalog.MODELLE[self.modell_id] @property def kontext(self) -> int: return int(self.modell["kontext"]) def faehigkeiten(self, basis: str) -> list[str]: schluessel = (basis, self.modell["hf"]) if getattr(self, "_faeh_fuer", None) != schluessel: self._faeh = karte.faehigkeiten(basis, self.modell) self._faeh_fuer = schluessel return self._faeh def _an_kontext_anpassen(self) -> None: # Ein Werkzeugergebnis darf hoechstens ein Viertel des Kontexts belegen. self.umgebung.max_ausgabe = min(werkzeuge.MAX_AUSGABE, int(self.kontext * ZEICHEN_JE_TOKEN * 0.25)) def modell_wechseln(self, mid: str) -> None: self.modell_id = mid self._an_kontext_anpassen() # ------------------------------------------------------------ Kontext def system(self) -> dict[str, str]: projekt = "" for name in ("KODE.md", "CLAUDE.md", "AGENTS.md"): p = self.projekt / name if p.is_file(): inhalt = p.read_text(encoding="utf-8", errors="replace")[:12000] projekt = f"\nProject notes from {name}:\n{inhalt}\n" break return {"role": "system", "content": SYSTEM.format( os=f"{platform.system()} {platform.release()}", cwd=self.umgebung.cwd, datum=date.today().isoformat(), projekt=projekt)} @staticmethod def _zeichen(nachrichten: list[dict[str, Any]]) -> int: n = 0 for m in nachrichten: n += len(m.get("content") or "") + 20 for t in m.get("tool_calls") or []: n += len(json.dumps(t, ensure_ascii=False)) return n def schaetzung(self) -> int: return int((self._zeichen([self.system()] + self.verlauf) + len(json.dumps(werkzeuge.SCHEMA))) / ZEICHEN_JE_TOKEN) def belegung(self) -> float: """Anteil des Kontexts: gemessen, wenn vorhanden, sonst geschaetzt.""" return max(self.stat.letzter_kontext, self.schaetzung()) / self.kontext def _alte_ergebnisse_kuerzen(self) -> int: gekuerzt = 0 for m in self.verlauf[:-6]: if m["role"] == "tool" and len(m.get("content") or "") > 400: m["content"] = m["content"][:300] + "\n[… gekürzt, um Kontext zu sparen - bei Bedarf neu lesen]" gekuerzt += 1 return gekuerzt def verdichten(self, ui: Oberflaeche, basis: str) -> None: """Ersetzt den Verlauf durch eine Zusammenfassung des Modells.""" if len(self.verlauf) < 3: return ui.hinweis("Kontext fast voll - Verlauf wird zusammengefasst …") t0 = time.monotonic() vorher = (len(self.verlauf), self.schaetzung()) auszug = [] for m in self.verlauf: if m["role"] == "tool": auszug.append(f"[tool result] {(m.get('content') or '')[:600]}") elif m.get("tool_calls"): for t in m["tool_calls"]: f = t.get("function", {}) auszug.append(f"[assistant called {f.get('name')}] " f"{json.dumps(f.get('arguments'), ensure_ascii=False)[:400]}") if m.get("content"): auszug.append(f"[assistant] {m['content'][:1500]}") else: auszug.append(f"[{m['role']}] {(m.get('content') or '')[:3000]}") text = "\n".join(auszug) grenze = int((self.kontext - RESERVE_ANTWORT) * ZEICHEN_JE_TOKEN * 0.8) text = text[-grenze:] anfrage = [{"role": "system", "content": "You summarize coding sessions."}, {"role": "user", "content": "Summarize this coding session so the work can continue without it. Include: the user's " "goal and requests, decisions, files touched and what changed, current state, open " "problems and next steps. Be specific (paths, function names, commands). German.\n\n" + text}] try: a = httpx.post(f"{basis}/api/chat", timeout=600, json={ "model": self.modell["hf"], "stream": False, "messages": anfrage, "options": {"num_ctx": self.kontext}, "keep_alive": -1, **({"think": False} if "thinking" in self.faehigkeiten(basis) else {})}) zusammenfassung = a.json()["message"]["content"] except (httpx.HTTPError, KeyError, ValueError) as f: protokoll.fehler("verdichten", f, nachrichten=vorher[0], token=vorher[1]) ui.hinweis(f"Zusammenfassen gescheitert ({f}) - älteste Nachrichten werden verworfen.") while len(self.verlauf) > 4 and self.belegung() > KUERZEN_AB: self.verlauf.pop(0) return self.verlauf = [{"role": "user", "content": "[Zusammenfassung des bisherigen Gesprächs]\n" + zusammenfassung}, {"role": "assistant", "content": "Verstanden, ich mache dort weiter."}] self.stat.letzter_kontext = 0 protokoll.ereignis("verdichten", nachrichten_vorher=vorher[0], token_vorher=vorher[1], token_nachher=self.schaetzung(), dauer_s=round(time.monotonic() - t0, 1), zusammenfassung=zusammenfassung) ui.hinweis("Verlauf zusammengefasst.") def _platz_schaffen(self, ui: Oberflaeche, basis: str) -> None: grenze = (self.kontext - RESERVE_ANTWORT) / self.kontext if self.schaetzung() / self.kontext > KUERZEN_AB: vorher = self.schaetzung() n = self._alte_ergebnisse_kuerzen() protokoll.ereignis("kontext_kuerzen", ergebnisse=n, token_vorher=vorher, token_nachher=self.schaetzung(), kontext=self.kontext) if self.schaetzung() / self.kontext > min(VERDICHTEN_AB, grenze): self.verdichten(ui, basis) # ------------------------------------------------------------ Aufruf def _strom(self, ui: Oberflaeche, basis: str) -> dict[str, Any]: nutzlast: dict[str, Any] = { "model": self.modell["hf"], "stream": True, "keep_alive": -1, "messages": [self.system()] + self.verlauf, "tools": werkzeuge.SCHEMA, "options": {"num_ctx": self.kontext, "temperature": 0.3}, } # Ohne think-Feld denken manche Modelle trotzdem (qwen3, gemessen # 16.10.2026). Gesetzt wird es nur, wenn Ollama "thinking" meldet - # sonst lehnt es den Aufruf ab. if "thinking" in self.faehigkeiten(basis): nutzlast["think"] = bool(self.modell.get("denkt")) inhalt, aufrufe, gedanken = [], [], [] gedacht = False letzter_puls = time.time() abgebrochen = False t0 = time.monotonic() erstes: float | None = None fertig: dict[str, Any] = {} protokoll.ereignis("anfrage", modell=self.modell_id, nachrichten=len(nutzlast["messages"]), token_geschaetzt=self.schaetzung(), num_ctx=self.kontext, think=nutzlast.get("think"), modus=self.modus) try: with httpx.stream("POST", f"{basis}/api/chat", json=nutzlast, timeout=httpx.Timeout(900, connect=30)) as a: if a.status_code != 200: raise RuntimeError(f"Ollama {a.status_code}: {a.read().decode()[:300]}") for zeile in a.iter_lines(): if not zeile.strip(): continue d = json.loads(zeile) if d.get("error"): raise RuntimeError(f"Ollama: {d['error']}") m = d.get("message") or {} if erstes is None and (m.get("thinking") or m.get("content") or m.get("tool_calls")): erstes = time.monotonic() - t0 if m.get("thinking"): gedacht = True gedanken.append(m["thinking"]) ui.denken(m["thinking"]) if m.get("content"): inhalt.append(m["content"]) ui.text(m["content"]) if m.get("tool_calls"): aufrufe.extend(m["tool_calls"]) if time.time() - letzter_puls > 30 and not konfig_fest(): karte.genutzt() letzter_puls = time.time() if d.get("done"): fertig = {k: d.get(k) for k in ("done_reason", "prompt_eval_count", "eval_count", "prompt_eval_duration", "eval_duration", "load_duration", "total_duration")} self.stat.prompt_token = d.get("prompt_eval_count") or 0 self.stat.antwort_token = d.get("eval_count") or 0 dauer = (d.get("eval_duration") or 0) / 1e9 self.stat.tok_s = self.stat.antwort_token / dauer if dauer else 0 self.stat.letzter_kontext = self.stat.prompt_token + self.stat.antwort_token except KeyboardInterrupt: abgebrochen = True except Exception as f: protokoll.fehler("strom", f, modell=self.modell_id, dauer_s=round(time.monotonic() - t0, 1), text_bisher="".join(inhalt)) raise ui.antwort_ende() text = "".join(inhalt) echte_aufrufe = len(aufrufe) if not aufrufe: # Ersatz: manche Vorlagen geben den Aufruf als Text aus, wenn # Ollamas Erkennung ihn nicht greift. for roh in _TEXT_AUFRUF.findall(text): try: j = json.loads(roh) aufrufe.append({"function": {"name": j.get("name"), "arguments": j.get("arguments") or {}}}) except ValueError: pass kaputte_args = 0 for t in aufrufe: f = t.setdefault("function", {}) if isinstance(f.get("arguments"), str): try: f["arguments"] = json.loads(f["arguments"]) except ValueError: kaputte_args += 1 protokoll.ereignis("argumente_kaputt", name=f.get("name"), roh=f["arguments"]) f["arguments"] = {} dauer = time.monotonic() - t0 prompt_s = (fertig.get("prompt_eval_duration") or 0) / 1e9 protokoll.ereignis( "antwort", modell=self.modell_id, dauer_s=round(dauer, 2), erstes_zeichen_s=round(erstes, 2) if erstes is not None else None, abgebrochen=abgebrochen, done_reason=fertig.get("done_reason"), prompt_token=fertig.get("prompt_eval_count"), antwort_token=fertig.get("eval_count"), tok_s=round(self.stat.tok_s, 1), prompt_tok_s=round((fertig.get("prompt_eval_count") or 0) / prompt_s, 1) if prompt_s else None, load_s=round((fertig.get("load_duration") or 0) / 1e9, 2), # Anteil des Kontexts: nahe 1 heisst, Ollama hat womoeglich vorne abgeschnitten kontext_anteil=round(((fertig.get("prompt_eval_count") or 0) + (fertig.get("eval_count") or 0)) / self.kontext, 3), token_geschaetzt=self.schaetzung(), text=text, gedanken="".join(gedanken), werkzeuge=[t["function"].get("name") for t in aufrufe], aufrufe_echt=echte_aufrufe, aufrufe_aus_text=len(aufrufe) - echte_aufrufe, argumente_kaputt=kaputte_args) return {"text": text, "aufrufe": aufrufe, "abgebrochen": abgebrochen, "gedacht": gedacht} def frage(self, eingabe: str, ui: Oberflaeche) -> None: z = karte.bereit() if not z: ui.hinweis("Keine Karte bereit - /start") return basis = z["basis"] protokoll.ereignis("eingabe", text=eingabe, cwd=str(self.umgebung.cwd), modell=self.modell_id, modus=self.modus, karte=z.get("karte"), nachrichten=len(self.verlauf)) self.verlauf.append({"role": "user", "content": eingabe}) if not konfig_fest(): karte.genutzt() t_auftrag = time.monotonic() abgelehnt: Counter = Counter() for schritt in range(MAX_SCHRITTE): self._platz_schaffen(ui, basis) r = self._strom(ui, basis) nachricht: dict[str, Any] = {"role": "assistant", "content": r["text"]} if r["aufrufe"]: nachricht["tool_calls"] = r["aufrufe"] self.verlauf.append(nachricht) if r["abgebrochen"]: self.verlauf.append({"role": "user", "content": "[Vom Nutzer abgebrochen.]"}) protokoll.ereignis("auftrag_ende", ausgang="abgebrochen", schritte=schritt + 1, dauer_s=round(time.monotonic() - t_auftrag, 1)) ui.hinweis("Abgebrochen.") return if not r["aufrufe"]: protokoll.ereignis("auftrag_ende", ausgang="fertig", schritte=schritt + 1, dauer_s=round(time.monotonic() - t_auftrag, 1)) return for t in r["aufrufe"]: name = t["function"].get("name") or "" args = t["function"].get("arguments") or {} schluessel = (name, json.dumps(args, sort_keys=True, ensure_ascii=False)) if abgelehnt[schluessel] >= 2: # Gemessen 16.10.2026: nach einer Ablehnung rief das Modell denselben # Befehl neunmal wieder auf. Ab der zweiten Wiederholung ist Schluss. protokoll.ereignis("auftrag_ende", ausgang="ablehnung_wiederholt", name=name, args=args, schritte=schritt + 1, dauer_s=round(time.monotonic() - t_auftrag, 1)) self.verlauf.append({"role": "tool", "tool_name": name, "content": "[ABGELEHNT] Wiederholt abgelehnt - Auftrag beendet."}) ui.hinweis("Das Modell wiederholt einen abgelehnten Aufruf - Auftrag angehalten.") return ergebnis = self._werkzeug(name, args, ui) if ergebnis.startswith("[ABGELEHNT]"): abgelehnt[schluessel] += 1 self.verlauf.append({"role": "tool", "tool_name": name, "content": ergebnis}) if ((ergebnis.startswith("[ABGELEHNT]") and "STOPP" in ergebnis) or ergebnis.startswith("[Vom Nutzer abgebrochen")): protokoll.ereignis("auftrag_ende", ausgang="gestoppt", schritte=schritt + 1, dauer_s=round(time.monotonic() - t_auftrag, 1)) return protokoll.ereignis("auftrag_ende", ausgang="schrittgrenze", schritte=MAX_SCHRITTE, dauer_s=round(time.monotonic() - t_auftrag, 1)) ui.hinweis(f"Nach {MAX_SCHRITTE} Schritten angehalten - mit einer neuen Nachricht weitermachen.") def _werkzeug(self, name: str, args: dict, ui: Oberflaeche) -> str: t0 = time.monotonic() self._entscheidung = "frei" cwd_vorher = str(self.umgebung.cwd) text = self._werkzeug_innen(name, args, ui) protokoll.ereignis( "werkzeug", name=name, args=args, entscheidung=self._entscheidung, modus=self.modus, dauer_s=round(time.monotonic() - t0, 2), cwd=cwd_vorher, fehler=text.startswith(("Fehler", "[ABGELEHNT]", "[Vom Nutzer")), exit_code=(int(text.rsplit("[Exit-Code ", 1)[1].split("]")[0]) if name == "bash" and "[Exit-Code " in text else None), zeitueberschreitung=name == "bash" and text.startswith("[Zeitüberschreitung"), laenge=len(text), ergebnis=text) return text def _werkzeug_innen(self, name: str, args: dict, ui: Oberflaeche) -> str: ui.werkzeug(name, args) if name not in {t["function"]["name"] for t in werkzeuge.SCHEMA}: text = f"Fehler: unbekanntes Werkzeug '{name}'." ui.ergebnis(name, args, text) return text if not freigabe.frei(self.modus, self.projekt, name, args): d = None if name in werkzeuge.DATEI_AENDERN: d, fehler = werkzeuge.vorschau_schreiben(self.umgebung, name, args) if fehler: ui.ergebnis(name, args, fehler) return fehler t_frage = time.monotonic() antwort, grund = ui.fragen(name, args, d) self._entscheidung = antwort protokoll.ereignis("freigabe", name=name, antwort=antwort, grund=grund, bedenkzeit_s=round(time.monotonic() - t_frage, 1)) if antwort == "nein": text = "[ABGELEHNT] Der Nutzer hat das nicht erlaubt." + (f" Begründung: {grund}" if grund else "") ui.ergebnis(name, args, text) return text if antwort == "stopp": text = "[ABGELEHNT] STOPP - der Nutzer hat angehalten. Warte auf neue Anweisungen." ui.ergebnis(name, args, text) return text if antwort == "immer" and name == "bash": freigabe.regel_hinzu(self.projekt, grund) elif antwort == "immer" and name in werkzeuge.DATEI_AENDERN: self.modus = "dateien" try: text = werkzeuge.ausfuehren(self.umgebung, name, args) except KeyboardInterrupt: text = "[Vom Nutzer abgebrochen während der Ausführung.]" ui.ergebnis(name, args, text) return text def konfig_fest() -> bool: return bool(konfig.FESTE_BASIS)