Spaces:
Running
Running
File size: 3,921 Bytes
b2c1c67 | 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 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | """Usage analytics endpoints: tokens, cost, latency, tools, models."""
import json
from collections import Counter
from datetime import datetime, timedelta, timezone
from fastapi import APIRouter
from sqlalchemy import func, select
from app.api.deps import DB, CurrentUser
from app.models import ChatSession, Document, Message
from app.schemas import (
AnalyticsOverview,
DailyPoint,
ModelUsage,
ToolUsage,
)
router = APIRouter(prefix="/analytics", tags=["analytics"])
@router.get("/overview", response_model=AnalyticsOverview)
async def overview(user: CurrentUser, db: DB) -> AnalyticsOverview:
session_ids = select(ChatSession.id).where(ChatSession.user_id == user.id)
totals = (
await db.execute(
select(
func.count(Message.id),
func.coalesce(func.sum(Message.input_tokens), 0),
func.coalesce(func.sum(Message.output_tokens), 0),
func.coalesce(func.sum(Message.cost_usd), 0.0),
).where(Message.session_id.in_(session_ids))
)
).one()
avg_latency = (
await db.execute(
select(func.coalesce(func.avg(Message.latency_ms), 0)).where(
Message.session_id.in_(session_ids),
Message.role == "assistant",
Message.latency_ms > 0,
)
)
).scalar_one()
total_sessions = (
await db.execute(
select(func.count(ChatSession.id)).where(ChatSession.user_id == user.id)
)
).scalar_one()
total_documents = (
await db.execute(
select(func.count(Document.id)).where(Document.user_id == user.id)
)
).scalar_one()
# Last 30 days of assistant messages for the daily series and breakdowns
cutoff = datetime.now(timezone.utc) - timedelta(days=30)
recent = (
(
await db.execute(
select(Message).where(
Message.session_id.in_(session_ids),
Message.created_at >= cutoff,
)
)
)
.scalars()
.all()
)
daily_map: dict[str, DailyPoint] = {}
for offset in range(29, -1, -1):
day = (datetime.now(timezone.utc) - timedelta(days=offset)).strftime("%Y-%m-%d")
daily_map[day] = DailyPoint(date=day, messages=0, tokens=0, cost_usd=0.0)
model_counter: dict[str, ModelUsage] = {}
tool_counter: Counter[str] = Counter()
for message in recent:
day = message.created_at.strftime("%Y-%m-%d")
if day in daily_map:
point = daily_map[day]
point.messages += 1
point.tokens += message.input_tokens + message.output_tokens
point.cost_usd = round(point.cost_usd + message.cost_usd, 6)
if message.role == "assistant" and message.model:
usage = model_counter.setdefault(
message.model, ModelUsage(model=message.model, messages=0, cost_usd=0.0)
)
usage.messages += 1
usage.cost_usd = round(usage.cost_usd + message.cost_usd, 6)
if message.tool_calls_json:
try:
for call in json.loads(message.tool_calls_json):
tool_counter[call.get("name", "unknown")] += 1
except json.JSONDecodeError:
pass
return AnalyticsOverview(
total_messages=totals[0],
total_sessions=total_sessions,
total_documents=total_documents,
input_tokens=totals[1],
output_tokens=totals[2],
cost_usd=round(totals[3], 6),
avg_latency_ms=round(avg_latency),
daily=list(daily_map.values()),
models=sorted(
model_counter.values(), key=lambda m: m.messages, reverse=True
),
tools=[
ToolUsage(name=name, count=count)
for name, count in tool_counter.most_common()
],
)
|