Spaces:
Sleeping
Sleeping
File size: 1,939 Bytes
b4b8242 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 |
from dataclasses import dataclass, field
from typing import Dict, Any, Optional, List
@dataclass
class Ledger:
spend_usd: float = 0.0
tokens_prompt: int = 0
tokens_completion: int = 0
tokens_cached_prompt: int = 0
tokens_reasoning: int = 0
latency_ms_total: int = 0
llm_calls: int = 0
tool_calls: int = 0
spend_by_agent: Dict[str, float] = field(default_factory=dict)
spend_by_model: Dict[str, float] = field(default_factory=dict)
alerts: List[str] = field(default_factory=list)
def add_cost(self, usd: float, agent_id: Optional[str], model_key: Optional[str]):
self.spend_usd += float(usd)
if agent_id:
self.spend_by_agent[agent_id] = self.spend_by_agent.get(agent_id, 0.0) + float(usd)
if model_key:
self.spend_by_model[model_key] = self.spend_by_model.get(model_key, 0.0) + float(usd)
def add_tokens(self, tokens: Dict[str, Any]):
self.tokens_prompt += int(tokens.get("prompt_tokens", 0))
self.tokens_completion += int(tokens.get("completion_tokens", 0))
self.tokens_cached_prompt += int(tokens.get("cached_prompt_tokens", 0))
self.tokens_reasoning += int(tokens.get("reasoning_tokens", 0))
def add_latency(self, latency_ms: int):
self.latency_ms_total += int(latency_ms)
def mark_llm_call(self):
self.llm_calls += 1
def mark_tool_call(self):
self.tool_calls += 1
def check_budget(self, soft: float, hard: float):
if self.spend_usd >= hard:
msg = f"BUDGET_HARD_EXCEEDED: spend=${self.spend_usd:.4f} >= ${hard:.2f}"
if not self.alerts or self.alerts[-1] != msg:
self.alerts.append(msg)
elif self.spend_usd >= soft:
msg = f"BUDGET_SOFT_EXCEEDED: spend=${self.spend_usd:.4f} >= ${soft:.2f}"
if not self.alerts or self.alerts[-1] != msg:
self.alerts.append(msg)
|