""" Agent Q3 [HQ] — Metrics Prometheus-compatible /metrics endpoint + internal tracking. """ import time from collections import defaultdict _counters = defaultdict(int) _latencies = defaultdict(list) _start_time = time.time() def track_request(target: str, latency_s: float): _counters[f"requests_total_{target}"] += 1 _latencies[target].append(latency_s) def get_metrics() -> dict: uptime = time.time() - _start_time avg_latency = { t: round(sum(v)/len(v), 3) if v else 0 for t, v in _latencies.items() } return { "uptime_seconds": round(uptime, 1), "request_counts": dict(_counters), "avg_latency_seconds": avg_latency, } def prometheus_format() -> str: lines = [f"# Agent Q3 [HQ] Metrics\nagentq3_uptime_seconds {round(time.time()-_start_time,1)}"] for k, v in _counters.items(): lines.append(f"agentq3_{k} {v}") return "\n".join(lines)