Der Positiv-Index

Jetzt der Concierge. Seine Aufgabe klingt nach einem Fall für ein Sprachmodell: „Welcher meiner Agenten kann das?“ Ist es aber nicht. Ein LLM-Aufruf an dieser Stelle kostet Latenz, kostet Token, ist nicht deterministisch und - das ist der eigentliche Einwand - nicht kalibriert. Er sagt zu allem ja.

Der Router dieses Kurses macht keinen einzigen LLM-Aufruf. Er bettet Text ein und rechnet Kosinus-Ähnlichkeit. Das ist alles.

Das Ziel

Geroutet wird Anfrage gegen Anfrage, nicht Anfrage gegen Beschreibung. Das ist der eine Trick, an dem die Trefferquote hängt, und er kostet kein Modelltraining, sondern nur die Entscheidung, das Richtige einzubetten.

Warum das so viel ausmacht: Nutzer sagen „ich brauch die rechnung von meier nochmal“. Die Beschreibung sagt „Agent für Dokumentenabruf aus dem DMS mit Volltextsuche“. Zwischen diesen beiden Sätzen liegt semantisch eine Menge Luft, obwohl sie inhaltlich exakt zusammengehören. Zwischen der Anfrage und dem Beispiel „hol mir nochmal die rechnung von“ liegt fast nichts. Das nennt sich asymmetrisches Retrieval, und die Gegenmaßnahme ist so simpel wie wirksam: Leg in den Index, was der Anfrage ähnlich sieht.

Schritt 1: Der Client

Eine Datei, zwei Funktionen, keine Anbieter-SDKs. Alles, was /v1/chat/completions und /v1/embeddings spricht, funktioniert:

# tiny/llm.py
from __future__ import annotations

import os

import httpx

BASE_URL = os.environ.get("TINY_BASE_URL", "http://localhost:11434/v1")
API_KEY = os.environ.get("TINY_API_KEY", "nicht-noetig")
CHAT_MODEL = os.environ.get("TINY_CHAT_MODEL", "qwen3:8b")
EMBED_MODEL = os.environ.get("TINY_EMBED_MODEL", "bge-m3")

_client = httpx.Client(
    base_url=BASE_URL,
    headers={"Authorization": f"Bearer {API_KEY}"},
    timeout=120,
)


def embed(texte: list[str]) -> list[list[float]]:
    """Einbettungen fuer eine Liste von Texten, Reihenfolge bleibt erhalten."""
    antwort = _client.post("/embeddings", json={"model": EMBED_MODEL, "input": texte})
    antwort.raise_for_status()
    daten = antwort.json()["data"]
    return [eintrag["embedding"] for eintrag in sorted(daten, key=lambda d: d["index"])]


def chat(nachrichten: list[dict], tools: list[dict] | None = None, **kwargs) -> dict:
    """Ein Chat-Completion-Aufruf. Gibt die erste Choice-Nachricht plus usage zurueck."""
    nutzlast = {"model": CHAT_MODEL, "messages": nachrichten, **kwargs}
    if tools:
        nutzlast["tools"] = tools
    antwort = _client.post("/chat/completions", json=nutzlast)
    antwort.raise_for_status()
    daten = antwort.json()
    return {"message": daten["choices"][0]["message"], "usage": daten.get("usage", {})}

Der sorted(...)-Aufruf ist kein Zierrat. Nicht jeder Server liefert die Einbettungen in der Reihenfolge der Eingabe zurück, und wenn sie einmal verrutscht sind, gehört jedes Beispiel dem falschen Agenten. Der Fehler ist still: Der Router funktioniert, nur schlecht.

Schritt 2: Der Index

Hier fällt die Entscheidung, die den Kurs von der Referenzarchitektur trennt: keine Vektordatenbank. Drei Cards mit je zehn Beispielen sind dreißig Vektoren. Dreißig. Eine NumPy-Matrix und ein Skalarprodukt erledigen das in Mikrosekunden.

