Der Runner

Der Concierge weiß jetzt, wer eine Anfrage bearbeiten soll. Jetzt bauen wir das, was danach kommt.

Die naheliegende Bauweise wäre: pro Tiny Agent ein Dienst, ein Container, ein Healthcheck, ein Port. Bei zwanzig Fähigkeiten sind das zwanzig Container, und der Betrieb frisst mehr Zeit als die Sache selbst.

Die Bauweise dieses Kurses ist eine andere, und sie beruht auf einer einzigen Beobachtung.

Das Ziel

Die Isolation kommt vom frischen Kontext, nicht vom eigenen Prozess. Ein Tiny Agent ist keine Anwendung, sondern eine Konfiguration: Systemprompt, Werkzeugauswahl, Ausgabeschema, Budget. Was ihn schützt, ist, dass sein Kontext bei jedem Aufruf leer anfängt - und dafür braucht es keinen zweiten Prozess.

Prozess-Isolation brauchst du erst, wenn fremder Code läuft oder zwei Agenten sich über Abhängigkeiten streiten. Beides ist hier nicht der Fall.

Schritt 1: Die Werkzeug-Registry

Der Runner hält alle Werkzeuge, die es im System gibt. Jede Card sagt, welche davon ihr Agent sehen darf.

# tiny/tools.py
from __future__ import annotations

import re
from typing import Callable

import httpx

TAGS_WEG = re.compile(r"<(script|style)[^>]*>.*?</\1>", re.S | re.I)
MARKUP_WEG = re.compile(r"<[^>]+>")


def http_get(url: str) -> str:
    """Laedt eine URL und gibt den rohen Text zurueck."""
    return httpx.get(url, timeout=30, follow_redirects=True).text


def html_to_text(html: str) -> str:
    """Wirft Markup weg und laesst den lesbaren Text uebrig."""
    return re.sub(r"\n{3,}", "\n\n", MARKUP_WEG.sub(" ", TAGS_WEG.sub(" ", html))).strip()


def docs_search(query: str) -> str:
    """Platzhalter fuer deine interne Suche - hier eine Datei im Ordner docs/."""
    from pathlib import Path

    treffer = [
        f"### {p.name}\n{p.read_text(encoding='utf-8')[:2000]}"
        for p in Path("docs").glob("*.md")
        if query.lower().split()[0] in p.read_text(encoding="utf-8").lower()
    ]
    return "\n\n".join(treffer) or "Keine Treffer."


# name -> (Funktion, JSON-Schema fuer den tools-Parameter)
REGISTRY: dict[str, tuple[Callable, dict]] = {
    "http_get": (
        http_get,
        {
            "type": "function",
            "function": {
                "name": "http_get",
                "description": "Laedt eine URL und gibt den Seiteninhalt zurueck.",
                "parameters": {
                    "type": "object",
                    "properties": {"url": {"type": "string"}},
                    "required": ["url"],
                },
            },
        },
    ),
    "html_to_text": (
        html_to_text,
        {
            "type": "function",
            "function": {
                "name": "html_to_text",
                "description": "Entfernt Markup aus HTML und gibt den Text zurueck.",
                "parameters": {
                    "type": "object",
                    "properties": {"html": {"type": "string"}},
                    "required": ["html"],
                },
            },
        },
    ),
    "docs_search": (
        docs_search,
        {
            "type": "function",
            "function": {
                "name": "docs_search",
                "description": "Durchsucht die interne Dokumentation nach einem Begriff.",
                "parameters": {
                    "type": "object",
                    "properties": {"query": {"type": "string"}},
                    "required": ["query"],
                },
            },
        },
    ),
}


def schemata_fuer(namen: list[str]) -> list[dict]:
    """Nur diese Werkzeuge gehen ins Schema. Nie mehr, nie andere."""
    fehlend = [n for n in namen if n not in REGISTRY]
    if fehlend:
        raise KeyError(f"Card verlangt unbekannte Werkzeuge: {fehlend}")
    return [REGISTRY[n][1] for n in namen]

Die letzte Funktion ist die eigentliche Pointe des Kapitels. Sie ist drei Zeilen lang und macht den Unterschied zwischen einem Kontext mit 200 Werkzeugen und einem mit zwei.

