Compétences : surveillance et observabilité
Après la publication d'une Skill, comment savoir si elle fonctionne bien ? Échoue-t-elle fréquemment ? Quelles étapes sont les plus lentes ?
Cet article présente comment mettre en place l'observabilité pour les Skills, en utilisant les journaux et les métriques pour comprendre l'état d'exécution.
Trois dimensions de l'observabilité
| Dimension | Réponse à la question | Implémentation |
|---|---|---|
| Journaux | Que s'est-il passé ? Qu'est-ce qui a mal tourné ? | Fichiers journaux JSON structurés |
| Métriques | Combien de fois a-t-il été exécuté ? Combien de temps a-t-il pris ? Quel est le taux de réussite ? | Ajouter au fichier de métriques |
| Tracez | Quelles étapes chaque exécution a-t-elle traversées ? Combien de temps chaque étape a-t-elle pris ? | Enregistrements d'horodatage au niveau des étapes |
Normes de sortie de journaux structurés
Les scripts de compétence doivent écrire les journaux d'exécution dans un fichier à un chemin fixe pour faciliter leur examen ultérieur.
Exemple
# Chemin du fichier : 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):
"""
Écrire des entrées de journal structurées
Paramètres :
level: Niveau de journalisation (info / warning / error)
event: Nom de l'événement (snake_case)
skill: Nom de la compétence
context: Tous les champs supplémentaires
"""
entry = {
"ts": datetime.now(timezone.utc).isoformat(),
"level": level,
"event": event,
"skill": skill,
**context
}
line = json.dumps(entry, ensure_ascii=False)
# Sortie à la fois vers stderr (visibilité en temps réel) et fichier journal (persistance)
print(line, file=sys.stderr)
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\n")
# Exemple d'utilisation
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 introuvable")
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):
"""
Écrire des entrées de journal structurées
Paramètres :
level: Niveau de journalisation (info / warning / error)
event: Nom de l'événement (snake_case)
skill: Nom de la compétence
context: Tous les champs supplémentaires
"""
entry = {
"ts": datetime.now(timezone.utc).isoformat(),
"level": level,
"event": event,
"skill": skill,
**context
}
line = json.dumps(entry, ensure_ascii=False)
# Sortie à la fois vers stderr (visibilité en temps réel) et fichier journal (persistance)
print(line, file=sys.stderr)
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\n")
# Exemple d'utilisation
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 introuvable")
{"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"}
Agrégation des métriques : comptage des exécutions et taux de réussite
Extraire les métriques clés des fichiers journaux pour comprendre l'état de santé global de la Skill.
Exemple
# Chemin du fichier : scripts/metrics_report.py
# Calculer les métriques d'exécution de la Skill à partir des fichiers journaux
import json
import os
from collections import defaultdict
LOG_FILE = "/home/claude/skill_run.log"
def generate_metrics_report(log_file: str) -> dict:
"""Lire les fichiers journaux et générer un résumé des métriques"""
if not os.path.exists(log_file):
return {"error": f"Le fichier journal n'existe pas : {log_file}"}
counts = defaultdict(int) # Nombre d'événements divers
errors = [] # Enregistrements d'erreurs
elapsed = [] # Liste des temps d'exécution (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:] # 5 dernières erreurs
}
if __name__ == "__main__":
report = generate_metrics_report(LOG_FILE)
print(json.dumps(report, ensure_ascii=False, indent=2))
# Calculer les métriques d'exécution de la Skill à partir des fichiers journaux
import json
import os
from collections import defaultdict
LOG_FILE = "/home/claude/skill_run.log"
def generate_metrics_report(log_file: str) -> dict:
"""Lire les fichiers journaux et générer un résumé des métriques"""
if not os.path.exists(log_file):
return {"error": f"Le fichier journal n'existe pas : {log_file}"}
counts = defaultdict(int) # Nombre d'événements divers
errors = [] # Enregistrements d'erreurs
elapsed = [] # Liste des temps d'exécution (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:] # 5 dernières erreurs
}
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"}
]
}
Traçage de l'exécution : enregistrement du temps pour chaque étape
Lorsqu'une exécution particulière est particulièrement lente, les relevés de temps au niveau des étapes peuvent rapidement identifier le goulot d'étranglement.
Exemple
# Chemin du fichier : scripts/tracer.py
import time
import json
import sys
from skill_logger import log
class ExecutionTracer:
"""Enregistrer le temps pris par chaque étape lors d'une exécution de Skill"""
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):
"""Marquer le début d'une étape, en enregistrant automatiquement la durée de l'étape précédente"""
now = time.perf_counter()
if self.steps:
# Calculer la durée de l'étape précédente
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):
"""Marquer la fin de l'exécution et produire un rapport de trace complet"""
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
# Exemple d'utilisation
if __name__ == "__main__":
tracer = ExecutionTracer("data-analyzer")
tracer.step("Lire le fichier")
time.sleep(0.3) # Simulez une opération chronophage
tracer.step("Nettoyage des données")
time.sleep(0.1)
tracer.step("Générer le rapport")
time.sleep(0.8)
report = tracer.finish()
print(json.dumps(report, ensure_ascii=False, indent=2))
import time
import json
import sys
from skill_logger import log
class ExecutionTracer:
"""Enregistrer le temps pris par chaque étape lors d'une exécution de Skill"""
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):
"""Marquer le début d'une étape, en enregistrant automatiquement la durée de l'étape précédente"""
now = time.perf_counter()
if self.steps:
# Calculer la durée de l'étape précédente
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):
"""Marquer la fin de l'exécution et produire un rapport de trace complet"""
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
# Exemple d'utilisation
if __name__ == "__main__":
tracer = ExecutionTracer("data-analyzer")
tracer.step("Lire le fichier")
time.sleep(0.3) # Simulez une opération chronophage
tracer.step("Nettoyage des données")
time.sleep(0.1)
tracer.step("Générer le rapport")
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}
]
}
Autres extensionsL'observabilité n'est pas quelque chose dont seuls les systèmes à grande échelle ont besoin. Même pour une compétence utilisée personnellement, un fichier journal clair peut souvent réduire le temps de dépannage de plusieurs heures à quelques minutes lorsque des problèmes surviennent.