polish admin leaderboard analytics

This commit is contained in:
Administrator committed 2026-06-04 21:09:36 +08:00
1 parent 5590f7d85a
commit fe0ede178e
4 files changed
+1122 -86

No files matched your search

@@ -19,7 +19,7 @@ from datetime import datetime, timedelta
from fastapi import APIRouter, HTTPException, Query, Request
from pydantic import BaseModel, Field
from sqlalchemy import and_, case, desc, func, select
from sqlalchemy import and_, case, desc, func, or_, select
from app.gateway.deps import (
get_agent_store,
@@ -32,7 +32,9 @@ from deerflow.config.paths import get_paths
from deerflow.runtime import serialize_channel_values
from app.gateway.routers.memory import _build_v1_provider, _is_v1
from deerflow.persistence.engine import get_session_factory
from deerflow.persistence.llm_metrics.model import LlmCallMetricRow
from deerflow.persistence.run.model import RunRow
from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRunRow
from deerflow.persistence.thread_meta.model import ThreadMetaRow
from deerflow.persistence.tool_metrics.model import ToolCallMetricRow
from deerflow.persistence.user.model import UserRow
@@ -82,6 +84,16 @@ class AdminLeaderboardSkill(BaseModel):
last_used_at: str = ""
class AdminLeaderboardDimension(BaseModel):
name: str
count: int = 0
user_count: int = 0
success_count: int = 0
error_count: int = 0
total_tokens: int = 0
last_seen_at: str = ""
class AdminLeaderboardQuestion(BaseModel):
text: str
user_id: str = ""
@@ -121,6 +133,9 @@ class AdminLeaderboardResponse(BaseModel):
users_by_tokens: list[AdminLeaderboardUser] = Field(default_factory=list)
skills: list[AdminLeaderboardSkill] = Field(default_factory=list)
tools: list[AdminLeaderboardSkill] = Field(default_factory=list)
models: list[AdminLeaderboardDimension] = Field(default_factory=list)
agents: list[AdminLeaderboardDimension] = Field(default_factory=list)
run_statuses: list[AdminLeaderboardDimension] = Field(default_factory=list)
recent_questions: list[AdminLeaderboardQuestion] = Field(default_factory=list)
frequent_questions: list[AdminLeaderboardQuestion] = Field(default_factory=list)
trends: list[AdminLeaderboardTrendPoint] = Field(default_factory=list)
@@ -134,6 +149,9 @@ class AdminUserAnalyticsResponse(BaseModel):
overview: AdminLeaderboardOverview
skills: list[AdminLeaderboardSkill] = Field(default_factory=list)
tools: list[AdminLeaderboardSkill] = Field(default_factory=list)
models: list[AdminLeaderboardDimension] = Field(default_factory=list)
agents: list[AdminLeaderboardDimension] = Field(default_factory=list)
run_statuses: list[AdminLeaderboardDimension] = Field(default_factory=list)
recent_questions: list[AdminLeaderboardQuestion] = Field(default_factory=list)
trends: list[AdminLeaderboardTrendPoint] = Field(default_factory=list)
@@ -181,7 +199,78 @@ def _tool_time_filter(since: datetime, until: datetime):
return and_(ToolCallMetricRow.created_at >= since, ToolCallMetricRow.created_at <= until)
async def _query_user_leaderboard(session, since: datetime, until: datetime, user_id: str | None = None) -> list[AdminLeaderboardUser]:
def _llm_time_filter(since: datetime, until: datetime):
return and_(LlmCallMetricRow.created_at >= since, LlmCallMetricRow.created_at <= until)
def _exclude_user_filter(column, exclude_user_ids: set[str]):
if not exclude_user_ids:
return None
return or_(column.is_(None), column.not_in(list(exclude_user_ids)))
def _apply_exclude(stmt, column, exclude_user_ids: set[str]):
condition = _exclude_user_filter(column, exclude_user_ids)
return stmt.where(condition) if condition is not None else stmt
def _admin_user_ids_stmt():
return select(UserRow.id).where(UserRow.system_role == "admin")
def _scheduled_run_ids_stmt():
return select(ScheduledTaskRunRow.agent_run_id).where(ScheduledTaskRunRow.agent_run_id.is_not(None))
def _apply_user_scope(stmt, column, *, exclude_user_ids: set[str], include_admins: bool):
stmt = _apply_exclude(stmt, column, exclude_user_ids)
if not include_admins:
stmt = stmt.where(or_(column.is_(None), column.not_in(_admin_user_ids_stmt())))
return stmt
def _apply_run_scope(
stmt,
*,
user_column,
run_id_column,
exclude_user_ids: set[str],
include_admins: bool,
include_failed: bool,
include_scheduled: bool,
):
stmt = _apply_user_scope(stmt, user_column, exclude_user_ids=exclude_user_ids, include_admins=include_admins)
if not include_failed:
stmt = stmt.where(RunRow.status != "error")
if not include_scheduled:
stmt = stmt.where(or_(run_id_column.is_(None), run_id_column.not_in(_scheduled_run_ids_stmt())))
return stmt
def _apply_tool_scope(
stmt,
*,
exclude_user_ids: set[str],
include_admins: bool,
include_scheduled: bool,
):
stmt = _apply_user_scope(stmt, ToolCallMetricRow.user_id, exclude_user_ids=exclude_user_ids, include_admins=include_admins)
if not include_scheduled:
stmt = stmt.where(or_(ToolCallMetricRow.run_id.is_(None), ToolCallMetricRow.run_id.not_in(_scheduled_run_ids_stmt())))
return stmt
async def _query_user_leaderboard(
session,
since: datetime,
until: datetime,
user_id: str | None = None,
exclude_user_ids: set[str] | None = None,
include_scheduled: bool = True,
include_admins: bool = True,
include_failed: bool = True,
) -> list[AdminLeaderboardUser]:
exclude_user_ids = exclude_user_ids or set()
run_agg = (
select(
RunRow.user_id.label("user_id"),
@@ -195,14 +284,27 @@ async def _query_user_leaderboard(session, since: datetime, until: datetime, use
)
.where(_run_time_filter(since, until))
.group_by(RunRow.user_id)
.subquery()
)
run_agg = _apply_run_scope(
run_agg,
user_column=RunRow.user_id,
run_id_column=RunRow.run_id,
exclude_user_ids=exclude_user_ids,
include_admins=include_admins,
include_failed=include_failed,
include_scheduled=include_scheduled,
).subquery()
thread_agg = (
select(ThreadMetaRow.user_id.label("user_id"), func.count(ThreadMetaRow.thread_id).label("thread_count"))
.where(and_(ThreadMetaRow.created_at >= since, ThreadMetaRow.created_at <= until))
.group_by(ThreadMetaRow.user_id)
.subquery()
)
thread_agg = _apply_user_scope(
thread_agg,
ThreadMetaRow.user_id,
exclude_user_ids=exclude_user_ids,
include_admins=include_admins,
).subquery()
tool_agg = (
select(
ToolCallMetricRow.user_id.label("user_id"),
@@ -211,8 +313,13 @@ async def _query_user_leaderboard(session, since: datetime, until: datetime, use
)
.where(_tool_time_filter(since, until))
.group_by(ToolCallMetricRow.user_id)
.subquery()
)
tool_agg = _apply_tool_scope(
tool_agg,
exclude_user_ids=exclude_user_ids,
include_admins=include_admins,
include_scheduled=include_scheduled,
).subquery()
stmt = (
select(
@@ -236,6 +343,7 @@ async def _query_user_leaderboard(session, since: datetime, until: datetime, use
)
if user_id:
stmt = stmt.where(UserRow.id == user_id)
stmt = _apply_user_scope(stmt, UserRow.id, exclude_user_ids=exclude_user_ids, include_admins=include_admins)
stmt = stmt.order_by(
desc(func.coalesce(run_agg.c.run_count, 0) + func.coalesce(tool_agg.c.tool_call_count, 0)),
desc(func.coalesce(run_agg.c.total_tokens, 0)),
@@ -261,7 +369,19 @@ async def _query_user_leaderboard(session, since: datetime, until: datetime, use
]
async def _query_call_rankings(session, since: datetime, until: datetime, *, user_id: str | None, field: str, limit: int) -> list[AdminLeaderboardSkill]:
async def _query_call_rankings(
session,
since: datetime,
until: datetime,
*,
user_id: str | None,
field: str,
limit: int,
exclude_user_ids: set[str] | None = None,
include_scheduled: bool = True,
include_admins: bool = True,
) -> list[AdminLeaderboardSkill]:
exclude_user_ids = exclude_user_ids or set()
name_col = ToolCallMetricRow.skill_name if field == "skill" else ToolCallMetricRow.tool_name
call_count = func.count(ToolCallMetricRow.id).label("call_count")
last_used_at = func.max(ToolCallMetricRow.created_at).label("last_used_at")
@@ -282,6 +402,12 @@ async def _query_call_rankings(session, since: datetime, until: datetime, *, use
)
if user_id:
stmt = stmt.where(ToolCallMetricRow.user_id == user_id)
stmt = _apply_tool_scope(
stmt,
exclude_user_ids=exclude_user_ids,
include_admins=include_admins,
include_scheduled=include_scheduled,
)
rows = (await session.execute(stmt)).all()
return [
AdminLeaderboardSkill(
@@ -297,7 +423,142 @@ async def _query_call_rankings(session, since: datetime, until: datetime, *, use
]
async def _query_recent_questions(session, since: datetime, until: datetime, *, user_id: str | None, limit: int) -> list[AdminLeaderboardQuestion]:
async def _query_run_dimensions(
session,
since: datetime,
until: datetime,
*,
user_id: str | None,
field: str,
limit: int,
exclude_user_ids: set[str] | None = None,
include_scheduled: bool = True,
include_admins: bool = True,
include_failed: bool = True,
) -> list[AdminLeaderboardDimension]:
exclude_user_ids = exclude_user_ids or set()
if field == "model":
name_col = RunRow.model_name
elif field == "agent":
name_col = RunRow.assistant_id
elif field == "status":
name_col = RunRow.status
else:
raise ValueError(f"Unknown run dimension: {field}")
count = func.count(RunRow.run_id).label("count")
last_seen_at = func.max(RunRow.updated_at).label("last_seen_at")
stmt = (
select(
name_col.label("name"),
count,
func.count(func.distinct(RunRow.user_id)).label("user_count"),
func.sum(case((RunRow.status == "success", 1), else_=0)).label("success_count"),
func.sum(case((RunRow.status == "error", 1), else_=0)).label("error_count"),
func.coalesce(func.sum(RunRow.total_tokens), 0).label("total_tokens"),
last_seen_at,
)
.where(_run_time_filter(since, until), name_col.is_not(None), name_col != "")
.group_by(name_col)
.order_by(count.desc(), last_seen_at.desc())
.limit(limit)
)
if user_id:
stmt = stmt.where(RunRow.user_id == user_id)
stmt = _apply_run_scope(
stmt,
user_column=RunRow.user_id,
run_id_column=RunRow.run_id,
exclude_user_ids=exclude_user_ids,
include_admins=include_admins,
include_failed=include_failed,
include_scheduled=include_scheduled,
)
rows = (await session.execute(stmt)).all()
return [
AdminLeaderboardDimension(
name=r[0] or "",
count=int(r[1] or 0),
user_count=int(r[2] or 0),
success_count=int(r[3] or 0),
error_count=int(r[4] or 0),
total_tokens=int(r[5] or 0),
last_seen_at=_iso(r[6]),
)
for r in rows
]
async def _query_model_rankings(
session,
since: datetime,
until: datetime,
*,
user_id: str | None,
limit: int,
exclude_user_ids: set[str] | None = None,
include_scheduled: bool = True,
include_admins: bool = True,
include_failed: bool = True,
) -> list[AdminLeaderboardDimension]:
exclude_user_ids = exclude_user_ids or set()
count = func.count(LlmCallMetricRow.id).label("count")
last_seen_at = func.max(LlmCallMetricRow.created_at).label("last_seen_at")
stmt = (
select(
LlmCallMetricRow.model_name.label("name"),
count,
func.count(func.distinct(LlmCallMetricRow.user_id)).label("user_count"),
func.sum(case((LlmCallMetricRow.status == "success", 1), else_=0)).label("success_count"),
func.sum(case((LlmCallMetricRow.status == "error", 1), else_=0)).label("error_count"),
func.coalesce(func.sum(LlmCallMetricRow.total_tokens), 0).label("total_tokens"),
last_seen_at,
)
.where(_llm_time_filter(since, until), LlmCallMetricRow.model_name.is_not(None), LlmCallMetricRow.model_name != "")
.group_by(LlmCallMetricRow.model_name)
.order_by(count.desc(), last_seen_at.desc())
.limit(limit)
)
if user_id:
stmt = stmt.where(LlmCallMetricRow.user_id == user_id)
stmt = _apply_user_scope(
stmt,
LlmCallMetricRow.user_id,
exclude_user_ids=exclude_user_ids,
include_admins=include_admins,
)
if not include_failed:
stmt = stmt.where(LlmCallMetricRow.status != "error")
if not include_scheduled:
stmt = stmt.where(or_(LlmCallMetricRow.run_id.is_(None), LlmCallMetricRow.run_id.not_in(_scheduled_run_ids_stmt())))
rows = (await session.execute(stmt)).all()
return [
AdminLeaderboardDimension(
name=r[0] or "",
count=int(r[1] or 0),
user_count=int(r[2] or 0),
success_count=int(r[3] or 0),
error_count=int(r[4] or 0),
total_tokens=int(r[5] or 0),
last_seen_at=_iso(r[6]),
)
for r in rows
]
async def _query_recent_questions(
session,
since: datetime,
until: datetime,
*,
user_id: str | None,
limit: int,
exclude_user_ids: set[str] | None = None,
include_scheduled: bool = True,
include_admins: bool = True,
include_failed: bool = True,
) -> list[AdminLeaderboardQuestion]:
exclude_user_ids = exclude_user_ids or set()
stmt = (
select(RunRow.first_human_message, RunRow.user_id, UserRow.email, RunRow.thread_id, RunRow.run_id, RunRow.created_at)
.outerjoin(UserRow, UserRow.id == RunRow.user_id)
@@ -307,6 +568,15 @@ async def _query_recent_questions(session, since: datetime, until: datetime, *,
)
if user_id:
stmt = stmt.where(RunRow.user_id == user_id)
stmt = _apply_run_scope(
stmt,
user_column=RunRow.user_id,
run_id_column=RunRow.run_id,
exclude_user_ids=exclude_user_ids,
include_admins=include_admins,
include_failed=include_failed,
include_scheduled=include_scheduled,
)
rows = (await session.execute(stmt)).all()
return [
AdminLeaderboardQuestion(
@@ -321,7 +591,17 @@ async def _query_recent_questions(session, since: datetime, until: datetime, *,
]
async def _query_frequent_questions(session, since: datetime, until: datetime, limit: int) -> list[AdminLeaderboardQuestion]:
async def _query_frequent_questions(
session,
since: datetime,
until: datetime,
limit: int,
exclude_user_ids: set[str] | None = None,
include_scheduled: bool = True,
include_admins: bool = True,
include_failed: bool = True,
) -> list[AdminLeaderboardQuestion]:
exclude_user_ids = exclude_user_ids or set()
stmt = (
select(
RunRow.first_human_message,
@@ -337,6 +617,15 @@ async def _query_frequent_questions(session, since: datetime, until: datetime, l
.order_by(desc("count"), desc("asked_at"))
.limit(limit)
)
stmt = _apply_run_scope(
stmt,
user_column=RunRow.user_id,
run_id_column=RunRow.run_id,
exclude_user_ids=exclude_user_ids,
include_admins=include_admins,
include_failed=include_failed,
include_scheduled=include_scheduled,
)
rows = (await session.execute(stmt)).all()
return [
AdminLeaderboardQuestion(
@@ -349,7 +638,18 @@ async def _query_frequent_questions(session, since: datetime, until: datetime, l
]
async def _query_trends(session, since: datetime, until: datetime, *, user_id: str | None = None) -> list[AdminLeaderboardTrendPoint]:
async def _query_trends(
session,
since: datetime,
until: datetime,
*,
user_id: str | None = None,
exclude_user_ids: set[str] | None = None,
include_scheduled: bool = True,
include_admins: bool = True,
include_failed: bool = True,
) -> list[AdminLeaderboardTrendPoint]:
exclude_user_ids = exclude_user_ids or set()
bucket = func.date(RunRow.created_at)
stmt = (
select(
@@ -365,6 +665,15 @@ async def _query_trends(session, since: datetime, until: datetime, *, user_id: s
)
if user_id:
stmt = stmt.where(RunRow.user_id == user_id)
stmt = _apply_run_scope(
stmt,
user_column=RunRow.user_id,
run_id_column=RunRow.run_id,
exclude_user_ids=exclude_user_ids,
include_admins=include_admins,
include_failed=include_failed,
include_scheduled=include_scheduled,
)
rows = (await session.execute(stmt)).all()
return [
AdminLeaderboardTrendPoint(
@@ -378,7 +687,18 @@ async def _query_trends(session, since: datetime, until: datetime, *, user_id: s
]
async def _query_overview(session, since: datetime, until: datetime, *, user_id: str | None = None) -> AdminLeaderboardOverview:
async def _query_overview(
session,
since: datetime,
until: datetime,
*,
user_id: str | None = None,
exclude_user_ids: set[str] | None = None,
include_scheduled: bool = True,
include_admins: bool = True,
include_failed: bool = True,
) -> AdminLeaderboardOverview:
exclude_user_ids = exclude_user_ids or set()
run_where = [_run_time_filter(since, until)]
thread_where = [and_(ThreadMetaRow.created_at >= since, ThreadMetaRow.created_at <= until)]
tool_where = [_tool_time_filter(since, until)]
@@ -386,8 +706,23 @@ async def _query_overview(session, since: datetime, until: datetime, *, user_id:
run_where.append(RunRow.user_id == user_id)
thread_where.append(ThreadMetaRow.user_id == user_id)
tool_where.append(ToolCallMetricRow.user_id == user_id)
if exclude_user_ids:
run_where.append(_exclude_user_filter(RunRow.user_id, exclude_user_ids))
thread_where.append(_exclude_user_filter(ThreadMetaRow.user_id, exclude_user_ids))
tool_where.append(_exclude_user_filter(ToolCallMetricRow.user_id, exclude_user_ids))
if not include_admins:
run_where.append(or_(RunRow.user_id.is_(None), RunRow.user_id.not_in(_admin_user_ids_stmt())))
thread_where.append(or_(ThreadMetaRow.user_id.is_(None), ThreadMetaRow.user_id.not_in(_admin_user_ids_stmt())))
tool_where.append(or_(ToolCallMetricRow.user_id.is_(None), ToolCallMetricRow.user_id.not_in(_admin_user_ids_stmt())))
if not include_failed:
run_where.append(RunRow.status != "error")
if not include_scheduled:
run_where.append(or_(RunRow.run_id.is_(None), RunRow.run_id.not_in(_scheduled_run_ids_stmt())))
tool_where.append(or_(ToolCallMetricRow.run_id.is_(None), ToolCallMetricRow.run_id.not_in(_scheduled_run_ids_stmt())))
total_users = 1 if user_id else (await session.execute(select(func.count(UserRow.id)))).scalar_one()
total_users_stmt = select(func.count(UserRow.id))
total_users_stmt = _apply_user_scope(total_users_stmt, UserRow.id, exclude_user_ids=exclude_user_ids, include_admins=include_admins)
total_users = 1 if user_id else (await session.execute(total_users_stmt)).scalar_one()
run_row = (await session.execute(
select(
func.count(RunRow.run_id),
@@ -424,6 +759,10 @@ async def get_user_leaderboard(
days: int = Query(default=30, ge=1, le=365),
since: str | None = Query(default=None, description="Inclusive range start, ISO datetime"),
until: str | None = Query(default=None, description="Inclusive range end, ISO datetime"),
exclude_user_ids: list[str] | None = Query(default=None, description="User ids excluded from this leaderboard aggregation"),
include_scheduled: bool = Query(default=True, description="Include runs created by scheduled tasks"),
include_admins: bool = Query(default=True, description="Include admin accounts in leaderboard aggregation"),
include_failed: bool = Query(default=True, description="Include failed runs and LLM calls"),
limit: int = Query(default=20, ge=5, le=100),
) -> AdminLeaderboardResponse:
"""Return admin analytics for user activity, Q&A, skills, questions, and trends."""
@@ -433,14 +772,31 @@ async def get_user_leaderboard(
raise HTTPException(status_code=503, detail="Analytics requires database persistence")
since_dt, until_dt, range_days = _resolve_range(days, since, until)
excluded = {uid for uid in (exclude_user_ids or []) if uid}
async with session_factory() as session:
overview = await _query_overview(session, since_dt, until_dt)
users = await _query_user_leaderboard(session, since_dt, until_dt)
skills = await _query_call_rankings(session, since_dt, until_dt, user_id=None, field="skill", limit=limit)
tools = await _query_call_rankings(session, since_dt, until_dt, user_id=None, field="tool", limit=limit)
recent_questions = await _query_recent_questions(session, since_dt, until_dt, user_id=None, limit=limit)
frequent_questions = await _query_frequent_questions(session, since_dt, until_dt, limit=limit)
trends = await _query_trends(session, since_dt, until_dt)
common = {
"exclude_user_ids": excluded,
"include_scheduled": include_scheduled,
"include_admins": include_admins,
"include_failed": include_failed,
}
tool_common = {
"exclude_user_ids": excluded,
"include_scheduled": include_scheduled,
"include_admins": include_admins,
}
overview = await _query_overview(session, since_dt, until_dt, **common)
users = await _query_user_leaderboard(session, since_dt, until_dt, **common)
skills = await _query_call_rankings(session, since_dt, until_dt, user_id=None, field="skill", limit=limit, **tool_common)
tools = await _query_call_rankings(session, since_dt, until_dt, user_id=None, field="tool", limit=limit, **tool_common)
models = await _query_model_rankings(session, since_dt, until_dt, user_id=None, limit=limit, **common)
if not models:
models = await _query_run_dimensions(session, since_dt, until_dt, user_id=None, field="model", limit=limit, **common)
agents = await _query_run_dimensions(session, since_dt, until_dt, user_id=None, field="agent", limit=limit, **common)
run_statuses = await _query_run_dimensions(session, since_dt, until_dt, user_id=None, field="status", limit=limit, **common)
recent_questions = await _query_recent_questions(session, since_dt, until_dt, user_id=None, limit=limit, **common)
frequent_questions = await _query_frequent_questions(session, since_dt, until_dt, limit=limit, **common)
trends = await _query_trends(session, since_dt, until_dt, **common)
return AdminLeaderboardResponse(
range_days=range_days,
@@ -452,6 +808,9 @@ async def get_user_leaderboard(
users_by_tokens=sorted(users, key=lambda u: u.total_tokens, reverse=True)[:limit],
skills=skills,
tools=tools,
models=models,
agents=agents,
run_statuses=run_statuses,
recent_questions=recent_questions,
frequent_questions=frequent_questions,
trends=trends,
@@ -481,6 +840,11 @@ async def get_user_analytics(
overview = await _query_overview(session, since_dt, until_dt, user_id=user_id)
skills = await _query_call_rankings(session, since_dt, until_dt, user_id=user_id, field="skill", limit=limit)
tools = await _query_call_rankings(session, since_dt, until_dt, user_id=user_id, field="tool", limit=limit)
models = await _query_model_rankings(session, since_dt, until_dt, user_id=user_id, limit=limit)
if not models:
models = await _query_run_dimensions(session, since_dt, until_dt, user_id=user_id, field="model", limit=limit)
agents = await _query_run_dimensions(session, since_dt, until_dt, user_id=user_id, field="agent", limit=limit)
run_statuses = await _query_run_dimensions(session, since_dt, until_dt, user_id=user_id, field="status", limit=limit)
recent_questions = await _query_recent_questions(session, since_dt, until_dt, user_id=user_id, limit=limit)
trends = await _query_trends(session, since_dt, until_dt, user_id=user_id)
@@ -492,6 +856,9 @@ async def get_user_analytics(
overview=overview,
skills=skills,
tools=tools,
models=models,
agents=agents,
run_statuses=run_statuses,
recent_questions=recent_questions,
trends=trends,
)