| from __future__ import annotations |
| import math, time |
| from sqlalchemy import select, func |
| from app.models.request_log import RequestLog |
| from app.models.model_catalog import ModelCatalog |
|
|
|
|
| def estimate_tokens(text: str) -> int: |
| words = len(text.split()) |
| chars = len(text) |
| return int(words * 1.35 + chars / 4) |
|
|
|
|
| async def estimate_cost(session, model: str, tokens_in: int, tokens_out: int) -> float: |
| result = await session.execute( |
| select(ModelCatalog).where(ModelCatalog.id == model) |
| ) |
| row = result.scalar_one_or_none() |
| if not row: |
| return 0.0 |
| cost_in = (tokens_in / 1_000_000) * row.input_price |
| cost_out = (tokens_out / 1_000_000) * row.output_price |
| return round(cost_in + cost_out, 6) |
|
|
|
|
| async def get_latency_leaderboard(session, hours: int = 24) -> list: |
| cutoff = time.time() - (hours * 3600) |
| result = await session.execute( |
| select( |
| RequestLog.model, |
| func.avg(RequestLog.latency_ms).label("avg_latency"), |
| func.count(RequestLog.id).label("request_count"), |
| ) |
| .where( |
| RequestLog.latency_ms > 0, |
| RequestLog.model.isnot(None), |
| ) |
| .group_by(RequestLog.model) |
| .order_by(func.avg(RequestLog.latency_ms).asc()) |
| .limit(20) |
| ) |
| return [ |
| { |
| "model": row[0], |
| "avg_latency_ms": round(row[1], 1), |
| "request_count": row[2], |
| } |
| for row in result.all() if row[0] |
| ] |
|
|
