Der Artefakt-Speicher

Kapitel 06 endet mit einem ungelösten Problem, und es ist das ernsteste des ganzen Systems.

Der Tiny Agent hat 38.000 Token gelesen und 118 zurückgegeben. Er hat also entschieden, was wichtig ist. Nur kennt er die Gesamtaufgabe nicht - er sieht einen Brief, nicht das Gespräch. Vielleicht wollte der Aufrufer anschließend noch wissen, welche Länder beteiligt waren. Der Agent hat diesen Absatz weggeworfen, weil im Brief nur nach Beginn und Ende gefragt war.

Kompression ohne Rückweg ist Datenverlust mit besserer Presse.

Das Ziel

Kompression ist erst dann vertretbar, wenn sie rückholbar ist. Der Aufrufer bekommt die kompakte Antwort und behält Zugriff auf das Original. Das ist der Unterschied zwischen einer Zusammenfassung und einem Verlust.

Die Idee in einem Satz

Nichts wegwerfen, nur nicht mitschicken.

Werkzeug-Ausgabe (38k Token)
   ├─> Volltext ins Dateisystem        -> artifact_id
   ├─> in Stuecke schneiden (~512 Token, 64 Ueberlappung)
   ├─> Stuecke einbetten               -> in-memory Matrix, wie in Kapitel 03
   └─> Handle an den Aufrufer          -> "art_7f3a1c"

Und dann expand(handle, frage): eine Vektorsuche innerhalb dieses einen Artefakts, die die drei passendsten Stücke zurückgibt. Typisch 400 bis 1.500 Token statt 38.000.

Schritt 1: Der Speicher

# tiny/artifacts.py
from __future__ import annotations

import hashlib
import json
import time
from dataclasses import dataclass
from pathlib import Path

import numpy as np

from tiny.llm import embed

SPEICHER = Path(".artifacts")
STUECK = 2000        # Zeichen, entspricht grob 500 Token
UEBERLAPPUNG = 250   # damit ein Satz an der Naht nicht verlorengeht
TTL_STUNDEN = 24


@dataclass
class Artefakt:
    handle: str
    quelle: str
    tokens_geschaetzt: int
    preview: str


def _stuecke(text: str) -> list[str]:
    schritt = STUECK - UEBERLAPPUNG
    return [text[i : i + STUECK] for i in range(0, len(text), schritt)] or [""]


