From f4ccef8c70f374ee1abe836eb68327bdc33b6528 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=89=E9=91=AB=E5=8D=87?= Date: Fri, 5 Jun 2026 21:06:23 +0800 Subject: [PATCH 1/7] fix: optimize leaderboard and agent handoff --- .../src/pages/AdminUserLeaderboardPage.tsx | 45 +++++- frontend-web/src/pages/AgentChatPage.tsx | 4 +- .../app/gateway/routers/admin_users.py | 145 +++++++++++------- 3 files changed, 135 insertions(+), 59 deletions(-) diff --git a/frontend-web/src/pages/AdminUserLeaderboardPage.tsx b/frontend-web/src/pages/AdminUserLeaderboardPage.tsx index cfd7a5d7..412f0b3c 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], @@ -1083,6 +1088,19 @@ export default function AdminUserLeaderboardPage() { const overview = data?.overview; 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 +1429,14 @@ export default function AdminUserLeaderboardPage() {
{isLoading && ( -
- - 加载统计数据 +
+
+ + 正在统计排行榜数据 +
+
+ 线上用户较多或时间范围较大时，统计时间可能较长，请耐心等待。为避免影响其他接口，系统会限制排行榜统计的并发数量。 +
)} @@ -1492,11 +1515,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..bdb020d2 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/admin_users.py +++ b/offline-backend-20260512/backend/app/gateway/routers/admin_users.py @@ -13,8 +13,10 @@ methods' ``user_id`` parameter is designed for. from __future__ import annotations +import asyncio import json import logging +import os import re from collections.abc import AsyncIterator from difflib import SequenceMatcher @@ -53,6 +55,20 @@ 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_semaphore = asyncio.Semaphore(_LEADERBOARD_MAX_CONCURRENT) + async def _require_admin(request: Request) -> None: """Reject non-admin callers. In no-auth mode there is no user → allow.""" @@ -983,59 +999,84 @@ 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 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, - "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, - } - 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) + 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 - 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, - ) + 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( From db1b32612b8de48bbb4decc112064cb3877e47ec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=89=E9=91=AB=E5=8D=87?= Date: Fri, 5 Jun 2026 21:15:20 +0800 Subject: [PATCH 2/7] fix: run leaderboard stats in background --- frontend-web/src/core/admin/api.ts | 3 + frontend-web/src/core/admin/hooks.ts | 4 + .../src/pages/AdminUserLeaderboardPage.tsx | 26 +- .../app/gateway/routers/admin_users.py | 237 ++++++++++++++++++ 4 files changed, 269 insertions(+), 1 deletion(-) 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 412f0b3c..0122e7b2 100644 --- a/frontend-web/src/pages/AdminUserLeaderboardPage.tsx +++ b/frontend-web/src/pages/AdminUserLeaderboardPage.tsx @@ -1087,6 +1087,9 @@ 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(() => { @@ -1446,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)}} +
+ )}
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 bdb020d2..829361d2 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/admin_users.py +++ b/offline-backend-20260512/backend/app/gateway/routers/admin_users.py @@ -18,6 +18,7 @@ import json import logging import os import re +import time from collections.abc import AsyncIterator from difflib import SequenceMatcher from hashlib import sha256 @@ -67,7 +68,14 @@ _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: @@ -154,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) @@ -981,6 +992,169 @@ async def _query_overview( ) +def _empty_leaderboard_response( + since_dt: datetime, + until_dt: datetime, + range_days: int, + *, + status: str, + message: str, +) -> AdminLeaderboardResponse: + return AdminLeaderboardResponse( + range_days=range_days, + since=_iso(since_dt), + until=_iso(until_dt), + status=status, + message=message, + overview=AdminLeaderboardOverview(), + ) + + +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, + "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, + } + 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), + 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], + 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, + ) + + +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, @@ -999,6 +1173,69 @@ 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 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(), From a41b5a6d4ce776852c925c356a8ebc4aac72a22e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=89=E9=91=AB=E5=8D=87?= Date: Fri, 5 Jun 2026 21:45:21 +0800 Subject: [PATCH 3/7] fix: improve agent handoff and leaderboard scheduling --- .../workspace/skills/skill-detail-dialog.tsx | 15 ++---- frontend-web/src/pages/AgentChatPage.tsx | 11 ++-- frontend-web/src/pages/ChatPage.tsx | 51 +++++++++++++++---- .../app/gateway/routers/admin_users.py | 24 +++++++++ 4 files changed, 75 insertions(+), 26 deletions(-) diff --git a/frontend-web/src/components/workspace/skills/skill-detail-dialog.tsx b/frontend-web/src/components/workspace/skills/skill-detail-dialog.tsx index 3dbe16ce..94148451 100644 --- a/frontend-web/src/components/workspace/skills/skill-detail-dialog.tsx +++ b/frontend-web/src/components/workspace/skills/skill-detail-dialog.tsx @@ -133,8 +133,7 @@ export function SkillDetailDialog({ return stripYamlFrontmatter(contentData?.content ?? ""); }, [contentData]); - const defaultTab: "detail" | "source" = - skill && !skill.detail ? "source" : "detail"; + const defaultTab: "detail" = "detail"; const selectedTab = activeTab ?? defaultTab; function startEditing() { @@ -234,10 +233,6 @@ export function SkillDetailDialog({ {tx.filesSectionTitle} )} - - - {tx.contentSectionTitle} -
@@ -337,13 +332,9 @@ export function SkillDetailDialog({

{tx.detailEmpty}

- {isOwner ? ( -

- Open the SKILL.md tab to view the skill source. -

- ) : ( + {isOwner && canManage ? (

{tx.detailAdminOnlyHint}

- )} + ) : null}
)} diff --git a/frontend-web/src/pages/AgentChatPage.tsx b/frontend-web/src/pages/AgentChatPage.tsx index 061c2ef1..3a834120 100644 --- a/frontend-web/src/pages/AgentChatPage.tsx +++ b/frontend-web/src/pages/AgentChatPage.tsx @@ -131,8 +131,8 @@ export default function AgentChatPage() { const pendingMessageHandled = useRef(false); useEffect(() => { if (pendingMessageHandled.current) return; - if (!agent_id || !isNewThread) return; - let pending: { agentId?: string; text?: string } | null = null; + if (!agent_id) return; + let pending: { agentId?: string; threadId?: string; text?: string } | null = null; try { const raw = sessionStorage.getItem("pendingAgentMessage"); if (raw) pending = JSON.parse(raw); @@ -140,10 +140,15 @@ export default function AgentChatPage() { pending = null; } if (!pending || pending.agentId !== agent_id || !pending.text) return; + if (pending.threadId) { + if (pending.threadId !== threadId) return; + } else if (!isNewThread) { + return; + } pendingMessageHandled.current = true; sessionStorage.removeItem("pendingAgentMessage"); handleSubmit({ text: pending.text, files: [] }); - }, [agent_id, isNewThread, handleSubmit]); + }, [agent_id, threadId, isNewThread, handleSubmit]); const handleClarificationSubmit = useCallback( (answer: string) => { diff --git a/frontend-web/src/pages/ChatPage.tsx b/frontend-web/src/pages/ChatPage.tsx index fd3f34cd..f96db7f6 100644 --- a/frontend-web/src/pages/ChatPage.tsx +++ b/frontend-web/src/pages/ChatPage.tsx @@ -1,9 +1,11 @@ "use client"; import { useRouter } from "next/navigation"; -import { useCallback, useEffect, useMemo, useState } from "react"; +import { useCallback, useEffect, useMemo, useRef, useState } from "react"; +import { toast } from "sonner"; import { type PromptInputMessage } from "@/components/ai-elements/prompt-input"; +import { getAPIClient } from "@/core/api"; import { usePromptPrefixSettings } from "@/core/system-settings/hooks"; import { useUserPreferencesQuery } from "@/core/user-preferences"; import { ArtifactTrigger, useArtifacts } from "@/components/workspace/artifacts"; @@ -58,6 +60,8 @@ export default function ChatPage() { const hideAdminPrompt = userPreferences?.hide_admin_prompt_prefix === true; const [promptPrefixOn, setPromptPrefixOn] = useState(false); const [selectedAgentId, setSelectedAgentId] = useState(null); + const [agentHandoffPending, setAgentHandoffPending] = useState(false); + const agentHandoffPendingRef = useRef(false); const { agents: selectorAgents, isLoading: selectorLoading } = useAgentSelector(); const { selectedArtifact } = useArtifacts(); @@ -148,15 +152,37 @@ export default function ChatPage() { message.text && (!message.files || message.files.length === 0) ) { - try { - sessionStorage.setItem( - "pendingAgentMessage", - JSON.stringify({ agentId: selectedAgentId, text: message.text }), - ); - } catch { - // sessionStorage may be unavailable in private mode; fall through. - } - router.push(`/page/workspace/agents/${selectedAgentId}/chats/new`); + if (agentHandoffPendingRef.current) return; + const text = message.text.trim(); + agentHandoffPendingRef.current = true; + setAgentHandoffPending(true); + setHasStartedConversation(true); + void (async () => { + try { + const created = await getAPIClient(isMock).threads.create({ + metadata: { agent_id: selectedAgentId }, + }); + sessionStorage.setItem( + "pendingAgentMessage", + JSON.stringify({ + agentId: selectedAgentId, + threadId: created.thread_id, + text, + }), + ); + saveLastAgentId(selectedAgentId); + router.push( + `/page/workspace/agents/${selectedAgentId}/chats/${created.thread_id}`, + ); + } catch (err) { + agentHandoffPendingRef.current = false; + setAgentHandoffPending(false); + setHasStartedConversation(false); + toast.error( + err instanceof Error ? err.message : "创建智能体对话失败", + ); + } + })(); return; } @@ -200,6 +226,8 @@ export default function ChatPage() { hideAdminPrompt, preferencesLoading, selectedAgentId, + agentHandoffPending, + isMock, router, settings.context.writing_mode, writingArtifactPath, @@ -345,7 +373,8 @@ export default function ChatPage() { context={settings.context} disabled={ env.NEXT_PUBLIC_STATIC_WEBSITE_ONLY === "true" || - isUploading + isUploading || + agentHandoffPending } onContextChange={(context) => setSettings("context", context) 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 829361d2..516b5ab8 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/admin_users.py +++ b/offline-backend-20260512/backend/app/gateway/routers/admin_users.py @@ -72,6 +72,14 @@ _LEADERBOARD_CACHE_TTL_SECONDS = max( 5.0, float(os.getenv("ADMIN_LEADERBOARD_CACHE_TTL_SECONDS", "300")), ) +_LEADERBOARD_QUERY_PAUSE_SECONDS = max( + 0.0, + float(os.getenv("ADMIN_LEADERBOARD_QUERY_PAUSE_SECONDS", "0.05")), +) +_LEADERBOARD_START_DELAY_SECONDS = max( + 0.0, + float(os.getenv("ADMIN_LEADERBOARD_START_DELAY_SECONDS", "0.2")), +) _leaderboard_semaphore = asyncio.Semaphore(_LEADERBOARD_MAX_CONCURRENT) _leaderboard_jobs: dict[str, asyncio.Task] = {} _leaderboard_cache: dict[str, dict] = {} @@ -1035,6 +1043,11 @@ def _leaderboard_cache_key( return sha256(json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest() +async def _leaderboard_pause() -> None: + if _LEADERBOARD_QUERY_PAUSE_SECONDS > 0: + await asyncio.sleep(_LEADERBOARD_QUERY_PAUSE_SECONDS) + + async def _build_user_leaderboard_response( session_factory, settings, @@ -1064,16 +1077,25 @@ async def _build_user_leaderboard_response( "cleanup_system_questions": settings.cleanup_system_questions, } overview = await _query_overview(session, since_dt, until_dt, **common) + await _leaderboard_pause() users = await _query_user_leaderboard(session, since_dt, until_dt, **common) + await _leaderboard_pause() skills = await _query_call_rankings(session, since_dt, until_dt, user_id=None, field="skill", limit=limit, **tool_common) + await _leaderboard_pause() tools = await _query_call_rankings(session, since_dt, until_dt, user_id=None, field="tool", limit=limit, **tool_common) + await _leaderboard_pause() 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) + await _leaderboard_pause() agents = await _query_run_dimensions(session, since_dt, until_dt, user_id=None, field="agent", limit=limit, **common) + await _leaderboard_pause() run_statuses = await _query_run_dimensions(session, since_dt, until_dt, user_id=None, field="status", limit=limit, **common) + await _leaderboard_pause() recent_questions = await _query_recent_questions(session, since_dt, until_dt, user_id=None, limit=limit, **question_common) + await _leaderboard_pause() frequent_questions = await _query_frequent_questions(session, since_dt, until_dt, limit=limit, **question_common) + await _leaderboard_pause() trends = await _query_trends(session, since_dt, until_dt, **common) return AdminLeaderboardResponse( @@ -1113,6 +1135,8 @@ async def _refresh_leaderboard_cache( ) -> None: acquired = False try: + if _LEADERBOARD_START_DELAY_SECONDS > 0: + await asyncio.sleep(_LEADERBOARD_START_DELAY_SECONDS) await _leaderboard_semaphore.acquire() acquired = True response = await asyncio.wait_for( From e5ba898fc78b84b9e103cc4fae4f600a020e742d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=89=E9=91=AB=E5=8D=87?= Date: Fri, 5 Jun 2026 22:11:31 +0800 Subject: [PATCH 4/7] hide user leaderboard entry --- frontend-web/src/components/page-sidebar.tsx | 9 ---- .../workspace/workspace-nav-menu.tsx | 53 ++++++++----------- .../src/core/page-layout/sidebar-menu.ts | 8 --- 3 files changed, 21 insertions(+), 49 deletions(-) diff --git a/frontend-web/src/components/page-sidebar.tsx b/frontend-web/src/components/page-sidebar.tsx index 5cebcc9f..10512957 100644 --- a/frontend-web/src/components/page-sidebar.tsx +++ b/frontend-web/src/components/page-sidebar.tsx @@ -8,7 +8,6 @@ import { Folder, FolderOpen, Settings, - Trophy, User, Wrench, } from "lucide-react"; @@ -261,14 +260,6 @@ export function PageSidebar({ {t.settings.sections.tools} - {isAdmin && ( - navigate("/page/strategy/admin/user-leaderboard")} - > - - 用户排行榜 - - )} openSettings("account")}> 个人信息 diff --git a/frontend-web/src/components/workspace/workspace-nav-menu.tsx b/frontend-web/src/components/workspace/workspace-nav-menu.tsx index eaf72024..29689e91 100644 --- a/frontend-web/src/components/workspace/workspace-nav-menu.tsx +++ b/frontend-web/src/components/workspace/workspace-nav-menu.tsx @@ -4,10 +4,10 @@ import { BellIcon, BotIcon, BrainIcon, - CalendarClockIcon, - ChevronsUpDown, - DatabaseZapIcon, - GaugeIcon, + CalendarClockIcon, + ChevronsUpDown, + DatabaseZapIcon, + GaugeIcon, HistoryIcon, MessageSquareCodeIcon, MessageSquarePlusIcon, @@ -16,10 +16,9 @@ import { SettingsIcon, ShieldCheckIcon, SparklesIcon, - TagsIcon, - TrophyIcon, - UsersIcon, -} from "lucide-react"; + TagsIcon, + UsersIcon, +} from "lucide-react"; import { useEffect, useState } from "react"; import { @@ -189,30 +188,20 @@ export function WorkspaceNavMenu() { 大模型调用统计 )} - {isAdmin && ( - - navigate("/page/strategy/admin/user-leaderboard") - } - > - - 用户排行榜 - - )} - {isAdmin && ( - - navigate("/page/workspace/settings/checkpoint-migration") - } - > - - Checkpoint 迁移 - - )} - {isAdmin && ( - navigate("/page/workspace/admin")} - > + {isAdmin && ( + + navigate("/page/workspace/settings/checkpoint-migration") + } + > + + Checkpoint 迁移 + + )} + {isAdmin && ( + navigate("/page/workspace/admin")} + > 用户中心 diff --git a/frontend-web/src/core/page-layout/sidebar-menu.ts b/frontend-web/src/core/page-layout/sidebar-menu.ts index 1a52e92f..b06ccd8b 100644 --- a/frontend-web/src/core/page-layout/sidebar-menu.ts +++ b/frontend-web/src/core/page-layout/sidebar-menu.ts @@ -35,7 +35,6 @@ import { ShieldCheck, Sparkles, Target, - Trophy, Users, } from "lucide-react"; @@ -244,13 +243,6 @@ export const sidebarMenuItems: MenuItem[] = [ path: "/page/strategy/settings/llm-metrics", adminOnly: true, }, - { - id: "mgmt-user-leaderboard", - label: "用户排行榜", - icon: Trophy, - path: "/page/strategy/admin/user-leaderboard", - adminOnly: true, - }, { id: "mgmt-users", label: "用户管理", From 4df71f32a838d231e2c9ef4048e9d50938f5b101 Mon Sep 17 00:00:00 2001 From: Administrator Date: Fri, 5 Jun 2026 22:44:07 +0800 Subject: [PATCH 5/7] fix: use sqlite backend config --- offline-backend-20260512/backend/config.yaml | 24 +++++++++++--------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/offline-backend-20260512/backend/config.yaml b/offline-backend-20260512/backend/config.yaml index 621f301f..fae7f00d 100644 --- a/offline-backend-20260512/backend/config.yaml +++ b/offline-backend-20260512/backend/config.yaml @@ -90,6 +90,16 @@ models: temperature: 1.0 supports_vision: true supports_thinking: true + - name: Qwen/Qwen3.6-27B + display_name: Qwen/Qwen3.6-27B + use: langchain_openai:ChatOpenAI + model: Qwen/Qwen3.6-27B + api_key: sk-ighylafootabtuthwueirmoycccyqtdgbifyhozduvolkdyt + base_url: https://api.siliconflow.cn/v1/ + max_tokens: 4096 + temperature: 1.0 + supports_vision: true + supports_thinking: true # 识图(图片识别)配置。 # - model_name:指定一个 models[] 中的模型名,作为“默认识图大模型”。view_image 工具会把 # 图片交给该模型识别,并把识别出的文字结果返回给主对话模型 —— 这样即使主对话模型不支持 @@ -213,21 +223,13 @@ auth_login: database: backend: sqlite - sqlite_dir: .deer-flow/data + sqlite_dir: /app/backend/.deer-flow/data pool_size: 10 echo_sql: false checkpointer: - type: postgres - # true: 新 checkpoint 写入 Postgres；false: 保留 PG 配置但继续使用 SQLite。 - # 也可临时用环境变量 DEER_FLOW_USE_PG_CHECKPOINTER=0 覆盖。 - use_postgres: true - connection_string: postgresql://postgres:postgres123@47.88.25.99:8432/deerflow - sqlite_connection_string: .deer-flow/data/deerflow.db - dual_read_enabled: true - dual_read_sqlite_paths: - - .deer-flow/data/deerflow.db - - ../data/.deer-flow/data/checkpoints.db + type: sqlite + connection_string: /app/backend/.deer-flow/data/deerflow.db # 记忆子系统:跨会话持久化用户画像与 agent 私有记忆。 # 三层覆盖优先级(高->低): From a1bcc6eed4d654950f66ce43da5ab572bc4e7c32 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Fri, 5 Jun 2026 23:45:28 +0800 Subject: [PATCH 6/7] =?UTF-8?q?feat:=20=E9=87=8D=E6=96=B0=E8=AE=BE?= =?UTF-8?q?=E8=AE=A1=E6=B5=8F=E8=A7=88=E5=99=A8=E6=A0=87=E7=AD=BE=E9=A1=B5?= =?UTF-8?q?=E5=9B=BE=E6=A0=87=EF=BC=88=E6=99=BA=E8=83=BD=E5=8A=A9=E6=89=8B?= =?UTF-8?q?=E4=B8=AD=E5=BF=83=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 玫红渐变磁贴 + 白色对话气泡 + 气泡内 AI 闪光星 + 角落小闪光，围绕「助手 / 多种智能工具」主题；并在 index.html 补上 favicon 链接。 Co-Authored-By: Claude Opus 4.8 (1M context) --- frontend-web/index.html | 3 ++- frontend-web/public/favicon.svg | 29 +++++++++++++++++++++++++++++ 2 files changed, 31 insertions(+), 1 deletion(-) create mode 100644 frontend-web/public/favicon.svg diff --git a/frontend-web/index.html b/frontend-web/index.html index f6fe9779..65c7d4e3 100644 --- a/frontend-web/index.html +++ b/frontend-web/index.html @@ -13,7 +13,8 @@ } catch (e) {} })(); - + + CM 助手 diff --git a/frontend-web/public/favicon.svg b/frontend-web/public/favicon.svg new file mode 100644 index 00000000..f708efd6 --- /dev/null +++ b/frontend-web/public/favicon.svg @@ -0,0 +1,29 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + From e7c745a51293f4bf6c138295504b504e73d9d867 Mon Sep 17 00:00:00 2001 From: Administrator Date: Sat, 6 Jun 2026 00:51:43 +0800 Subject: [PATCH 7/7] =?UTF-8?q?feat:=20=E6=8E=92=E8=A1=8C=E6=A6=9C?= =?UTF-8?q?=E6=97=A5=E5=BF=AB=E7=85=A7=E5=88=86=E6=9E=90=20+=20=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=BA=93=E8=BF=9E=E6=8E=A5=E6=B1=A0/=E7=B4=A2?= =?UTF-8?q?=E5=BC=95=E8=B0=83=E4=BC=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 排行榜改为按天预计算快照(admin_leaderboard_daily_stats),请求只读取合并,重统计移至后台调度;过去日不可变、当天 TTL 刷新,避免在线聚合长时间占用数据库 - 为 runs/tool_call_metrics/llm_call_metrics/threads_meta 增加面向时间段+用户聚合的复合索引(model __table_args__ + migration 20260608_01) - 连接池可调:DatabaseConfig 新增 max_overflow / pool_timeout,接入 postgres/mysql 引擎;config.yaml 上调 pool_size=20/max_overflow=20 - 新增手动执行的索引脚本 scripts/sql/leaderboard_indexes_mysql.sql(幂等,适用于已存在数据、create_all 不补索引且 alembic 关闭的部署) Co-Authored-By: Claude Opus 4.8 (1M context) --- .../workspace/agents/agent-card.tsx | 36 +- .../settings/skill-settings-page.tsx | 36 +- .../workspace/tags/tag-picker-dialog.tsx | 6 +- .../workspace/workspace-nav-menu.tsx | 14 +- frontend-web/src/core/admin/api.ts | 38 + frontend-web/src/core/admin/hooks.ts | 12 + .../src/pages/AdminUserActivityPage.tsx | 194 ++- .../src/pages/ScheduledTaskRunDetailPage.tsx | 108 +- frontend-web/src/pages/ScheduledTasksPage.tsx | 28 +- frontend-web/src/pages/StrategyRoutes.tsx | 8 +- frontend-web/src/pages/WorkspaceRoutes.tsx | 10 +- .../backend/app/gateway/app.py | 286 +++- .../app/gateway/routers/admin_users.py | 1209 +++++++++++++---- .../backend/app/gateway/routers/tags.py | 30 +- offline-backend-20260512/backend/config.yaml | 4 +- .../deerflow/config/database_config.py | 108 +- .../persistence/admin_stats/__init__.py | 5 + .../deerflow/persistence/admin_stats/model.py | 31 + .../harness/deerflow/persistence/engine.py | 178 +-- .../deerflow/persistence/llm_metrics/model.py | 6 +- ...260608_01_admin_leaderboard_daily_stats.py | 85 ++ .../deerflow/persistence/models/__init__.py | 2 + .../harness/deerflow/persistence/run/model.py | 9 +- .../deerflow/persistence/thread_meta/model.py | 10 +- .../persistence/tool_metrics/model.py | 6 +- .../scripts/sql/leaderboard_indexes_mysql.sql | 85 ++ 26 files changed, 1980 insertions(+), 564 deletions(-) create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/admin_stats/__init__.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/admin_stats/model.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260608_01_admin_leaderboard_daily_stats.py create mode 100644 offline-backend-20260512/backend/scripts/sql/leaderboard_indexes_mysql.sql diff --git a/frontend-web/src/components/workspace/agents/agent-card.tsx b/frontend-web/src/components/workspace/agents/agent-card.tsx index c60e2084..60be77f8 100644 --- a/frontend-web/src/components/workspace/agents/agent-card.tsx +++ b/frontend-web/src/components/workspace/agents/agent-card.tsx @@ -585,17 +585,15 @@ export function AgentCard({ agent, layout = "card" }: AgentCardProps) { )} - {isAdmin && ( - - )} + {isOwned && ( - )} + {canManage && (
{!userId && ( @@ -229,6 +246,8 @@ export default function AdminUserActivityPage() { type="button" onClick={() => { setThreadId(thread.thread_id); + setQaQuery(thread.thread_id); + setQaPage(1); setTab("qa"); }} className={cn( @@ -270,28 +289,63 @@ export default function AdminUserActivityPage() { {tab === "qa" && ( -
- {!threadId && ( -
- - 请从中间列选择一个会话 +
+
+
+
+ + { + setQaQuery(event.target.value); + setThreadId(null); + setQaPage(1); + }} + placeholder="搜索问题、回答、运行 ID 或会话 ID" + className="pl-9" + /> +
+ {qaQuery && ( + + )}
- )} - {threadId && messagesLoading && ( +
+ 共 {fmtNumber(qaData?.total ?? 0)} 条问答 + {threadId && 已按会话过滤} + {qaFetching && !qaLoading && ( + + 更新中 + + )} +
+
+ + {qaLoading && (
加载中
)} - {threadId && !messagesLoading && ( + {!qaLoading && (qaData?.items.length ?? 0) === 0 && ( +
+ + 暂无匹配的问答记录 +
+ )} + {!qaLoading && (qaData?.items.length ?? 0) > 0 && (
- {(messages ?? []).map((m, i) => ( - + {(qaData?.items ?? []).map((item) => ( + ))} - {(messages ?? []).every((m) => !roleOf(m)) && ( -
- 该会话暂无问答记录 -
- )} +
)}
diff --git a/frontend-web/src/pages/ScheduledTaskRunDetailPage.tsx b/frontend-web/src/pages/ScheduledTaskRunDetailPage.tsx index 2f9cad01..9002fbcb 100644 --- a/frontend-web/src/pages/ScheduledTaskRunDetailPage.tsx +++ b/frontend-web/src/pages/ScheduledTaskRunDetailPage.tsx @@ -1,7 +1,7 @@ "use client"; import { useCallback, useEffect, useMemo, useState } from "react"; -import { Activity, ChevronDown, ChevronLeft, Download, Edit3, FileText, Heart, Loader2, Save, Code2, Trash2, X } from "lucide-react"; +import { Activity, ChevronDown, ChevronLeft, Download, Edit3, FileText, Heart, Loader2, Maximize2, Save, Code2, Trash2, X } from "lucide-react"; import { useParams, useSearchParams } from "react-router-dom"; import { useRouteNavigate as useNavigate } from "@/core/route-base"; import { toast } from "sonner"; @@ -9,6 +9,7 @@ import { toast } from "sonner"; import { Badge } from "@/components/ui/badge"; import { Button } from "@/components/ui/button"; import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card"; +import { Dialog, DialogContent, DialogHeader, DialogTitle } from "@/components/ui/dialog"; import { ScrollArea } from "@/components/ui/scroll-area"; import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs"; import { Textarea } from "@/components/ui/textarea"; @@ -101,6 +102,32 @@ function RenderPreview({ file }: { file: ScheduledTaskResultFileContent }) { return 
{file.content}
; } +type FullscreenPreview = + | { mode: "html"; title: string; html: string } + | { mode: "markdown"; title: string; content: string } + | { mode: "text"; title: string; content: string } + | { mode: "file"; title: string; file: ScheduledTaskResultFileContent }; + +function FullscreenPreviewContent({ preview }: { preview: FullscreenPreview }) { + if (preview.mode === "html") { + return ; + } + if (preview.mode === "markdown") { + return ( + + ); + } + if (preview.mode === "file") { + return ; + } + return 
{preview.content || "无正文内容"}
; +} + type EditState = { draft: string; saving: boolean }; /** Flatten a LangChain-style message content (string | parts | nested) to text. */ @@ -218,6 +245,7 @@ export default function ScheduledTaskRunDetailPage() { const [filesLoading, setFilesLoading] = useState(false); const [showMeta, setShowMeta] = useState(false); const [edits, setEdits] = useState>({}); + const [fullscreenPreview, setFullscreenPreview] = useState(null); const canEditRun = run?.can_edit === true; const canDeleteRun = canEditRun; @@ -428,7 +456,27 @@ export default function ScheduledTaskRunDetailPage() { }; return ( - + <> + !open && setFullscreenPreview(null)}> + + + + {fullscreenPreview?.title ?? "预览"} + + + {fullscreenPreview && ( +
+ +
+ )} + + +
@@ -544,6 +592,23 @@ export default function ScheduledTaskRunDetailPage() { {isHtmlRun && (
+ +
{result.format === "markdown" || result.format === "mixed" ? (
+ {file.previewable && !isEditing && ( + + )} {isMarkdown && (
- + + ); } diff --git a/frontend-web/src/pages/ScheduledTasksPage.tsx b/frontend-web/src/pages/ScheduledTasksPage.tsx index 418f807e..6a527443 100644 --- a/frontend-web/src/pages/ScheduledTasksPage.tsx +++ b/frontend-web/src/pages/ScheduledTasksPage.tsx @@ -45,7 +45,6 @@ import { TagChipList } from "@/components/workspace/tags/tag-chip"; import { TagPickerDialog } from "@/components/workspace/tags/tag-picker-dialog"; import { TagSearchBar } from "@/components/workspace/tags/tag-search-bar"; import { useAgents } from "@/core/agents"; -import { getStoredAuth } from "@/core/auth"; import { useModels } from "@/core/models/hooks"; import { downloadScheduledTaskRun, @@ -188,7 +187,6 @@ export default function ScheduledTasksPage() { const [subscribeTarget, setSubscribeTarget] = useState(null); const [subscribeNotify, setSubscribeNotify] = useState(false); const [tagPickerTask, setTagPickerTask] = useState(null); - const isAdmin = useMemo(() => getStoredAuth()?.system_role === "admin", []); const sortedTasks = useMemo( () => @@ -552,19 +550,17 @@ export default function ScheduledTasksPage() { )}
- {isAdmin && ( - - )} + {isOwner ? ( <>
diff --git a/frontend-web/src/pages/WorkspaceRoutes.tsx b/frontend-web/src/pages/WorkspaceRoutes.tsx index 923eb137..2f2a40c0 100644 --- a/frontend-web/src/pages/WorkspaceRoutes.tsx +++ b/frontend-web/src/pages/WorkspaceRoutes.tsx @@ -420,13 +420,9 @@ export default function WorkspaceRoutes({ - - - ) : ( - - ) + + + } /> None: pass -async def _curator_scheduler_loop() -> None: +async def _curator_scheduler_loop() -> None: """常驻后台任务:每隔 _CURATOR_CHECK_INTERVAL_SECONDS 检查一次需不需要进化。 多 worker 部署下用文件锁做 leader 选举 —— 只有抢到锁的那个 worker 进程真正 @@ -404,11 +404,248 @@ async def _curator_scheduler_loop() -> None: except Exception: logger.exception("Curator scheduler tick failed (non-fatal)") await asyncio.sleep(_CURATOR_CHECK_INTERVAL_SECONDS) - finally: - _release_singleton_lock(lock_handle) - - -def _warn_if_multi_worker() -> None: + finally: + _release_singleton_lock(lock_handle) + + +def _env_bool(name: str, default: bool) -> bool: + raw = os.getenv(name) + if raw is None: + return default + return raw.strip().lower() not in {"0", "false", "no", "off"} + + +def _env_int(name: str, default: int, *, alias: str | None = None, minimum: int = 1) -> int: + raw = os.getenv(name) + if raw is None and alias: + raw = os.getenv(alias) + if raw is None: + raw = str(default) + try: + value = int(raw) + except ValueError: + return default + return max(minimum, value) + + +def _env_float(name: str, default: float, *, alias: str | None = None, minimum: float = 0.0) -> float: + raw = os.getenv(name) + if raw is None and alias: + raw = os.getenv(alias) + if raw is None: + raw = str(default) + try: + value = float(raw) + except ValueError: + return default + return max(minimum, value) + + +_LEADERBOARD_SNAPSHOT_BACKGROUND_ENABLED = _env_bool("ADMIN_LEADERBOARD_BACKGROUND_ENABLED", True) +_LEADERBOARD_SNAPSHOT_PREWARM_DAYS = _env_int( + "ADMIN_LEADERBOARD_BACKGROUND_PREWARM_DAYS", + 30, +) +_LEADERBOARD_SNAPSHOT_BATCH_SIZE = _env_int( + "ADMIN_LEADERBOARD_BACKGROUND_BATCH_SIZE", + 1, +) +_LEADERBOARD_SNAPSHOT_INTERVAL_SECONDS = _env_float( + "ADMIN_LEADERBOARD_BACKGROUND_INTERVAL_SECONDS", + 300.0, +) +_LEADERBOARD_SNAPSHOT_SLEEP_SECONDS = _env_float( + "ADMIN_LEADERBOARD_BACKGROUND_SLEEP_SECONDS", + 2.0, +) +_LEADERBOARD_SNAPSHOT_RUNNING_TTL_SECONDS = _env_int( + "ADMIN_LEADERBOARD_BACKGROUND_RUNNING_TTL_SECONDS", + 1800, +) +_LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS = _env_float( + "ADMIN_LEADERBOARD_BACKGROUND_STARTUP_DELAY_SECONDS", + 60.0, +) + + +def _recent_leaderboard_stat_dates(days: int) -> list[str]: + from datetime import datetime, timedelta + + from deerflow.persistence.types import BEIJING_TZ + + today = datetime.now(BEIJING_TZ).date() + start = today - timedelta(days=max(1, days) - 1) + return [(start + timedelta(days=i)).isoformat() for i in range((today - start).days + 1)] + + +async def _leaderboard_snapshot_due_dates( + session_factory, + *, + settings_hash: str, + limit: int, + running_ttl_seconds: int, +) -> list[str]: + from datetime import UTC, datetime, timedelta + + from sqlalchemy import and_, or_, select + + from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow + + cutoff = datetime.now(UTC) - timedelta(seconds=running_ttl_seconds) + async with session_factory() as session: + rows = ( + await session.execute( + select(AdminLeaderboardDailyStatRow.stat_date) + .where( + AdminLeaderboardDailyStatRow.settings_hash == settings_hash, + or_( + AdminLeaderboardDailyStatRow.status.in_(["queued", "error"]), + and_( + AdminLeaderboardDailyStatRow.status == "running", + AdminLeaderboardDailyStatRow.updated_at < cutoff, + ), + ), + ) + .order_by(AdminLeaderboardDailyStatRow.stat_date.desc()) + .limit(limit) + ) + ).all() + return [str(row[0]) for row in rows] + + +async def _claim_leaderboard_snapshot( + session_factory, + *, + stat_date: str, + settings_hash: str, + running_ttl_seconds: int, +) -> bool: + from datetime import UTC, datetime + + from app.gateway.routers.admin_users import _snapshot_is_fresh + from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow + from deerflow.persistence.types import BEIJING_TZ + + async with session_factory() as session: + row = await session.get( + AdminLeaderboardDailyStatRow, + {"stat_date": stat_date, "settings_hash": settings_hash}, + ) + if row is None or _snapshot_is_fresh(row, stat_date): + return False + if row.status == "running": + updated_at = row.updated_at + if isinstance(updated_at, datetime): + if updated_at.tzinfo is None: + updated_at = updated_at.replace(tzinfo=BEIJING_TZ) + age = (datetime.now(BEIJING_TZ) - updated_at.astimezone(BEIJING_TZ)).total_seconds() + if age < running_ttl_seconds: + return False + row.status = "running" + row.message = "排行榜后台统计任务已认领。" + row.error = None + row.updated_at = datetime.now(UTC) + await session.commit() + return True + + +async def _run_leaderboard_snapshot_tick() -> tuple[int, int]: + from app.gateway.routers.admin_users import ( + _daily_job_key, + _queue_daily_leaderboard_snapshots, + _refresh_daily_leaderboard_snapshot, + _resolve_leaderboard_snapshot_scope, + ) + from deerflow.config.system_settings import load_system_settings + from deerflow.persistence.engine import get_session_factory + + session_factory = get_session_factory() + if session_factory is None: + return 0, 0 + + settings = load_system_settings().leaderboard + excluded, include_scheduled, include_admins, include_failed, settings_hash = _resolve_leaderboard_snapshot_scope(settings) + stat_dates = _recent_leaderboard_stat_dates(_LEADERBOARD_SNAPSHOT_PREWARM_DAYS) + queued_dates = await _queue_daily_leaderboard_snapshots( + session_factory, + stat_dates, + settings_hash, + message="排行榜后台统计任务定期预热/补算队列。", + ) + + due_dates = await _leaderboard_snapshot_due_dates( + session_factory, + settings_hash=settings_hash, + limit=_LEADERBOARD_SNAPSHOT_BATCH_SIZE, + running_ttl_seconds=_LEADERBOARD_SNAPSHOT_RUNNING_TTL_SECONDS, + ) + processed = 0 + for stat_date in due_dates: + claimed = await _claim_leaderboard_snapshot( + session_factory, + stat_date=stat_date, + settings_hash=settings_hash, + running_ttl_seconds=_LEADERBOARD_SNAPSHOT_RUNNING_TTL_SECONDS, + ) + if not claimed: + continue + logger.info("Leaderboard snapshot scheduler: building date=%s hash=%s", stat_date, settings_hash[:10]) + await _refresh_daily_leaderboard_snapshot( + _daily_job_key(stat_date, settings_hash), + session_factory=session_factory, + settings=settings, + stat_date=stat_date, + settings_hash=settings_hash, + excluded=excluded, + include_scheduled=include_scheduled, + include_admins=include_admins, + include_failed=include_failed, + ) + processed += 1 + if _LEADERBOARD_SNAPSHOT_SLEEP_SECONDS > 0 and processed < len(due_dates): + await asyncio.sleep(_LEADERBOARD_SNAPSHOT_SLEEP_SECONDS) + return processed, len(queued_dates) + + +async def _leaderboard_snapshot_scheduler_loop() -> None: + """Background task inside the existing gateway process for daily snapshots.""" + if not _LEADERBOARD_SNAPSHOT_BACKGROUND_ENABLED: + logger.info("Leaderboard snapshot scheduler disabled by ADMIN_LEADERBOARD_BACKGROUND_ENABLED") + return + + from deerflow.config.runtime_paths import runtime_home + + lock_handle = _acquire_singleton_lock(runtime_home() / ".leaderboard_snapshot_scheduler.lock") + if lock_handle is None: + logger.info("Leaderboard snapshot scheduler: another gateway worker holds the lock; this worker skips") + return + + logger.info( + "Leaderboard snapshot scheduler started: prewarm_days=%d batch_size=%d interval=%.1fs", + _LEADERBOARD_SNAPSHOT_PREWARM_DAYS, + _LEADERBOARD_SNAPSHOT_BATCH_SIZE, + _LEADERBOARD_SNAPSHOT_INTERVAL_SECONDS, + ) + try: + if _LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS > 0: + await asyncio.sleep(_LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS) + while True: + try: + processed, queued = await _run_leaderboard_snapshot_tick() + if processed or queued: + logger.info( + "Leaderboard snapshot scheduler tick: queued=%d processed=%d", + queued, + processed, + ) + except Exception: + logger.exception("Leaderboard snapshot scheduler tick failed (non-fatal)") + await asyncio.sleep(_LEADERBOARD_SNAPSHOT_INTERVAL_SECONDS) + finally: + _release_singleton_lock(lock_handle) + + +def _warn_if_multi_worker() -> None: """检测 uvicorn / gunicorn 多 worker 部署并打 ERROR 日志。 主 agent / canvas / AI 写作流式响应在 Gateway 进程内通过 ``StreamBridge`` @@ -562,16 +799,21 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: # 默认每天凌晨 3 点扫一次,删 retention_days(默认 7 天)以前的会话,联动 # 清理 MySQL 业务表 + SQLite checkpointer 里的 thread 状态。多 worker # 下用文件锁选 leader,只有抢到锁的 worker 真正跑清理。 - try: - from app.gateway.ai_writing_cleanup import cleanup_scheduler_loop - - app.state.ai_writing_cleanup_task = asyncio.create_task(cleanup_scheduler_loop(app)) - except Exception: - logger.exception("AI writing cleanup scheduler failed to start (non-fatal)") - - # Start IM channel service if any channels are configured - try: - from app.channels.service import start_channel_service + try: + from app.gateway.ai_writing_cleanup import cleanup_scheduler_loop + + app.state.ai_writing_cleanup_task = asyncio.create_task(cleanup_scheduler_loop(app)) + except Exception: + logger.exception("AI writing cleanup scheduler failed to start (non-fatal)") + + try: + app.state.leaderboard_snapshot_scheduler_task = asyncio.create_task(_leaderboard_snapshot_scheduler_loop()) + except Exception: + logger.exception("Leaderboard snapshot scheduler failed to start (non-fatal)") + + # Start IM channel service if any channels are configured + try: + from app.channels.service import start_channel_service channel_service = await start_channel_service(app.state.config) logger.info("Channel service started: %s", channel_service.get_status()) @@ -615,7 +857,17 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: except asyncio.CancelledError: pass except Exception: - logger.exception("AI writing cleanup scheduler shutdown error") + logger.exception("AI writing cleanup scheduler shutdown error") + + leaderboard_snapshot_task = getattr(app.state, "leaderboard_snapshot_scheduler_task", None) + if leaderboard_snapshot_task is not None: + leaderboard_snapshot_task.cancel() + try: + await leaderboard_snapshot_task + except asyncio.CancelledError: + pass + except Exception: + logger.exception("Leaderboard snapshot scheduler shutdown error") # 阶段 7 删冗余：AI 写作 graph 不再自管 checkpointer / in-flight task； # thread 状态由 LangGraph Server 主 checkpointer 接管，lifespan 出口由 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 516b5ab8..753471ea 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/admin_users.py +++ b/offline-backend-20260512/backend/app/gateway/routers/admin_users.py @@ -18,17 +18,16 @@ import json import logging import os import re -import time from collections.abc import AsyncIterator from difflib import SequenceMatcher from hashlib import sha256 -from datetime import datetime, timedelta +from datetime import UTC, datetime, timedelta, time as datetime_time from fastapi import APIRouter, Depends, HTTPException, Query, Request from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field -from sqlalchemy import and_, case, desc, func, not_, or_, select +from sqlalchemy import and_, case, desc, func, not_, or_, select, text from langchain_core.messages import HumanMessage, SystemMessage from app.gateway.deps import ( @@ -46,11 +45,13 @@ from deerflow.models import create_chat_model 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.admin_stats.model import AdminLeaderboardDailyStatRow 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.types import BEIJING_TZ from deerflow.persistence.user.model import UserRow logger = logging.getLogger(__name__) @@ -64,14 +65,6 @@ _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_QUERY_PAUSE_SECONDS = max( 0.0, float(os.getenv("ADMIN_LEADERBOARD_QUERY_PAUSE_SECONDS", "0.05")), @@ -80,10 +73,32 @@ _LEADERBOARD_START_DELAY_SECONDS = max( 0.0, float(os.getenv("ADMIN_LEADERBOARD_START_DELAY_SECONDS", "0.2")), ) +_LEADERBOARD_DAILY_SNAPSHOT_LIMIT = max( + 100, + int(os.getenv("ADMIN_LEADERBOARD_DAILY_SNAPSHOT_LIMIT", "1000")), +) +_LEADERBOARD_TODAY_TTL_SECONDS = max( + 30.0, + float(os.getenv("ADMIN_LEADERBOARD_TODAY_TTL_SECONDS", "300")), +) +_LEADERBOARD_QUERY_TIMEOUT_MS = max( + 500, + int(os.getenv("ADMIN_LEADERBOARD_QUERY_TIMEOUT_MS", "8000")), +) +_LEADERBOARD_LOCK_WAIT_SECONDS = max( + 1, + int(os.getenv("ADMIN_LEADERBOARD_LOCK_WAIT_SECONDS", "5")), +) +_LEADERBOARD_QUEUE_ON_REQUEST_ENABLED = os.getenv( + "ADMIN_LEADERBOARD_QUEUE_ON_REQUEST_ENABLED", + os.getenv("ADMIN_LEADERBOARD_WEB_QUEUE_ENABLED", "true"), +).strip().lower() not in { + "0", + "false", + "no", + "off", +} _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: @@ -209,6 +224,45 @@ class AdminLeaderboardReportResponse(BaseModel): model_name: str | None = None +class AdminLeaderboardSnapshotStatusItem(BaseModel): + stat_date: str + status: str = "missing" + has_payload: bool = False + generated_at: str = "" + updated_at: str = "" + message: str = "" + error: str = "" + + +class AdminLeaderboardSnapshotStatusResponse(BaseModel): + settings_hash: str + since: str + until: str + total_days: int + ready_days: int + queued_days: int + running_days: int + error_days: int + missing_days: int + items: list[AdminLeaderboardSnapshotStatusItem] = Field(default_factory=list) + + +class AdminLeaderboardSnapshotBackfillRequest(BaseModel): + days: int = Field(default=30, ge=1, le=365) + since: str | None = Field(default=None, description="Inclusive range start, ISO datetime") + until: str | None = Field(default=None, description="Inclusive range end, ISO datetime") + force: bool = Field(default=False, description="Queue dates even when a ready snapshot already exists") + + +class AdminLeaderboardSnapshotBackfillResponse(BaseModel): + settings_hash: str + since: str + until: str + total_days: int + queued_dates: list[str] = Field(default_factory=list) + message: str = "" + + def _now() -> datetime: return datetime.now() @@ -240,6 +294,56 @@ def _resolve_range(days: int, since: str | None, until: str | None) -> tuple[dat return start, end, span_days +def _as_beijing(value: datetime) -> datetime: + if value.tzinfo is None: + value = value.replace(tzinfo=UTC) + return value.astimezone(BEIJING_TZ) + + +def _today_stat_date() -> str: + return datetime.now(BEIJING_TZ).date().isoformat() + + +def _stat_dates_for_range(since: datetime, until: datetime) -> list[str]: + start_date = _as_beijing(since).date() + end_date = _as_beijing(until).date() + if start_date > end_date: + raise HTTPException(status_code=400, detail="since must be before until") + count = min(366, (end_date - start_date).days + 1) + return [(start_date + timedelta(days=i)).isoformat() for i in range(count)] + + +def _bounds_for_stat_date(stat_date: str) -> tuple[datetime, datetime]: + try: + day = datetime.strptime(stat_date, "%Y-%m-%d").date() + except ValueError as exc: + raise HTTPException(status_code=400, detail=f"Invalid stat date: {stat_date}") from exc + start = datetime.combine(day, datetime_time.min).replace(tzinfo=BEIJING_TZ) + end = datetime.combine(day, datetime_time.max).replace(tzinfo=BEIJING_TZ) + return start, end + + +async def _try_set_session_value(session, sql: str) -> None: + try: + await session.execute(text(sql)) + except Exception: + logger.debug("Failed to apply analytics session setting: %s", sql, exc_info=True) + try: + await session.rollback() + except Exception: + logger.debug("Failed to rollback after analytics session setting failure", exc_info=True) + + +async def _set_analytics_read_timeout(session) -> None: + conn = await session.connection() + dialect = conn.dialect.name + if dialect == "mysql": + await _try_set_session_value(session, f"SET SESSION max_execution_time = {_LEADERBOARD_QUERY_TIMEOUT_MS}") + await _try_set_session_value(session, f"SET SESSION innodb_lock_wait_timeout = {_LEADERBOARD_LOCK_WAIT_SECONDS}") + elif dialect == "postgresql": + await _try_set_session_value(session, f"SET LOCAL statement_timeout = {_LEADERBOARD_QUERY_TIMEOUT_MS}") + + def _question_count_expr(): return func.sum(case((RunRow.first_human_message.is_not(None), 1), else_=0)) @@ -1018,31 +1122,6 @@ def _empty_leaderboard_response( ) -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 _leaderboard_pause() -> None: if _LEADERBOARD_QUERY_PAUSE_SECONDS > 0: await asyncio.sleep(_LEADERBOARD_QUERY_PAUSE_SECONDS) @@ -1061,6 +1140,7 @@ async def _build_user_leaderboard_response( include_failed: bool, ) -> AdminLeaderboardResponse: async with session_factory() as session: + await _set_analytics_read_timeout(session) common = { "exclude_user_ids": excluded, "include_scheduled": include_scheduled, @@ -1119,15 +1199,523 @@ async def _build_user_leaderboard_response( ) -async def _refresh_leaderboard_cache( - cache_key: str, +def _leaderboard_daily_settings_hash( *, - session_factory, + excluded: set[str], + include_scheduled: bool, + include_admins: bool, + include_failed: bool, + cleanup_system_questions: bool, +) -> str: + payload = { + "version": 2, + "snapshot_limit": _LEADERBOARD_DAILY_SNAPSHOT_LIMIT, + "exclude_user_ids": 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() + + +def _resolve_leaderboard_snapshot_scope( settings, + *, + exclude_user_ids: list[str] | None = None, + include_scheduled: bool | None = None, + include_admins: bool | None = None, + include_failed: bool | None = None, +) -> tuple[set[str], bool, bool, bool, str]: + 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 + settings_hash = _leaderboard_daily_settings_hash( + excluded=excluded, + include_scheduled=resolved_include_scheduled, + include_admins=resolved_include_admins, + include_failed=resolved_include_failed, + cleanup_system_questions=settings.cleanup_system_questions, + ) + return excluded, resolved_include_scheduled, resolved_include_admins, resolved_include_failed, settings_hash + + +def _daily_job_key(stat_date: str, settings_hash: str) -> str: + return f"{stat_date}:{settings_hash}" + + +def _sort_datetime(value: str | None) -> datetime: + if not value: + return datetime.min + text_value = str(value).strip() + if text_value.endswith("Z"): + text_value = text_value[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text_value) + except ValueError: + return datetime.min + if parsed.tzinfo is not None: + return parsed.astimezone(UTC).replace(tzinfo=None) + return parsed + + +def _latest_iso(left: str | None, right: str | None) -> str: + if not left: + return str(right or "") + if not right: + return str(left or "") + return str(right if _sort_datetime(right) > _sort_datetime(left) else left) + + +def _snapshot_is_fresh(row: AdminLeaderboardDailyStatRow | None, stat_date: str) -> bool: + if row is None or not row.payload_json or row.status != "ready": + return False + if stat_date != _today_stat_date(): + return True + generated_at = row.generated_at + if not isinstance(generated_at, datetime): + return False + age = (datetime.now(BEIJING_TZ) - _as_beijing(generated_at)).total_seconds() + return age <= _LEADERBOARD_TODAY_TTL_SECONDS + + +def _snapshot_response_from_row(row: AdminLeaderboardDailyStatRow | None) -> AdminLeaderboardResponse | None: + if row is None or not row.payload_json: + return None + payload = row.payload_json + if isinstance(payload, str): + try: + payload = json.loads(payload) + except json.JSONDecodeError: + return None + try: + return AdminLeaderboardResponse.model_validate(payload) + except Exception: + logger.warning("Invalid leaderboard snapshot payload: %s/%s", row.stat_date, row.settings_hash, exc_info=True) + return None + + +def _merge_users_from_daily(responses: list[AdminLeaderboardResponse], limit: int) -> list[AdminLeaderboardUser]: + users_by_id: dict[str, dict] = {} + for response in responses: + day_users: dict[str, AdminLeaderboardUser] = {} + for ranking in (response.users_by_activity, response.users_by_questions, response.users_by_tokens): + for user in ranking: + if user.user_id: + day_users[user.user_id] = user + for user in day_users.values(): + item = users_by_id.setdefault( + user.user_id, + { + "user_id": user.user_id, + "email": user.email, + "system_role": user.system_role or "user", + "thread_count": 0, + "run_count": 0, + "qa_count": 0, + "llm_call_count": 0, + "tool_call_count": 0, + "skill_call_count": 0, + "total_tokens": 0, + "input_tokens": 0, + "output_tokens": 0, + "last_active_at": "", + }, + ) + if user.email: + item["email"] = user.email + if user.system_role: + item["system_role"] = user.system_role + for field in ( + "thread_count", + "run_count", + "qa_count", + "llm_call_count", + "tool_call_count", + "skill_call_count", + "total_tokens", + "input_tokens", + "output_tokens", + ): + item[field] += int(getattr(user, field) or 0) + item["last_active_at"] = _latest_iso(item["last_active_at"], user.last_active_at) + users = [AdminLeaderboardUser(**item) for item in users_by_id.values()] + return sorted(users, key=lambda u: (u.run_count + u.tool_call_count, u.total_tokens), reverse=True)[:limit] + + +def _merge_call_rankings_from_daily( + responses: list[AdminLeaderboardResponse], + field_name: str, + limit: int, +) -> list[AdminLeaderboardSkill]: + items_by_name: dict[str, dict] = {} + for response in responses: + for item in getattr(response, field_name): + if not item.name: + continue + merged = items_by_name.setdefault( + item.name, + { + "name": item.name, + "call_count": 0, + "user_count": 0, + "success_count": 0, + "error_count": 0, + "duration_weight": 0, + "last_used_at": "", + }, + ) + calls = int(item.call_count or 0) + merged["call_count"] += calls + merged["user_count"] += int(item.user_count or 0) + merged["success_count"] += int(item.success_count or 0) + merged["error_count"] += int(item.error_count or 0) + merged["duration_weight"] += int(item.avg_duration_ms or 0) * calls + merged["last_used_at"] = _latest_iso(merged["last_used_at"], item.last_used_at) + merged_items = [ + AdminLeaderboardSkill( + name=item["name"], + call_count=item["call_count"], + user_count=item["user_count"], + success_count=item["success_count"], + error_count=item["error_count"], + avg_duration_ms=int(item["duration_weight"] / item["call_count"]) if item["call_count"] else 0, + last_used_at=item["last_used_at"], + ) + for item in items_by_name.values() + ] + return sorted(merged_items, key=lambda item: (item.call_count, _sort_datetime(item.last_used_at)), reverse=True)[:limit] + + +def _merge_dimensions_from_daily( + responses: list[AdminLeaderboardResponse], + field_name: str, + limit: int, +) -> list[AdminLeaderboardDimension]: + items_by_name: dict[str, dict] = {} + for response in responses: + for item in getattr(response, field_name): + if not item.name: + continue + merged = items_by_name.setdefault( + item.name, + { + "name": item.name, + "count": 0, + "user_count": 0, + "success_count": 0, + "error_count": 0, + "total_tokens": 0, + "last_seen_at": "", + }, + ) + for field in ("count", "user_count", "success_count", "error_count", "total_tokens"): + merged[field] += int(getattr(item, field) or 0) + merged["last_seen_at"] = _latest_iso(merged["last_seen_at"], item.last_seen_at) + merged_items = [AdminLeaderboardDimension(**item) for item in items_by_name.values()] + return sorted(merged_items, key=lambda item: (item.count, _sort_datetime(item.last_seen_at)), reverse=True)[:limit] + + +def _merge_recent_questions_from_daily( + responses: list[AdminLeaderboardResponse], + limit: int, +) -> list[AdminLeaderboardQuestion]: + questions: list[AdminLeaderboardQuestion] = [] + seen: set[str] = set() + for response in responses: + for question in response.recent_questions: + key = question.run_id or f"{question.user_id}:{question.asked_at}:{question.text}" + if key in seen: + continue + seen.add(key) + questions.append(question) + questions.sort(key=lambda item: _sort_datetime(item.asked_at), reverse=True) + return questions[:limit] + + +_FREQUENT_QUESTION_COUNT_RE = re.compile(r"（相似问题\s+(\d+)\s+次）$") + + +def _split_frequent_question_count(text: str) -> tuple[str, int]: + match = _FREQUENT_QUESTION_COUNT_RE.search(text) + if not match: + return text, 1 + return _FREQUENT_QUESTION_COUNT_RE.sub("", text).strip(), max(1, int(match.group(1))) + + +def _merge_frequent_questions_from_daily( + responses: list[AdminLeaderboardResponse], + limit: int, +) -> list[AdminLeaderboardQuestion]: + groups: list[dict] = [] + for response in responses: + for question in response.frequent_questions: + base_text, count = _split_frequent_question_count(question.text) + key = _question_similarity_key(base_text) + if not key: + continue + target = None + for group in groups: + score = SequenceMatcher(None, key, group["key"]).ratio() + if score >= 0.72 or key in group["key"] or group["key"] in key: + target = group + break + if target is None: + groups.append( + { + "key": key, + "text": base_text, + "count": count, + "user_id": question.user_id, + "email": question.email, + "thread_id": question.thread_id, + "run_id": question.run_id, + "asked_at": question.asked_at, + } + ) + else: + target["count"] += count + if _sort_datetime(question.asked_at) > _sort_datetime(target.get("asked_at")): + target.update( + { + "text": base_text, + "user_id": question.user_id, + "email": question.email, + "thread_id": question.thread_id, + "run_id": question.run_id, + "asked_at": question.asked_at, + } + ) + groups.sort(key=lambda item: (item["count"], _sort_datetime(item.get("asked_at"))), reverse=True) + questions: list[AdminLeaderboardQuestion] = [] + for group in groups[:limit]: + text_value = group["text"][:460] + if group["count"] > 1: + text_value = f"{text_value}（相似问题 {group['count']} 次）" + questions.append( + AdminLeaderboardQuestion( + text=text_value, + user_id=group["user_id"], + email=group["email"], + thread_id=group["thread_id"], + run_id=group["run_id"], + asked_at=group["asked_at"], + ) + ) + return questions + + +def _merge_trends_from_daily(responses: list[AdminLeaderboardResponse]) -> list[AdminLeaderboardTrendPoint]: + trends_by_date: dict[str, AdminLeaderboardTrendPoint] = {} + for response in responses: + for point in response.trends: + item = trends_by_date.setdefault(point.date, AdminLeaderboardTrendPoint(date=point.date)) + item.runs += int(point.runs or 0) + item.questions += int(point.questions or 0) + item.tokens += int(point.tokens or 0) + item.users += int(point.users or 0) + return sorted(trends_by_date.values(), key=lambda item: item.date) + + +def _merge_daily_leaderboard_payloads( + responses: list[AdminLeaderboardResponse], + *, since_dt: datetime, until_dt: datetime, range_days: int, limit: int, + status: str, + message: str, +) -> AdminLeaderboardResponse: + users = _merge_users_from_daily(responses, _LEADERBOARD_DAILY_SNAPSHOT_LIMIT) + active_user_ids = { + user.user_id + for user in users + if user.user_id and (user.run_count or user.qa_count or user.tool_call_count or user.thread_count) + } + overview = AdminLeaderboardOverview( + total_users=max((item.overview.total_users for item in responses), default=0), + active_users=len(active_user_ids) if active_user_ids else sum(item.overview.active_users for item in responses), + total_threads=sum(item.overview.total_threads for item in responses), + total_runs=sum(item.overview.total_runs for item in responses), + total_questions=sum(item.overview.total_questions for item in responses), + total_tokens=sum(item.overview.total_tokens for item in responses), + total_skill_calls=sum(item.overview.total_skill_calls for item in responses), + total_tool_calls=sum(item.overview.total_tool_calls for item in responses), + error_runs=sum(item.overview.error_runs for item in responses), + ) + generated_at = "" + for response in responses: + generated_at = _latest_iso(generated_at, response.generated_at) + return AdminLeaderboardResponse( + range_days=range_days, + since=_iso(since_dt), + until=_iso(until_dt), + status=status, + generated_at=generated_at or _iso(datetime.now(BEIJING_TZ)), + message=message, + 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=_merge_call_rankings_from_daily(responses, "skills", limit), + tools=_merge_call_rankings_from_daily(responses, "tools", limit), + models=_merge_dimensions_from_daily(responses, "models", limit), + agents=_merge_dimensions_from_daily(responses, "agents", limit), + run_statuses=_merge_dimensions_from_daily(responses, "run_statuses", limit), + recent_questions=_merge_recent_questions_from_daily(responses, limit), + frequent_questions=_merge_frequent_questions_from_daily(responses, limit), + trends=_merge_trends_from_daily(responses), + ) + + +async def _load_daily_snapshot_rows( + session_factory, + stat_dates: list[str], + settings_hash: str, +) -> dict[str, AdminLeaderboardDailyStatRow]: + if not stat_dates: + return {} + async with session_factory() as session: + rows = ( + await session.execute( + select(AdminLeaderboardDailyStatRow).where( + AdminLeaderboardDailyStatRow.stat_date.in_(stat_dates), + AdminLeaderboardDailyStatRow.settings_hash == settings_hash, + ) + ) + ).scalars().all() + return {row.stat_date: row for row in rows} + + +async def _store_daily_snapshot( + session_factory, + stat_date: str, + settings_hash: str, + *, + status: str, + payload: dict | None = None, + message: str | None = None, + error: str | None = None, + generated_at: datetime | None = None, +) -> None: + async with session_factory() as session: + now = datetime.now(UTC) + row = await session.get( + AdminLeaderboardDailyStatRow, + {"stat_date": stat_date, "settings_hash": settings_hash}, + ) + if row is None: + row = AdminLeaderboardDailyStatRow( + stat_date=stat_date, + settings_hash=settings_hash, + status=status, + payload_json=payload or {}, + message=message, + error=error, + generated_at=generated_at, + created_at=now, + updated_at=now, + ) + session.add(row) + else: + row.status = status + if payload is not None: + row.payload_json = payload + if message is not None: + row.message = message + row.error = error + if generated_at is not None: + row.generated_at = generated_at + row.updated_at = now + await session.commit() + + +async def _queue_daily_leaderboard_snapshots( + session_factory, + stat_dates: list[str], + settings_hash: str, + *, + force: bool = False, + message: str = "等待排行榜后台统计任务处理。", +) -> list[str]: + queued: list[str] = [] + if not stat_dates: + return queued + async with session_factory() as session: + now = datetime.now(UTC) + rows = ( + await session.execute( + select(AdminLeaderboardDailyStatRow).where( + AdminLeaderboardDailyStatRow.stat_date.in_(stat_dates), + AdminLeaderboardDailyStatRow.settings_hash == settings_hash, + ) + ) + ).scalars().all() + rows_by_date = {row.stat_date: row for row in rows} + for stat_date in stat_dates: + row = rows_by_date.get(stat_date) + if row is None: + row = AdminLeaderboardDailyStatRow( + stat_date=stat_date, + settings_hash=settings_hash, + status="queued", + payload_json={}, + message=message, + created_at=now, + updated_at=now, + ) + session.add(row) + queued.append(stat_date) + continue + if not force and _snapshot_is_fresh(row, stat_date): + continue + if row.status == "running": + continue + row.status = "queued" + row.message = message + row.error = None + row.updated_at = now + queued.append(stat_date) + if queued: + await session.commit() + return queued + + +async def _build_daily_leaderboard_snapshot( + session_factory, + settings, + stat_date: str, + *, + excluded: set[str], + include_scheduled: bool, + include_admins: bool, + include_failed: bool, +) -> AdminLeaderboardResponse: + since_dt, until_dt = _bounds_for_stat_date(stat_date) + return await _build_user_leaderboard_response( + session_factory, + settings, + since_dt, + until_dt, + 1, + _LEADERBOARD_DAILY_SNAPSHOT_LIMIT, + excluded, + include_scheduled, + include_admins, + include_failed, + ) + + +async def _refresh_daily_leaderboard_snapshot( + job_key: str, + *, + session_factory, + settings, + stat_date: str, + settings_hash: str, excluded: set[str], include_scheduled: bool, include_admins: bool, @@ -1137,47 +1725,140 @@ async def _refresh_leaderboard_cache( try: if _LEADERBOARD_START_DELAY_SECONDS > 0: await asyncio.sleep(_LEADERBOARD_START_DELAY_SECONDS) + await _store_daily_snapshot( + session_factory, + stat_date, + settings_hash, + status="running", + message="正在生成当天排行榜快照。", + ) await _leaderboard_semaphore.acquire() acquired = True response = await asyncio.wait_for( - _build_user_leaderboard_response( + _build_daily_leaderboard_snapshot( session_factory, settings, - since_dt, - until_dt, - range_days, - limit, - excluded, - include_scheduled, - include_admins, - include_failed, + stat_date, + excluded=excluded, + include_scheduled=include_scheduled, + include_admins=include_admins, + include_failed=include_failed, ), timeout=_LEADERBOARD_TIMEOUT_SECONDS, ) - _leaderboard_cache[cache_key] = { - "created_at": time.monotonic(), - "response": response, - } + generated_at = datetime.now(BEIJING_TZ) + response.generated_at = _iso(generated_at) + await _store_daily_snapshot( + session_factory, + stat_date, + settings_hash, + status="ready", + payload=response.model_dump(mode="json"), + message="", + error=None, + generated_at=generated_at, + ) 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) + logger.warning("Daily leaderboard snapshot timed out: %s", job_key) + await _store_daily_snapshot( + session_factory, + stat_date, + settings_hash, + status="error", + message="当天排行榜快照统计超时，后台会在下次访问时重试。", + error="timeout", + ) + except Exception as exc: + logger.exception("Daily leaderboard snapshot failed: %s", job_key) + await _store_daily_snapshot( + session_factory, + stat_date, + settings_hash, + status="error", + message="当天排行榜快照统计失败，后台会在下次访问时重试。", + error=str(exc), + ) 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) +async def _load_or_schedule_daily_leaderboard( + session_factory, + settings, + since_dt: datetime, + until_dt: datetime, + *, + limit: int, + exclude_user_ids: list[str] | None = None, + include_scheduled: bool | None = None, + include_admins: bool | None = None, + include_failed: bool | None = None, +) -> AdminLeaderboardResponse: + stat_dates = _stat_dates_for_range(since_dt, until_dt) + range_days = len(stat_dates) + ( + _excluded, + _resolved_include_scheduled, + _resolved_include_admins, + _resolved_include_failed, + settings_hash, + ) = _resolve_leaderboard_snapshot_scope( + settings, + exclude_user_ids=exclude_user_ids, + include_scheduled=include_scheduled, + include_admins=include_admins, + include_failed=include_failed, + ) + rows = await _load_daily_snapshot_rows(session_factory, stat_dates, settings_hash) + responses: list[AdminLeaderboardResponse] = [] + refresh_dates: list[str] = [] + for stat_date in stat_dates: + row = rows.get(stat_date) + response = _snapshot_response_from_row(row) + if response is not None: + responses.append(response) + if not _snapshot_is_fresh(row, stat_date): + refresh_dates.append(stat_date) + + queued_dates: list[str] = [] + if refresh_dates and _LEADERBOARD_QUEUE_ON_REQUEST_ENABLED: + queued_dates = await _queue_daily_leaderboard_snapshots( + session_factory, + refresh_dates, + settings_hash, + message="排行榜页面请求发现该日快照缺失或过期，已加入后台统计队列。", ) + if responses: + status = "refreshing" if refresh_dates else "ready" + message = "" + if refresh_dates: + if _LEADERBOARD_QUEUE_ON_REQUEST_ENABLED: + message = f"缺失/过期的 {len(refresh_dates)} 天快照已加入队列，当前展示已完成的 {len(responses)} 天数据。后台统计任务会继续补齐。" + else: + message = f"缺失/过期的 {len(refresh_dates)} 天快照尚未生成，当前展示已完成的 {len(responses)} 天数据。" + return _merge_daily_leaderboard_payloads( + responses, + since_dt=since_dt, + until_dt=until_dt, + range_days=range_days, + limit=limit, + status=status, + message=message, + ) + return _empty_leaderboard_response( + since_dt, + until_dt, + range_days, + status="running", + message=( + f"该范围暂无可用快照，已加入 {len(queued_dates)} 天统计队列。后台统计任务会继续补齐。" + if _LEADERBOARD_QUEUE_ON_REQUEST_ENABLED + else "该范围暂无可用快照，请开启排行榜后台统计任务或手动补算对应日期。" + ), + ) + @router.get("/leaderboard", response_model=AdminLeaderboardResponse) async def get_user_leaderboard( @@ -1198,146 +1879,103 @@ async def get_user_leaderboard( 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 = _resolve_range(days, since, until) + return await _load_or_schedule_daily_leaderboard( + session_factory, + settings, since_dt, until_dt, - range_days, - status="running", - message="排行榜统计已在后台运行。线上用户较多时会慢一些，页面会自动刷新结果。", + limit=limit, + exclude_user_ids=exclude_user_ids, + include_scheduled=include_scheduled, + include_admins=include_admins, + include_failed=include_failed, ) - 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) +@router.get("/leaderboard/snapshots", response_model=AdminLeaderboardSnapshotStatusResponse) +async def get_leaderboard_snapshot_status( + 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"), +) -> AdminLeaderboardSnapshotStatusResponse: + """Return per-day snapshot status for the current leaderboard settings.""" + 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), - 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, + settings = load_system_settings().leaderboard + since_dt, until_dt, _range_days = _resolve_range(days, since, until) + stat_dates = _stat_dates_for_range(since_dt, until_dt) + *_scope, settings_hash = _resolve_leaderboard_snapshot_scope(settings) + rows = await _load_daily_snapshot_rows(session_factory, stat_dates, settings_hash) + items: list[AdminLeaderboardSnapshotStatusItem] = [] + counts = {"ready": 0, "queued": 0, "running": 0, "error": 0, "missing": 0} + for stat_date in stat_dates: + row = rows.get(stat_date) + if row is None: + counts["missing"] += 1 + items.append(AdminLeaderboardSnapshotStatusItem(stat_date=stat_date)) + continue + status = row.status or "missing" + counts[status if status in counts else "missing"] += 1 + items.append( + AdminLeaderboardSnapshotStatusItem( + stat_date=stat_date, + status=status, + has_payload=bool(row.payload_json), + generated_at=_iso(row.generated_at), + updated_at=_iso(row.updated_at), + message=row.message or "", + error=row.error or "", + ) ) + return AdminLeaderboardSnapshotStatusResponse( + settings_hash=settings_hash, + since=_iso(since_dt), + until=_iso(until_dt), + total_days=len(stat_dates), + ready_days=counts["ready"], + queued_days=counts["queued"], + running_days=counts["running"], + error_days=counts["error"], + missing_days=counts["missing"], + items=items, + ) - 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() + +@router.post("/leaderboard/snapshots/backfill", response_model=AdminLeaderboardSnapshotBackfillResponse) +async def queue_leaderboard_snapshot_backfill( + body: AdminLeaderboardSnapshotBackfillRequest, + request: Request, +) -> AdminLeaderboardSnapshotBackfillResponse: + """Queue a date range for the leaderboard background snapshot scheduler.""" + 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(body.days, body.since, body.until) + stat_dates = _stat_dates_for_range(since_dt, until_dt) + *_scope, settings_hash = _resolve_leaderboard_snapshot_scope(settings) + queued_dates = await _queue_daily_leaderboard_snapshots( + session_factory, + stat_dates, + settings_hash, + force=body.force, + message="管理员手动补算请求已加入排行榜后台统计队列。", + ) + return AdminLeaderboardSnapshotBackfillResponse( + settings_hash=settings_hash, + since=_iso(since_dt), + until=_iso(until_dt), + total_days=len(stat_dates), + queued_dates=queued_dates, + message=f"已加入 {len(queued_dates)} 天快照队列，后台统计任务会继续补齐。", + ) async def _build_leaderboard_report_prompt( @@ -1347,38 +1985,26 @@ async def _build_leaderboard_report_prompt( until_dt: datetime, limit: int, ) -> str: - excluded = {uid for uid in settings.exclude_user_ids if uid} - common = { - "exclude_user_ids": excluded, - "include_scheduled": settings.include_scheduled, - "include_admins": settings.include_admins, - "include_failed": settings.include_failed, - } - tool_common = { - "exclude_user_ids": excluded, - "include_scheduled": settings.include_scheduled, - "include_admins": settings.include_admins, - } - question_common = { - **common, - "cleanup_system_questions": settings.cleanup_system_questions, - } + snapshot = await _load_or_schedule_daily_leaderboard( + session_factory, + settings, + since_dt, + until_dt, + limit=max(limit, 30), + ) + if snapshot.status == "running" and snapshot.overview.total_runs == 0: + raise HTTPException(status_code=202, detail="排行榜快照正在后台生成，请稍后再生成报告。") - async with session_factory() as session: - 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) - - top_users = sorted(users, key=lambda u: (u.run_count + u.tool_call_count, u.total_tokens), reverse=True)[:12] + overview = snapshot.overview + skills = snapshot.skills + tools = snapshot.tools + models = snapshot.models + agents = snapshot.agents + run_statuses = snapshot.run_statuses + recent_questions = snapshot.recent_questions + frequent_questions = snapshot.frequent_questions + trends = snapshot.trends + top_users = snapshot.users_by_activity[:12] context = { "range": {"since": _iso(since_dt), "until": _iso(until_dt)}, "overview": overview.model_dump(), @@ -1566,6 +2192,7 @@ async def get_user_analytics( } since_dt, until_dt, range_days = _resolve_range(days, since, until) async with session_factory() as session: + await _set_analytics_read_timeout(session) 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") @@ -1596,6 +2223,128 @@ async def get_user_analytics( ) +# ── Q&A audit list ───────────────────────────────────────────────────────── + + +class AdminUserQaItem(BaseModel): + run_id: str + thread_id: str = "" + thread_title: str = "" + assistant_id: str = "" + status: str = "" + model_name: str = "" + question: str = "" + answer: str = "" + total_tokens: int = 0 + created_at: str = "" + updated_at: str = "" + + +class AdminUserQaPageResponse(BaseModel): + user_id: str + page: int + page_size: int + total: int + items: list[AdminUserQaItem] = Field(default_factory=list) + + +@router.get("/{user_id}/qa", response_model=AdminUserQaPageResponse) +async def list_user_qa( + user_id: str, + request: Request, + page: int = Query(default=1, ge=1), + page_size: int = Query(default=20, ge=1, le=100), + q: str | None = Query(default=None, description="Search question, answer, run id, or thread id"), + since: str | None = Query(default=None, description="Inclusive range start, ISO datetime"), + until: str | None = Query(default=None, description="Inclusive range end, ISO datetime"), +) -> AdminUserQaPageResponse: + """Return one user's Q&A runs from lightweight run summary fields.""" + await _require_admin(request) + session_factory = get_session_factory() + if session_factory is None: + raise HTTPException(status_code=503, detail="Q&A audit requires database persistence") + + filters = [ + RunRow.user_id == user_id, + RunRow.first_human_message.is_not(None), + RunRow.first_human_message != "", + ] + if since or until: + since_dt, until_dt, _range_days = _resolve_range(365, since, until) + filters.append(_run_time_filter(since_dt, until_dt)) + needle = (q or "").strip() + if needle: + like = f"%{needle}%" + filters.append( + or_( + RunRow.first_human_message.like(like), + RunRow.last_ai_message.like(like), + RunRow.run_id.like(like), + RunRow.thread_id.like(like), + ) + ) + + offset = (page - 1) * page_size + async with session_factory() as session: + await _set_analytics_read_timeout(session) + total = ( + await session.execute( + select(func.count(RunRow.run_id)).where(*filters) + ) + ).scalar_one() + rows = ( + await session.execute( + select( + RunRow.run_id, + RunRow.thread_id, + ThreadMetaRow.display_name, + RunRow.assistant_id, + RunRow.status, + RunRow.model_name, + RunRow.first_human_message, + RunRow.last_ai_message, + RunRow.total_tokens, + RunRow.created_at, + RunRow.updated_at, + ) + .outerjoin( + ThreadMetaRow, + and_( + ThreadMetaRow.thread_id == RunRow.thread_id, + ThreadMetaRow.user_id == RunRow.user_id, + ), + ) + .where(*filters) + .order_by(RunRow.created_at.desc(), RunRow.run_id.desc()) + .limit(page_size) + .offset(offset) + ) + ).all() + + return AdminUserQaPageResponse( + user_id=user_id, + page=page, + page_size=page_size, + total=int(total or 0), + items=[ + AdminUserQaItem( + run_id=r[0] or "", + thread_id=r[1] or "", + thread_title=r[2] or "", + assistant_id=r[3] or "", + status=r[4] or "", + model_name=r[5] or "", + question=_clean_question_text(r[6])[:2000], + answer=str(r[7] or "").strip()[:6000], + total_tokens=int(r[8] or 0), + created_at=_iso(r[9]), + updated_at=_iso(r[10]), + ) + for r in rows + ], + ) + + # ── Threads ──────────────────────────────────────────────────────────────── diff --git a/offline-backend-20260512/backend/app/gateway/routers/tags.py b/offline-backend-20260512/backend/app/gateway/routers/tags.py index a8960602..e85a5464 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/tags.py +++ b/offline-backend-20260512/backend/app/gateway/routers/tags.py @@ -1,9 +1,9 @@ """Tag management API. Tags are split into three types (``agent`` / ``skill`` / ``scheduled_task``). -A tag may only be applied to a resource of its own type. Tag CRUD and tag -assignment are admin-only; listing tags and reading a resource's tags is open -to everyone (the frontend needs them to render chips and search filters). +A tag may only be applied to a resource of its own type. Tag CRUD, listing, +and assignment are available to ordinary users so teams can organize resources +without asking an administrator to tag every item. """ from __future__ import annotations @@ -11,10 +11,10 @@ from __future__ import annotations import logging import uuid -from fastapi import APIRouter, Depends, HTTPException, Query, Request +from fastapi import APIRouter, Depends, HTTPException, Query from pydantic import BaseModel, Field -from app.gateway.deps import get_optional_user_from_request, get_tag_store +from app.gateway.deps import get_tag_store from deerflow.persistence.tags import TAG_TYPES, TagStore logger = logging.getLogger(__name__) @@ -22,18 +22,6 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/tags", tags=["tags"]) -# --------------------------------------------------------------------------- auth - - -async def _require_admin(request: Request) -> None: - """Reject non-admins. Lets the call through when auth is disabled.""" - user = await get_optional_user_from_request(request) - if user is None: - return - if getattr(user, "system_role", None) != "admin": - raise HTTPException(status_code=403, detail="Tag management is admin-only") - - def _validate_tag_type(tag_type: str) -> str: if tag_type not in TAG_TYPES: raise HTTPException(status_code=422, detail=f"Invalid tag type '{tag_type}'. Allowed: {', '.join(TAG_TYPES)}") @@ -96,10 +84,8 @@ async def list_tags( @router.post("", response_model=TagResponse, status_code=201, summary="Create Tag") async def create_tag( body: TagCreateRequest, - request: Request, store: TagStore = Depends(get_tag_store), ) -> TagResponse: - await _require_admin(request) _validate_tag_type(body.tag_type) try: record = await store.create_tag( @@ -125,10 +111,8 @@ async def create_tag( async def update_tag( tag_id: str, body: TagUpdateRequest, - request: Request, store: TagStore = Depends(get_tag_store), ) -> TagResponse: - await _require_admin(request) update_data = body.model_dump(exclude_unset=True) if "name" in update_data and update_data["name"] is not None: update_data["name"] = update_data["name"].strip() @@ -149,10 +133,8 @@ async def update_tag( @router.delete("/{tag_id}", status_code=204, summary="Delete Tag") async def delete_tag( tag_id: str, - request: Request, store: TagStore = Depends(get_tag_store), ) -> None: - await _require_admin(request) try: deleted = await store.delete_tag(tag_id) if not deleted: @@ -195,10 +177,8 @@ async def set_resource_tags( target_type: str, target_id: str, body: TagAssignmentRequest, - request: Request, store: TagStore = Depends(get_tag_store), ) -> TagListResponse: - await _require_admin(request) _validate_tag_type(target_type) try: records = await store.set_target_tags(target_type, target_id, body.tag_ids) diff --git a/offline-backend-20260512/backend/config.yaml b/offline-backend-20260512/backend/config.yaml index fae7f00d..8f727d9c 100644 --- a/offline-backend-20260512/backend/config.yaml +++ b/offline-backend-20260512/backend/config.yaml @@ -224,7 +224,9 @@ auth_login: database: backend: sqlite sqlite_dir: /app/backend/.deer-flow/data - pool_size: 10 + pool_size: 20 # 基础连接池(postgres/mysql 生效;sqlite 忽略) + max_overflow: 20 # 高峰期可临时超出 pool_size 的额外连接数,真实上限 = pool_size + max_overflow + pool_timeout: 30 # 池满时等待空闲连接的秒数,超时报错 echo_sql: false checkpointer: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/config/database_config.py b/offline-backend-20260512/backend/packages/harness/deerflow/config/database_config.py index bd78d38d..2ed2100c 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/config/database_config.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/config/database_config.py @@ -11,13 +11,13 @@ blocking, making a unified file safe for both workloads. Writers that contend for the lock wait via the default 5-second sqlite3 busy timeout rather than failing immediately. -Postgres mode: both use the same database URL but maintain independent -connection pools with different lifecycles. - -MySQL mode: application ORM data uses MySQL. LangGraph's checkpointer does -not have a built-in MySQL backend in this project, so deployments that choose -MySQL for business tables should configure the legacy ``checkpointer:`` -section separately (usually sqlite) to preserve conversation checkpoints. +Postgres mode: both use the same database URL but maintain independent +connection pools with different lifecycles. + +MySQL mode: application ORM data uses MySQL. LangGraph's checkpointer does +not have a built-in MySQL backend in this project, so deployments that choose +MySQL for business tables should configure the legacy ``checkpointer:`` +section separately (usually sqlite) to preserve conversation checkpoints. Memory mode: checkpointer uses MemorySaver, app uses in-memory stores. No database is initialized. @@ -26,12 +26,12 @@ Sensitive values (postgres_url) should use $VAR syntax in config.yaml to reference environment variables from .env: database: - backend: postgres - postgres_url: $DATABASE_URL - -The $VAR resolution is handled by AppConfig.resolve_env_variables() -before this config is instantiated -- DatabaseConfig itself does not -need to do any environment variable processing. + backend: postgres + postgres_url: $DATABASE_URL + +The $VAR resolution is handled by AppConfig.resolve_env_variables() +before this config is instantiated -- DatabaseConfig itself does not +need to do any environment variable processing. """ from __future__ import annotations @@ -42,40 +42,54 @@ from typing import Literal from pydantic import BaseModel, Field -class DatabaseConfig(BaseModel): - backend: Literal["memory", "sqlite", "postgres", "mysql"] = Field( - default="memory", - description=("Storage backend for application data. 'memory' for development (no persistence across restarts), 'sqlite' for single-node deployment, 'postgres' for production multi-node deployment, 'mysql' for MySQL-backed business tables."), - ) +class DatabaseConfig(BaseModel): + backend: Literal["memory", "sqlite", "postgres", "mysql"] = Field( + default="memory", + description=("Storage backend for application data. 'memory' for development (no persistence across restarts), 'sqlite' for single-node deployment, 'postgres' for production multi-node deployment, 'mysql' for MySQL-backed business tables."), + ) sqlite_dir: str = Field( default=".deer-flow/data", description=("Directory for the SQLite database file. Both checkpointer and application data share {sqlite_dir}/deerflow.db."), ) - postgres_url: str = Field( - default="", - description=( - "PostgreSQL connection URL, shared by checkpointer and app. " + postgres_url: str = Field( + default="", + description=( + "PostgreSQL connection URL, shared by checkpointer and app. " "Use $DATABASE_URL in config.yaml to reference .env. " "Example: postgresql://user:pass@host:5432/deerflow " "(the +asyncpg driver suffix is added automatically where needed)." - ), - ) - mysql_url: str = Field( - default="", - description=( - "MySQL connection URL for the application ORM. " - "Use $MYSQL_DATABASE_URL in config.yaml to reference .env. " - "Example: mysql+asyncmy://user:pass@host:3306/deerflow?charset=utf8mb4. " - "If mysql:// is provided, the +asyncmy driver suffix is added automatically." - ), - ) + ), + ) + mysql_url: str = Field( + default="", + description=( + "MySQL connection URL for the application ORM. " + "Use $MYSQL_DATABASE_URL in config.yaml to reference .env. " + "Example: mysql+asyncmy://user:pass@host:3306/deerflow?charset=utf8mb4. " + "If mysql:// is provided, the +asyncmy driver suffix is added automatically." + ), + ) echo_sql: bool = Field( default=False, description="Echo all SQL statements to log (debug only).", ) pool_size: int = Field( default=5, - description="Connection pool size for the app ORM engine (postgres/mysql only).", + description="Connection pool size for the app ORM engine (postgres/mysql only).", + ) + max_overflow: int = Field( + default=10, + description=( + "Extra connections allowed beyond pool_size under load (postgres/mysql only). " + "The real connection ceiling is pool_size + max_overflow. SQLAlchemy default is 10." + ), + ) + pool_timeout: int = Field( + default=30, + description=( + "Seconds a caller waits for a free pooled connection before erroring (postgres/mysql only). " + "SQLAlchemy default is 30." + ), ) # -- Derived helpers (not user-configured) -- @@ -108,16 +122,16 @@ class DatabaseConfig(BaseModel): """SQLAlchemy async URL for the application ORM engine.""" if self.backend == "sqlite": return f"sqlite+aiosqlite:///{self.sqlite_path}" - if self.backend == "postgres": - url = self.postgres_url - if url.startswith("postgresql://"): - url = url.replace("postgresql://", "postgresql+asyncpg://", 1) - return url - if self.backend == "mysql": - url = self.mysql_url - if not url: - raise ValueError("database.mysql_url is required for backend='mysql'") - if url.startswith("mysql://"): - url = url.replace("mysql://", "mysql+asyncmy://", 1) - return url - raise ValueError(f"No SQLAlchemy URL for backend={self.backend!r}") + if self.backend == "postgres": + url = self.postgres_url + if url.startswith("postgresql://"): + url = url.replace("postgresql://", "postgresql+asyncpg://", 1) + return url + if self.backend == "mysql": + url = self.mysql_url + if not url: + raise ValueError("database.mysql_url is required for backend='mysql'") + if url.startswith("mysql://"): + url = url.replace("mysql://", "mysql+asyncmy://", 1) + return url + raise ValueError(f"No SQLAlchemy URL for backend={self.backend!r}") diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/admin_stats/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/admin_stats/__init__.py new file mode 100644 index 00000000..e5dec995 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/admin_stats/__init__.py @@ -0,0 +1,5 @@ +"""Persistence models for admin analytics snapshots.""" + +from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow + +__all__ = ["AdminLeaderboardDailyStatRow"] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/admin_stats/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/admin_stats/model.py new file mode 100644 index 00000000..434e02f9 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/admin_stats/model.py @@ -0,0 +1,31 @@ +"""ORM models for admin analytics snapshots.""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from sqlalchemy import Index, String, Text +from sqlalchemy.orm import Mapped, mapped_column + +from deerflow.persistence.base import Base +from deerflow.persistence.types import BeijingDateTime, PortableJSON + + +class AdminLeaderboardDailyStatRow(Base): + """One precomputed leaderboard snapshot for a single local calendar day.""" + + __tablename__ = "admin_leaderboard_daily_stats" + __table_args__ = ( + Index("ix_admin_leaderboard_daily_stats_status", "status"), + Index("ix_admin_leaderboard_daily_stats_updated_at", "updated_at"), + ) + + stat_date: Mapped[str] = mapped_column(String(10), primary_key=True) + settings_hash: Mapped[str] = mapped_column(String(64), primary_key=True) + status: Mapped[str] = mapped_column(String(16), default="ready") + payload_json: Mapped[dict] = mapped_column(PortableJSON(), default=dict) + message: Mapped[str | None] = mapped_column(Text) + error: Mapped[str | None] = mapped_column(Text) + generated_at: Mapped[datetime | None] = mapped_column(BeijingDateTime()) + created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC)) + updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC)) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py index 384f86af..a65f6136 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py @@ -11,16 +11,16 @@ None and fall back to in-memory implementations. from __future__ import annotations import asyncio -import json -import logging -import os -import sys -from importlib.util import find_spec -from pathlib import Path - -from sqlalchemy import text -from sqlalchemy.engine import make_url -from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine +import json +import logging +import os +import sys +from importlib.util import find_spec +from pathlib import Path + +from sqlalchemy import text +from sqlalchemy.engine import make_url +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine def _json_serializer(obj: object) -> str: @@ -133,7 +133,7 @@ async def _run_alembic_upgrade(url: str) -> None: logger.exception("Auto-applying Alembic migrations failed; continuing with create_all schema") -async def _auto_create_postgres_db(url: str) -> None: +async def _auto_create_postgres_db(url: str) -> None: """Connect to the ``postgres`` maintenance DB and CREATE DATABASE. The target database name is extracted from *url*. The connection is @@ -156,55 +156,57 @@ async def _auto_create_postgres_db(url: str) -> None: async with maint_engine.connect() as conn: await conn.execute(text(f'CREATE DATABASE "{db_name}"')) logger.info("Auto-created PostgreSQL database: %s", db_name) - finally: - await maint_engine.dispose() - - -async def _auto_create_mysql_db(url: str) -> None: - """Create the target MySQL database when it is missing. - - ``create_all`` can create tables, but MySQL requires the database/schema to - exist before the application URL can connect. We connect to the built-in - ``mysql`` maintenance database on the same server and issue a conservative - ``CREATE DATABASE IF NOT EXISTS``. - """ - - parsed = make_url(url) - db_name = parsed.database - if not db_name: - raise ValueError("Cannot auto-create MySQL database: no database name in URL") - - maint_url = parsed.set(database="mysql") - maint_engine = create_async_engine( - maint_url.render_as_string(hide_password=False), - isolation_level="AUTOCOMMIT", - pool_pre_ping=True, - ) - quoted = "`" + db_name.replace("`", "``") + "`" - try: - async with maint_engine.connect() as conn: - await conn.execute( - text( - f"CREATE DATABASE IF NOT EXISTS {quoted} " - "CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci" - ) - ) - logger.info("Auto-created MySQL database if missing: %s", db_name) - finally: - await maint_engine.dispose() - - -def _is_mysql_unknown_database_error(exc: Exception) -> bool: - message = str(exc).lower() - return "unknown database" in message or "1049" in message - - -async def init_engine( + finally: + await maint_engine.dispose() + + +async def _auto_create_mysql_db(url: str) -> None: + """Create the target MySQL database when it is missing. + + ``create_all`` can create tables, but MySQL requires the database/schema to + exist before the application URL can connect. We connect to the built-in + ``mysql`` maintenance database on the same server and issue a conservative + ``CREATE DATABASE IF NOT EXISTS``. + """ + + parsed = make_url(url) + db_name = parsed.database + if not db_name: + raise ValueError("Cannot auto-create MySQL database: no database name in URL") + + maint_url = parsed.set(database="mysql") + maint_engine = create_async_engine( + maint_url.render_as_string(hide_password=False), + isolation_level="AUTOCOMMIT", + pool_pre_ping=True, + ) + quoted = "`" + db_name.replace("`", "``") + "`" + try: + async with maint_engine.connect() as conn: + await conn.execute( + text( + f"CREATE DATABASE IF NOT EXISTS {quoted} " + "CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci" + ) + ) + logger.info("Auto-created MySQL database if missing: %s", db_name) + finally: + await maint_engine.dispose() + + +def _is_mysql_unknown_database_error(exc: Exception) -> bool: + message = str(exc).lower() + return "unknown database" in message or "1049" in message + + +async def init_engine( backend: str, *, url: str = "", echo: bool = False, pool_size: int = 5, + max_overflow: int = 10, + pool_timeout: int = 30, sqlite_dir: str = "", ) -> None: """Create the async engine and session factory, then auto-create tables. @@ -213,7 +215,9 @@ async def init_engine( backend: "memory", "sqlite", "postgres", or "mysql". url: SQLAlchemy async URL (for sqlite/postgres/mysql). echo: Echo SQL to log. - pool_size: Postgres connection pool size. + pool_size: Base connection pool size (postgres/mysql). + max_overflow: Extra connections allowed beyond pool_size under load (postgres/mysql). + pool_timeout: Seconds to wait for a free pooled connection before erroring (postgres/mysql). sqlite_dir: Directory to create for SQLite (ensured to exist). """ global _engine, _session_factory @@ -271,6 +275,8 @@ async def init_engine( url, echo=echo, pool_size=pool_size, + max_overflow=max_overflow, + pool_timeout=pool_timeout, pool_pre_ping=True, json_serializer=_json_serializer, ) @@ -279,6 +285,8 @@ async def init_engine( url, echo=echo, pool_size=pool_size, + max_overflow=max_overflow, + pool_timeout=pool_timeout, pool_pre_ping=True, pool_recycle=1800, json_serializer=_json_serializer, @@ -302,37 +310,39 @@ async def init_engine( # This is expected during initial scaffolding or minimal installs. logger.debug("deerflow.persistence.models not found; skipping auto-create tables") - try: - async with _engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) - except Exception as exc: - if backend == "postgres" and "does not exist" in str(exc): + try: + async with _engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + except Exception as exc: + if backend == "postgres" and "does not exist" in str(exc): # Database not yet created — attempt to auto-create it, then retry. await _auto_create_postgres_db(url) # Rebuild engine against the now-existing database await _engine.dispose() - _engine = create_async_engine(url, echo=echo, pool_size=pool_size, pool_pre_ping=True, json_serializer=_json_serializer) - _session_factory = async_sessionmaker(_engine, expire_on_commit=False) - async with _engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) - elif backend == "mysql" and _is_mysql_unknown_database_error(exc): - # Database not yet created. Create it, rebuild the app engine - # against the target DB, then let SQLAlchemy create all tables. - await _auto_create_mysql_db(url) - await _engine.dispose() - _engine = create_async_engine( - url, - echo=echo, - pool_size=pool_size, - pool_pre_ping=True, - pool_recycle=1800, - json_serializer=_json_serializer, - ) - _session_factory = async_sessionmaker(_engine, expire_on_commit=False) - async with _engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) - else: - raise + _engine = create_async_engine(url, echo=echo, pool_size=pool_size, max_overflow=max_overflow, pool_timeout=pool_timeout, pool_pre_ping=True, json_serializer=_json_serializer) + _session_factory = async_sessionmaker(_engine, expire_on_commit=False) + async with _engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + elif backend == "mysql" and _is_mysql_unknown_database_error(exc): + # Database not yet created. Create it, rebuild the app engine + # against the target DB, then let SQLAlchemy create all tables. + await _auto_create_mysql_db(url) + await _engine.dispose() + _engine = create_async_engine( + url, + echo=echo, + pool_size=pool_size, + max_overflow=max_overflow, + pool_timeout=pool_timeout, + pool_pre_ping=True, + pool_recycle=1800, + json_serializer=_json_serializer, + ) + _session_factory = async_sessionmaker(_engine, expire_on_commit=False) + async with _engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + else: + raise # Apply in-place schema fixes (column widenings, etc.) that ``create_all`` # cannot do on its own. Failures don't abort startup — see the helper. @@ -351,6 +361,8 @@ async def init_engine_from_config(config) -> None: url=config.app_sqlalchemy_url, echo=config.echo_sql, pool_size=config.pool_size, + max_overflow=getattr(config, "max_overflow", 10), + pool_timeout=getattr(config, "pool_timeout", 30), sqlite_dir=config.sqlite_dir if config.backend == "sqlite" else "", ) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/llm_metrics/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/llm_metrics/model.py index 9d16e1ba..e6882fa9 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/llm_metrics/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/llm_metrics/model.py @@ -4,7 +4,7 @@ from __future__ import annotations from datetime import UTC, datetime -from sqlalchemy import Float, Integer, String, Text +from sqlalchemy import Float, Index, Integer, String, Text from sqlalchemy.orm import Mapped, mapped_column from deerflow.persistence.base import Base @@ -15,6 +15,10 @@ class LlmCallMetricRow(Base): """One row per completed (or failed) LLM provider invocation.""" __tablename__ = "llm_call_metrics" + __table_args__ = ( + Index("ix_llm_call_metrics_created_user", "created_at", "user_id"), + Index("ix_llm_call_metrics_run_created", "run_id", "created_at"), + ) id: Mapped[str] = mapped_column(String(64), primary_key=True) # Wall-clock time the LLM call ended. Indexed because admin views always diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260608_01_admin_leaderboard_daily_stats.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260608_01_admin_leaderboard_daily_stats.py new file mode 100644 index 00000000..50a14e64 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260608_01_admin_leaderboard_daily_stats.py @@ -0,0 +1,85 @@ +"""Create admin leaderboard daily snapshots and analytics indexes. + +Revision ID: 20260608_01 +Revises: 20260607_02 +Create Date: 2026-06-08 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op +from sqlalchemy import inspect + +from deerflow.persistence.types import PortableJSON + + +revision = "20260608_01" +down_revision = "20260607_02" +branch_labels = None +depends_on = None + + +def _has_table(table_name: str) -> bool: + inspector = inspect(op.get_bind()) + return table_name in inspector.get_table_names() + + +def _has_index(table_name: str, index_name: str) -> bool: + if not _has_table(table_name): + return False + inspector = inspect(op.get_bind()) + return any(idx["name"] == index_name for idx in inspector.get_indexes(table_name)) + + +def _create_index(table_name: str, index_name: str, columns: list[str]) -> None: + if _has_table(table_name) and not _has_index(table_name, index_name): + op.create_index(index_name, table_name, columns) + + +def _drop_index(table_name: str, index_name: str) -> None: + if _has_table(table_name) and _has_index(table_name, index_name): + op.drop_index(index_name, table_name=table_name) + + +def upgrade() -> None: + if not _has_table("admin_leaderboard_daily_stats"): + op.create_table( + "admin_leaderboard_daily_stats", + sa.Column("stat_date", sa.String(10), nullable=False), + sa.Column("settings_hash", sa.String(64), nullable=False), + sa.Column("status", sa.String(16), nullable=False, server_default="ready"), + sa.Column("payload_json", PortableJSON(), nullable=True), + sa.Column("message", sa.Text(), nullable=True), + sa.Column("error", sa.Text(), nullable=True), + sa.Column("generated_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("stat_date", "settings_hash", name="pk_admin_leaderboard_daily_stats"), + ) + op.create_index("ix_admin_leaderboard_daily_stats_status", "admin_leaderboard_daily_stats", ["status"]) + op.create_index("ix_admin_leaderboard_daily_stats_updated_at", "admin_leaderboard_daily_stats", ["updated_at"]) + + _create_index("runs", "ix_runs_created_user_status", ["created_at", "user_id", "status"]) + _create_index("runs", "ix_runs_user_created", ["user_id", "created_at"]) + _create_index("runs", "ix_runs_thread_created", ["thread_id", "created_at"]) + _create_index("tool_call_metrics", "ix_tool_call_metrics_created_user", ["created_at", "user_id"]) + _create_index("tool_call_metrics", "ix_tool_call_metrics_run_created", ["run_id", "created_at"]) + _create_index("llm_call_metrics", "ix_llm_call_metrics_created_user", ["created_at", "user_id"]) + _create_index("llm_call_metrics", "ix_llm_call_metrics_run_created", ["run_id", "created_at"]) + _create_index("threads_meta", "ix_threads_meta_created_user", ["created_at", "user_id"]) + _create_index("threads_meta", "ix_threads_meta_user_created", ["user_id", "created_at"]) + + +def downgrade() -> None: + _drop_index("threads_meta", "ix_threads_meta_user_created") + _drop_index("threads_meta", "ix_threads_meta_created_user") + _drop_index("llm_call_metrics", "ix_llm_call_metrics_run_created") + _drop_index("llm_call_metrics", "ix_llm_call_metrics_created_user") + _drop_index("tool_call_metrics", "ix_tool_call_metrics_run_created") + _drop_index("tool_call_metrics", "ix_tool_call_metrics_created_user") + _drop_index("runs", "ix_runs_thread_created") + _drop_index("runs", "ix_runs_user_created") + _drop_index("runs", "ix_runs_created_user_status") + if _has_table("admin_leaderboard_daily_stats"): + op.drop_table("admin_leaderboard_daily_stats") diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py index 4ac8baa7..45122546 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py @@ -21,6 +21,7 @@ from deerflow.persistence.agents.model import ( AgentRow, AgentSkillRow, ) +from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow from deerflow.persistence.memory_config.model import UserMemoryConfigRow from deerflow.persistence.feedback.model import FeedbackRow from deerflow.persistence.llm_metrics.model import LlmCallMetricRow @@ -58,6 +59,7 @@ __all__ = [ "AgentFavoriteRow", "AgentRow", "AgentSkillRow", + "AdminLeaderboardDailyStatRow", "LlmCallMetricRow", "NotificationRow", "RecommendedQuestionRow", diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/run/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/run/model.py index 175e7ed7..8f75f10a 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/run/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/run/model.py @@ -4,7 +4,7 @@ from __future__ import annotations from datetime import UTC, datetime -from sqlalchemy import Index, String, Text +from sqlalchemy import Index, String, Text from sqlalchemy.orm import Mapped, mapped_column from deerflow.persistence.base import Base @@ -47,4 +47,9 @@ class RunRow(Base): created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC)) updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC)) - __table_args__ = (Index("ix_runs_thread_status", "thread_id", "status"),) + __table_args__ = ( + Index("ix_runs_thread_status", "thread_id", "status"), + Index("ix_runs_created_user_status", "created_at", "user_id", "status"), + Index("ix_runs_user_created", "user_id", "created_at"), + Index("ix_runs_thread_created", "thread_id", "created_at"), + ) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/model.py index 6b0331cb..594b38f6 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/model.py @@ -4,15 +4,19 @@ from __future__ import annotations from datetime import UTC, datetime -from sqlalchemy import String +from sqlalchemy import Index, String from sqlalchemy.orm import Mapped, mapped_column from deerflow.persistence.base import Base from deerflow.persistence.types import BeijingDateTime, PortableJSON -class ThreadMetaRow(Base): - __tablename__ = "threads_meta" +class ThreadMetaRow(Base): + __tablename__ = "threads_meta" + __table_args__ = ( + Index("ix_threads_meta_created_user", "created_at", "user_id"), + Index("ix_threads_meta_user_created", "user_id", "created_at"), + ) thread_id: Mapped[str] = mapped_column(String(64), primary_key=True) assistant_id: Mapped[str | None] = mapped_column(String(128), index=True) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/tool_metrics/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/tool_metrics/model.py index bd70c038..b55d407f 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/tool_metrics/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/tool_metrics/model.py @@ -4,7 +4,7 @@ from __future__ import annotations from datetime import UTC, datetime -from sqlalchemy import Integer, String, Text +from sqlalchemy import Index, Integer, String, Text from sqlalchemy.orm import Mapped, mapped_column from deerflow.persistence.base import Base @@ -21,6 +21,10 @@ class ToolCallMetricRow(Base): """ __tablename__ = "tool_call_metrics" + __table_args__ = ( + Index("ix_tool_call_metrics_created_user", "created_at", "user_id"), + Index("ix_tool_call_metrics_run_created", "run_id", "created_at"), + ) id: Mapped[str] = mapped_column(String(64), primary_key=True) created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), index=True) diff --git a/offline-backend-20260512/backend/scripts/sql/leaderboard_indexes_mysql.sql b/offline-backend-20260512/backend/scripts/sql/leaderboard_indexes_mysql.sql new file mode 100644 index 00000000..8c4151c3 --- /dev/null +++ b/offline-backend-20260512/backend/scripts/sql/leaderboard_indexes_mysql.sql @@ -0,0 +1,85 @@ +-- ============================================================================ +-- 排行榜(admin leaderboard)统计所需的复合索引 —— MySQL 手动执行版 +-- ============================================================================ +-- +-- 背景: +-- 这些索引声明在各 ORM model 的 __table_args__ 里,但 Base.metadata.create_all +-- 对【已存在的表】不会补建索引,而 alembic 自动迁移在本项目里是关闭的 +-- (engine.py 中 `# await _run_alembic_upgrade(url)` 被注释)。 +-- 因此在【已有数据的 MySQL 部署】上,这些索引很可能并未真正创建,排行榜的 +-- 单日聚合查询会退化为全表扫描,拖慢并占用连接。 +-- +-- 用法: +-- 1) 先选中目标库(与 database.mysql_url 里的库名一致): +-- USE your_deerflow_db; +-- 2) 直接整文件执行即可。本脚本是【幂等】的:已存在的索引会跳过,不会报错, +-- 可重复执行。 +-- +-- 注意: +-- * 大表上建索引可能耗时;MySQL 8.0 在线 DDL 默认 ALGORITHM=INPLACE 一般不阻塞 +-- 写入,但仍建议在低峰期执行,或先在备库/预发验证。 +-- * 涉及列均为 String(64) / String(20) 等短列,utf8mb4 下复合索引长度远低于 +-- 3072 字节上限,无需前缀索引。 +-- * 索引名与 migration 20260608_01 保持一致,日后若启用 alembic 也不会重复创建。 +-- ============================================================================ + +-- ---- 通用幂等封装:仅当索引不存在时才创建 -------------------------------------- +DELIMITER // +DROP PROCEDURE IF EXISTS _create_index_if_missing // +CREATE PROCEDURE _create_index_if_missing( + IN p_table VARCHAR(64), + IN p_index VARCHAR(64), + IN p_columns VARCHAR(255) +) +BEGIN + DECLARE v_exists INT DEFAULT 0; + SELECT COUNT(*) INTO v_exists + FROM information_schema.statistics + WHERE table_schema = DATABASE() + AND table_name = p_table + AND index_name = p_index; + + IF v_exists = 0 THEN + SET @ddl = CONCAT('CREATE INDEX `', p_index, '` ON `', p_table, '` (', p_columns, ')'); + PREPARE stmt FROM @ddl; + EXECUTE stmt; + DEALLOCATE PREPARE stmt; + SELECT CONCAT('CREATED ', p_table, '.', p_index) AS result; + ELSE + SELECT CONCAT('SKIPPED ', p_table, '.', p_index, ' (already exists)') AS result; + END IF; +END // +DELIMITER ; + +-- ---- runs --------------------------------------------------------------------- +CALL _create_index_if_missing('runs', 'ix_runs_created_user_status', '`created_at`, `user_id`, `status`'); +CALL _create_index_if_missing('runs', 'ix_runs_user_created', '`user_id`, `created_at`'); +CALL _create_index_if_missing('runs', 'ix_runs_thread_created', '`thread_id`, `created_at`'); + +-- ---- tool_call_metrics ------------------------------------------------------- +CALL _create_index_if_missing('tool_call_metrics', 'ix_tool_call_metrics_created_user', '`created_at`, `user_id`'); +CALL _create_index_if_missing('tool_call_metrics', 'ix_tool_call_metrics_run_created', '`run_id`, `created_at`'); + +-- ---- llm_call_metrics -------------------------------------------------------- +CALL _create_index_if_missing('llm_call_metrics', 'ix_llm_call_metrics_created_user', '`created_at`, `user_id`'); +CALL _create_index_if_missing('llm_call_metrics', 'ix_llm_call_metrics_run_created', '`run_id`, `created_at`'); + +-- ---- threads_meta ------------------------------------------------------------ +CALL _create_index_if_missing('threads_meta', 'ix_threads_meta_created_user', '`created_at`, `user_id`'); +CALL _create_index_if_missing('threads_meta', 'ix_threads_meta_user_created', '`user_id`, `created_at`'); + +-- ---- 清理临时存储过程 -------------------------------------------------------- +DROP PROCEDURE IF EXISTS _create_index_if_missing; + +-- ---- 执行后自检:确认 9 个索引都已存在 --------------------------------------- +SELECT table_name, index_name + FROM information_schema.statistics + WHERE table_schema = DATABASE() + AND index_name IN ( + 'ix_runs_created_user_status', 'ix_runs_user_created', 'ix_runs_thread_created', + 'ix_tool_call_metrics_created_user', 'ix_tool_call_metrics_run_created', + 'ix_llm_call_metrics_created_user', 'ix_llm_call_metrics_run_created', + 'ix_threads_meta_created_user', 'ix_threads_meta_user_created' + ) + GROUP BY table_name, index_name + ORDER BY table_name, index_name;