Skills Monitoring und Observability

Woher wissen Sie nach der Veröffentlichung eines Skills, ob er gut funktioniert? Schlägt er häufig fehl? Welche Schritte sind am langsamsten?

Dieser Artikel zeigt, wie man Observability für Skills aufbaut, indem Logs und Metriken genutzt werden, um den Laufzeitstatus zu verstehen.

Drei Dimensionen der Observability

DimensionFrage beantwortetImplementierung
ProtokolleWas ist passiert? Was ist schiefgelaufen?Strukturierte JSON-Logdateien
MetrikenWie oft wurde es ausgeführt? Wie lange hat es gedauert? Wie hoch ist die Erfolgsrate?An Metrikdatei anhängen
VerfolgenWelche Schritte hat jede Ausführung durchlaufen? Wie lange hat jeder Schritt gedauert?Zeitstempelaufzeichnungen auf Schritt-Ebene

Standards für strukturierte Log-Ausgabe

Skill-Skripte sollten Laufzeitprotokolle in eine Datei unter einem festen Pfad schreiben, damit sie später leicht überprüft werden können.

Beispiel

# Dateipfad: scripts/skill_logger.py
import json
import os
import sys
from datetime import datetime, timezone

LOG_FILE = "/home/claude/skill_run.log"

def log(level: str, event: str, skill: str = "", **context):
    """
Strukturierte Log-Einträge schreiben

Parameter:
level: Log-Level (info / warning / error)
event: Ereignisname (snake_case)
skill: Skill-Name
context: Beliebige zusätzliche Felder
    """
    entry = {
        "ts":    datetime.now(timezone.utc).isoformat(),
        "level": level,
        "event": event,
        "skill": skill,
        **context
    }
    line = json.dumps(entry, ensure_ascii=False)

    # Ausgabe sowohl auf stderr (Echtzeitsichtbarkeit) als auch in Logdatei (Persistenz)
    print(line, file=sys.stderr)
    with open(LOG_FILE, "a", encoding="utf-8") as f:
        f.write(line + "\n")

# Verwendungsbeispiel
if __name__ == "__main__":
    log("info",    "skill_start",    skill="data-analyzer",
        file="/mnt/user-data/uploads/example.csv")
    log("info",    "step_complete",  skill="data-analyzer",
        step="clean", rows_removed=3, elapsed_ms=240)
    log("info",    "skill_complete", skill="data-analyzer",
        output="/mnt/user-data/outputs/report.xlsx", elapsed_ms=1830)
    log("error",   "step_failed",    skill="data-analyzer",
        step="generate_chart", error="openpyxl nicht gefunden")
{"ts": "2026-05-18T10:23:05+00:00", "level": "info",  "event": "skill_start",    "skill": "data-analyzer", "file": "/mnt/user-data/uploads/example.csv"}
{"ts": "2026-05-18T10:23:05+00:00", "level": "info",  "event": "step_complete",  "skill": "data-analyzer", "step": "clean", "rows_removed": 3, "elapsed_ms": 240}
{"ts": "2026-05-18T10:23:07+00:00", "level": "info",  "event": "skill_complete", "skill": "data-analyzer", "output": "/mnt/user-data/outputs/report.xlsx", "elapsed_ms": 1830}
{"ts": "2026-05-18T10:23:07+00:00", "level": "error", "event": "step_failed",    "skill": "data-analyzer", "step": "generate_chart", "error": "openpyxl not found"}

Metrik-Aggregation: Zählen von Ausführungen und Erfolgsquote

Schlüsselmetriken aus Logdateien extrahieren, um den Gesamtzustand des Skills zu verstehen.

Beispiel

# Dateipfad: scripts/metrics_report.py
# Skill-Laufzeitmetriken aus Logdateien berechnen

import json
import os
from collections import defaultdict

LOG_FILE = "/home/claude/skill_run.log"