Dass sie bei unbekannten Namen wirft statt sie zu überspringen, ist Absicht: Eine Card, die ein Werkzeug verlangt, das es nicht gibt, ist kaputt. Still weniger Werkzeuge zu injizieren, würde einen Agenten erzeugen, der ohne erkennbaren Grund schlecht arbeitet.

Schritt 2: Der Systemprompt aus der Card

Der Agent bekommt keinen handgeschriebenen Prompt. Er bekommt seine eigene Card, in Sätze übersetzt:

# tiny/runner.py
from __future__ import annotations

import json

from tiny.cards import Card
from tiny.llm import chat
from tiny.tools import REGISTRY, schemata_fuer


def systemprompt(card: Card) -> str:
    verboten = "\n".join(f"- {z}" for z in card.does_not)
    felder = ", ".join(card.returns_required)
    return (
        f"Du bist '{card.id}', ein spezialisierter Agent.\n\n"
        f"AUFGABE\n{card.summary}\n\n"
        f"DAS TUST DU NICHT\n{verboten}\n\n"
        f"AUSGABE\n"
        f"Antworte am Ende mit einem einzigen JSON-Objekt, das genau diese "
        f"Felder enthaelt: {felder}. Kein Fliesstext davor oder danach.\n"
        f"Wenn du die Aufgabe fuer nicht zustaendig haeltst, gib "
        f'{{"status": "refused"}} zurueck.\n\n'
        f"Fasse dich kurz. Deine Aufgabe ist nicht, alles zu berichten, "
        f"sondern das Gefragte zu verdichten."
    )

Der letzte Satz ist die Selbstbeschreibung eines Tiny Agent, und in Kapitel 06 lernst du, warum er allein nichts wert ist.

Schritt 3: Die Invocation-Schleife

# tiny/runner.py (Fortsetzung)
class Ergebnis(dict):
    """Das, was der Agent zurueckgibt - plus was der Lauf gekostet hat."""


def invoke(card: Card, auftrag: str, zaehler: dict | None = None) -> Ergebnis:
    zaehler = zaehler if zaehler is not None else {}
    tools = schemata_fuer(card.tools)
    erlaubt = set(card.tools)

    # FRISCH. Kein Dialogverlauf, keine Historie, nichts vom letzten Aufruf.
    ctx = [
        {"role": "system", "content": systemprompt(card)},
        {"role": "user", "content": auftrag},
    ]

    for _ in range(card.budget.max_tool_calls):
        antwort = chat(ctx, tools=tools, max_tokens=card.budget.max_output_tokens)
        nachricht = antwort["message"]
        ctx.append(nachricht)

        aufrufe = nachricht.get("tool_calls") or []
        if not aufrufe:
            return Ergebnis(
                status="ok",
                text=nachricht.get("content", ""),
                usage=antwort["usage"],
            )

        for aufruf in aufrufe:
            name = aufruf["function"]["name"]
            argumente = json.loads(aufruf["function"]["arguments"] or "{}")

            if name not in erlaubt:
                # GEISTERAUFRUF: ein Werkzeug, das dieser Agent nicht hat.
                zaehler["ghost_calls"] = zaehler.get("ghost_calls", 0) + 1
                zaehler.setdefault("ghost_namen", []).append(name)
                inhalt = (
                    f"FEHLER: Das Werkzeug '{name}' hast du nicht. "
                    f"Dir stehen zur Verfuegung: {sorted(erlaubt)}."
                )
            else:
                funktion = REGISTRY[name][0]
                try:
                    inhalt = str(funktion(**argumente))
                except Exception as fehler:                     # noqa: BLE001
                    inhalt = f"FEHLER beim Aufruf von {name}: {fehler}"

            ctx.append(
                {
                    "role": "tool",
                    "tool_call_id": aufruf["id"],
                    "content": inhalt[:12000],   # Notbremse, die richtige Loesung folgt
                }
            )

    return Ergebnis(status="partial", text="", grund="max_tool_calls erreicht")

Drei Stellen darin sind wichtiger, als sie aussehen.