# tiny/index.py
from __future__ import annotations

from dataclasses import dataclass

import numpy as np

from tiny.llm import embed


@dataclass
class Treffer:
    agent_id: str
    text: str
    score: float


class Index:
    """Ein Positiv- oder Veto-Index: Beispieltexte plus ihre Herkunft."""

    def __init__(self) -> None:
        self._agenten: list[str] = []
        self._texte: list[str] = []
        self._matrix: np.ndarray | None = None

    def hinzufuegen(self, agent_id: str, texte: list[str]) -> None:
        self._agenten.extend([agent_id] * len(texte))
        self._texte.extend(texte)
        self._matrix = None  # veraltet, wird beim naechsten bauen() neu gerechnet

    def bauen(self) -> None:
        vektoren = np.array(embed(self._texte), dtype=np.float32)
        # Normieren, damit das Skalarprodukt direkt die Kosinus-Aehnlichkeit ist.
        laengen = np.linalg.norm(vektoren, axis=1, keepdims=True)
        self._matrix = vektoren / np.maximum(laengen, 1e-12)

    def suchen(self, anfrage: str, limit: int = 10) -> list[Treffer]:
        if self._matrix is None:
            self.bauen()
        q = np.array(embed([anfrage])[0], dtype=np.float32)
        q /= max(float(np.linalg.norm(q)), 1e-12)
        scores = self._matrix @ q
        beste = np.argsort(-scores)[:limit]
        return [
            Treffer(self._agenten[i], self._texte[i], float(scores[i])) for i in beste
        ]

Das ist der ganze Vektorspeicher. Elf Zeilen Rechnung.

Schritt 3: Der Router

# tiny/concierge.py
from __future__ import annotations

from dataclasses import dataclass

from tiny.cards import Card, registry_laden
from tiny.index import Index


@dataclass
class Entscheidung:
    art: str                       # "match" | "abstain" | "ambiguous", vollstaendig ab Kapitel 04
    agent_id: str | None
    score: float
    alternativen: list[tuple[str, float]]


class Concierge:
    def __init__(self, cards: dict[str, Card]) -> None:
        self.cards = cards
        self.positiv = Index()
        for card in cards.values():
            self.positiv.hinzufuegen(card.id, card.examples)
        self.positiv.bauen()

    def route(self, ziel: str) -> Entscheidung:
        treffer = self.positiv.suchen(ziel, limit=20)

        # Bester Treffer je Agent: ein Agent mit vielen aehnlichen Beispielen
        # soll nicht gewinnen, weil er viele hat, sondern weil einer passt.
        beste: dict[str, float] = {}
        for t in treffer:
            beste[t.agent_id] = max(beste.get(t.agent_id, 0.0), t.score)

        rangliste = sorted(beste.items(), key=lambda p: -p[1])
        if not rangliste:
            return Entscheidung("abstain", None, 0.0, [])
        agent_id, score = rangliste[0]
        return Entscheidung("match", agent_id, score, rangliste[1:3])


def concierge_laden(ordner: str = "registry") -> Concierge:
    return Concierge(registry_laden(ordner))

Die Zeile mit max(...) ist die einzige Stelle mit einer echten Entscheidung darin. Ein Agent mit fünfzehn Beispielen hätte sonst allein deshalb mehr Treffer in den Top 20 als einer mit acht - und würde über die schiere Anzahl gewinnen. Was zählt, ist der beste Treffer, nicht die Menge.

Schritt 4: Messen statt glauben

Ein Router, den man an drei Anfragen ausprobiert, ist ein Router, von dem man nichts weiß. Schreib dir ein Testset: zwölf Anfragen, die keine der Card-Beispiele wörtlich wiederholen, mit der Antwort, die du erwartest.

# scripts/03_router_eval.py
from tiny.concierge import concierge_laden

