Навыки: мониторинг и наблюдаемость
После публикации навыка как понять, хорошо ли он работает? Часто ли он даёт сбои? Какие шаги самые медленные?
В этой статье рассказывается, как построить наблюдаемость для навыков с использованием логов и метрик для понимания состояния во время выполнения.
Три измерения наблюдаемости
| Измерение | Вопрос, на который дан ответ | Реализация |
|---|---|---|
| Журналы | Что произошло? Что пошло не так? | Структурированные лог-файлы JSON |
| Метрики | Сколько раз это выполнялось? Сколько времени это заняло? Каков процент успеха? | Добавить в файл метрик |
| Трассируйте | Какие шаги прошло каждое выполнение? Сколько времени занял каждый шаг? | Записи отметок времени на уровне шагов |
Стандарты структурированного вывода логов
Скрипты навыков должны записывать логи выполнения в файл по фиксированному пути, чтобы их было легко просмотреть позже.
Пример
# Путь к файлу: 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):
"""
Записывать структурированные записи журнала
Параметры:
level: Уровень логирования (info / warning / error)
event: Имя события (snake_case)
skill: Имя навыка
context: Любые дополнительные поля
"""
entry = {
"ts": datetime.now(timezone.utc).isoformat(),
"level": level,
"event": event,
"skill": skill,
**context
}
line = json.dumps(entry, ensure_ascii=False)
# Вывод как в stderr (видимость в реальном времени), так и в файл журнала (сохранение)
print(line, file=sys.stderr)
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\n")
# Пример использования
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 не найден")
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):
"""
Записывать структурированные записи журнала
Параметры:
level: Уровень логирования (info / warning / error)
event: Имя события (snake_case)
skill: Имя навыка
context: Любые дополнительные поля
"""
entry = {
"ts": datetime.now(timezone.utc).isoformat(),
"level": level,
"event": event,
"skill": skill,
**context
}
line = json.dumps(entry, ensure_ascii=False)
# Вывод как в stderr (видимость в реальном времени), так и в файл журнала (сохранение)
print(line, file=sys.stderr)
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(line + "\n")
# Пример использования
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 не найден")
{"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"}
Агрегация метрик: подсчёт выполнений и процент успеха
Извлеките ключевые метрики из файлов журнала, чтобы понять общее состояние Skill.
Пример
# Путь к файлу: scripts/metrics_report.py
# Вычислить метрики времени выполнения Skill из файлов журнала
import json
import os
from collections import defaultdict
LOG_FILE = "/home/claude/skill_run.log"
def generate_metrics_report(log_file: str) -> dict:
"""Читает файлы журналов и создаёт сводку метрик"""
if not os.path.exists(log_file):
return {"error": f"Файл журнала не существует: {log_file}"}
counts = defaultdict(int) # Количество различных событий
errors = [] # Записи об ошибках
elapsed = [] # Список времени выполнения (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 ошибок
}
if __name__ == "__main__":
report = generate_metrics_report(LOG_FILE)
print(json.dumps(report, ensure_ascii=False, indent=2))
# Вычислить метрики времени выполнения Skill из файлов журнала
import json
import os
from collections import defaultdict
LOG_FILE = "/home/claude/skill_run.log"
def generate_metrics_report(log_file: str) -> dict:
"""Читает файлы журналов и создаёт сводку метрик"""
if not os.path.exists(log_file):
return {"error": f"Файл журнала не существует: {log_file}"}
counts = defaultdict(int) # Количество различных событий
errors = [] # Записи об ошибках
elapsed = [] # Список времени выполнения (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 ошибок
}
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"}
]
}
Трассировка выполнения: запись времени для каждого шага
Когда конкретное выполнение особенно медленное, записи времени на уровне шагов могут быстро определить узкое место.
Пример
# Путь к файлу: scripts/tracer.py
import time
import json
import sys
from skill_logger import log
class ExecutionTracer:
"""Записывает время, затрачиваемое на каждый шаг выполнения 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):
"""Отмечает начало шага, автоматически записывая длительность предыдущего шага"""
now = time.perf_counter()
if self.steps:
# Вычислить длительность предыдущего шага
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):
"""Отметить завершение выполнения и вывести полный отчёт трассировки"""
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
# Пример использования
if __name__ == "__main__":
tracer = ExecutionTracer("data-analyzer")
tracer.step("Чтение файла")
time.sleep(0.3) # Имитация длительной операции
tracer.step("Очистка данных")
time.sleep(0.1)
tracer.step("Генерация отчёта")
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:
"""Записывает время, затрачиваемое на каждый шаг выполнения 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):
"""Отмечает начало шага, автоматически записывая длительность предыдущего шага"""
now = time.perf_counter()
if self.steps:
# Вычислить длительность предыдущего шага
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):
"""Отметить завершение выполнения и вывести полный отчёт трассировки"""
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
# Пример использования
if __name__ == "__main__":
tracer = ExecutionTracer("data-analyzer")
tracer.step("Чтение файла")
time.sleep(0.3) # Имитация длительной операции
tracer.step("Очистка данных")
time.sleep(0.1)
tracer.step("Генерация отчёта")
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}
]
}
Другие расширенияНаблюдаемость нужна не только крупномасштабным системам. Даже если Skill используется лично, понятный файл журнала часто может сократить время устранения неполадок с часов до минут, когда возникают проблемы.