Cost Tracking¶
Отслеживание расходов на AI API (OpenAI, Anthropic, etc.).
Возможности¶
- Трекинг по провайдерам и моделям
- Расчёт стоимости по токенам
- Лимиты и алерты
- История и аналитика
- Export в разных форматах
Использование¶
Базовый трекинг¶
from kit.pipeline.cost import CostTracker
tracker = CostTracker()
# После вызова API
tracker.track(
provider="openai",
model="gpt-4",
input_tokens=500,
output_tokens=1000
)
tracker.track(
provider="anthropic",
model="claude-3-sonnet",
input_tokens=800,
output_tokens=1500
)
# Общая стоимость
print(f"Total: ${tracker.total:.4f}")
# По провайдерам
print(tracker.by_provider())
# {"openai": 0.045, "anthropic": 0.018}
Цены моделей¶
# Встроенные цены (обновляются)
tracker = CostTracker()
# Кастомные цены
tracker = CostTracker(prices={
"openai": {
"gpt-4": {"input": 0.03, "output": 0.06},
"gpt-4o": {"input": 0.005, "output": 0.015},
"gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
},
"anthropic": {
"claude-3-opus": {"input": 0.015, "output": 0.075},
"claude-3-sonnet": {"input": 0.003, "output": 0.015},
"claude-3-haiku": {"input": 0.00025, "output": 0.00125},
}
})
Лимиты и алерты¶
tracker = CostTracker(
daily_limit=10.0, # $10 в день
monthly_limit=100.0 # $100 в месяц
)
@tracker.on_limit_reached
async def handle_limit(limit_type: str, current: float, limit: float):
await notify_admin(f"Cost limit reached: {limit_type}")
await disable_generation()
# Проверка лимита перед запросом
if tracker.would_exceed_limit("openai", "gpt-4", estimated_tokens=2000):
raise CostLimitError("Daily limit would be exceeded")
Сессии¶
# Трекинг по сессиям/пользователям
tracker.start_session("user_123")
tracker.track("openai", "gpt-4", 500, 1000)
tracker.track("openai", "gpt-4", 300, 800)
session_cost = tracker.end_session("user_123")
print(f"Session cost: ${session_cost:.4f}")
История и аналитика¶
# Последние операции
history = tracker.get_history(limit=100)
# Статистика за период
stats = tracker.get_stats(
start_date=datetime(2024, 1, 1),
end_date=datetime(2024, 1, 31)
)
print(f"January total: ${stats['total']:.2f}")
print(f"Average per day: ${stats['avg_daily']:.2f}")
print(f"Most used model: {stats['top_model']}")
# Export
tracker.export_csv("costs_january.csv")
tracker.export_json("costs_january.json")
API Reference¶
CostTracker¶
class CostTracker:
def __init__(
self,
prices: Dict = None,
daily_limit: float = None,
monthly_limit: float = None,
storage_path: str = None
)
def track(
self,
provider: str,
model: str,
input_tokens: int,
output_tokens: int,
metadata: Dict = None
) -> float # Returns cost
@property
def total(self) -> float
def by_provider(self) -> Dict[str, float]
def by_model(self) -> Dict[str, float]
def would_exceed_limit(
self,
provider: str,
model: str,
estimated_tokens: int
) -> bool
def start_session(self, session_id: str) -> None
def end_session(self, session_id: str) -> float
def get_history(self, limit: int = None) -> List[Dict]
def get_stats(self, start_date: datetime = None, end_date: datetime = None) -> Dict
def export_csv(self, path: str) -> None
def export_json(self, path: str) -> None
def on_limit_reached(self, handler: Callable) -> None
def reset_daily(self) -> None
def reset_monthly(self) -> None
CostEntry¶
@dataclass
class CostEntry:
provider: str
model: str
input_tokens: int
output_tokens: int
cost: float
timestamp: datetime
session_id: Optional[str] = None
metadata: Dict = None
Примеры из production¶
Autoshorts — контроль расходов на генерацию¶
class VideoGenerator:
def __init__(self):
self.cost_tracker = CostTracker(
daily_limit=50.0,
monthly_limit=500.0
)
self.cost_tracker.on_limit_reached(self.pause_generation)
async def generate(self, user_id: str, prompt: str):
self.cost_tracker.start_session(user_id)
try:
# Script generation
script = await self.llm.generate(prompt)
self.cost_tracker.track(
"openai", "gpt-4o",
script.usage.input_tokens,
script.usage.output_tokens
)
# Image prompts
for scene in script.scenes:
prompt = await self.llm.generate(scene)
self.cost_tracker.track(
"openai", "gpt-4o-mini",
prompt.usage.input_tokens,
prompt.usage.output_tokens
)
session_cost = self.cost_tracker.end_session(user_id)
# Сохраняем стоимость для биллинга
await self.billing.charge(user_id, session_cost)
except CostLimitError:
await self.notify_user(user_id, "Service temporarily unavailable")
async def pause_generation(self, limit_type: str, current: float, limit: float):
self.generation_enabled = False
await self.notify_admin(f"Generation paused: {limit_type} limit")
SaaS — биллинг пользователей¶
class UserCostManager:
def __init__(self):
self.trackers = {} # user_id -> CostTracker
def get_tracker(self, user_id: str) -> CostTracker:
if user_id not in self.trackers:
user = get_user(user_id)
self.trackers[user_id] = CostTracker(
daily_limit=user.plan.daily_limit,
monthly_limit=user.plan.monthly_limit
)
return self.trackers[user_id]
async def track_usage(self, user_id: str, provider: str, model: str, tokens: dict):
tracker = self.get_tracker(user_id)
cost = tracker.track(
provider, model,
tokens['input'], tokens['output']
)
# Update user balance
await update_balance(user_id, -cost)
return cost
async def get_user_stats(self, user_id: str) -> dict:
tracker = self.get_tracker(user_id)
return {
"today": tracker.today_total,
"month": tracker.month_total,
"daily_limit": tracker.daily_limit,
"monthly_limit": tracker.monthly_limit,
"history": tracker.get_history(limit=50)
}