Spaces:
Running
Running
| """ | |
| 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)", "—", "—", "—", "—"]] | |