def ablegen(text: str, quelle: str) -> Artefakt:
    SPEICHER.mkdir(exist_ok=True)
    handle = "art_" + hashlib.sha256(f"{quelle}{text[:400]}".encode()).hexdigest()[:8]
    ordner = SPEICHER / handle
    ordner.mkdir(exist_ok=True)

    (ordner / "volltext.txt").write_text(text, encoding="utf-8")
    stuecke = _stuecke(text)
    (ordner / "stuecke.json").write_text(json.dumps(stuecke), encoding="utf-8")

    vektoren = np.array(embed(stuecke), dtype=np.float32)
    vektoren /= np.maximum(np.linalg.norm(vektoren, axis=1, keepdims=True), 1e-12)
    np.save(ordner / "vektoren.npy", vektoren)

    (ordner / "meta.json").write_text(
        json.dumps({"quelle": quelle, "erstellt": time.time()}), encoding="utf-8"
    )
    return Artefakt(handle, quelle, len(text) // 4, text[:200].replace("\n", " "))


def expand(handle: str, frage: str, k: int = 3) -> str:
    """Vektorsuche INNERHALB eines Artefakts. Der Rest bleibt liegen."""
    ordner = SPEICHER / handle
    if not ordner.exists():
        return f"Artefakt {handle} gibt es nicht (mehr)."

    stuecke = json.loads((ordner / "stuecke.json").read_text(encoding="utf-8"))
    vektoren = np.load(ordner / "vektoren.npy")

    q = np.array(embed([frage])[0], dtype=np.float32)
    q /= max(float(np.linalg.norm(q)), 1e-12)
    beste = np.argsort(-(vektoren @ q))[:k]
    return "\n\n---\n\n".join(stuecke[i] for i in sorted(beste))


def aufraeumen() -> int:
    """Artefakte sind Sitzungs-Cache, kein Archiv."""
    grenze = time.time() - TTL_STUNDEN * 3600
    entfernt = 0
    for ordner in SPEICHER.glob("art_*"):
        meta = json.loads((ordner / "meta.json").read_text(encoding="utf-8"))
        if meta["erstellt"] < grenze:
            for datei in ordner.iterdir():
                datei.unlink()
            ordner.rmdir()
            entfernt += 1
    return entfernt

Zwei Entscheidungen darin sind bewusst, nicht bequem.

Die Überlappung. Ohne sie fällt ein Satz, der genau auf der Schnittkante liegt, in beiden Stücken halb an - und wird in keinem gefunden. 250 Zeichen sind grob ein bis zwei Sätze und kosten 12 Prozent mehr Speicher. Ein guter Handel.

Die Verfallszeit. Artefakte sind Sitzungs-Cache. Wer sie behält, hat nach zwei Wochen ein unbeabsichtigtes Archiv gescrapter Fremdinhalte auf der Platte - mit allem, was das rechtlich und praktisch bedeutet. aufraeumen() gehört in einen Cronjob, spätestens am Tag nach dem Prototyp.

Schritt 2: Den Runner anschließen

Erinnerst du dich an die hässliche 12.000-Zeichen-Notbremse aus Kapitel 05? Sie wird jetzt ersetzt:

# tiny/runner.py, in der Werkzeugschleife
from tiny.artifacts import ablegen

ARTEFAKT_SCHWELLE = 16000   # Zeichen, grob 4000 Token

...
            else:
                funktion = REGISTRY[name][0]
                inhalt = str(funktion(**argumente))

            if len(inhalt) > ARTEFAKT_SCHWELLE:
                artefakt = ablegen(inhalt, quelle=f"{name}:{argumente}")
                gesammelte_artefakte.append(artefakt)
                inhalt = (
                    inhalt[:ARTEFAKT_SCHWELLE]
                    + f"\n\n[gekuerzt. Vollstaendig unter handle={artefakt.handle}]"
                )

Das schützt zuerst einmal den Tiny Agent selbst - sonst frisst ein einziger Scrape seinen eigenen Kontext auf, bevor er zum Verdichten kommt. Und es erzeugt das Handle, das gleich weiterwandert.

Im Digest kommt das Handle mit nach oben:

    return Ergebnis(
        ...,
        artifacts=[
            {
                "handle": a.handle,
                "quelle": a.quelle,
                "tokens": a.tokens_geschaetzt,
                "preview": a.preview,
            }
            for a in gesammelte_artefakte
        ],
    )

Die vier Felder kosten zusammen etwa 40 Token. Dafür weiß der Aufrufer, dass da noch 38.000 liegen, woher sie stammen und wie sie ungefähr anfangen. Das reicht, um zu entscheiden, ob sich Nachfragen lohnt.

Schritt 3: Das dritte Werkzeug des Aufrufers

Der aufrufende Agent hat jetzt genau drei Werkzeuge, egal wie viele Fähigkeiten registriert sind:

# tiny/host.py
from tiny.artifacts import expand
from tiny.concierge import concierge_laden
from tiny.runner import delegate

_concierge = concierge_laden()


def find_capability(goal: str) -> dict:
    """Kann das hier ueberhaupt jemand? Kein LLM-Aufruf, kalibrierte Antwort."""
    e = _concierge.route(goal)
    if e.art == "abstain":
        return {"can_do": False, "confidence": round(e.score, 3)}
    card = _concierge.cards[e.agent_id]
    return {
        "can_do": True,
        "agent": card.id,
        "summary": card.summary,
        "limitations": card.does_not,
        "confidence": round(e.score, 3),
        "ambiguous": e.art == "ambiguous",
        "alternativen": [a for a, _ in e.alternativen],
    }


def host_delegate(auftrag: str) -> dict:
    return dict(delegate(_concierge, auftrag))


def host_expand(handle: str, frage: str) -> str:
    return expand(handle, frage)

Drei Funktionen, drei Werkzeug-Definitionen im Hauptkontext. Nach der Rechnung aus Kapitel 01 sind das rund 600 Token - konstant, ob drei Fähigkeiten registriert sind oder fünfhundert. Das ist die Zielgröße der ganzen Architektur, und jetzt steht sie da.

find_capability ist dabei die deklarative Antwort auf die Frage, die man zuerst dialogisch lösen möchte: den Agenten einfach fragen, ob er das kann. Das wäre ein LLM-Aufruf, eine Runde Latenz und eine unkalibrierte Antwort. Hier ist es eine Vektorsuche mit einer Zahl dahinter.

Der Meilenstein

Der Test besteht aus zwei Aufrufen, und der zweite ist der Punkt.

# scripts/07_expand_test.py
from tiny.host import host_delegate, host_expand

ergebnis = host_delegate(
    "lies https://de.wikipedia.org/wiki/Zweiter_Weltkrieg und sag mir, "
    "wann der krieg begann und wann er endete"
)
print("Digest :", ergebnis["summary"][:120])
print("Felder :", ergebnis["fields"])
for a in ergebnis["artifacts"]:
    print(f"Artefakt: {a['handle']}  ~{a['tokens']} Token  {a['quelle'][:40]}")

# Diese Information stand NICHT im Digest - danach war ja nicht gefragt.
handle = ergebnis["artifacts"][0]["handle"]
nachschlag = host_expand(handle, "welche laender waren an der wannseekonferenz beteiligt")
print(f"\nNachschlag: ~{len(nachschlag) // 4} Token")
print(nachschlag[:400])

Digest : Der Zweite Weltkrieg begann am 1. September 1939 mit dem Ueberfall auf Polen ...
Felder : {'beginn': '1. September 1939', 'ende': '2. September 1945'}
Artefakt: art_7f3a1c  ~38412 Token  http_get:{'url': 'https://de.wikipe

Nachschlag: ~610 Token
... die Wannseekonferenz am 20. Januar 1942 ...

Jetzt die Buchhaltung, denn nur sie beweist etwas:

Der Stand ist erreicht, wenn du eine Information nachholen kannst, die im Digest nicht vorkam - und die 38.000 Token dabei nie in deinem Hauptkontext auftauchen. Sie liegen auf der Platte. Erreichbar, aber nicht anwesend.

Das ist der Unterschied, um den es die ganze Zeit ging.