Der frische Kontext. ctx entsteht in der Funktion und stirbt mit ihr. Kein Zustand, kein Cache, keine Historie. Genau das ist die Isolation, für die andere Architekturen Container bauen.

Die Geisteraufruf-Zählung. Ein Modell, das ein Werkzeug aufruft, das es nicht hat, ist der häufigste Fehler beim Tool-Calling überhaupt. Die meisten Systeme schlucken ihn: Sie geben eine Fehlermeldung zurück und gehen weiter. Wir tun das auch - aber wir zählen dabei. Die Zahl ist eine der wenigen direkt vergleichbaren Metriken, die es in diesem Feld gibt, und sie fällt hier im Betrieb kostenlos an.

Die 12.000-Zeichen-Notbremse. Sie ist mit Absicht hässlich, weil sie ein Platzhalter ist. Sie verhindert, dass ein einziger Scrape den Kontext des Tiny Agent selbst auffrisst - der Schutz gilt nämlich nicht nur für den Hauptkontext. In Kapitel 07 wird daraus etwas Vernünftiges: ein Handle statt eines Abschnitts.

Schritt 4: Concierge und Runner verbinden

# tiny/runner.py (Schluss)
def delegate(concierge, auftrag: str, zaehler: dict | None = None) -> Ergebnis:
    entscheidung = concierge.route(auftrag)
    if entscheidung.art == "abstain":
        return Ergebnis(status="abstain", text="Dafuer habe ich niemanden.")
    card = concierge.cards[entscheidung.agent_id]
    if card.side_effects != "none":
        # Schreibende Agenten laufen nicht ungefragt. Im Kurs reicht die
        # Rueckmeldung an den Aufrufer, im Betrieb fragt der Host den Nutzer.
        return Ergebnis(status="needs_confirmation", agent=card.id)
    ergebnis = invoke(card, auftrag, zaehler)
    ergebnis["agent"] = card.id
    ergebnis["routing_score"] = round(entscheidung.score, 4)
    return ergebnis

Die side_effects-Abfrage ist vier Zeilen und verhindert die unangenehmste Klasse von Fehlern in solchen Systemen: dass ein Router, der sich irrt, dabei etwas schreibt. Ein falsch geroutetes Lesen kostet Token. Ein falsch geroutetes Schreiben kostet einen Datensatz.

Der Meilenstein

# scripts/05_runner_test.py
from tiny.concierge import concierge_laden
from tiny.runner import delegate

concierge = concierge_laden()
zaehler: dict = {}

for auftrag in [
    "lies https://de.wikipedia.org/wiki/Werkzeug und sag mir in zwei saetzen worum es geht",
    "wie war nochmal der ablauf beim onboarding",
    "trag mir bitte die firma mustermann gmbh als neuen kunden ein",
    "mach mir ein butterbrot",
]:
    e = delegate(concierge, auftrag, zaehler)
    print(f"[{e['status']:<18}] {e.get('agent', '-'):<14} {auftrag[:52]}")

print(f"\nGeisteraufrufe: {zaehler.get('ghost_calls', 0)} {zaehler.get('ghost_namen', [])}")

[ok                ] web-reader     lies https://de.wikipedia.org/wiki/Werkzeug und sag ...
[ok                ] docs-finder    wie war nochmal der ablauf beim onboarding
[needs_confirmation] crm-writer     trag mir bitte die firma mustermann gmbh als neuen k...
[abstain           ] -              mach mir ein butterbrot

Geisteraufrufe: 1 ['web_search']

Der Stand ist erreicht, wenn drei Dinge gleichzeitig gelten: Ein Runner-Prozess bedient alle drei Cards, jeder Aufruf sieht nur seine eigenen ein bis zwei Werkzeuge, und der Geisteraufruf-Zähler steht nicht auf einem Zufallswert, sondern auf dem, was wirklich passiert ist.

Wenn dein Zähler hoch ist, ist das kein Fehlschlag. Kleine Modelle erfinden gerne web_search, weil sie es aus dem Training kennen. Notier die Zahl - in Kapitel 08 vergleichst du sie mit der Variante, in der alle Werkzeuge in einem Kontext liegen. Das ist einer der Punkte, an denen sich die ganze These entscheidet.