214 lines
7.4 KiB
Python
214 lines
7.4 KiB
Python
import logging
|
|
import statistics
|
|
import time
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from rp.config import TOKEN_THROUGHPUT_TARGET
|
|
|
|
logger = logging.getLogger("rp")
|
|
|
|
|
|
@dataclass
|
|
class RequestMetrics:
|
|
timestamp: float
|
|
tokens_input: int
|
|
tokens_output: int
|
|
tokens_cached: int
|
|
duration: float
|
|
cost: float
|
|
cache_hit: bool
|
|
tool_count: int
|
|
error_count: int
|
|
model: str
|
|
|
|
@property
|
|
def tokens_per_sec(self) -> float:
|
|
if self.duration > 0:
|
|
return self.tokens_output / self.duration
|
|
return 0.0
|
|
|
|
@property
|
|
def total_tokens(self) -> int:
|
|
return self.tokens_input + self.tokens_output
|
|
|
|
|
|
@dataclass
|
|
class Alert:
|
|
timestamp: float
|
|
alert_type: str
|
|
message: str
|
|
severity: str
|
|
metrics: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
class MetricsCollector:
|
|
def __init__(self):
|
|
self.requests: List[RequestMetrics] = []
|
|
self.alerts: List[Alert] = []
|
|
self.tool_metrics: Dict[str, Dict[str, Any]] = defaultdict(
|
|
lambda: {'total_calls': 0, 'total_duration': 0.0, 'errors': 0}
|
|
)
|
|
self.start_time = time.time()
|
|
|
|
def record_request(self, metrics: RequestMetrics):
|
|
self.requests.append(metrics)
|
|
self._check_alerts(metrics)
|
|
|
|
def record_tool_call(self, tool_name: str, duration: float, success: bool):
|
|
self.tool_metrics[tool_name]['total_calls'] += 1
|
|
self.tool_metrics[tool_name]['total_duration'] += duration
|
|
if not success:
|
|
self.tool_metrics[tool_name]['errors'] += 1
|
|
|
|
def _check_alerts(self, metrics: RequestMetrics):
|
|
if metrics.tokens_per_sec < TOKEN_THROUGHPUT_TARGET * 0.7:
|
|
self.alerts.append(Alert(
|
|
timestamp=time.time(),
|
|
alert_type='low_throughput',
|
|
message=f"Throughput below target: {metrics.tokens_per_sec:.1f} tok/sec (target: {TOKEN_THROUGHPUT_TARGET})",
|
|
severity='warning',
|
|
metrics={'tokens_per_sec': metrics.tokens_per_sec}
|
|
))
|
|
if metrics.duration > 60:
|
|
self.alerts.append(Alert(
|
|
timestamp=time.time(),
|
|
alert_type='high_latency',
|
|
message=f"Request latency p99 > 60s: {metrics.duration:.1f}s",
|
|
severity='warning',
|
|
metrics={'duration': metrics.duration}
|
|
))
|
|
if metrics.error_count > 0:
|
|
error_rate = self._calculate_error_rate()
|
|
if error_rate > 0.05:
|
|
self.alerts.append(Alert(
|
|
timestamp=time.time(),
|
|
alert_type='high_error_rate',
|
|
message=f"Error rate > 5%: {error_rate:.1%}",
|
|
severity='error',
|
|
metrics={'error_rate': error_rate}
|
|
))
|
|
|
|
def _calculate_error_rate(self) -> float:
|
|
if not self.requests:
|
|
return 0.0
|
|
errors = sum(1 for r in self.requests if r.error_count > 0)
|
|
return errors / len(self.requests)
|
|
|
|
def get_throughput_stats(self) -> Dict[str, float]:
|
|
if not self.requests:
|
|
return {'avg': 0, 'min': 0, 'max': 0}
|
|
throughputs = [r.tokens_per_sec for r in self.requests]
|
|
return {
|
|
'avg': statistics.mean(throughputs),
|
|
'min': min(throughputs),
|
|
'max': max(throughputs),
|
|
'target': TOKEN_THROUGHPUT_TARGET,
|
|
'meeting_target': statistics.mean(throughputs) >= TOKEN_THROUGHPUT_TARGET * 0.9
|
|
}
|
|
|
|
def get_latency_stats(self) -> Dict[str, float]:
|
|
if not self.requests:
|
|
return {'p50': 0, 'p95': 0, 'p99': 0, 'avg': 0}
|
|
durations = sorted([r.duration for r in self.requests])
|
|
n = len(durations)
|
|
return {
|
|
'p50': durations[n // 2] if n > 0 else 0,
|
|
'p95': durations[int(n * 0.95)] if n >= 20 else durations[-1] if n > 0 else 0,
|
|
'p99': durations[int(n * 0.99)] if n >= 100 else durations[-1] if n > 0 else 0,
|
|
'avg': statistics.mean(durations)
|
|
}
|
|
|
|
def get_cost_stats(self) -> Dict[str, float]:
|
|
if not self.requests:
|
|
return {'total': 0, 'avg': 0}
|
|
costs = [r.cost for r in self.requests]
|
|
return {
|
|
'total': sum(costs),
|
|
'avg': statistics.mean(costs),
|
|
'min': min(costs),
|
|
'max': max(costs)
|
|
}
|
|
|
|
def get_cache_stats(self) -> Dict[str, Any]:
|
|
if not self.requests:
|
|
return {'hit_rate': 0, 'hits': 0, 'misses': 0}
|
|
hits = sum(1 for r in self.requests if r.cache_hit)
|
|
misses = len(self.requests) - hits
|
|
return {
|
|
'hit_rate': hits / len(self.requests) if self.requests else 0,
|
|
'hits': hits,
|
|
'misses': misses,
|
|
'cached_tokens': sum(r.tokens_cached for r in self.requests)
|
|
}
|
|
|
|
def get_context_usage(self) -> Dict[str, float]:
|
|
if not self.requests:
|
|
return {'avg_input': 0, 'avg_output': 0, 'avg_total': 0}
|
|
return {
|
|
'avg_input': statistics.mean([r.tokens_input for r in self.requests]),
|
|
'avg_output': statistics.mean([r.tokens_output for r in self.requests]),
|
|
'avg_total': statistics.mean([r.total_tokens for r in self.requests])
|
|
}
|
|
|
|
def get_summary(self) -> Dict[str, Any]:
|
|
return {
|
|
'total_requests': len(self.requests),
|
|
'session_duration': time.time() - self.start_time,
|
|
'throughput': self.get_throughput_stats(),
|
|
'latency': self.get_latency_stats(),
|
|
'cost': self.get_cost_stats(),
|
|
'cache': self.get_cache_stats(),
|
|
'context': self.get_context_usage(),
|
|
'tools': dict(self.tool_metrics),
|
|
'alerts': len(self.alerts)
|
|
}
|
|
|
|
def get_recent_alerts(self, limit: int = 10) -> List[Dict[str, Any]]:
|
|
recent = self.alerts[-limit:] if self.alerts else []
|
|
return [
|
|
{
|
|
'timestamp': a.timestamp,
|
|
'type': a.alert_type,
|
|
'message': a.message,
|
|
'severity': a.severity
|
|
}
|
|
for a in reversed(recent)
|
|
]
|
|
|
|
def format_summary(self) -> str:
|
|
summary = self.get_summary()
|
|
lines = [
|
|
"=== Session Metrics ===",
|
|
f"Requests: {summary['total_requests']}",
|
|
f"Duration: {summary['session_duration']:.1f}s",
|
|
"",
|
|
"Throughput:",
|
|
f" Average: {summary['throughput']['avg']:.1f} tok/sec",
|
|
f" Target: {summary['throughput']['target']} tok/sec",
|
|
"",
|
|
"Latency:",
|
|
f" p50: {summary['latency']['p50']:.2f}s",
|
|
f" p95: {summary['latency']['p95']:.2f}s",
|
|
f" p99: {summary['latency']['p99']:.2f}s",
|
|
"",
|
|
"Cost:",
|
|
f" Total: ${summary['cost']['total']:.4f}",
|
|
f" Average: ${summary['cost']['avg']:.6f}",
|
|
"",
|
|
"Cache:",
|
|
f" Hit Rate: {summary['cache']['hit_rate']:.1%}",
|
|
f" Cached Tokens: {summary['cache']['cached_tokens']}",
|
|
]
|
|
if summary['alerts'] > 0:
|
|
lines.extend([
|
|
"",
|
|
f"Alerts: {summary['alerts']} (see /metrics alerts)"
|
|
])
|
|
return "\n".join(lines)
|
|
|
|
|
|
def create_metrics_collector() -> MetricsCollector:
|
|
return MetricsCollector()
|