butler/tools/tool_metrics.py
Rama main · réplica de lectura del repositorio original.
"""Tool performance metrics — collection and reporting.
Tracks tool call statistics including:
- Call count and success rate
- Average duration
- Cache hit rate
- Error patterns
"""
from __future__ import annotations
import logging
import threading
import time
from collections import defaultdict
from typing import Any
logger = logging.getLogger(__name__)
class ToolMetrics:
"""Collects and reports tool performance metrics."""
def __init__(self):
self._metrics: dict[str, dict[str, Any]] = defaultdict(self._default_metric)
self._lock = threading.Lock()
self._session_start = time.time()
def _default_metric(self) -> dict[str, Any]:
"""Default metric structure for a new tool."""
return {
"call_count": 0,
"success_count": 0,
"error_count": 0,
"total_duration_ms": 0,
"min_duration_ms": float("inf"),
"max_duration_ms": 0,
"cache_hits": 0,
"cache_misses": 0,
"duplicate_blocks": 0,
"errors": defaultdict(int), # error_type → count
}
def record(
self,
tool_name: str,
duration_ms: float,
success: bool,
cache_hit: bool = False,
error_type: str | None = None,
) -> None:
"""Record a tool call outcome.
Args:
tool_name: Name of the tool
duration_ms: Duration in milliseconds
success: Whether the call succeeded
cache_hit: Whether result was from cache
error_type: Error type if failed
"""
with self._lock:
m = self._metrics[tool_name]
m["call_count"] += 1
m["total_duration_ms"] += duration_ms
m["min_duration_ms"] = min(m["min_duration_ms"], duration_ms)
m["max_duration_ms"] = max(m["max_duration_ms"], duration_ms)
if success:
m["success_count"] += 1
else:
m["error_count"] += 1
if error_type:
m["errors"][error_type] += 1
if cache_hit:
m["cache_hits"] += 1
else:
m["cache_misses"] += 1
def record_duplicate_block(self, tool_name: str) -> None:
"""Record a duplicate call block."""
with self._lock:
m = self._metrics[tool_name]
m["duplicate_blocks"] += 1
def get_tool_stats(self, tool_name: str) -> dict[str, Any]:
"""Get statistics for a specific tool."""
with self._lock:
m = self._metrics.get(tool_name)
if not m:
return {}
call_count = m["call_count"]
if call_count == 0:
return {"tool_name": tool_name, "call_count": 0}
success_rate = m["success_count"] / call_count
avg_duration = m["total_duration_ms"] / call_count
cache_total = m["cache_hits"] + m["cache_misses"]
cache_hit_rate = m["cache_hits"] / cache_total if cache_total > 0 else 0
return {
"tool_name": tool_name,
"call_count": call_count,
"success_count": m["success_count"],
"error_count": m["error_count"],
"success_rate": round(success_rate, 4),
"avg_duration_ms": round(avg_duration, 2),
"min_duration_ms": round(m["min_duration_ms"], 2) if m["min_duration_ms"] != float("inf") else 0,
"max_duration_ms": round(m["max_duration_ms"], 2),
"cache_hits": m["cache_hits"],
"cache_misses": m["cache_misses"],
"cache_hit_rate": round(cache_hit_rate, 4),
"duplicate_blocks": m["duplicate_blocks"],
"errors": dict(m["errors"]),
}
def get_all_stats(self) -> dict[str, Any]:
"""Get statistics for all tools."""
with self._lock:
tools = {}
for tool_name in self._metrics:
tools[tool_name] = self.get_tool_stats(tool_name)
# Overall summary
total_calls = sum(m["call_count"] for m in self._metrics.values())
total_success = sum(m["success_count"] for m in self._metrics.values())
total_errors = sum(m["error_count"] for m in self._metrics.values())
total_cache_hits = sum(m["cache_hits"] for m in self._metrics.values())
session_duration = time.time() - self._session_start
return {
"session_duration_seconds": round(session_duration, 2),
"total_tools": len(self._metrics),
"total_calls": total_calls,
"total_success": total_success,
"total_errors": total_errors,
"overall_success_rate": round(total_success / total_calls, 4) if total_calls > 0 else 0,
"total_cache_hits": total_cache_hits,
"overall_cache_hit_rate": round(total_cache_hits / total_calls, 4) if total_calls > 0 else 0,
"tools": tools,
}
def get_top_tools(self, by: str = "call_count", limit: int = 10) -> list[dict[str, Any]]:
"""Get top tools by a metric."""
with self._lock:
stats = [(name, self.get_tool_stats(name)) for name in self._metrics]
stats.sort(key=lambda x: x[1].get(by, 0), reverse=True)
return [s[1] for s in stats[:limit]]
def get_slowest_tools(self, limit: int = 10) -> list[dict[str, Any]]:
"""Get slowest tools by average duration."""
with self._lock:
stats = [(name, self.get_tool_stats(name)) for name in self._metrics]
stats.sort(key=lambda x: x[1].get("avg_duration_ms", 0), reverse=True)
return [s[1] for s in stats[:limit]]
def get_error_prone_tools(self, limit: int = 10) -> list[dict[str, Any]]:
"""Get tools with most errors."""
with self._lock:
stats = [(name, self.get_tool_stats(name)) for name in self._metrics]
stats.sort(key=lambda x: x[1].get("error_count", 0), reverse=True)
return [s[1] for s in stats[:limit]]
def reset(self) -> None:
"""Reset all metrics."""
with self._lock:
self._metrics.clear()
self._session_start = time.time()
class ToolCallRecorder:
"""Records tool calls with timing."""
def __init__(self, metrics: ToolMetrics | None = None):
self._metrics = metrics or get_tool_metrics()
self._start_time: float | None = None
self._tool_name: str | None = None
def start(self, tool_name: str) -> None:
"""Start recording a tool call."""
self._tool_name = tool_name
self._start_time = time.time()
def end(
self,
success: bool = True,
cache_hit: bool = False,
error_type: str | None = None,
) -> None:
"""End recording and submit metrics."""
if self._start_time is None or self._tool_name is None:
return
duration_ms = (time.time() - self._start_time) * 1000
self._metrics.record(
self._tool_name,
duration_ms,
success,
cache_hit,
error_type,
)
self._start_time = None
self._tool_name = None
# Singleton
_metrics: ToolMetrics | None = None
def get_tool_metrics() -> ToolMetrics:
"""Get the singleton metrics instance."""
global _metrics
if _metrics is None:
_metrics = ToolMetrics()
return _metrics
def record_tool_call(
tool_name: str,
duration_ms: float,
success: bool,
cache_hit: bool = False,
error_type: str | None = None,
) -> None:
"""Convenience function to record a tool call."""
get_tool_metrics().record(tool_name, duration_ms, success, cache_hit, error_type)