Spaces:
Sleeping
Sleeping
File size: 3,935 Bytes
d4b664a | 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 | import logging
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from backend.database.db import get_db
from backend.analytics.metrics import get_dashboard_metrics, get_evaluation_history
from backend.database.models import EvaluationResult, FailureLog
from sqlalchemy import desc
router = APIRouter(prefix="/analytics", tags=["analytics"])
logger = logging.getLogger(__name__)
@router.get("/dashboard")
async def dashboard(db: Session = Depends(get_db)):
return get_dashboard_metrics(db)
@router.get("/history")
async def history(limit: int = Query(50, le=200), db: Session = Depends(get_db)):
return get_evaluation_history(db, limit=limit)
@router.get("/evaluation/{eval_id}")
async def get_evaluation(eval_id: int, db: Session = Depends(get_db)):
ev = db.query(EvaluationResult).filter(EvaluationResult.id == eval_id).first()
if not ev:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Evaluation not found")
failure = db.query(FailureLog).filter(FailureLog.eval_id == eval_id).first()
return {
"id": ev.id,
"session_id": ev.session_id,
"query": ev.query,
"response": ev.response,
"code_eval": {
"result": ev.code_eval_result,
"details": ev.code_eval_details
},
"scores": {
"policy_compliance": ev.policy_compliance,
"faithfulness": ev.faithfulness,
"relevance": ev.relevance,
"tone": ev.tone,
"correctness": ev.correctness
},
"judge_verdict": ev.judge_verdict,
"judge_reasoning": ev.judge_reasoning,
"hallucination": {
"detected": ev.hallucination_detected,
"details": ev.hallucination_details,
"severity": ev.hallucination_severity
},
"trust_score": ev.trust_score,
"final_verdict": ev.final_verdict,
"is_retry": ev.is_retry,
"deployment_ready": ev.deployment_ready,
"failure_analysis": {
"primary_failure_reason": failure.primary_failure_reason if failure else None,
"policy_violations": failure.policy_violations if failure else [],
"hallucinations": failure.hallucinations if failure else [],
"improvement_suggestions": failure.improvement_suggestions if failure else [],
"corrected_response": failure.corrected_response if failure else None,
"severity": failure.severity if failure else None
} if failure else None,
"created_at": ev.created_at.isoformat()
}
@router.get("/failures")
async def get_failures(limit: int = Query(20, le=100), db: Session = Depends(get_db)):
failures = (
db.query(FailureLog)
.order_by(desc(FailureLog.created_at))
.limit(limit)
.all()
)
return [
{
"id": f.id,
"eval_id": f.eval_id,
"query": f.query,
"response": f.response,
"primary_failure_reason": f.primary_failure_reason,
"policy_violations": f.policy_violations,
"hallucinations": f.hallucinations,
"improvement_suggestions": f.improvement_suggestions,
"corrected_response": f.corrected_response,
"severity": f.severity,
"created_at": f.created_at.isoformat()
}
for f in failures
]
@router.get("/policies")
async def get_policies():
from backend.prompts.policies import COMPANY_POLICIES
return {
k: {
"title": v["title"],
"content": v["content"]
}
for k, v in COMPANY_POLICIES.items()
}
@router.get("/dataset")
async def get_dataset():
from backend.datasets.queries import CUSTOMER_QUERIES_DATASET
return {
"total": len(CUSTOMER_QUERIES_DATASET),
"queries": CUSTOMER_QUERIES_DATASET
}
|