enhance admin leaderboard and previews

offline-backend-20260512/backend/app/gateway/routers/admin_users.py

@@ -16,22 +16,27 @@ from __future__ import annotations
 import json
 import logging
 import re
+from collections.abc import AsyncIterator
 from difflib import SequenceMatcher
+from hashlib import sha256
 
 from datetime import datetime, timedelta
 
-from fastapi import APIRouter, HTTPException, Query, Request
+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, or_, select
 from langchain_core.messages import HumanMessage, SystemMessage
 
 from app.gateway.deps import (
+    get_config,
     get_agent_store,
     get_checkpointer,
     get_current_user_from_request,
     get_thread_store,
 )
 from deerflow.agents.memory import BuiltinFileProvider
+from deerflow.config.app_config import AppConfig
 from deerflow.config.paths import get_paths
 from deerflow.config.system_settings import load_system_settings
 from deerflow.models import create_chat_model
@@ -165,6 +170,8 @@ class AdminUserAnalyticsResponse(BaseModel):
 class AdminLeaderboardReportResponse(BaseModel):
     report: str
     generated_at: str
+    cached: bool = False
+    model_name: str | None = None
 
 
 def _now() -> datetime:
@@ -242,6 +249,108 @@ def _message_content_to_text(content) -> str:
     return str(content or "")
 
 
+def _sse(event: str, data: dict) -> str:
+    return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
+
+
+def _extract_chunk_parts(chunk: object) -> tuple[str, str]:
+    """Return (thinking_delta, text_delta) from a LangChain streaming chunk."""
+    thinking = ""
+    text = ""
+    content = getattr(chunk, "content", "")
+    if isinstance(content, str):
+        text = content
+    elif isinstance(content, list):
+        for block in content:
+            if isinstance(block, str):
+                text += block
+            elif isinstance(block, dict):
+                block_type = block.get("type", "")
+                if block_type in {"thinking", "thinking_delta"}:
+                    thinking += block.get("thinking", "")
+                elif block_type in {"text", "text_delta"}:
+                    text += block.get("text", "")
+                elif block_type == "reasoning":
+                    thinking += block.get("content", "")
+    if not thinking:
+        kwargs = getattr(chunk, "additional_kwargs", {}) or {}
+        reasoning = kwargs.get("reasoning_content") or ""
+        if isinstance(reasoning, str):
+            thinking = reasoning
+    return thinking, text
+
+
+def _report_cache_file():
+    return get_paths().base_dir / "admin_leaderboard_reports.json"
+
+
+def _load_report_cache() -> dict:
+    path = _report_cache_file()
+    if not path.exists():
+        return {}
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+    except Exception:
+        logger.warning("Failed to read leaderboard report cache: %s", path, exc_info=True)
+        return {}
+    return data if isinstance(data, dict) else {}
+
+
+def _save_report_cache(cache: dict) -> None:
+    path = _report_cache_file()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(cache, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+def _resolve_report_model_name(config: AppConfig, model_name: str | None) -> str | None:
+    cleaned = (model_name or "").strip()
+    if cleaned:
+        return cleaned
+    return config.models[0].name if config.models else None
+
+
+def _report_cache_key(
+    *,
+    since: datetime,
+    until: datetime,
+    limit: int,
+    model_name: str | None,
+    settings,
+) -> str:
+    payload = {
+        "version": 3,
+        "since": _iso(since),
+        "until": _iso(until),
+        "limit": limit,
+        "model_name": model_name,
+        "settings": {
+            "exclude_user_ids": sorted(uid for uid in settings.exclude_user_ids if uid),
+            "include_scheduled": settings.include_scheduled,
+            "include_admins": settings.include_admins,
+            "include_failed": settings.include_failed,
+            "cleanup_system_questions": settings.cleanup_system_questions,
+        },
+    }
+    raw = json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str)
+    return sha256(raw.encode("utf-8")).hexdigest()
+
+
+def _cached_report(cache_key: str) -> dict | None:
+    item = _load_report_cache().get(cache_key)
+    if isinstance(item, dict) and isinstance(item.get("report"), str):
+        return item
+    return None
+
+
+def _store_report_cache(cache_key: str, item: dict) -> None:
+    cache = _load_report_cache()
+    cache[cache_key] = item
+    if len(cache) > 80:
+        ordered = sorted(cache.items(), key=lambda kv: str(kv[1].get("generated_at") or ""))
+        cache = dict(ordered[-80:])
+    _save_report_cache(cache)
+
+
 def _run_time_filter(since: datetime, until: datetime):
     return and_(RunRow.created_at >= since, RunRow.created_at <= until)
 
