Spaces:
Paused
Paused
File size: 3,772 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 | 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()
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()
],
)
|