Agent-Monitoring

Wissen

Ein Agent in der Produktion ohne Monitoring ist wie ein Auto ohne Tacho: Du weißt nicht, wie schnell du fährst, wie viel Sprit du verbrauchst oder ob der Motor überhitzt. In diesem Abschnitt baust du ein Monitoring-System, das dir jederzeit zeigt, was dein Multi-Agent System tut.

Die vier Säulen des Agent-Monitorings

Token-Tracking implementieren

Jeder API-Aufruf gibt die verbrauchten Tokens zurück. Diese musst du systematisch erfassen:

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class TokenUsage:
    agent_name: str
    model: str
    input_tokens: int
    output_tokens: int
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())

    @property
    def total_tokens(self) -> int:
        return self.input_tokens + self.output_tokens

    @property
    def estimated_cost(self) -> float:
        """Schätzt die Kosten basierend auf dem Modell."""
        pricing = {
            "claude-opus-5": (5.0, 25.0),      # $/1M tokens (input, output)
            "claude-sonnet-5": (3.0, 15.0),
            "claude-haiku-4-5": (1.0, 5.0),
        }
        input_price, output_price = pricing.get(self.model, (3.0, 15.0))
        return (
            (self.input_tokens / 1_000_000) * input_price +
            (self.output_tokens / 1_000_000) * output_price
        )

class TokenTracker:
    def __init__(self):
        self.records: list[TokenUsage] = []

    def record(self, agent_name: str, model: str, response) -> TokenUsage:
        usage = TokenUsage(
            agent_name=agent_name,
            model=model,
            input_tokens=response.usage.input_tokens,
            output_tokens=response.usage.output_tokens
        )
        self.records.append(usage)
        return usage

    def total_cost(self) -> float:
        return sum(r.estimated_cost for r in self.records)

    def cost_by_agent(self) -> dict[str, float]:
        costs: dict[str, float] = {}
        for r in self.records:
            costs[r.agent_name] = costs.get(r.agent_name, 0) + r.estimated_cost
        return costs

Verstehen

Error-Handling mit Retry-Logik

In der Produktion schlagen API-Calls fehl -- Rate Limits, Timeouts, Server-Fehler. Dein System muss das abfangen:

import time
import logging
from functools import wraps

logger = logging.getLogger("agent-system")

def retry_with_backoff(max_retries: int = 3, base_delay: float = 1.0):
    """Decorator für automatische Retries mit exponentiellem Backoff."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        logger.error(f"Alle {max_retries} Versuche fehlgeschlagen: {e}")
                        raise
                    delay = base_delay * (2 ** attempt)
                    logger.warning(
                        f"Versuch {attempt + 1} fehlgeschlagen: {e}. "
                        f"Retry in {delay}s..."
                    )
                    time.sleep(delay)
        return wrapper
    return decorator

# Anwendung im ProductionAgent
class ProductionAgent:
    @retry_with_backoff(max_retries=3, base_delay=1.0)
    def process(self, task: str) -> str:
        response = client.messages.create(
            model=self.model,
            max_tokens=2048,
            system=self.system_prompt,
            messages=[{"role": "user", "content": task}]
        )
        # Token-Tracking
        self.tracker.record(self.name, self.model, response)
        return response.content[0].text

Alerting-System

Definiere Schwellwerte, bei deren Überschreitung du benachrichtigt wirst:

@dataclass
class AlertRule:
    name: str
    metric: str              # "cost_per_hour", "error_rate", "latency_p95"
    threshold: float
    window_minutes: int = 60
    action: str = "log"      # "log", "email", "slack", "pagerduty"

PRODUCTION_ALERTS = [
    AlertRule(name="Kosten-Alarm", metric="cost_per_hour", threshold=5.0),
    AlertRule(name="Fehler-Alarm", metric="error_rate", threshold=0.1),
    AlertRule(name="Latenz-Alarm", metric="latency_p95", threshold=30.0),
]

Das Production Dashboard

Mit dem folgenden Dashboard siehst du Token-Verbrauch, Kosten und Fehlerraten deines Multi-Agent Systems auf einen Blick:

Anwenden

Reflektieren

Monitoring ist keine optionale Ergänzung, sondern die Grundlage für jede Produktionsentscheidung. Ohne Token-Tracking weißt du nicht, ob dein Budget reicht. Ohne Error-Logging findest du Bugs nicht. Ohne Alerting merkst du Probleme erst, wenn es zu spät ist. Investiere die Zeit in gutes Monitoring -- es ist die billigste Versicherung, die du für dein Agent-System kaufen kannst.