diff --git a/frontend-web/src/core/admin/api.ts b/frontend-web/src/core/admin/api.ts index d56004d9..e54b494b 100644 --- a/frontend-web/src/core/admin/api.ts +++ b/frontend-web/src/core/admin/api.ts @@ -104,6 +104,9 @@ export interface AdminLeaderboardResponse { range_days: number; since: string; until: string; + status?: "ready" | "running" | "refreshing"; + generated_at?: string; + message?: string; overview: AdminLeaderboardOverview; users_by_activity: AdminLeaderboardUser[]; users_by_questions: AdminLeaderboardUser[]; diff --git a/frontend-web/src/core/admin/hooks.ts b/frontend-web/src/core/admin/hooks.ts index 9d19457b..243c06a5 100644 --- a/frontend-web/src/core/admin/hooks.ts +++ b/frontend-web/src/core/admin/hooks.ts @@ -33,6 +33,10 @@ export function useUserLeaderboard( limit, ], queryFn: () => getUserLeaderboard(range, limit), + refetchInterval: (query) => { + const status = query.state.data?.status; + return status === "running" || status === "refreshing" ? 3000 : false; + }, }); } diff --git a/frontend-web/src/pages/AdminUserLeaderboardPage.tsx b/frontend-web/src/pages/AdminUserLeaderboardPage.tsx index cfd7a5d7..0122e7b2 100644 --- a/frontend-web/src/pages/AdminUserLeaderboardPage.tsx +++ b/frontend-web/src/pages/AdminUserLeaderboardPage.tsx @@ -39,6 +39,7 @@ import { DialogTrigger, } from "@/components/ui/dialog"; import { Input } from "@/components/ui/input"; +import { Pagination } from "@/components/ui/pagination"; import { ScrollArea } from "@/components/ui/scroll-area"; import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select"; import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs"; @@ -63,6 +64,7 @@ type UserRankTab = "activity" | "questions" | "tokens" | "skills" | "efficiency" type RangeState = { since: string; until: string; days?: number }; type LeaderboardAccent = "crimson" | "rose" | "amber" | "teal"; const DEFAULT_REPORT_MODEL = "__default__"; +const LEADERBOARD_PAGE_SIZE = 20; const accentClasses: Record = { crimson: { @@ -722,11 +724,13 @@ function UserTable({ users, tab, selectedUserId, + rankOffset = 0, onSelect, }: { users: AdminLeaderboardUser[]; tab: UserRankTab; selectedUserId: string | null; + rankOffset?: number; onSelect: (user: AdminLeaderboardUser) => void; }) { const max = Math.max(1, ...users.map((user) => userScore(user, tab))); @@ -749,7 +753,7 @@ function UserTable({ )} >
- {index + 1} + {rankOffset + index + 1}
@@ -1041,6 +1045,7 @@ export default function AdminUserLeaderboardPage() { const [groupRules, setGroupRules] = useState("zero,two"); const [showGroupStats, setShowGroupStats] = useState(false); const [selectedUserId, setSelectedUserId] = useState(null); + const [userPage, setUserPage] = useState(1); const apiRange = useMemo( () => rangeForApi(range), [range], @@ -1082,7 +1087,23 @@ export default function AdminUserLeaderboardPage() { }, [data, query, tab]); const overview = data?.overview; + const leaderboardStatus = data?.status; + const leaderboardPending = leaderboardStatus === "running" || leaderboardStatus === "refreshing"; + const leaderboardHasRows = rankedUsers.length > 0 || (overview?.total_runs ?? 0) > 0 || (overview?.total_users ?? 0) > 0; const topUsers = rankedUsers.slice(0, 3); + const userPageCount = Math.max(1, Math.ceil(rankedUsers.length / LEADERBOARD_PAGE_SIZE)); + const visibleUsers = useMemo(() => { + const start = (userPage - 1) * LEADERBOARD_PAGE_SIZE; + return rankedUsers.slice(start, start + LEADERBOARD_PAGE_SIZE); + }, [rankedUsers, userPage]); + + useEffect(() => { + setUserPage(1); + }, [apiRange.since, apiRange.until, query, tab]); + + useEffect(() => { + setUserPage((page) => Math.min(page, userPageCount)); + }, [userPageCount]); useEffect(() => { if (rankedUsers.length === 0) { @@ -1411,9 +1432,14 @@ export default function AdminUserLeaderboardPage() {
{isLoading && ( -
- - 加载统计数据 +
+
+ + 正在统计排行榜数据 +
+
+ 线上用户较多或时间范围较大时，统计时间可能较长，请耐心等待。为避免影响其他接口，系统会限制排行榜统计的并发数量。 +
)} @@ -1423,8 +1449,29 @@ export default function AdminUserLeaderboardPage() {
)} - {!isLoading && data && overview && ( + {!isLoading && !error && leaderboardPending && !leaderboardHasRows && ( +
+
+ + 排行榜正在后台统计 +
+
+ {data?.message || "统计任务已放到后台执行，页面会自动刷新结果。这样排行榜慢慢算，不会长时间占住当前接口请求。"} +
+
+ )} + + {!isLoading && data && overview && (!leaderboardPending || leaderboardHasRows) && ( <> + {leaderboardPending && leaderboardHasRows && ( +
+
+ + {data.message || "正在后台刷新排行榜，当前展示缓存结果。"} +
+ {data.generated_at && 缓存生成于 {fmtTime(data.generated_at)}} +
+ )}
@@ -1492,11 +1539,23 @@ export default function AdminUserLeaderboardPage() {
setSelectedUserId(user.user_id)} /> +
+ + 共 {fmtNumber(rankedUsers.length)} 位用户，当前第 {userPage} / {userPageCount} 页 + + +
diff --git a/frontend-web/src/pages/AgentChatPage.tsx b/frontend-web/src/pages/AgentChatPage.tsx index 2dca0d8f..061c2ef1 100644 --- a/frontend-web/src/pages/AgentChatPage.tsx +++ b/frontend-web/src/pages/AgentChatPage.tsx @@ -131,7 +131,7 @@ export default function AgentChatPage() { const pendingMessageHandled = useRef(false); useEffect(() => { if (pendingMessageHandled.current) return; - if (!agent_id || !isNewThread || hasStartedConversation) return; + if (!agent_id || !isNewThread) return; let pending: { agentId?: string; text?: string } | null = null; try { const raw = sessionStorage.getItem("pendingAgentMessage"); @@ -143,7 +143,7 @@ export default function AgentChatPage() { pendingMessageHandled.current = true; sessionStorage.removeItem("pendingAgentMessage"); handleSubmit({ text: pending.text, files: [] }); - }, [agent_id, isNewThread, hasStartedConversation, handleSubmit]); + }, [agent_id, isNewThread, handleSubmit]); const handleClarificationSubmit = useCallback( (answer: string) => { diff --git a/offline-backend-20260512/backend/app/gateway/routers/admin_users.py b/offline-backend-20260512/backend/app/gateway/routers/admin_users.py index c7ef2b53..829361d2 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/admin_users.py +++ b/offline-backend-20260512/backend/app/gateway/routers/admin_users.py @@ -13,9 +13,12 @@ methods' ``user_id`` parameter is designed for. from __future__ import annotations +import asyncio import json import logging +import os import re +import time from collections.abc import AsyncIterator from difflib import SequenceMatcher from hashlib import sha256 @@ -53,6 +56,27 @@ from deerflow.persistence.user.model import UserRow logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/admin/users", tags=["admin-users"]) +_LEADERBOARD_MAX_CONCURRENT = max( + 1, + int(os.getenv("ADMIN_LEADERBOARD_MAX_CONCURRENT", "1")), +) +_LEADERBOARD_TIMEOUT_SECONDS = max( + 5.0, + float(os.getenv("ADMIN_LEADERBOARD_TIMEOUT_SECONDS", "45")), +) +_LEADERBOARD_ACQUIRE_TIMEOUT_SECONDS = max( + 0.1, + float(os.getenv("ADMIN_LEADERBOARD_ACQUIRE_TIMEOUT_SECONDS", "0.5")), +) +_LEADERBOARD_CACHE_TTL_SECONDS = max( + 5.0, + float(os.getenv("ADMIN_LEADERBOARD_CACHE_TTL_SECONDS", "300")), +) +_leaderboard_semaphore = asyncio.Semaphore(_LEADERBOARD_MAX_CONCURRENT) +_leaderboard_jobs: dict[str, asyncio.Task] = {} +_leaderboard_cache: dict[str, dict] = {} +_leaderboard_jobs_lock = asyncio.Lock() + async def _require_admin(request: Request) -> None: """Reject non-admin callers. In no-auth mode there is no user → allow.""" @@ -138,6 +162,9 @@ class AdminLeaderboardResponse(BaseModel): range_days: int since: str until: str + status: str = "ready" + generated_at: str = "" + message: str = "" overview: AdminLeaderboardOverview users_by_activity: list[AdminLeaderboardUser] = Field(default_factory=list) users_by_questions: list[AdminLeaderboardUser] = Field(default_factory=list) @@ -965,31 +992,61 @@ async def _query_overview( ) -@router.get("/leaderboard", response_model=AdminLeaderboardResponse) -async def get_user_leaderboard( - request: Request, - 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 | 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), +def _empty_leaderboard_response( + since_dt: datetime, + until_dt: datetime, + range_days: int, + *, + status: str, + message: str, ) -> AdminLeaderboardResponse: - """Return admin analytics for user activity, Q&A, skills, questions, and trends.""" - await _require_admin(request) - session_factory = get_session_factory() - if session_factory is None: - raise HTTPException(status_code=503, detail="Analytics requires database persistence") + return AdminLeaderboardResponse( + range_days=range_days, + since=_iso(since_dt), + until=_iso(until_dt), + status=status, + message=message, + overview=AdminLeaderboardOverview(), + ) - settings = load_system_settings().leaderboard - since_dt, until_dt, range_days = _resolve_range(days, since, until) - 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 + +def _leaderboard_cache_key( + *, + since_dt: datetime, + until_dt: datetime, + limit: int, + excluded: set[str], + include_scheduled: bool, + include_admins: bool, + include_failed: bool, + cleanup_system_questions: bool, +) -> str: + payload = { + "version": 2, + "since": _iso(since_dt), + "until": _iso(until_dt), + "limit": limit, + "excluded": sorted(excluded), + "include_scheduled": include_scheduled, + "include_admins": include_admins, + "include_failed": include_failed, + "cleanup_system_questions": cleanup_system_questions, + } + return sha256(json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest() + + +async def _build_user_leaderboard_response( + session_factory, + settings, + since_dt: datetime, + until_dt: datetime, + range_days: int, + limit: int, + excluded: set[str], + include_scheduled: bool, + include_admins: bool, + include_failed: bool, +) -> AdminLeaderboardResponse: async with session_factory() as session: common = { "exclude_user_ids": excluded, @@ -1023,6 +1080,8 @@ async def get_user_leaderboard( range_days=range_days, since=_iso(since_dt), until=_iso(until_dt), + status="ready", + generated_at=_iso(datetime.utcnow()), overview=overview, users_by_activity=sorted(users, key=lambda u: (u.run_count + u.tool_call_count, u.total_tokens), reverse=True)[:limit], users_by_questions=sorted(users, key=lambda u: (u.qa_count, u.run_count), reverse=True)[:limit], @@ -1038,6 +1097,225 @@ async def get_user_leaderboard( ) +async def _refresh_leaderboard_cache( + cache_key: str, + *, + session_factory, + settings, + since_dt: datetime, + until_dt: datetime, + range_days: int, + limit: int, + excluded: set[str], + include_scheduled: bool, + include_admins: bool, + include_failed: bool, +) -> None: + acquired = False + try: + await _leaderboard_semaphore.acquire() + acquired = True + response = await asyncio.wait_for( + _build_user_leaderboard_response( + session_factory, + settings, + since_dt, + until_dt, + range_days, + limit, + excluded, + include_scheduled, + include_admins, + include_failed, + ), + timeout=_LEADERBOARD_TIMEOUT_SECONDS, + ) + _leaderboard_cache[cache_key] = { + "created_at": time.monotonic(), + "response": response, + } + except asyncio.TimeoutError: + logger.warning("Leaderboard refresh timed out for key=%s", cache_key) + except Exception: + logger.exception("Leaderboard refresh failed for key=%s", cache_key) + finally: + if acquired: + _leaderboard_semaphore.release() + async with _leaderboard_jobs_lock: + _leaderboard_jobs.pop(cache_key, None) + + +async def _ensure_leaderboard_refresh(cache_key: str, **kwargs) -> None: + async with _leaderboard_jobs_lock: + task = _leaderboard_jobs.get(cache_key) + if task and not task.done(): + return + _leaderboard_jobs[cache_key] = asyncio.create_task( + _refresh_leaderboard_cache(cache_key, **kwargs) + ) + + +@router.get("/leaderboard", response_model=AdminLeaderboardResponse) +async def get_user_leaderboard( + request: Request, + 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 | 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.""" + await _require_admin(request) + session_factory = get_session_factory() + 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 settings.exclude_user_ids if uid} + excluded.update({uid for uid in (exclude_user_ids or []) if uid}) + resolved_include_scheduled = settings.include_scheduled if include_scheduled is None else include_scheduled + resolved_include_admins = settings.include_admins if include_admins is None else include_admins + resolved_include_failed = settings.include_failed if include_failed is None else include_failed + cache_key = _leaderboard_cache_key( + since_dt=since_dt, + until_dt=until_dt, + limit=limit, + excluded=excluded, + include_scheduled=resolved_include_scheduled, + include_admins=resolved_include_admins, + include_failed=resolved_include_failed, + cleanup_system_questions=settings.cleanup_system_questions, + ) + + cached = _leaderboard_cache.get(cache_key) + if cached: + response = cached["response"].model_copy() + age = time.monotonic() - float(cached["created_at"]) + if age <= _LEADERBOARD_CACHE_TTL_SECONDS: + response.status = "ready" + return response + response.status = "refreshing" + response.message = "排行榜缓存已返回，后台正在刷新最新统计。" + await _ensure_leaderboard_refresh( + cache_key, + session_factory=session_factory, + settings=settings, + since_dt=since_dt, + until_dt=until_dt, + range_days=range_days, + limit=limit, + excluded=excluded, + include_scheduled=resolved_include_scheduled, + include_admins=resolved_include_admins, + include_failed=resolved_include_failed, + ) + return response + + await _ensure_leaderboard_refresh( + cache_key, + session_factory=session_factory, + settings=settings, + since_dt=since_dt, + until_dt=until_dt, + range_days=range_days, + limit=limit, + excluded=excluded, + include_scheduled=resolved_include_scheduled, + include_admins=resolved_include_admins, + include_failed=resolved_include_failed, + ) + return _empty_leaderboard_response( + since_dt, + until_dt, + range_days, + status="running", + message="排行榜统计已在后台运行。线上用户较多时会慢一些，页面会自动刷新结果。", + ) + + try: + await asyncio.wait_for( + _leaderboard_semaphore.acquire(), + timeout=_LEADERBOARD_ACQUIRE_TIMEOUT_SECONDS, + ) + except TimeoutError as exc: + raise HTTPException( + status_code=429, + detail="排行榜正在统计中，请稍后再试。线上用户较多时统计时间可能较长，请耐心等待。", + ) from exc + + async def build_response() -> AdminLeaderboardResponse: + settings = load_system_settings().leaderboard + since_dt, until_dt, range_days = _resolve_range(days, since, until) + excluded = {uid for uid in settings.exclude_user_ids if uid} + excluded.update({uid for uid in (exclude_user_ids or []) if uid}) + resolved_include_scheduled = settings.include_scheduled if include_scheduled is None else include_scheduled + resolved_include_admins = settings.include_admins if include_admins is None else include_admins + resolved_include_failed = settings.include_failed if include_failed is None else include_failed + async with session_factory() as session: + common = { + "exclude_user_ids": excluded, + "include_scheduled": resolved_include_scheduled, + "include_admins": resolved_include_admins, + "include_failed": resolved_include_failed, + } + tool_common = { + "exclude_user_ids": excluded, + "include_scheduled": resolved_include_scheduled, + "include_admins": resolved_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) + 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, **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( + range_days=range_days, + since=_iso(since_dt), + until=_iso(until_dt), + overview=overview, + users_by_activity=sorted(users, key=lambda u: (u.run_count + u.tool_call_count, u.total_tokens), reverse=True)[:limit], + users_by_questions=sorted(users, key=lambda u: (u.qa_count, u.run_count), reverse=True)[:limit], + 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, + ) + + try: + return await asyncio.wait_for( + build_response(), + timeout=_LEADERBOARD_TIMEOUT_SECONDS, + ) + except TimeoutError as exc: + raise HTTPException( + status_code=504, + detail="排行榜统计时间过长，请缩短时间范围后重试，或稍后再查看。", + ) from exc + finally: + _leaderboard_semaphore.release() + + async def _build_leaderboard_report_prompt( session_factory, settings,