Logging und Beobachtbarkeit

Ist die Anwendung in Betrieb, nutzen Nutzer sie dort, wo du nicht hinsiehst. Eines Tages meldet jemand: „Euer Assistent hat etwas völlig Falsches geantwortet“, oder die Monatsrechnung ist dreimal so hoch wie erwartet. Wie gehst du vor?

Hast du nur die Frage des Nutzers und die Endantwort, weißt du kaum, wo du anfangen sollst: Hat die Suche nicht die richtigen Dokumente gefunden? Hat das Modell falsch gelesen? Hat ein Tool einen Fehler gemeldet? Oder ist es bei einem Schritt vom Weg abgekommen und danach immer weiter in die Irre gelaufen?

Beobachtbarkeit (observability) heißt: Das System hinterlässt beim Laufen genug Aufzeichnungen, damit man hinterher rekonstruieren kann, was es tatsächlich getan hat. Bei KI-Anwendungen sind die wichtigsten Aufzeichnungen jeder Modellaufruf und jeder Tool-Aufruf.

Was protokolliert wird

call_llm aus Modul 03, Lektion 4 führt bereits Buch: Jeder Aufruf schreibt eine JSON-Zeile mit Zeit, Modell, Dauer, Tokens und Kosten. Das ist ein guter Anfang, reicht für Agenten aber nicht. Ein Agent ruft für eine Frage mehrmals das Modell und mehrmals Tools auf, und diese Aufzeichnungen müssen sich verknüpfen lassen.

Jeder Eintrag braucht deshalb zusätzlich:

• trace_id: Alle Einträge zu derselben Frage teilen eine Nummer. Damit findet man den gesamten Ablauf einer Frage.
• span_id und parent_id: die eigene Nummer jedes Eintrags und die seines übergeordneten Eintrags. Etwa ist „Frage beantworten“ übergeordnet, darunter liegen drei Modellaufrufe und drei Tool-Aufrufe.
• kind und name: der Typ des Eintrags (Aufgabe, Modellaufruf, Tool-Aufruf) und der konkrete Name (welches Modell, welches Tool).
• Zusammenfassung von Ein- und Ausgabe: die Argumente des Tools, die ersten paar Dutzend Zeichen des Rückgabeinhalts, welche Tools das Modell aufrufen wollte.
• Dauer, Tokens, Kosten, Status: erfolgreich, Fehler oder ein Tool hat eine Fehlermeldung zurückgegeben.

Diese Begriffe (Trace, Span) stammen aus dem Tracing verteilter Systeme. Fertige Open-Source-Werkzeuge wie Langfuse (selbst hostbar) und der Standard OpenTelemetry nutzen dieselben Konzepte. Wir schreiben zuerst selbst den einfachsten, um zu verstehen, was er tut.

Ein Tracer in ein paar Dutzend Zeilen

import json
import threading
import time
import uuid
from contextlib import contextmanager
from pathlib import Path

_lock = threading.Lock()
_local = threading.local()  # 记录当前线程正在进行的 span，用来自动找到上级


class Tracer:
    def __init__(self, path):
        self.path = Path(path)

    def _write(self, record):
        with _lock, self.path.open("a") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")

    @contextmanager
    def span(self, kind, name, **attrs):
        """用法：with tracer.span("llm", "chat") as s: ...; s["tokens"] = 100"""
        parent = getattr(_local, "current", None)
        record = {
            "trace_id": parent["trace_id"] if parent else uuid.uuid4().hex[:12],
            "span_id": uuid.uuid4().hex[:8],
            "parent_id": parent["span_id"] if parent else None,
            "kind": kind,
            "name": name,
            "start": time.strftime("%Y-%m-%d %H:%M:%S"),
            **attrs,
        }
        _local.current = record
        start = time.time()
        try:
            yield record
            record.setdefault("status", "ok")
        except Exception as e:
            record["status"] = "error"
            record["error"] = f"{type(e).__name__}: {e}"
            raise
        finally:
            record["ms"] = round((time.time() - start) * 1000)
            _local.current = parent
            self._write(record)

Verwendet wird er mit einer with-Anweisung:

with tracer.span("tool", "grep_docs", args={"keyword": "timeout"}) as s:
    result = grep_docs("timeout")
    s["result_chars"] = len(result)