@@ -360,7 +469,7 @@ async def _query_user_leaderboard(
         select(
             ToolCallMetricRow.user_id.label("user_id"),
             func.count(ToolCallMetricRow.id).label("tool_call_count"),
-            func.sum(case((func.coalesce(func.nullif(ToolCallMetricRow.skill_name, ""), func.nullif(ToolCallMetricRow.tool_name, "")).is_not(None), 1), else_=0)).label("skill_call_count"),
+            func.sum(case((func.nullif(ToolCallMetricRow.skill_name, "").is_not(None), 1), else_=0)).label("skill_call_count"),
         )
         .where(_tool_time_filter(since, until))
         .group_by(ToolCallMetricRow.user_id)
@@ -433,11 +542,7 @@ async def _query_call_rankings(
     include_admins: bool = True,
 ) -> list[AdminLeaderboardSkill]:
     exclude_user_ids = exclude_user_ids or set()
-    name_col = (
-        func.coalesce(func.nullif(ToolCallMetricRow.skill_name, ""), func.nullif(ToolCallMetricRow.tool_name, ""))
-        if field == "skill"
-        else ToolCallMetricRow.tool_name
-    )
+    name_col = ToolCallMetricRow.skill_name if field == "skill" else ToolCallMetricRow.tool_name
     call_count = func.count(ToolCallMetricRow.id).label("call_count")
     last_used_at = func.max(ToolCallMetricRow.created_at).label("last_used_at")
     stmt = (
@@ -822,7 +927,7 @@ async def _query_overview(
     tool_row = (await session.execute(
         select(
             func.count(ToolCallMetricRow.id),
-            func.sum(case((func.coalesce(func.nullif(ToolCallMetricRow.skill_name, ""), func.nullif(ToolCallMetricRow.tool_name, "")).is_not(None), 1), else_=0)),
+            func.sum(case((func.nullif(ToolCallMetricRow.skill_name, "").is_not(None), 1), else_=0)),
         ).where(*tool_where)
     )).one()
 
@@ -912,22 +1017,13 @@ async def get_user_leaderboard(
     )
 
 
-@router.post("/leaderboard/report", response_model=AdminLeaderboardReportResponse)
-async def generate_user_leaderboard_report(
-    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"),
-    limit: int = Query(default=30, ge=10, le=100),
-) -> AdminLeaderboardReportResponse:
-    """Generate an LLM-written admin report for the current leaderboard window."""
-    await _require_admin(request)
-    session_factory = get_session_factory()
-    if session_factory is None:
-        raise HTTPException(status_code=503, detail="Analytics requires database persistence")
-
-    settings = load_system_settings().leaderboard
-    since_dt, until_dt, _range_days = _resolve_range(days, since, until)
+async def _build_leaderboard_report_prompt(
+    session_factory,
+    settings,
+    since_dt: datetime,
+    until_dt: datetime,
+    limit: int,
+) -> str:
     excluded = {uid for uid in settings.exclude_user_ids if uid}
     common = {
         "exclude_user_ids": excluded,
@@ -951,6 +1047,10 @@ async def generate_user_leaderboard_report(
         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)
@@ -963,11 +1063,13 @@ async def generate_user_leaderboard_report(
         "top_skills": [s.model_dump() for s in skills[:12]],
         "top_tools": [t.model_dump() for t in tools[:12]],
         "top_models": [m.model_dump() for m in models[:12]],
-        "frequent_questions": [q.model_dump() for q in frequent_questions[:20]],
-        "recent_questions": [q.model_dump() for q in recent_questions[:30]],
-        "trends": [t.model_dump() for t in trends[-14:]],
+        "agents": [a.model_dump() for a in agents[:12]],
+        "run_statuses": [s.model_dump() for s in run_statuses[:12]],
+        "frequent_questions": [q.model_dump() for q in frequent_questions[:30]],
+        "recent_questions": [q.model_dump() for q in recent_questions[:40]],
+        "trends": [t.model_dump() for t in trends[-21:]],
     }
-    prompt = (
+    return (
         "请基于以下管理员排行榜统计数据，生成一份中文运营分析报告。"
         "报告需要有结论，不要只罗列数字。必须分析："
         "1) 谁使用次数最多、问答最多、Token 消耗最高；"
@@ -976,12 +1078,46 @@ async def generate_user_leaderboard_report(
         "4) 从问题重复、失败率、追问密度等信号推断用户是否满意系统回答，注明这是推断；"
         "5) 哪些技能/工具/模型最常用，可能说明什么；"
         "6) 给管理员 3-5 条可执行建议。"
-        "输出结构：摘要、关键发现、用户与分组观察、问题主题、满意度推断、建议。"
+        "输出结构：摘要、关键发现、用户观察、问题主题、满意度推断、建议。"
         "\n\n统计数据如下：\n"
         f"{json.dumps(context, ensure_ascii=False, default=str)}"
     )
+
+
+@router.post("/leaderboard/report", response_model=AdminLeaderboardReportResponse)
+async def generate_user_leaderboard_report(
+    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"),
+    limit: int = Query(default=30, ge=10, le=100),
+    model_name: str | None = Query(default=None, description="Model name used to generate the report"),
+    config: AppConfig = Depends(get_config),
+) -> AdminLeaderboardReportResponse:
+    """Generate an LLM-written admin report for the current leaderboard window."""
+    await _require_admin(request)
+    session_factory = get_session_factory()
+    if session_factory is None:
+        raise HTTPException(status_code=503, detail="Analytics requires database persistence")
+
+    settings = load_system_settings().leaderboard
+    since_dt, until_dt, _range_days = _resolve_range(days, since, until)
+    resolved_model_name = _resolve_report_model_name(config, model_name)
+    cache_key = _report_cache_key(since=since_dt, until=until_dt, limit=limit, model_name=resolved_model_name, settings=settings)
+    cached = _cached_report(cache_key)
+    if cached:
+        return AdminLeaderboardReportResponse(
+            report=cached["report"],
+            generated_at=str(cached.get("generated_at") or ""),
+            cached=True,
+            model_name=str(cached.get("model_name") or resolved_model_name or ""),
+        )
+
+    prompt = await _build_leaderboard_report_prompt(session_factory, settings, since_dt, until_dt, limit)
     try:
-        model = create_chat_model(name=None, thinking_enabled=False)
+        model_cfg = config.get_model_config(resolved_model_name) if resolved_model_name else None
+        thinking_enabled = bool(model_cfg and getattr(model_cfg, "supports_thinking", False))
+        model = create_chat_model(name=resolved_model_name, thinking_enabled=thinking_enabled, app_config=config)
         response = await model.ainvoke(
             [
                 SystemMessage(content="你是企业 AI 系统的运营分析师，擅长从用户行为、问题文本和调用指标中生成管理报告。"),
@@ -993,9 +1129,85 @@ async def generate_user_leaderboard_report(
         logger.exception("Failed to generate leaderboard report")
         raise HTTPException(status_code=503, detail=f"Report generation failed: {exc}") from exc
 
+    report = _message_content_to_text(getattr(response, "content", response)).strip()
+    generated_at = _iso(_now())
+    _store_report_cache(cache_key, {"report": report, "generated_at": generated_at, "model_name": resolved_model_name})
     return AdminLeaderboardReportResponse(
-        report=_message_content_to_text(getattr(response, "content", response)).strip(),
-        generated_at=_iso(_now()),
+        report=report,
+        generated_at=generated_at,
+        cached=False,
+        model_name=resolved_model_name,
+    )
+
+
+@router.post("/leaderboard/report/stream")
+async def stream_user_leaderboard_report(
+    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"),
+    limit: int = Query(default=30, ge=10, le=100),
+    model_name: str | None = Query(default=None, description="Model name used to generate the report"),
+    config: AppConfig = Depends(get_config),
+) -> StreamingResponse:
+    """Stream an LLM-written admin report, returning cached reports when available."""
+    await _require_admin(request)
+    session_factory = get_session_factory()
+    if session_factory is None:
+        raise HTTPException(status_code=503, detail="Analytics requires database persistence")
+
+    settings = load_system_settings().leaderboard
+    since_dt, until_dt, _range_days = _resolve_range(days, since, until)
+    resolved_model_name = _resolve_report_model_name(config, model_name)
+    cache_key = _report_cache_key(since=since_dt, until=until_dt, limit=limit, model_name=resolved_model_name, settings=settings)
+
+    async def generate() -> AsyncIterator[str]:
+        cached = _cached_report(cache_key)
+        if cached:
+            yield _sse(
+                "cached",
+                {
+                    "report": cached["report"],
+                    "generated_at": cached.get("generated_at"),
+                    "model_name": cached.get("model_name") or resolved_model_name,
+                },
+            )
+            yield _sse("done", {"cached": True, "generated_at": cached.get("generated_at"), "model_name": cached.get("model_name") or resolved_model_name})
+            return
+
+        report_parts: list[str] = []
+        generated_at = ""
+        try:
+            yield _sse("meta", {"cached": False, "model_name": resolved_model_name})
+            prompt = await _build_leaderboard_report_prompt(session_factory, settings, since_dt, until_dt, limit)
+            model_cfg = config.get_model_config(resolved_model_name) if resolved_model_name else None
+            thinking_enabled = bool(model_cfg and getattr(model_cfg, "supports_thinking", False))
+            model = create_chat_model(name=resolved_model_name, thinking_enabled=thinking_enabled, app_config=config)
+            async for chunk in model.astream(
+                [
+                    SystemMessage(content="你是企业 AI 系统的运营分析师，擅长从用户行为、问题文本和调用指标中生成管理报告。"),
+                    HumanMessage(content=prompt),
+                ],
+                config={"run_name": "admin_user_leaderboard_report"},
+            ):
+                thinking_delta, text_delta = _extract_chunk_parts(chunk)
+                if thinking_delta:
+                    yield _sse("thinking", {"chunk": thinking_delta})
+                if text_delta:
+                    report_parts.append(text_delta)
+                    yield _sse("text", {"chunk": text_delta})
+            report = "".join(report_parts).strip()
+            generated_at = _iso(_now())
+            _store_report_cache(cache_key, {"report": report, "generated_at": generated_at, "model_name": resolved_model_name})
+            yield _sse("done", {"cached": False, "generated_at": generated_at, "model_name": resolved_model_name, "report": report})
+        except Exception as exc:
+            logger.exception("Failed to stream leaderboard report")
+            yield _sse("error", {"message": str(exc), "generated_at": generated_at, "model_name": resolved_model_name})
+
+    return StreamingResponse(
+        generate(),
+        media_type="text/event-stream",
+        headers={"X-Accel-Buffering": "no", "Cache-Control": "no-cache"},
     )