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