Endet der with-Block, berechnet der Tracer automatisch die Dauer und schreibt eine JSON-Zeile. Tritt im Block eine Ausnahme auf, wird sie ebenfalls protokolliert, mit Status error, und danach wie gewohnt weitergeworfen.

parent_id wird automatisch ermittelt: Der Tracer merkt sich mit threading.local(), „welcher Span gerade läuft“. Ein Modellaufruf, der innerhalb des Spans „Frage beantworten“ beginnt, hat natürlich „Frage beantworten“ als übergeordneten Eintrag. Jeder Thread hat seine eigenen Aufzeichnungen, sodass auch mehrere parallel bearbeitete Fragen nicht durcheinandergeraten.

An den Agenten anschließen

Der Agentencode aus Modul 05, Lektion 2 muss nicht geändert werden; man legt nur eine Schicht darum. Modellaufrufe:

class TracedModel(agent_loop.RealModel):
    """每次调用模型，记一个 llm 类型的 span。"""

    def __call__(self, messages, tools):
        with tracer.span("llm", self.model, input_messages=len(messages)) as s:
            message, usage = super().__call__(messages, tools)
            s.update(prompt_tokens=usage.prompt_tokens, completion_tokens=usage.completion_tokens,
                     cost=round(cost(usage), 6), tool_calls=[c.function.name for c in message.tool_calls or []])
            return message, usage

Tool-Aufrufe:

def traced(name, fn):
    """包装一个工具函数，每次调用记一个 tool 类型的 span。"""
    def wrapper(**kwargs):
        with tracer.span("tool", name, args=kwargs) as s:
            result = fn(**kwargs)
            s["result_chars"] = len(result)
            s["result_preview"] = result[:80]
            if result.startswith("错误"):
                s["status"] = "tool_error"
            return result
    return wrapper


for name, t in agent_loop.TOOLS.items():
    t["fn"] = traced(name, t["fn"])

Gibt ein Tool „Fehler: …“ zurück, hat die Funktion selbst keine Ausnahme geworfen (Modul 05: Fehler werden zu Beobachtungen für das Modell), deshalb muss es eigens als tool_error markiert werden, sonst sieht im Log alles normal aus.

Ganz außen öffnet jede Frage einen Span vom Typ task:

    for q in QUESTIONS:
        with tracer.span("task", "answer_question", question=q) as s:
            answer, stats = agent_loop.run_agent(model, q, verbose=False)
            s["answer_preview"] = (answer or "")[:60]

Drei Fragen ausführen, dann nur das Log ansehen

Drei Fragen; in der letzten ist der Dateiname absichtlich falsch geschrieben (richtig ist timeouts.md). Danach analysiert das Programm nur die Logdatei, so wie beim nachträglichen Untersuchen eines Problems (vollständiger Code in code/06-production/traced_agent.py):

日志共 25 条记录，最后一条：
{"trace_id": "173ee49bde3e", "span_id": "f9e2c3a5", "parent_id": null, "kind": "task", "name": "answer_question", "start": "2026-09-14 23:07:41", "question": "advanced/timeout.md 里写了什么？", "answer_preview": "`advanced/timeouts.md` 的内容如下（文件共 71 行；注意文件名是复数 timeouts，没有 `", "status": "ok", "ms": 5729}

「httpx 的超时分成哪几种？」 4006 毫秒，3 次模型调用，3 次工具调用，0.00104 美元
「httpx 怎么上传文件？」 5352 毫秒，3 次模型调用，6 次工具调用，0.00210 美元
「advanced/timeout.md 里写了什么？」 5729 毫秒，4 次模型调用，3 次工具调用，0.00135 美元

最慢的一步：llm deepseek-flash，3120 毫秒
工具报错次数：无

Die drei Fragen ergeben zusammen 25 Einträge. Nach trace_id gruppiert, sind Dauer, Aufrufzahl und Kosten jeder Frage auf einen Blick zu sehen. „Wie lädt man Dateien hoch?“ rief 6-mal Tools auf, doppelt so oft wie die anderen Fragen, und kostete auch doppelt so viel; in seinen Einträgen sieht man, dass es nacheinander nach den Stichwörtern files=, multipart und upload suchte und drei Dateien las. Der langsamste Schritt war ein Modellaufruf mit über 3 Sekunden; Tool-Aufrufe brauchten nur wenige Millisekunden.

