Spaces:
Sleeping
Sleeping
File size: 4,531 Bytes
c509b44 c16e1c9 c509b44 c16e1c9 aa63765 c16e1c9 c509b44 c16e1c9 c509b44 c16e1c9 c509b44 c16e1c9 c509b44 c16e1c9 aa63765 c16e1c9 c509b44 c16e1c9 c509b44 c16e1c9 c509b44 c16e1c9 c509b44 c16e1c9 aa63765 c16e1c9 c509b44 c16e1c9 c509b44 c16e1c9 c509b44 c16e1c9 c509b44 c16e1c9 aa63765 c16e1c9 c509b44 c16e1c9 c509b44 c16e1c9 c509b44 c16e1c9 c509b44 c16e1c9 |
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 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 |
from fastapi import APIRouter, Header, HTTPException, Query
from typing import Optional
from datetime import datetime, timedelta
from ..storage.analytics_store import AnalyticsStore
router = APIRouter()
# Initialize analytics store
analytics_store = AnalyticsStore()
@router.get("/overview")
async def analytics_overview(
x_tenant_id: str = Header(None),
days: int = Query(30, description="Number of days to look back")
):
"""
Returns an overview of analytics for the dashboard.
Includes total queries, tool usage, red-flag count, and active users.
"""
if not x_tenant_id:
raise HTTPException(status_code=400, detail="Missing tenant ID")
since_timestamp = int((datetime.now() - timedelta(days=days)).timestamp()) if days else None
tool_usage = analytics_store.get_tool_usage_stats(x_tenant_id, since_timestamp)
activity = analytics_store.get_activity_summary(x_tenant_id, since_timestamp)
rag_quality = analytics_store.get_rag_quality_metrics(x_tenant_id, since_timestamp)
return {
"tenant_id": x_tenant_id,
"overview": {
"total_queries": activity["total_queries"],
"tool_usage": tool_usage,
"redflag_count": activity["redflag_count"],
"active_users": activity["active_users"],
"last_query": activity["last_query"],
"rag_quality": rag_quality
}
}
@router.get("/tool-usage")
async def analytics_tool_usage(
x_tenant_id: str = Header(None),
days: int = Query(30, description="Number of days to look back")
):
"""
Returns how often each tool (RAG, Web, Admin, LLM) was used with detailed stats.
Includes counts, latency, tokens, and success/error rates.
"""
if not x_tenant_id:
raise HTTPException(status_code=400, detail="Missing tenant ID")
since_timestamp = int((datetime.now() - timedelta(days=days)).timestamp()) if days else None
tool_usage = analytics_store.get_tool_usage_stats(x_tenant_id, since_timestamp)
return {
"tenant_id": x_tenant_id,
"tool_usage": tool_usage,
"period_days": days
}
@router.get("/redflags")
async def analytics_redflags(
x_tenant_id: str = Header(None),
limit: int = Query(50, description="Maximum number of violations to return"),
days: int = Query(30, description="Number of days to look back")
):
"""
Returns red-flag violations for this tenant.
Includes rule details, severity, confidence, and timestamps.
"""
if not x_tenant_id:
raise HTTPException(status_code=400, detail="Missing tenant ID")
since_timestamp = int((datetime.now() - timedelta(days=days)).timestamp()) if days else None
redflags = analytics_store.get_redflag_violations(x_tenant_id, limit, since_timestamp)
# Convert timestamps to ISO format
for violation in redflags:
if "timestamp" in violation:
violation["timestamp_iso"] = datetime.fromtimestamp(violation["timestamp"]).isoformat()
return {
"tenant_id": x_tenant_id,
"redflags": redflags,
"count": len(redflags)
}
@router.get("/activity")
async def analytics_activity(
x_tenant_id: str = Header(None),
days: int = Query(30, description="Number of days to look back")
):
"""
Returns general tenant activity statistics.
Includes total queries, active users, and last query timestamp.
"""
if not x_tenant_id:
raise HTTPException(status_code=400, detail="Missing tenant ID")
since_timestamp = int((datetime.now() - timedelta(days=days)).timestamp()) if days else None
activity = analytics_store.get_activity_summary(x_tenant_id, since_timestamp)
return {
"tenant_id": x_tenant_id,
"activity": activity,
"period_days": days
}
@router.get("/rag-quality")
async def analytics_rag_quality(
x_tenant_id: str = Header(None),
days: int = Query(30, description="Number of days to look back")
):
"""
Returns RAG quality metrics including recall/precision indicators.
Includes average hits, scores, and latency.
"""
if not x_tenant_id:
raise HTTPException(status_code=400, detail="Missing tenant ID")
since_timestamp = int((datetime.now() - timedelta(days=days)).timestamp()) if days else None
rag_quality = analytics_store.get_rag_quality_metrics(x_tenant_id, since_timestamp)
return {
"tenant_id": x_tenant_id,
"rag_quality": rag_quality,
"period_days": days
}
|