Coverage for agentos/observability/cost_analytics.py: 31%
236 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-08 10:59 +0800
1"""
2AgentOS v0.70 — 成本分析与运营仪表板。
3基因来源: OpenAI Usage Dashboard + Grafana
5提供:
6- 按模型/按天/按session的多维度成本统计
7- Token消耗趋势分析
8- 预算预警系统
9- 成本预测（简单滑动平均）
10"""
12from __future__ import annotations
14import json
15import os
16import threading
17import time
18from collections import defaultdict
19from dataclasses import dataclass
21from agentos.cost.tracker import PRICING, CostTracker
24@dataclass
25class CostEntry:
26 """单次调用的成本记录。"""
28 timestamp: float
29 model: str
30 session_id: str
31 input_tokens: int
32 output_tokens: int
33 cost_usd: float
34 duration_ms: float = 0.0
37@dataclass
38class DailySummary:
39 """日成本摘要。"""
41 date: str
42 model: str
43 calls: int = 0
44 total_input_tokens: int = 0
45 total_output_tokens: int = 0
46 total_cost_usd: float = 0.0
47 avg_duration_ms: float = 0.0
50@dataclass
51class CostBreakdown:
52 """单次调用的详细成本分解。"""
54 model: str
55 input_tokens: int
56 output_tokens: int
57 input_cost_usd: float
58 output_cost_usd: float
59 total_cost_usd: float
60 token_cost_ratio: str = "" # e.g. "1:2.5"
62 def __post_init__(self):
63 if self.output_cost_usd > 0:
64 r = self.input_cost_usd / self.output_cost_usd
65 self.token_cost_ratio = f"1:{r:.1f}" if r > 1 else f"{1/r:.1f}:1"
68@dataclass
69class CostSession:
70 """单次会话的成本摘要。"""
72 session_id: str
73 model: str = ""
74 calls: int = 0
75 total_input_tokens: int = 0
76 total_output_tokens: int = 0
77 total_cost_usd: float = 0.0
78 start_time: float = 0.0
79 end_time: float = 0.0
80 status: str = "active"
83@dataclass
84class BudgetAlert:
85 """预算告警。"""
87 triggered: bool
88 current_cost: float
89 budget: float
90 pct_used: float
91 projected_daily: float
92 message: str
95class CostAnalytics:
96 """
97 成本分析引擎 — 多维度聚合、趋势、预算管理。
98 """
100 def __init__(
101 self,
102 cost_tracker: CostTracker,
103 budget_monthly: float = 0.0,
104 warn_threshold: float = 0.8,
105 persist_path: str = "",
106 ):
107 self.tracker = cost_tracker
108 self.budget_monthly = budget_monthly
109 self.warn_threshold = warn_threshold
110 self.persist_path = persist_path
111 self._entries: list[CostEntry] = []
112 self._lock = threading.Lock()
113 self._load()
115 def record(
116 self,
117 model: str,
118 session_id: str,
119 input_tokens: int,
120 output_tokens: int,
121 duration_ms: float = 0.0,
122 ):
123 """记录一次调用成本。"""
124 price = PRICING.get(model, {})
125 cost = input_tokens / 1_000_000 * price.get(
126 "input", 0
127 ) + output_tokens / 1_000_000 * price.get("output", 0)
128 entry = CostEntry(
129 timestamp=time.time(),
130 model=model,
131 session_id=session_id,
132 input_tokens=input_tokens,
133 output_tokens=output_tokens,
134 cost_usd=cost,
135 duration_ms=duration_ms,
136 )
137 with self._lock:
138 self._entries.append(entry)
140 # Periodic save (every 50 entries)
141 if len(self._entries) % 50 == 0:
142 self._save()
144 # ── 按模型汇总 ───────────────────────────────
146 def by_model(self, hours: float = 24.0) -> list[dict]:
147 """最近N小时的模型成本分布。"""
148 cutoff = time.time() - hours * 3600
149 agg: dict[str, dict] = defaultdict(
150 lambda: {
151 "model": "",
152 "calls": 0,
153 "input_tokens": 0,
154 "output_tokens": 0,
155 "cost_usd": 0.0,
156 }
157 )
158 with self._lock:
159 for e in self._entries:
160 if e.timestamp < cutoff:
161 continue
162 d = agg[e.model]
163 d["model"] = e.model
164 d["calls"] += 1
165 d["input_tokens"] += e.input_tokens
166 d["output_tokens"] += e.output_tokens
167 d["cost_usd"] += e.cost_usd
169 return sorted(agg.values(), key=lambda x: x["cost_usd"], reverse=True)
171 # ── 按天汇总 ────────────────────────────────
173 def daily_breakdown(self, days: int = 7) -> list[DailySummary]:
174 """最近N天的每日成本明细。"""
175 from datetime import datetime
177 datetime.now().date()
178 summaries: dict[tuple[str, str], DailySummary] = {}
180 with self._lock:
181 for e in self._entries:
182 d = datetime.fromtimestamp(e.timestamp).strftime("%Y-%m-%d")
183 key = (d, e.model)
184 if key not in summaries:
185 summaries[key] = DailySummary(date=d, model=e.model)
186 s = summaries[key]
187 s.calls += 1
188 s.total_input_tokens += e.input_tokens
189 s.total_output_tokens += e.output_tokens
190 s.total_cost_usd += e.cost_usd
191 if s.avg_duration_ms == 0:
192 s.avg_duration_ms = e.duration_ms
193 else:
194 s.avg_duration_ms = (s.avg_duration_ms + e.duration_ms) / 2
196 # Sort by date desc, model
197 result = sorted(summaries.values(), key=lambda x: (x.date, x.model), reverse=True)
198 return result
200 # ── 按Session汇总 ────────────────────────────
202 def by_session(self, top_n: int = 10) -> list[dict]:
203 """最贵的N个session。"""
204 agg: dict[str, dict] = defaultdict(
205 lambda: {"session_id": "", "calls": 0, "cost_usd": 0.0, "tokens": 0}
206 )
207 with self._lock:
208 for e in self._entries:
209 d = agg[e.session_id]
210 d["session_id"] = e.session_id
211 d["calls"] += 1
212 d["cost_usd"] += e.cost_usd
213 d["tokens"] += e.input_tokens + e.output_tokens
215 sorted_sessions = sorted(agg.values(), key=lambda x: x["cost_usd"], reverse=True)
216 return sorted_sessions[:top_n]
218 # ── 趋势分析 ─────────────────────────────────
220 def trend(self, metric: str = "cost", window: int = 7) -> list[dict]:
221 """
222 趋势分析（滑动平均）。
223 metric: cost | tokens | calls
224 """
225 daily = self.daily_breakdown(days=window * 2)
226 # Aggregate by date
227 by_date: dict[str, dict] = defaultdict(
228 lambda: {"date": "", "cost": 0.0, "tokens": 0, "calls": 0}
229 )
230 for s in daily:
231 d = by_date[s.date]
232 d["date"] = s.date
233 d["cost"] += s.total_cost_usd
234 d["tokens"] += s.total_input_tokens + s.total_output_tokens
235 d["calls"] += s.calls
237 dates = sorted(by_date.keys())[-window * 2 :]
238 values = [by_date[d].get(metric, 0) for d in dates]
240 # Simple moving average (window=3)
241 trend_data = []
242 for i, d in enumerate(dates):
243 w_start = max(0, i - 2)
244 w_vals = values[w_start : i + 1]
245 trend_data.append(
246 {
247 "date": d,
248 "value": values[i],
249 "sma": sum(w_vals) / len(w_vals),
250 }
251 )
252 return trend_data
254 # ── 预算预警 ─────────────────────────────────
256 def check_budget(self) -> BudgetAlert:
257 """检查预算是否超过阈值。"""
258 if self.budget_monthly <= 0:
259 return BudgetAlert(False, self.tracker.total_cost, 0, 0, 0, "")
261 current = self.tracker.total_cost
262 pct = current / self.budget_monthly
264 # Projection: based on past 7 days average
265 daily_data = self.daily_breakdown(days=7)
266 daily_costs = defaultdict(float)
267 for s in daily_data:
268 daily_costs[s.date] += s.total_cost_usd
269 if daily_costs:
270 avg_daily = sum(daily_costs.values()) / len(daily_costs)
271 else:
272 avg_daily = 0
274 from datetime import datetime
276 days_left = 31 - datetime.now().day
277 projected = avg_daily * max(days_left, 1)
279 triggered = pct > self.warn_threshold
280 message = ""
281 if triggered:
282 projected_total = current + projected
283 message = (
284 f"成本告警: 已消耗 ${current:.4f} ({pct:.0%})，" f"预计月末 ${projected_total:.4f}"
285 )
287 return BudgetAlert(
288 triggered=triggered,
289 current_cost=current,
290 budget=self.budget_monthly,
291 pct_used=pct,
292 projected_daily=avg_daily,
293 message=message,
294 )
296 # ── Summary ──────────────────────────────────
298 @property
299 def total_cost(self) -> float:
300 return self.tracker.total_cost
302 @property
303 def total_calls(self) -> int:
304 with self._lock:
305 return len(self._entries)
307 def summary(self) -> str:
308 models = self.by_model(hours=24)
309 lines = [
310 f"总成本: ${self.total_cost:.4f}",
311 f"总调用: {self.total_calls} 次",
312 "",
313 "最近24h模型分布:",
314 ]
315 for m in models[:5]:
316 lines.append(
317 f" {m['model']}: ${m['cost_usd']:.4f} "
318 f"({m['calls']}次, {m['input_tokens']}+{m['output_tokens']} tokens)"
319 )
320 if self.budget_monthly > 0:
321 alert = self.check_budget()
322 lines.append(f"\n月度预算: ${self.budget_monthly:.2f} (已用 {alert.pct_used:.1%})")
323 if alert.triggered:
324 lines.append(f" {alert.message}")
325 return "\n".join(lines)
327 # ── 详细分解 ─────────────────────────────────
329 def get_breakdown(self, session_id: str = "", hours: float = 24.0) -> list[CostBreakdown]:
330 """获取指定会话或最近的详细成本分解。"""
331 cutoff = time.time() - hours * 3600
332 results = []
333 with self._lock:
334 entries = self._entries
335 if session_id:
336 entries = [e for e in entries if e.session_id == session_id]
337 for e in entries:
338 if e.timestamp < cutoff:
339 continue
340 price = PRICING.get(e.model, {})
341 input_cost = e.input_tokens / 1_000_000 * price.get("input", 0)
342 output_cost = e.output_tokens / 1_000_000 * price.get("output", 0)
343 results.append(
344 CostBreakdown(
345 model=e.model,
346 input_tokens=e.input_tokens,
347 output_tokens=e.output_tokens,
348 input_cost_usd=input_cost,
349 output_cost_usd=output_cost,
350 total_cost_usd=e.cost_usd,
351 )
352 )
353 return sorted(results, key=lambda x: x.total_cost_usd, reverse=True)
355 def get_session(self, session_id: str) -> CostSession | None:
356 """获取指定会话的成本摘要。"""
357 with self._lock:
358 matches = [e for e in self._entries if e.session_id == session_id]
359 if not matches:
360 return None
361 models = set(e.model for e in matches)
362 timestamps = [e.timestamp for e in matches]
363 total_input = sum(e.input_tokens for e in matches)
364 total_output = sum(e.output_tokens for e in matches)
365 total_cost = sum(e.cost_usd for e in matches)
366 return CostSession(
367 session_id=session_id,
368 model=", ".join(sorted(models)),
369 calls=len(matches),
370 total_input_tokens=total_input,
371 total_output_tokens=total_output,
372 total_cost_usd=total_cost,
373 start_time=min(timestamps),
374 end_time=max(timestamps),
375 )
377 # ── Scores 成本关联 ──────────────────────────
379 def cost_by_score_tier(self, scores: dict[float, list[str]]) -> dict[str, float]:
380 """按评分层级聚合成本（与 ScoringEngine 联动）。"""
381 tiers = {}
382 for score, session_ids in scores.items():
383 tier_name = f"score_{score:.1f}"
384 tier_cost = 0.0
385 with self._lock:
386 for e in self._entries:
387 if e.session_id in session_ids:
388 tier_cost += e.cost_usd
389 tiers[tier_name] = tier_cost
390 return tiers
392 # ── Persistence ──────────────────────────────
394 def _save(self):
395 if not self.persist_path:
396 return
397 try:
398 data = []
399 with self._lock:
400 for e in self._entries[-10000:]: # Keep last 10k
401 data.append(
402 {
403 "ts": e.timestamp,
404 "model": e.model,
405 "session_id": e.session_id,
406 "input_tokens": e.input_tokens,
407 "output_tokens": e.output_tokens,
408 "cost_usd": e.cost_usd,
409 "duration_ms": e.duration_ms,
410 }
411 )
412 os.makedirs(os.path.dirname(self.persist_path) or ".", exist_ok=True)
413 with open(self.persist_path, "w") as f:
414 json.dump(data, f)
415 except Exception:
416 pass
418 def _load(self):
419 if not self.persist_path or not os.path.exists(self.persist_path):
420 return
421 try:
422 with open(self.persist_path) as f:
423 data = json.load(f)
424 with self._lock:
425 self._entries = [
426 CostEntry(
427 timestamp=d["ts"],
428 model=d["model"],
429 session_id=d["session_id"],
430 input_tokens=d["input_tokens"],
431 output_tokens=d["output_tokens"],
432 cost_usd=d["cost_usd"],
433 duration_ms=d.get("duration_ms", 0),
434 )
435 for d in data
436 ]
437 except Exception:
438 pass