Spaces:
Sleeping
Sleeping
File size: 5,383 Bytes
c509b44 b65ef75 c16e1c9 c509b44 c16e1c9 aa63765 c16e1c9 c509b44 b65ef75 c16e1c9 c509b44 c16e1c9 b65ef75 c16e1c9 c509b44 c16e1c9 c509b44 c16e1c9 aa63765 c16e1c9 c509b44 b65ef75 c16e1c9 c509b44 c16e1c9 b65ef75 c16e1c9 c509b44 c16e1c9 c509b44 c16e1c9 aa63765 c16e1c9 c509b44 b65ef75 c16e1c9 c509b44 c16e1c9 b65ef75 c16e1c9 c509b44 c16e1c9 c509b44 c16e1c9 aa63765 c16e1c9 c509b44 b65ef75 c16e1c9 adf80ee c509b44 b65ef75 c509b44 adf80ee c509b44 adf80ee c509b44 b65ef75 c509b44 c16e1c9 b65ef75 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 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 |
from fastapi import APIRouter, Header, HTTPException, Query
from typing import Optional
from datetime import datetime, timedelta
from ..storage.analytics_store import AnalyticsStore
from ..utils.access_control import require_api_permission
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"),
x_user_role: str = Header("viewer")
):
"""
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")
require_api_permission(x_user_role, "view_analytics")
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"),
x_user_role: str = Header("viewer")
):
"""
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")
require_api_permission(x_user_role, "view_analytics")
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"),
x_user_role: str = Header("viewer")
):
"""
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")
require_api_permission(x_user_role, "view_analytics")
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"),
x_user_role: str = Header("viewer")
):
"""
Returns general tenant activity statistics.
Includes total queries, active users, last query timestamp, and individual activity records for heatmap visualization.
"""
if not x_tenant_id:
raise HTTPException(status_code=400, detail="Missing tenant ID")
require_api_permission(x_user_role, "view_analytics")
since_timestamp = int((datetime.now() - timedelta(days=days)).timestamp()) if days else None
activity = analytics_store.get_activity_summary(x_tenant_id, since_timestamp)
# Also fetch individual activity records for heatmap visualization
activities = analytics_store.get_activity_records(x_tenant_id, since_timestamp)
return {
"tenant_id": x_tenant_id,
"activity": activity,
"activities": activities, # Individual records with timestamps for heatmap
"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"),
x_user_role: str = Header("viewer")
):
"""
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")
require_api_permission(x_user_role, "view_analytics")
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
}
|