Spaces:
Running
Running
File size: 3,982 Bytes
85f900d | 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 | """
ui.tabs.analytics_tab
=====================
Tab 3 — Analytics
Query analytics pulled from the SQLite audit log:
- Summary stats: total queries, avg latency, avg citations (7-day window)
- Queries by day (table)
- KB inventory table with doc/chunk counts
"""
from __future__ import annotations
import logging
from pathlib import Path
import gradio as gr
logger = logging.getLogger(__name__)
def build_analytics_tab(kb_manager, db_path: Path) -> None:
"""
Build and render the Analytics tab.
Args:
kb_manager: KBManager for KB stats.
db_path: Path to the central SQLite database for query stats.
"""
gr.Markdown("## 📊 Analytics")
gr.Markdown("*Query volume, latency breakdown, and knowledge base inventory.*")
# ── Top-level metric cards ──────────────────────────────────────────
with gr.Row():
total_queries_label = gr.Label(label="Total Queries (7d)", value="—")
avg_latency_label = gr.Label(label="Avg End-to-End Latency (ms)", value="—")
avg_citations_label = gr.Label(label="Avg Citations per Answer", value="—")
kb_count_label = gr.Label(label="Knowledge Bases", value="—")
refresh_btn = gr.Button("🔄 Refresh Stats", variant="secondary")
gr.Markdown("---")
gr.Markdown("### Query Volume (last 7 days)")
queries_by_day_df = gr.Dataframe(
headers=["Date", "Queries"],
value=[["—", "—"]],
interactive=False,
)
gr.Markdown("### Knowledge Base Inventory")
kb_inventory_df = gr.Dataframe(
headers=["KB Name", "Display Name", "Documents", "Chunks", "Protected"],
value=_get_kb_inventory(kb_manager),
interactive=False,
)
# ── Load stats on refresh ───────────────────────────────────────────
def _refresh_stats():
from voicevault.storage import sqlite_store as db_mod
try:
stats = db_mod.get_query_stats(db_path, days=7)
except Exception as exc:
logger.warning("Could not load query stats: %s", exc)
stats = {
"total_queries": 0,
"avg_latency_ms": 0,
"avg_citation_count": 0.0,
"queries_by_day": [],
}
total = str(stats.get("total_queries") or 0)
latency = str(int(stats.get("avg_latency_ms") or 0)) + " ms"
citations = f"{stats.get('avg_citation_count') or 0:.1f}"
by_day = stats.get("queries_by_day") or []
day_rows = [[r["date"], str(r["count"])] for r in by_day] or [["—", "—"]]
kbs = _get_kb_inventory(kb_manager)
kb_count = str(len(kbs))
return (
gr.update(value=total),
gr.update(value=latency),
gr.update(value=citations),
gr.update(value=kb_count),
gr.update(value=day_rows),
gr.update(value=kbs if kbs else [["(none)", "—", "0", "0", "No"]]),
)
refresh_btn.click(
fn=_refresh_stats,
outputs=[
total_queries_label,
avg_latency_label,
avg_citations_label,
kb_count_label,
queries_by_day_df,
kb_inventory_df,
],
)
def _get_kb_inventory(kb_manager) -> list[list]:
"""Return rows for the KB inventory dataframe."""
try:
kbs = kb_manager.list_kbs()
if not kbs:
return [["(none)", "—", "0", "0", "No"]]
return [
[
kb.kb_name,
kb.display_name,
str(kb.doc_count),
str(kb.chunk_count),
"🔒 Yes" if kb.is_protected else "No",
]
for kb in kbs
]
except Exception:
return [["(error)", "—", "—", "—", "—"]]
|