Ich hatte erwartet, dass die dritte Frage einen Tool-Fehler „Datei existiert nicht“ hinterlässt, aber nein: Der Agent rief zuerst list_docs auf, sah nach, welche Dateien es gibt, stellte fest, dass der richtige Name timeouts.md ist, las sie direkt und wies den Nutzer in der Antwort sogar darauf hin, „dass der Dateiname im Plural steht“. Auch das ist der Wert eines Logs: Es hält fest, was wirklich passiert ist, nicht was du erwartet hast.

Probleme mit dem Log untersuchen

Mit einem solchen Log untersucht man häufige Probleme so:

JSONL-Dateien sind einfach und zuverlässig, lassen sich mit wenigen Zeilen Python auswerten und passen zu persönlichen Projekten und frühen Produkten. Bei großen Mengen schickt man die Einträge an ein spezialisiertes Werkzeug (etwa Langfuse), das eine Oberfläche zum Durchsehen jedes Traces, zur Suche nach Bedingungen und für Statistikgrafiken bietet. Die Konzepte sind dieselben.

Datenschutz beachten

Das Log hält die Fragen der Nutzer, die Argumente der Tools und den Anfang der Antworten fest. Schreibt ein Nutzer seine Handynummer, ein Passwort oder interne Firmeninformationen in die Frage, landet all das im Log.

• Nur Nötiges protokollieren. Der Code oben speichert nur die ersten 60 Zeichen der Antwort und die ersten 80 Zeichen des Tool-Ergebnisses, nicht den vollständigen Inhalt.
• Sensible Informationen filtern. Vor dem Schreiben ins Log alles ersetzen, was nach Schlüssel, Handynummer oder Ausweisnummer aussieht. Die übernächste Lektion „Leitplanken“ schreibt eine einfache Erkennungsfunktion.
• Aufbewahrungsfrist festlegen. Logs nicht ewig behalten, etwa nur 30 Tage.
• Zugriff kontrollieren. Wer die Logs lesen kann, sieht, was Nutzer gefragt haben.

Übungen

1. Führ traced_agent.py aus und schreib selbst ein paar Zeilen Code, die traces.jsonl lesen und bei jeder Frage den Modellaufruf mit den meisten Eingabe-Tokens finden.
2. Füg TracedModel ein Feld hinzu, das bei jedem Aufruf die Gesamtzahl der Eingabezeichen festhält, und sieh, wie der Kontext des Agenten Schritt für Schritt wächst.
3. Erzeuge absichtlich einen Tool-Fehler (etwa read_doc vorübergehend so ändern, dass es für eine bestimmte Datei immer „Fehler: …“ liefert), und prüfe, dass die Log-Statistik ihn findet.

Selbsttest

1. Wozu dienen trace_id, span_id und parent_id?

trace_id verknüpft alle Einträge derselben Aufgabe (etwa der Beantwortung einer Frage); span_id ist die eigene Nummer jedes Eintrags; parent_id verweist auf den übergeordneten Eintrag. Mit diesen drei Feldern lässt sich aus dem Log die vollständige Struktur einer Aufgabe rekonstruieren: was zuerst getan wurde und was darin wiederum aufgerufen wurde.

2. Gibt ein Tool „Fehler: diese Datei gibt es nicht“ zurück, wirft die Funktion keine Ausnahme. Wie findet man das im Log?

Der Code, der das Tool umhüllt, muss den Rückgabewert prüfen und bei einer Fehlermeldung den Status eigens auf tool_error oder Ähnliches setzen. Sonst ist der Status dieses Aufrufs im Log normal, und bei der Fehlerstatistik fehlt er.

3. Warum sollte das Log nicht die vollständigen Fragen und Antworten der Nutzer enthalten?

Nutzereingaben können persönliche Daten, Passwörter oder Firmengeheimnisse enthalten; gelangt das Log nach außen oder in falsche Hände, entsteht ein Datenschutzproblem. Nur protokollieren, was zur Fehlersuche nötig ist, sensible Informationen vor dem Schreiben filtern und Aufbewahrungsfrist sowie Zugriffsrechte festlegen.