5. März 2026

Debugging und Observability in Autonomen Agentensystemen

Ein autonomer Agent, der still scheitert, ist schlimmer als gar kein Agent. Wenn eine traditionelle Funktion eine Ausnahme wirft, erhält man einen Stack Trace. Wenn ein Agent über zwanzig Tool-Aufrufe und drei Modell-Invokationen hinweg einen falschen Weg einschlägt, erhält man eine falsche Antwort — ohne offensichtliche Erklärung.

Das Debuggen von Agenten erfordert ein anderes mentales Modell. Das System führt keinen deterministischen Pfad aus; es trifft eine Reihe von Entscheidungen. Observability bedeutet, diese Entscheidungen zu erfassen — nicht nur Ein- und Ausgaben, sondern das Reasoning, das sie verbindet.

Warum Traditionelles Debugging bei Agenten Versagt

Standard-Logging erfasst, was passiert ist. Agenten-Observability erfordert die Erfassung des Warum — was das Modell gefolgert hat, welches Tool es warum gewählt hat und von welchem Zwischenzustand es ausgegangen ist.

Die Fehlermodalitäten sind ebenfalls unterschiedlich:

Stille Halluzination: Der Agent produziert selbstsicher eine falsche Antwort, ohne Unsicherheit zu signalisieren.
Entscheidungsdrift: Jeder Schritt wirkt lokal vernünftig, aber die Sequenz entfernt sich vom Ziel.
Tool-Missbrauch: Der Agent ruft das richtige Tool mit subtil falschen Parametern auf.
Endlosschleifen: Der Agent steckt im Wiederholen eines fehlgeschlagenen Ansatzes fest.
Kontext-Vergiftung: Schlechte Ausgaben eines frühen Schritts korrumpieren das gesamte nachfolgende Reasoning.

Keines davon erzeugt eine Ausnahme. Sie erzeugen falsches Verhalten, das nur sichtbar wird, wenn man den vollständigen Ausführungstrace rekonstruiert.

Strukturiertes Logging für Agenten-Entscheidungen

Der erste Schritt ist es, jede Agenten-Interaktion in strukturierten Logs zu verpacken. Logge keine rohen API-Antworten — logge semantische Events.

import json
import time
import uuid
from dataclasses import dataclass, asdict
from typing import Any
import anthropic

client = anthropic.Anthropic()

@dataclass
class AgentEvent:
    trace_id: str
    step: int
    event_type: str      # "llm_call", "tool_call", "tool_result", "decision", "error"
    model: str | None
    input_tokens: int | None
    output_tokens: int | None
    latency_ms: float | None
    content: dict[str, Any]
    timestamp: float

def log_event(event: AgentEvent):
    print(json.dumps(asdict(event)))  # Ersetze durch dein Log-Ziel

class TracedAgent:
    def __init__(self, trace_id: str | None = None):
        self.trace_id = trace_id or str(uuid.uuid4())
        self.step = 0
        self.tools = []

    def add_tool(self, name: str, description: str, input_schema: dict):
        self.tools.append({
            "name": name,
            "description": description,
            "input_schema": input_schema
        })

    def call(self, messages: list[dict], system: str = "") -> str:
        self.step += 1
        start = time.monotonic()

        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=4096,
            system=system,
            tools=self.tools,
            messages=messages
        )

        latency_ms = (time.monotonic() - start) * 1000

        log_event(AgentEvent(
            trace_id=self.trace_id,
            step=self.step,
            event_type="llm_call",
            model="claude-opus-4-6",
            input_tokens=response.usage.input_tokens,
            output_tokens=response.usage.output_tokens,
            latency_ms=latency_ms,
            content={
                "stop_reason": response.stop_reason,
                "text_blocks": [b.text for b in response.content if b.type == "text"],
                "tool_calls": [
                    {"name": b.name, "input": b.input}
                    for b in response.content if b.type == "tool_use"
                ]
            },
            timestamp=time.time()
        ))

        return response

Einen Vollständigen Trace Aufbauen

Eine einzelne Log-Zeile reicht nicht aus — du brauchst den vollständigen Ausführungstrace, der jede Entscheidung mit ihrem Ergebnis verbindet:

from typing import Callable

def run_traced_agent(
    task: str,
    tools: dict[str, Callable],
    tool_schemas: list[dict],
    system: str,
    max_steps: int = 20,
) -> dict:
    agent = TracedAgent()
    for schema in tool_schemas:
        agent.add_tool(**schema)

    messages = [{"role": "user", "content": task}]
    trace = {"trace_id": agent.trace_id, "task": task, "steps": []}
    step_count = 0

    while step_count < max_steps:
        step_count += 1
        response = agent.call(messages, system=system)

        step_record = {
            "step": step_count,
            "stop_reason": response.stop_reason,
            "model_output": [],
            "tool_results": []
        }

        if response.stop_reason == "end_turn":
            for block in response.content:
                if block.type == "text":
                    step_record["model_output"].append(block.text)
            trace["steps"].append(step_record)
            trace["final_answer"] = step_record["model_output"][-1] if step_record["model_output"] else ""
            break

        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                step_record["model_output"].append({
                    "tool": block.name,
                    "input": block.input
                })

                tool_fn = tools.get(block.name)
                if not tool_fn:
                    result = f"Fehler: unbekanntes Tool '{block.name}'"
                else:
                    try:
                        result = tool_fn(**block.input)
                    except Exception as e:
                        result = f"Tool-Fehler: {e}"

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(result)
                })
                step_record["tool_results"].append({
                    "tool": block.name,
                    "result_preview": str(result)[:200]
                })

        trace["steps"].append(step_record)
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})

    else:
        trace["error"] = f"Maximale Schrittzahl überschritten ({max_steps})"

    return trace