TESTSET = [
    ("zieh mir mal die preisliste von deren website", "web-reader"),
    ("steht in dem artikel was ueber lieferzeiten", "web-reader"),
    ("was schreiben die auf ihrer startseite ueber sich", "web-reader"),
    ("gibts auf der seite eine telefonnummer", "web-reader"),
    ("leg mir bitte den neuen kunden an", "crm-writer"),
    ("notier bei der firma dass wir angerufen haben", "crm-writer"),
    ("trag die neue adresse beim kunden ein", "crm-writer"),
    ("setz den status von dem deal auf gewonnen", "crm-writer"),
    ("wie laeuft das nochmal mit der reisekostenabrechnung", "docs-finder"),
    ("wo finde ich die anleitung zum vpn", "docs-finder"),
    ("was steht bei uns intern zur elternzeit", "docs-finder"),
    ("gibt es ein dokument zum onboarding", "docs-finder"),
]


def main() -> None:
    concierge = concierge_laden()
    richtig = 0
    for anfrage, erwartet in TESTSET:
        e = concierge.route(anfrage)
        ok = e.agent_id == erwartet
        richtig += ok
        marke = "  " if ok else "XX"
        print(f"{marke} {e.score:.3f}  {e.agent_id or '-':<14} (erwartet {erwartet:<14}) {anfrage}")
    print(f"\nRecall@1: {richtig}/{len(TESTSET)} = {richtig / len(TESTSET):.2f}")


if __name__ == "__main__":
    main()

Der Meilenstein

python scripts/03_router_eval.py

   0.734  web-reader     (erwartet web-reader    ) zieh mir mal die preisliste von deren website
   0.681  web-reader     (erwartet web-reader    ) steht in dem artikel was ueber lieferzeiten
   ...
XX 0.612  web-reader     (erwartet docs-finder   ) gibt es ein dokument zum onboarding

Recall@1: 11/12 = 0.92

Über 0,90 ist ein guter Stand für diesen Schritt. Wenn du deutlich darunter liegst, ist die Ursache fast immer eine von dreien:

Das Embedding-Modell ist englischlastig. Prüf es in zehn Sekunden: Bette „leg den kunden an“ und „create the customer“ ein und rechne die Ähnlichkeit. Liegt sie unter 0,7, versteht dein Modell die Brücke zwischen den Sprachen schlecht - und dann versteht es auch deutsche Umgangssprache schlecht. Nimm bge-m3.

Die Beispiele klingen wie Dokumentation. Wenn in deiner Card „Ermittelt Preisinformationen von Webseiten“ steht statt „was kostet das bei denen laut website“, hast du genau den Fehler gemacht, gegen den dieses Kapitel gebaut ist.

Zwei Cards überlappen wirklich. Wenn docs-finder und web-reader beide „lies mir das mal vor“ enthalten, ist das kein Routerfehler, sondern ein Zuschnittfehler. Der Router kann nicht trennen, was du nicht getrennt hast.

Wann eine Vektordatenbank sich lohnt

Die Rechnung ist einfach: Deine Matrix hat N × D Fließkommazahlen. Bei 500 Fähigkeiten mit je 15 Beispielen und 1024 Dimensionen sind das 7.500 × 1.024 × 4 Byte, also rund 30 MB im Speicher, und eine Suche kostet ein Skalarprodukt über 7.500 Vektoren - unter einer Millisekunde. Bis dahin ist Qdrant eine Abhängigkeit ohne Gegenleistung.

Ab wann sie sich lohnt, hängt nicht an der Zahl, sondern an drei Eigenschaften, die dein NumPy-Index nicht hat: Persistenz (dein Index wird bei jedem Start neu eingebettet - bei 7.500 Beispielen sind das Sekunden und API-Kosten), Nebenläufigkeit (mehrere Prozesse, ein Index) und Filter vor der Suche (nur Agenten eines Mandanten). Sobald du eins davon brauchst, nimm eine Datenbank. Vorher ist der Einzeiler ehrlicher.