wsk/deerflow-code
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增批量人物蒸馏脚本，精简排行榜相关接口

Browse Source 
- batch-distill: 独立单文件脚本，按时间窗调用线上服务批量蒸馏人物档案
- 精简 admin_users / public_leaderboard 路由及前端排行榜页面

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
吉鑫升andClaude Opus 4.8 committed 2026-06-11 09:34:34 +08:00
1 parent 3a04ba5726
commit 01696f75e9
8 files changed
+552 -418
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

offline-backend-20260512/backend/app/gateway/routers/admin_users.py
+4 -253
View File
@@ -20,7 +20,6 @@ import os
import re
from collections.abc import AsyncIterator
from contextvars import ContextVar
from difflib import SequenceMatcher
from hashlib import sha256
from time import perf_counter
from typing import Any
@@ -269,15 +268,6 @@ class AdminLeaderboardDimension(BaseModel):
last_seen_at: str = ""
class AdminLeaderboardQuestion(BaseModel):
text: str
user_id: str = ""
email: str = ""
thread_id: str = ""
run_id: str = ""
asked_at: str = ""
class AdminLeaderboardTrendPoint(BaseModel):
date: str
runs: int = 0
@@ -314,8 +304,6 @@ class AdminLeaderboardResponse(BaseModel):
models: list[AdminLeaderboardDimension] = Field(default_factory=list)
agents: list[AdminLeaderboardDimension] = Field(default_factory=list)
run_statuses: list[AdminLeaderboardDimension] = Field(default_factory=list)
recent_questions: list[AdminLeaderboardQuestion] = Field(default_factory=list)
frequent_questions: list[AdminLeaderboardQuestion] = Field(default_factory=list)
trends: list[AdminLeaderboardTrendPoint] = Field(default_factory=list)
@@ -330,7 +318,6 @@ class AdminUserAnalyticsResponse(BaseModel):
models: list[AdminLeaderboardDimension] = Field(default_factory=list)
agents: list[AdminLeaderboardDimension] = Field(default_factory=list)
run_statuses: list[AdminLeaderboardDimension] = Field(default_factory=list)
recent_questions: list[AdminLeaderboardQuestion] = Field(default_factory=list)
trends: list[AdminLeaderboardTrendPoint] = Field(default_factory=list)
@@ -1014,129 +1001,6 @@ async def _query_model_rankings(
]
async def _query_recent_questions(
session,
since: datetime,
until: datetime,
*,
user_id: str | None,
limit: int,
exclude_user_ids: set[str] | None = None,
include_scheduled: bool = True,
include_admins: bool = True,
include_failed: bool = True,
cleanup_system_questions: bool = True,
) -> list[AdminLeaderboardQuestion]:
exclude_user_ids = exclude_user_ids or set()
stmt = (
select(RunRow.first_human_message, RunRow.user_id, UserRow.email, RunRow.thread_id, RunRow.run_id, 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 != "")
.order_by(RunRow.created_at.desc())
.limit(limit)
)
if user_id:
stmt = stmt.where(RunRow.user_id == user_id)
stmt = _apply_run_scope(
stmt,
user_column=RunRow.user_id,
run_id_column=RunRow.run_id,
exclude_user_ids=exclude_user_ids,
include_admins=include_admins,
include_failed=include_failed,
include_scheduled=include_scheduled,
)
rows = (await _execute_leaderboard_query(session, "recent_questions", stmt)).all()
questions: list[AdminLeaderboardQuestion] = []
for r in rows:
text = _clean_question_text(r[0]) if cleanup_system_questions else str(r[0] or "").strip()
if cleanup_system_questions and not _is_displayable_question(text):
continue
questions.append(
AdminLeaderboardQuestion(
text=text[:500],
user_id=r[1] or "",
email=r[2] or "",
thread_id=r[3] or "",
run_id=r[4] or "",
asked_at=_iso(r[5]),
)
)
return questions
async def _query_frequent_questions(
session,
since: datetime,
until: datetime,
limit: int,
exclude_user_ids: set[str] | None = None,
include_scheduled: bool = True,
include_admins: bool = True,
include_failed: bool = True,
cleanup_system_questions: bool = True,
) -> list[AdminLeaderboardQuestion]:
exclude_user_ids = exclude_user_ids or set()
stmt = (
select(
RunRow.first_human_message,
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 != "")
.order_by(RunRow.created_at.desc())
.limit(min(max(limit * 20, 200), 1000))
)
stmt = _apply_run_scope(
stmt,
user_column=RunRow.user_id,
run_id_column=RunRow.run_id,
exclude_user_ids=exclude_user_ids,
include_admins=include_admins,
include_failed=include_failed,
include_scheduled=include_scheduled,
)
rows = (await _execute_leaderboard_query(session, "frequent_questions", stmt)).all()
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,
user_id=group["user_id"],
email=group["email"],
asked_at=_iso(group.get("asked_at")),
)
)
return questions
async def _query_trends(
session,
since: datetime,
@@ -1306,10 +1170,6 @@ async def _build_user_leaderboard_response(
"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)
await _leaderboard_pause()
users = await _query_user_leaderboard(session, since_dt, until_dt, **common)
@@ -1326,10 +1186,6 @@ async def _build_user_leaderboard_response(
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(
@@ -1347,8 +1203,6 @@ async def _build_user_leaderboard_response(
models=models,
agents=agents,
run_statuses=run_statuses,
recent_questions=recent_questions,
frequent_questions=frequent_questions,
trends=trends,
)
@@ -1572,95 +1426,6 @@ def _merge_dimensions_from_daily(
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:
@@ -1719,8 +1484,6 @@ def _merge_daily_leaderboard_payloads(
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),
)
@@ -2267,8 +2030,6 @@ async def _build_leaderboard_report_prompt(
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 = {
@@ -2280,8 +2041,6 @@ async def _build_leaderboard_report_prompt(
"top_models": [m.model_dump() for m in models[:12]],
"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:]],
}
return (
@@ -2289,12 +2048,10 @@ async def _build_leaderboard_report_prompt(
"报告需要有结论，不要只罗列数字；请使用 Markdown 标题、列表和表格，让前端可以直接渲染。"
"重点要求："
"1) 你是专门负责分析用户行为的专家，不要自称企业 AI 应用运营分析师；"
"2) 优先突出用户最近常用问题、最近关注的实时问题、事件、主题和业务方向，必须引用 recent_questions 中的近期问题作为依据；"
"3) 单独分析 frequent_questions，归纳高频关注簇，说明这些高频问题反映出的需求强度和重复频率；"
"4) 结合 top_users、top_skills、top_tools、top_models 和 trends，说明使用频率、使用增长、常用能力和重点用户行为画像；"
"5) 可以提及使用强度、关注变化和行为机会，但不要把报告重点放在系统本身的问题、故障、失败率或回答质量缺陷上；"
"6) 给管理员 3-5 条可执行建议，建议应围绕用户关注主题、资源配置、内容供给、培训引导和高频能力优化。"
"输出结构：摘要、最近关注、高频问题与频率、重点用户行为、常用能力、趋势观察、行动建议。"
"2) 结合 top_users、top_skills、top_tools、top_models 和 trends，说明使用频率、使用增长、常用能力和重点用户行为画像；"
"3) 可以提及使用强度、关注变化和行为机会，但不要把报告重点放在系统本身的问题、故障、失败率或回答质量缺陷上；"
"4) 给管理员 3-5 条可执行建议，建议应围绕资源配置、内容供给、培训引导和高频能力优化。"
"输出结构：摘要、重点用户行为、常用能力、趋势观察、行动建议。"
"\n\n统计数据如下：\n"
f"{json.dumps(context, ensure_ascii=False, default=str)}"
)
@@ -2452,10 +2209,6 @@ async def get_user_analytics(
"include_scheduled": settings.include_scheduled,
"include_admins": settings.include_admins,
}
question_common = {
**common,
"cleanup_system_questions": settings.cleanup_system_questions,
}
since_dt, until_dt, range_days = _resolve_range(days, since, until)
async with session_factory() as session:
await _set_analytics_read_timeout(session)
@@ -2470,7 +2223,6 @@ async def get_user_analytics(
models = await _query_run_dimensions(session, since_dt, until_dt, user_id=user_id, field="model", limit=limit, **common)
agents = await _query_run_dimensions(session, since_dt, until_dt, user_id=user_id, field="agent", limit=limit, **common)
run_statuses = await _query_run_dimensions(session, since_dt, until_dt, user_id=user_id, field="status", limit=limit, **common)
recent_questions = await _query_recent_questions(session, since_dt, until_dt, user_id=user_id, limit=limit, **question_common)
trends = await _query_trends(session, since_dt, until_dt, user_id=user_id, **common)
return AdminUserAnalyticsResponse(
@@ -2484,7 +2236,6 @@ async def get_user_analytics(
models=models,
agents=agents,
run_statuses=run_statuses,
recent_questions=recent_questions,
trends=trends,
)
offline-backend-20260512/backend/app/gateway/routers/public_leaderboard.py
-7
View File
@@ -62,13 +62,6 @@ def _anonymize(resp: AdminLeaderboardResponse) -> AdminLeaderboardResponse:
user.email = _mask_email(user.email)
user.user_id = hashed(user.user_id)
for bucket in (resp.recent_questions, resp.frequent_questions):
for question in bucket:
question.email = _mask_email(question.email)
question.user_id = hashed(question.user_id)
question.thread_id = ""
question.run_id = ""
return resp