Coverage for agentos/observability/metrics.py: 48%

1""" 

2AgentOS v0.70 — 性能指标与可观测性增强。 

3基因来源: Prometheus metrics + OpenTelemetry 

4 

5提供: 

6- 延迟分位数 (p50/p95/p99) 

7- 吞吐量统计 (RPS) 

8- 错误率追踪 

9- 缓存命中率 

10- TTL-based环形缓冲区 

11""" 

12 

13from __future__ import annotations 

14 

15import threading 

16import time 

17from collections import deque 

18from dataclasses import dataclass, field 

19 

20 

21@dataclass 

22class MetricSnapshot: 

23 """指标快照 — 用于导出/序列化。""" 

24 timestamp: float = field(default_factory=time.time) 

25 histograms: dict[str, dict] = field(default_factory=dict) 

26 counters: dict[str, int] = field(default_factory=dict) 

27 gauges: dict[str, float] = field(default_factory=dict) 

28 derived_metrics: dict[str, float] = field(default_factory=dict) 

29 

30 def to_json(self) -> str: 

31 import json 

32 return json.dumps({ 

33 "ts": self.timestamp, 

34 "h": self.histograms, 

35 "c": self.counters, 

36 "g": self.gauges, 

37 "d": self.derived_metrics, 

38 }) 

39 

40 @classmethod 

41 def from_collector(cls, collector: "MetricsCollector") -> "MetricSnapshot": 

42 s = collector.snapshot() 

43 return cls( 

44 histograms={"step": s["latency_step_ms"], "model": s["latency_model_ms"], "tool": s["latency_tool_ms"]}, 

45 counters={"steps": collector.steps_total.value, "model_calls": collector.model_calls_total.value, 

46 "tool_calls": collector.tool_calls_total.value, "errors": collector.errors_total.value, 

47 "cache_hits": collector.cache_hits.value, "cache_misses": collector.cache_misses.value}, 

48 gauges={"active_agents": collector.active_agents.value, "queue_depth": collector.queue_depth.value}, 

49 derived_metrics={"rps": s["throughput"]["rps"], "error_rate": s["error_rate"], "cache_hit_rate": s["cache_hit_rate"]}, 

50 ) 

51 

52 

53@dataclass 

54class MetricPoint: 

55 """指标数据点。""" 

56 timestamp: float 

57 value: float 

58 labels: dict[str, str] = field(default_factory=dict) 

59 

60 

61@dataclass 

62class Histogram: 

63 """滑动窗口直方图 — 计算分位数。""" 

64 name: str 

65 window_seconds: float = 300.0 

66 max_size: int = 10000 

67 _points: deque = field(default_factory=deque) 

68 _lock: threading.Lock = field(default_factory=threading.Lock) 

69 

70 def observe(self, value: float, **labels): 

71 with self._lock: 

72 self._points.append(MetricPoint(timestamp=time.time(), value=value, labels=labels)) 

73 self._prune() 

74 if len(self._points) > self.max_size: 

75 self._points.popleft() 

76 

77 def _prune(self): 

78 cutoff = time.time() - self.window_seconds 

79 while self._points and self._points[0].timestamp < cutoff: 

80 self._points.popleft() 

81 

82 @property 

83 def count(self) -> int: 

84 with self._lock: 

85 self._prune() 

86 return len(self._points) 

87 

88 def quantile(self, q: float) -> float: 

89 """计算分位数 0.5=p50, 0.95=p95, 0.99=p99。""" 

90 with self._lock: 

91 self._prune() 

92 if not self._points: 

93 return 0.0 

94 values = sorted(p.value for p in self._points) 

95 idx = int(len(values) * q) 

96 if idx >= len(values): 

97 idx = len(values) - 1 

98 return values[idx] 

99 

100 @property 

101 def p50(self) -> float: 

102 return self.quantile(0.5) 

103 

104 @property 

105 def p95(self) -> float: 

106 return self.quantile(0.95) 

107 

108 @property 

109 def p99(self) -> float: 

110 return self.quantile(0.99) 

111 

112 @property 

113 def avg(self) -> float: 

114 with self._lock: 

115 self._prune() 

116 if not self._points: 

117 return 0.0 

118 return sum(p.value for p in self._points) / len(self._points) 

119 

120 @property 

121 def min_val(self) -> float: 

122 with self._lock: 

123 self._prune() 

124 if not self._points: 

125 return 0.0 

126 return min(p.value for p in self._points) 

127 

128 @property 

129 def max_val(self) -> float: 

130 with self._lock: 

131 self._prune() 

132 if not self._points: 

133 return 0.0 

134 return max(p.value for p in self._points) 

135 

136 def stats(self) -> dict: 

137 return { 

138 "name": self.name, 

139 "count": self.count, 

140 "avg": self.avg, 

141 "p50": self.p50, 

142 "p95": self.p95, 

143 "p99": self.p99, 

144 "min": self.min_val, 

145 "max": self.max_val, 

146 "window_seconds": self.window_seconds, 

147 } 

148 

149 

150@dataclass 

151class Counter: 

152 """单调递增计数器。""" 

153 name: str 

154 _value: int = 0 

155 _labels: dict[str, str] = field(default_factory=dict) 

156 

157 def inc(self, amount: int = 1): 

