enhance leaderboard reporting and grouping

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

@@ -13,14 +13,17 @@ methods' ``user_id`` parameter is designed for.
 
 from __future__ import annotations
 
+import json
 import logging
 import re
+from difflib import SequenceMatcher
 
 from datetime import datetime, timedelta
 
 from fastapi import APIRouter, HTTPException, Query, Request
 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_agent_store,
@@ -31,6 +34,7 @@ from app.gateway.deps import (
 from deerflow.agents.memory import BuiltinFileProvider
 from deerflow.config.paths import get_paths
 from deerflow.config.system_settings import load_system_settings
+from deerflow.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
@@ -158,6 +162,11 @@ class AdminUserAnalyticsResponse(BaseModel):
     trends: list[AdminLeaderboardTrendPoint] = Field(default_factory=list)
 
 
+class AdminLeaderboardReportResponse(BaseModel):
+    report: str
+    generated_at: str
+
+
 def _now() -> datetime:
     return datetime.now()
 
@@ -215,6 +224,24 @@ def _is_displayable_question(value: str | None) -> bool:
     return lowered not in {"uploaded_files", "attachments", "files"}
 
 
+def _question_similarity_key(value: str) -> str:
+    return re.sub(r"[\W_]+", "", value.lower())
+
+
+def _message_content_to_text(content) -> str:
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        parts: list[str] = []
+        for item in content:
+            if isinstance(item, dict):
+                parts.append(str(item.get("text") or item.get("content") or ""))
+            else:
+                parts.append(str(item))
+        return "\n".join(part for part in parts if part.strip())
+    return str(content or "")
+
+
 def _run_time_filter(since: datetime, until: datetime):
     return and_(RunRow.created_at >= since, RunRow.created_at <= until)
 
@@ -640,17 +667,14 @@ async def _query_frequent_questions(
     stmt = (
         select(
             RunRow.first_human_message,
-            func.count(RunRow.run_id).label("count"),
-            func.max(RunRow.user_id).label("user_id"),
-            func.max(UserRow.email).label("email"),
-            func.max(RunRow.created_at).label("asked_at"),
+            RunRow.user_id,
+            UserRow.email,
+            RunRow.created_at,
         )
         .outerjoin(UserRow, UserRow.id == RunRow.user_id)
         .where(_run_time_filter(since, until), RunRow.first_human_message.is_not(None), RunRow.first_human_message != "")
-        .group_by(RunRow.first_human_message)
-        .having(func.count(RunRow.run_id) > 1)
-        .order_by(desc("count"), desc("asked_at"))
-        .limit(limit)
+        .order_by(RunRow.created_at.desc())
+        .limit(min(max(limit * 20, 200), 1000))
     )
     stmt = _apply_run_scope(
         stmt,
@@ -662,17 +686,39 @@ async def _query_frequent_questions(
         include_scheduled=include_scheduled,
     )
     rows = (await session.execute(stmt)).all()
-    questions: list[AdminLeaderboardQuestion] = []
+    groups: list[dict] = []
     for r in rows:
         text = _clean_question_text(r[0]) if cleanup_system_questions else str(r[0] or "").strip()
         if cleanup_system_questions and not _is_displayable_question(text):
             continue
+        key = _question_similarity_key(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": text, "count": 1, "user_id": r[1] or "", "email": r[2] or "", "asked_at": r[3]})
+        else:
+            target["count"] += 1
+            if r[3] and (target.get("asked_at") is None or r[3] > target["asked_at"]):
+                target.update({"text": text, "user_id": r[1] or "", "email": r[2] or "", "asked_at": r[3]})
+
+    groups.sort(key=lambda item: (item["count"], _iso(item.get("asked_at"))), reverse=True)
+    questions: list[AdminLeaderboardQuestion] = []
+    for group in groups[:limit]:
+        text = group["text"][:460]
+        if group["count"] > 1:
+            text = f"{text}（相似问题 {group['count']} 次）"
         questions.append(
             AdminLeaderboardQuestion(
-                text=text[:500],
-                user_id=r[2] or "",
-                email=r[3] or "",
-                asked_at=_iso(r[4]),
+                text=text,
+                user_id=group["user_id"],
+                email=group["email"],
+                asked_at=_iso(group.get("asked_at")),
             )
         )
     return questions
@@ -866,6 +912,93 @@ 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)
+    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,
+    }
+
+    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)
+        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]
+    context = {
+        "range": {"since": _iso(since_dt), "until": _iso(until_dt)},
+        "overview": overview.model_dump(),
+        "top_users": [u.model_dump() for u in top_users],
+        "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:]],
+    }
+    prompt = (
+        "请基于以下管理员排行榜统计数据，生成一份中文运营分析报告。"
+        "报告需要有结论，不要只罗列数字。必须分析："
+        "1) 谁使用次数最多、问答最多、Token 消耗最高；"
+        "2) 用户通常在问什么、相似问题聚类反映了哪些关注方向；"
+        "3) 当前所有用户关注的事件/主题/业务方向；"
+        "4) 从问题重复、失败率、追问密度等信号推断用户是否满意系统回答，注明这是推断；"
+        "5) 哪些技能/工具/模型最常用，可能说明什么；"
+        "6) 给管理员 3-5 条可执行建议。"
+        "输出结构：摘要、关键发现、用户与分组观察、问题主题、满意度推断、建议。"
+        "\n\n统计数据如下：\n"
+        f"{json.dumps(context, ensure_ascii=False, default=str)}"
+    )
+    try:
+        model = create_chat_model(name=None, thinking_enabled=False)
+        response = await model.ainvoke(
+            [
+                SystemMessage(content="你是企业 AI 系统的运营分析师，擅长从用户行为、问题文本和调用指标中生成管理报告。"),
+                HumanMessage(content=prompt),
+            ],
+            config={"run_name": "admin_user_leaderboard_report"},
+        )
+    except Exception as exc:
+        logger.exception("Failed to generate leaderboard report")
+        raise HTTPException(status_code=503, detail=f"Report generation failed: {exc}") from exc
+
+    return AdminLeaderboardReportResponse(
+        report=_message_content_to_text(getattr(response, "content", response)).strip(),
+        generated_at=_iso(_now()),
+    )
+
+
 @router.get("/{user_id}/analytics", response_model=AdminUserAnalyticsResponse)
 async def get_user_analytics(
     user_id: str,