refine admin leaderboard experience
This commit is contained in:
1 parent
fe0ede178e
commit
88c83b4dd7
13 files changed
+754
-405
No files matched your search
@@ -14,6 +14,7 @@ methods' ``user_id`` parameter is designed for.
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
@@ -29,6 +30,7 @@ from app.gateway.deps import (
|
||||
)
|
||||
from deerflow.agents.memory import BuiltinFileProvider
|
||||
from deerflow.config.paths import get_paths
|
||||
from deerflow.config.system_settings import load_system_settings
|
||||
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
|
||||
@@ -191,6 +193,28 @@ def _question_count_expr():
|
||||
return func.sum(case((RunRow.first_human_message.is_not(None), 1), else_=0))
|
||||
|
||||
|
||||
_SYSTEM_MARKER_RE = re.compile(r"<(?:uploaded_files|attachments|files|system|metadata)>.*?</(?:uploaded_files|attachments|files|system|metadata)>", re.I | re.S)
|
||||
_TAG_RE = re.compile(r"</?[^>]+>")
|
||||
|
||||
|
||||
def _clean_question_text(value: str | None) -> str:
|
||||
text = str(value or "").strip()
|
||||
if not text:
|
||||
return ""
|
||||
text = _SYSTEM_MARKER_RE.sub(" ", text)
|
||||
text = _TAG_RE.sub(" ", text)
|
||||
text = re.sub(r"\s+", " ", text).strip()
|
||||
return text
|
||||
|
||||
|
||||
def _is_displayable_question(value: str | None) -> bool:
|
||||
text = _clean_question_text(value)
|
||||
if not text:
|
||||
return False
|
||||
lowered = text.lower()
|
||||
return lowered not in {"uploaded_files", "attachments", "files"}
|
||||
|
||||
|
||||
def _run_time_filter(since: datetime, until: datetime):
|
||||
return and_(RunRow.created_at >= since, RunRow.created_at <= until)
|
||||
|
||||
@@ -309,7 +333,7 @@ async def _query_user_leaderboard(
|
||||
select(
|
||||
ToolCallMetricRow.user_id.label("user_id"),
|
||||
func.count(ToolCallMetricRow.id).label("tool_call_count"),
|
||||
func.sum(case((ToolCallMetricRow.skill_name.is_not(None), 1), else_=0)).label("skill_call_count"),
|
||||
func.sum(case((func.coalesce(func.nullif(ToolCallMetricRow.skill_name, ""), func.nullif(ToolCallMetricRow.tool_name, "")).is_not(None), 1), else_=0)).label("skill_call_count"),
|
||||
)
|
||||
.where(_tool_time_filter(since, until))
|
||||
.group_by(ToolCallMetricRow.user_id)
|
||||
@@ -382,7 +406,11 @@ async def _query_call_rankings(
|
||||
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
|
||||
name_col = (
|
||||
func.coalesce(func.nullif(ToolCallMetricRow.skill_name, ""), func.nullif(ToolCallMetricRow.tool_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")
|
||||
stmt = (
|
||||
@@ -557,6 +585,7 @@ async def _query_recent_questions(
|
||||
include_scheduled: bool = True,
|
||||
include_admins: bool = True,
|
||||
include_failed: bool = True,
|
||||
cleanup_system_questions: bool = True,
|
||||
) -> list[AdminLeaderboardQuestion]:
|
||||
exclude_user_ids = exclude_user_ids or set()
|
||||
stmt = (
|
||||
@@ -578,17 +607,22 @@ async def _query_recent_questions(
|
||||
include_scheduled=include_scheduled,
|
||||
)
|
||||
rows = (await session.execute(stmt)).all()
|
||||
return [
|
||||
AdminLeaderboardQuestion(
|
||||
text=(r[0] or "")[:500],
|
||||
user_id=r[1] or "",
|
||||
email=r[2] or "",
|
||||
thread_id=r[3] or "",
|
||||
run_id=r[4] or "",
|
||||
asked_at=_iso(r[5]),
|
||||
questions: list[AdminLeaderboardQuestion] = []
|
||||
for r in rows:
|
||||
text = _clean_question_text(r[0]) if cleanup_system_questions else str(r[0] or "").strip()
|
||||
if cleanup_system_questions and not _is_displayable_question(text):
|
||||
continue
|
||||
questions.append(
|
||||
AdminLeaderboardQuestion(
|
||||
text=text[:500],
|
||||
user_id=r[1] or "",
|
||||
email=r[2] or "",
|
||||
thread_id=r[3] or "",
|
||||
run_id=r[4] or "",
|
||||
asked_at=_iso(r[5]),
|
||||
)
|
||||
)
|
||||
for r in rows
|
||||
]
|
||||
return questions
|
||||
|
||||
|
||||
async def _query_frequent_questions(
|
||||
@@ -600,6 +634,7 @@ async def _query_frequent_questions(
|
||||
include_scheduled: bool = True,
|
||||
include_admins: bool = True,
|
||||
include_failed: bool = True,
|
||||
cleanup_system_questions: bool = True,
|
||||
) -> list[AdminLeaderboardQuestion]:
|
||||
exclude_user_ids = exclude_user_ids or set()
|
||||
stmt = (
|
||||
@@ -627,15 +662,20 @@ async def _query_frequent_questions(
|
||||
include_scheduled=include_scheduled,
|
||||
)
|
||||
rows = (await session.execute(stmt)).all()
|
||||
return [
|
||||
AdminLeaderboardQuestion(
|
||||
text=(r[0] or "")[:500],
|
||||
user_id=r[2] or "",
|
||||
email=r[3] or "",
|
||||
asked_at=_iso(r[4]),
|
||||
questions: list[AdminLeaderboardQuestion] = []
|
||||
for r in rows:
|
||||
text = _clean_question_text(r[0]) if cleanup_system_questions else str(r[0] or "").strip()
|
||||
if cleanup_system_questions and not _is_displayable_question(text):
|
||||
continue
|
||||
questions.append(
|
||||
AdminLeaderboardQuestion(
|
||||
text=text[:500],
|
||||
user_id=r[2] or "",
|
||||
email=r[3] or "",
|
||||
asked_at=_iso(r[4]),
|
||||
)
|
||||
)
|
||||
for r in rows
|
||||
]
|
||||
return questions
|
||||
|
||||
|
||||
async def _query_trends(
|
||||
@@ -736,7 +776,7 @@ async def _query_overview(
|
||||
tool_row = (await session.execute(
|
||||
select(
|
||||
func.count(ToolCallMetricRow.id),
|
||||
func.sum(case((ToolCallMetricRow.skill_name.is_not(None), 1), else_=0)),
|
||||
func.sum(case((func.coalesce(func.nullif(ToolCallMetricRow.skill_name, ""), func.nullif(ToolCallMetricRow.tool_name, "")).is_not(None), 1), else_=0)),
|
||||
).where(*tool_where)
|
||||
)).one()
|
||||
|
||||
@@ -760,9 +800,9 @@ async def get_user_leaderboard(
|
||||
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"),
|
||||
include_scheduled: bool | None = Query(default=None, description="Include runs created by scheduled tasks"),
|
||||
include_admins: bool | None = Query(default=None, description="Include admin accounts in leaderboard aggregation"),
|
||||
include_failed: bool | None = Query(default=None, 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."""
|
||||
@@ -771,8 +811,13 @@ async def get_user_leaderboard(
|
||||
if session_factory is None:
|
||||
raise HTTPException(status_code=503, detail="Analytics requires database persistence")
|
||||
|
||||
settings = load_system_settings().leaderboard
|
||||
since_dt, until_dt, range_days = _resolve_range(days, since, until)
|
||||
excluded = {uid for uid in (exclude_user_ids or []) if uid}
|
||||
excluded = {uid for uid in settings.exclude_user_ids if uid}
|
||||
excluded.update({uid for uid in (exclude_user_ids or []) if uid})
|
||||
include_scheduled = settings.include_scheduled if include_scheduled is None else include_scheduled
|
||||
include_admins = settings.include_admins if include_admins is None else include_admins
|
||||
include_failed = settings.include_failed if include_failed is None else include_failed
|
||||
async with session_factory() as session:
|
||||
common = {
|
||||
"exclude_user_ids": excluded,
|
||||
@@ -785,6 +830,10 @@ async def get_user_leaderboard(
|
||||
"include_scheduled": include_scheduled,
|
||||
"include_admins": include_admins,
|
||||
}
|
||||
question_common = {
|
||||
**common,
|
||||
"cleanup_system_questions": settings.cleanup_system_questions,
|
||||
}
|
||||
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)
|
||||
@@ -794,8 +843,8 @@ async def get_user_leaderboard(
|
||||
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)
|
||||
recent_questions = await _query_recent_questions(session, since_dt, until_dt, user_id=None, limit=limit, **question_common)
|
||||
frequent_questions = await _query_frequent_questions(session, since_dt, until_dt, limit=limit, **question_common)
|
||||
trends = await _query_trends(session, since_dt, until_dt, **common)
|
||||
|
||||
return AdminLeaderboardResponse(
|
||||
@@ -832,21 +881,35 @@ async def get_user_analytics(
|
||||
if session_factory is None:
|
||||
raise HTTPException(status_code=503, detail="Analytics requires database persistence")
|
||||
|
||||
settings = load_system_settings().leaderboard
|
||||
common = {
|
||||
"include_scheduled": settings.include_scheduled,
|
||||
"include_admins": settings.include_admins,
|
||||
"include_failed": settings.include_failed,
|
||||
}
|
||||
tool_common = {
|
||||
"include_scheduled": settings.include_scheduled,
|
||||
"include_admins": settings.include_admins,
|
||||
}
|
||||
question_common = {
|
||||
**common,
|
||||
"cleanup_system_questions": settings.cleanup_system_questions,
|
||||
}
|
||||
since_dt, until_dt, range_days = _resolve_range(days, since, until)
|
||||
async with session_factory() as session:
|
||||
users = await _query_user_leaderboard(session, since_dt, until_dt, user_id=user_id)
|
||||
users = await _query_user_leaderboard(session, since_dt, until_dt, user_id=user_id, **common)
|
||||
if not users:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
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)
|
||||
overview = await _query_overview(session, since_dt, until_dt, user_id=user_id, **common)
|
||||
skills = await _query_call_rankings(session, since_dt, until_dt, user_id=user_id, field="skill", limit=limit, **tool_common)
|
||||
tools = await _query_call_rankings(session, since_dt, until_dt, user_id=user_id, field="tool", limit=limit, **tool_common)
|
||||
models = await _query_model_rankings(session, since_dt, until_dt, user_id=user_id, limit=limit, **common)
|
||||
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)
|
||||
models = await _query_run_dimensions(session, since_dt, until_dt, user_id=user_id, field="model", limit=limit, **common)
|
||||
agents = await _query_run_dimensions(session, since_dt, until_dt, user_id=user_id, field="agent", limit=limit, **common)
|
||||
run_statuses = await _query_run_dimensions(session, since_dt, until_dt, user_id=user_id, field="status", limit=limit, **common)
|
||||
recent_questions = await _query_recent_questions(session, since_dt, until_dt, user_id=user_id, limit=limit, **question_common)
|
||||
trends = await _query_trends(session, since_dt, until_dt, user_id=user_id, **common)
|
||||
|
||||
return AdminUserAnalyticsResponse(
|
||||
range_days=range_days,
|
||||
|
||||
@@ -12,6 +12,7 @@ from app.gateway.auth.password import hash_password_async
|
||||
from app.gateway.deps import get_optional_user_from_request
|
||||
from deerflow.config import get_app_config
|
||||
from deerflow.config.system_settings import (
|
||||
LeaderboardSettings,
|
||||
PromptPrefixSettings,
|
||||
SystemSettings,
|
||||
load_system_settings,
|
||||
@@ -58,6 +59,22 @@ class CitationDisplayUpdateRequest(BaseModel):
|
||||
cleanup_words: list[str] | None = None
|
||||
|
||||
|
||||
class LeaderboardSettingsResponse(BaseModel):
|
||||
exclude_user_ids: list[str] = Field(default_factory=list)
|
||||
include_scheduled: bool = True
|
||||
include_admins: bool = True
|
||||
include_failed: bool = True
|
||||
cleanup_system_questions: bool = True
|
||||
|
||||
|
||||
class LeaderboardSettingsUpdateRequest(BaseModel):
|
||||
exclude_user_ids: list[str] | None = None
|
||||
include_scheduled: bool | None = None
|
||||
include_admins: bool | None = None
|
||||
include_failed: bool | None = None
|
||||
cleanup_system_questions: bool | None = None
|
||||
|
||||
|
||||
def _to_response(settings: PromptPrefixSettings) -> PromptPrefixResponse:
|
||||
return PromptPrefixResponse(
|
||||
enabled=settings.enabled,
|
||||
@@ -92,6 +109,44 @@ async def update_prompt_prefix(body: PromptPrefixUpdateRequest, request: Request
|
||||
return _to_response(saved.prompt_prefix)
|
||||
|
||||
|
||||
def _leaderboard_to_response(settings: LeaderboardSettings) -> LeaderboardSettingsResponse:
|
||||
return LeaderboardSettingsResponse(
|
||||
exclude_user_ids=list(dict.fromkeys([uid for uid in settings.exclude_user_ids if uid])),
|
||||
include_scheduled=settings.include_scheduled,
|
||||
include_admins=settings.include_admins,
|
||||
include_failed=settings.include_failed,
|
||||
cleanup_system_questions=settings.cleanup_system_questions,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/leaderboard", response_model=LeaderboardSettingsResponse)
|
||||
async def get_leaderboard_settings(request: Request) -> LeaderboardSettingsResponse:
|
||||
"""Return global leaderboard aggregation settings. Admin only."""
|
||||
await _require_admin(request)
|
||||
return _leaderboard_to_response(load_system_settings().leaderboard)
|
||||
|
||||
|
||||
@router.put("/leaderboard", response_model=LeaderboardSettingsResponse)
|
||||
async def update_leaderboard_settings(
|
||||
body: LeaderboardSettingsUpdateRequest,
|
||||
request: Request,
|
||||
) -> LeaderboardSettingsResponse:
|
||||
"""Update global leaderboard aggregation settings. Admin only."""
|
||||
await _require_admin(request)
|
||||
|
||||
current = load_system_settings()
|
||||
fields = body.model_dump(exclude_unset=True)
|
||||
update: dict[str, object] = {}
|
||||
if "exclude_user_ids" in fields and fields["exclude_user_ids"] is not None:
|
||||
update["exclude_user_ids"] = list(dict.fromkeys([str(uid).strip() for uid in fields["exclude_user_ids"] if str(uid).strip()]))
|
||||
for key in ("include_scheduled", "include_admins", "include_failed", "cleanup_system_questions"):
|
||||
if key in fields and fields[key] is not None:
|
||||
update[key] = bool(fields[key])
|
||||
leaderboard = current.leaderboard.model_copy(update=update)
|
||||
saved = save_system_settings(current.model_copy(update={"leaderboard": leaderboard}))
|
||||
return _leaderboard_to_response(saved.leaderboard)
|
||||
|
||||
|
||||
@router.get("/citation-display", response_model=CitationDisplayResponse)
|
||||
async def get_citation_display() -> CitationDisplayResponse:
|
||||
"""Return global citation display cleanup settings."""
|
||||
|
||||
@@ -78,6 +78,17 @@ class PrivateSquareSettings(BaseModel):
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
|
||||
class LeaderboardSettings(BaseModel):
|
||||
"""Admin-configurable defaults for the user leaderboard analytics."""
|
||||
|
||||
exclude_user_ids: list[str] = Field(default_factory=list, description="User ids excluded from leaderboard aggregation.")
|
||||
include_scheduled: bool = Field(default=True, description="Whether scheduled task runs participate in leaderboard statistics.")
|
||||
include_admins: bool = Field(default=True, description="Whether admin accounts participate in leaderboard statistics.")
|
||||
include_failed: bool = Field(default=True, description="Whether failed runs and provider calls participate in leaderboard statistics.")
|
||||
cleanup_system_questions: bool = Field(default=True, description="Hide attachment/system marker questions such as <uploaded_files>.")
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
|
||||
class SystemSettings(BaseModel):
|
||||
"""Top-level system settings document."""
|
||||
|
||||
@@ -85,6 +96,7 @@ class SystemSettings(BaseModel):
|
||||
private_square: PrivateSquareSettings = Field(default_factory=PrivateSquareSettings)
|
||||
appearance: AppearanceSettings = Field(default_factory=AppearanceSettings)
|
||||
citation_display: CitationDisplaySettings = Field(default_factory=CitationDisplaySettings)
|
||||
leaderboard: LeaderboardSettings = Field(default_factory=LeaderboardSettings)
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
|
||||
|
||||
Reference in new issue
Block a user