158 self._value += amount 

159 

160 @property 

161 def value(self) -> int: 

162 return self._value 

163 

164 

165@dataclass 

166class Gauge: 

167 """可增可减的仪表值。""" 

168 name: str 

169 _value: float = 0.0 

170 _labels: dict[str, str] = field(default_factory=dict) 

171 

172 def set(self, value: float): 

173 self._value = value 

174 

175 def inc(self, amount: float = 1.0): 

176 self._value += amount 

177 

178 def dec(self, amount: float = 1.0): 

179 self._value -= amount 

180 

181 @property 

182 def value(self) -> float: 

183 return self._value 

184 

185 

186class MetricsCollector: 

187 """ 

188 统一指标收集器。 

189 内置: latency, throughput, error_rate, cache_hit_rate。 

190 """ 

191 

192 def __init__(self, window_seconds: float = 300.0): 

193 self.window_seconds = window_seconds 

194 

195 # Histograms 

196 self.latency_step = Histogram("step_latency", window_seconds) 

197 self.latency_model = Histogram("model_latency", window_seconds) 

198 self.latency_tool = Histogram("tool_latency", window_seconds) 

199 

200 # Counters 

201 self.steps_total = Counter("steps_total") 

202 self.model_calls_total = Counter("model_calls_total") 

203 self.tool_calls_total = Counter("tool_calls_total") 

204 self.errors_total = Counter("errors_total") 

205 self.cache_hits = Counter("cache_hits") 

206 self.cache_misses = Counter("cache_misses") 

207 

208 # Gauges 

209 self.active_agents = Gauge("active_agents") 

210 self.queue_depth = Gauge("queue_depth") 

211 self.memory_used_mb = Gauge("memory_used_mb") 

212 

213 self._start_time = time.time() 

214 

215 # ── Recording ──────────────────────────────── 

216 

217 def record_step_latency(self, duration_ms: float): 

218 self.latency_step.observe(duration_ms) 

219 self.steps_total.inc() 

220 

221 def record_model_latency(self, duration_ms: float, model: str = ""): 

222 self.latency_model.observe(duration_ms, model=model) 

223 self.model_calls_total.inc() 

224 

225 def record_tool_latency(self, duration_ms: float, tool: str = ""): 

226 self.latency_tool.observe(duration_ms, tool=tool) 

227 self.tool_calls_total.inc() 

228 

229 def record_error(self): 

230 self.errors_total.inc() 

231 

232 def record_cache_hit(self): 

233 self.cache_hits.inc() 

234 

235 def record_cache_miss(self): 

236 self.cache_misses.inc() 

237 

238 # ── Derived Metrics ────────────────────────── 

239 

240 @property 

241 def uptime_seconds(self) -> float: 

242 return time.time() - self._start_time 

243 

244 @property 

245 def rps(self) -> float: 

246 """请求速率 (steps/sec over window)。""" 

247 if self.uptime_seconds < 1: 

248 return self.steps_total.value 

249 return self.steps_total.value / self.uptime_seconds 

250 

251 @property 

252 def error_rate(self) -> float: 

253 total = self.steps_total.value + self.errors_total.value 

254 if total == 0: 

255 return 0.0 

256 return self.errors_total.value / total 

257 

258 @property 

259 def cache_hit_rate(self) -> float: 

260 total = self.cache_hits.value + self.cache_misses.value 

261 if total == 0: 

262 return 0.0 

263 return self.cache_hits.value / total 

264 

265 # ── Snapshot ───────────────────────────────── 

266 

267 def snapshot(self) -> dict: 

268 return { 

269 "uptime_seconds": self.uptime_seconds, 

270 "throughput": { 

271 "rps": round(self.rps, 2), 

272 "steps_total": self.steps_total.value, 

273 "model_calls": self.model_calls_total.value, 

274 "tool_calls": self.tool_calls_total.value, 

275 }, 

276 "latency_step_ms": self.latency_step.stats(), 

277 "latency_model_ms": self.latency_model.stats(), 

278 "latency_tool_ms": self.latency_tool.stats(), 

279 "error_rate": round(self.error_rate, 4), 

280 "errors_total": self.errors_total.value, 

281 "cache_hit_rate": round(self.cache_hit_rate, 2), 

282 "cache_hits": self.cache_hits.value, 

283 "cache_misses": self.cache_misses.value, 

284 "active_agents": self.active_agents.value, 

285 "queue_depth": self.queue_depth.value, 

286 } 

287 

288 def summary(self) -> str: 

289 s = self.snapshot() 

290 lines = [ 

291 f"运行时间: {s['uptime_seconds']:.0f}s", 

292 f"吞吐: {s['throughput']['rps']} rps ({s['throughput']['steps_total']} steps)", 

293 f"延迟: p50={s['latency_step_ms']['p50']:.0f}ms p95={s['latency_step_ms']['p95']:.0f}ms p99={s['latency_step_ms']['p99']:.0f}ms", 

294 f"错误率: {s['error_rate']:.2%} ({s['errors_total']} errors)", 

295 f"缓存命中率: {s['cache_hit_rate']:.1%} ({s['cache_hits']}/{s['cache_hits'] + s['cache_misses']})", 

296 ] 

297 return "\n".join(lines)