def generate_metrics_report(log_file: str) -> dict:
    """Logdateien lesen und eine Zusammenfassung der Metriken erstellen"""
    if not os.path.exists(log_file):
        return {"error": f"Logdatei existiert nicht: {log_file}"}

    counts  = defaultdict(int)        # Anzahl verschiedener Ereignisse
    errors  = []                       # Fehlerdatensätze
    elapsed = []                       # Liste der Ausführungszeiten (ms)

    with open(log_file, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                entry = json.loads(line)
            except json.JSONDecodeError:
                continue

            event = entry.get("event", "")
            counts[event] += 1

            if entry.get("level") == "error":
                errors.append({
                    "ts":    entry.get("ts"),
                    "event": event,
                    "error": entry.get("error", ""),
                    "step":  entry.get("step", "")
                })

            if event == "skill_complete" and "elapsed_ms" in entry:
                elapsed.append(entry["elapsed_ms"])

    total_runs   = counts.get("skill_start", 0)
    total_errors = len(errors)
    success_rate = round((total_runs - total_errors) / total_runs * 100, 1) \
                   if total_runs > 0 else 0

    return {
        "total_runs":    total_runs,
        "total_errors":  total_errors,
        "success_rate":  f"{success_rate}%",
        "avg_elapsed_ms": round(sum(elapsed) / len(elapsed)) if elapsed else 0,
        "recent_errors": errors[-5:]   # Letzte 5 Fehler
    }

if __name__ == "__main__":
    report = generate_metrics_report(LOG_FILE)
    print(json.dumps(report, ensure_ascii=False, indent=2))
{
  "total_runs":     42,
  "total_errors":   3,
  "success_rate":   "92.9%",
  "avg_elapsed_ms": 1650,
  "recent_errors": [
    {"ts": "2026-05-18T09:12:00+00:00", "event": "step_failed", "step": "generate_chart", "error": "openpyxl not found"}
  ]
}

Ausführungsverfolgung: Zeitaufzeichnung für jeden Schritt

Wenn eine bestimmte Ausführung besonders langsam ist, können Zeitmessungen auf Schrittebene schnell den Engpass identifizieren.

Beispiel

# Dateipfad: scripts/tracer.py
import time
import json
import sys
from skill_logger import log

class ExecutionTracer:
    """Erfasse die Zeit, die jeder Schritt in einer Skill-Ausführung benötigt"""

    def __init__(self, skill_name: str):
        self.skill_name = skill_name
        self.steps      = []
        self.start_time = time.perf_counter()

    def step(self, step_name: str):
        """Markiere den Beginn eines Schritts und zeichne automatisch die Dauer des vorherigen Schritts auf"""
        now = time.perf_counter()
        if self.steps:
            # Berechne die Dauer des vorherigen Schritts
            self.steps[-1]["elapsed_ms"] = int((now - self.steps[-1]["_start"]) * 1000)
            del self.steps[-1]["_start"]

        self.steps.append({"name": step_name, "_start": now})
        log("info", "step_start", skill=self.skill_name, step=step_name)

    def finish(self):
        """Markiert das Ende der Ausführung und gibt einen vollständigen Trace-Bericht aus"""
        now = time.perf_counter()
        if self.steps:
            self.steps[-1]["elapsed_ms"] = int((now - self.steps[-1]["_start"]) * 1000)
            del self.steps[-1]["_start"]

        total_ms = int((now - self.start_time) * 1000)
        report   = {"skill": self.skill_name,
                    "total_ms": total_ms, "steps": self.steps}

        log("info", "trace_complete", skill=self.skill_name,
            total_ms=total_ms, steps=self.steps)
        return report

# Verwendungsbeispiel
if __name__ == "__main__":
    tracer = ExecutionTracer("data-analyzer")

    tracer.step("Datei lesen")
    time.sleep(0.3)   # Simuliere einen zeitaufwändigen Vorgang

    tracer.step("Datenbereinigung")
    time.sleep(0.1)

    tracer.step("Bericht erstellen")
    time.sleep(0.8)

    report = tracer.finish()
    print(json.dumps(report, ensure_ascii=False, indent=2))
{
  "skill": "data-analyzer",
  "total_ms": 1203,
  "steps": [
    {"name": "读取文件",  "elapsed_ms": 312},
    {"name": "数据清洗",  "elapsed_ms": 108},
    {"name": "生成报告",  "elapsed_ms": 783}
  ]
}

Observability ist nicht nur für groß angelegte Systeme erforderlich. Selbst bei einem persönlich genutzten Skill kann eine klare Logdatei die Fehlersuche bei Problemen oft von Stunden auf Minuten verkürzen.

Andere Erweiterungen