Schleifen-Erkennung

Endlosschleifen sind ein häufiger Fehlermodus. Erkenne sie, indem du den Tool-Aufruf-Pattern jedes LLM-Aufrufs mit einem Fingerabdruck versiehst:

def detect_loop(trace: dict, window: int = 4) -> bool:
    steps = trace["steps"]
    if len(steps) < window:
        return False

    def step_signature(step: dict) -> str:
        tools_called = sorted(
            t["tool"] if isinstance(t, dict) else t
            for t in step.get("model_output", [])
            if isinstance(t, dict) and "tool" in t
        )
        return "|".join(tools_called)

    recent = [step_signature(s) for s in steps[-window:]]
    if len(set(recent)) == 1 and recent[0]:
        return True

    if len(steps) >= 4:
        pattern = [step_signature(s) for s in steps[-4:]]
        if pattern[0] == pattern[2] and pattern[1] == pattern[3]:
            return True

    return False

Metriken für die Produktion

from collections import Counter

def compute_trace_metrics(trace: dict) -> dict:
    steps = trace["steps"]
    errors = [s for s in steps if "error" in s]

    tool_calls_by_name: Counter = Counter()
    for step in steps:
        for output in step.get("model_output", []):
            if isinstance(output, dict) and "tool" in output:
                tool_calls_by_name[output["tool"]] += 1

    return {
        "trace_id": trace["trace_id"],
        "total_steps": len(steps),
        "error_steps": len(errors),
        "tool_call_distribution": dict(tool_calls_by_name),
        "completed": "final_answer" in trace,
        "loop_detected": detect_loop(trace),
    }

Wichtige Signale für Alarme:

Schleifenrate > 5% der Runs — der Agent steckt fest
Fehlerrate pro Tool > Schwellenwert — ein Tool ist kaputt
Durchschnittliche Schrittanzahl steigt — Aufgaben werden schwieriger
p99-Latenz Spike — ein Modell-Endpunkt ist langsam

OpenTelemetry-Integration

Für Teams, die bereits OpenTelemetry nutzen, Agenten-Traces als Spans emittieren:

from opentelemetry import trace as otel_trace

tracer = otel_trace.get_tracer("agent")

def run_with_otel(task: str, tools: dict, tool_schemas: list, system: str):
    with tracer.start_as_current_span("agent.run") as root_span:
        root_span.set_attribute("agent.task", task[:200])

        agent = TracedAgent()
        for schema in tool_schemas:
            agent.add_tool(**schema)

        messages = [{"role": "user", "content": task}]

        for step in range(20):
            with tracer.start_as_current_span(f"agent.step.{step}") as step_span:
                response = agent.call(messages, system=system)
                step_span.set_attribute("llm.stop_reason", response.stop_reason)
                step_span.set_attribute("llm.input_tokens", response.usage.input_tokens)

                if response.stop_reason == "end_turn":
                    break

PII-Schwärzung in Logs

Agenten-Logs enthalten oft sensible Daten. Vor der Ausgabe an externe Systeme schwärzen:

import re

PII_PATTERNS = [
    (re.compile(r'\b[\w.+-]+@[\w-]+\.[a-z]{2,}\b'), '[EMAIL]'),
    (re.compile(r'\b\d{3,4}[-\s]?\d{6,7}\b'), '[TELEFON]'),
    (re.compile(r'\bsk-[a-zA-Z0-9]{20,}\b'), '[API_SCHLÜSSEL]'),
]

def redact(text: str) -> str:
    for pattern, replacement in PII_PATTERNS:
        text = pattern.sub(replacement, text)
    return text

Die Drei Wichtigsten Metriken

Aufgaben-Abschlussrate — welcher Anteil der Runs erreicht final_answer vs. max_steps oder einen Fehler. Baseline pro Aufgabentyp etablieren.

Token-Kosten pro Aufgabe — Summe von input_tokens + output_tokens über alle Schritte. Ein Kostenanstieg von 20% ohne Änderung der Abschlussrate signalisiert meist Prompt-Degradation.

Tool-Fehlerrate — error_steps / total_steps. Spitzen in dieser Metrik zeigen direkt auf ein kaputtes Tool oder eine kaputte API.

Observability in Agentensystemen ist keine Option — sie ist der Unterschied zwischen einem System, das du iterieren kannst, und einem, das du nur neu starten kannst, wenn es bricht. Beginne mit strukturierten Events und Trace-IDs. Füge Schleifen-Erkennung hinzu. Veröffentliche Metriken. Die Investition zahlt sich beim ersten Produktionsausfall aus, wenn du genau rekonstruieren kannst, was passiert ist, statt raten zu müssen.