diff --git a/batch-distill/README.md b/batch-distill/README.md new file mode 100644 index 00000000..5fa50926 --- /dev/null +++ b/batch-distill/README.md @@ -0,0 +1,33 @@ +# 批量人物蒸馏 → llmwiki + +一个**完全独立的单文件脚本 + 内置定时器**：读 Excel 名单，串行调用线上 DeerFlow 服务问答接口， +复用「女娲 / huashu-nuwa」技能逐个蒸馏人物，**由脚本抓取回复并保存**成 llmwiki 风格 markdown。 + +- **不联网**：只用已配置的知识库查询技能（`knowledge-search-v2`）检索资料。 +- **定时执行**：每天 **23:00 ~ 次日 08:00** 自动跑，到点暂停、次日继续。 +- **md 由 py 保存**：agent 只产出内容，脚本抓最终回复写成 `.md`（不让项目自己写文件）。 + +## 你只需要做两件事 + +1. **改一行地址**：打开 `distill_people.py`，把顶部 `BASE_URL = "..."` 改成你的线上服务地址。 +2. **放名单**：同目录放 `people.xlsx`（或 `people.csv`），**第一列 = 人名**，其余列=可选提示。 + +其余（用户名 `distiller`、口令 `123ewq`、技能名、时间窗）都已内置，不用填。 + +## 运行 + +```bash +python distill_people.py # 常驻，按 23:00~08:00 窗口自动执行（推荐挂后台） +python distill_people.py --now # 忽略时间窗，立刻把名单跑完（测试用） +python distill_people.py --redo # 忽略已完成，全部重做 +``` + +读 `.xlsx` 需要 `openpyxl`；用 `.csv` 则零额外依赖。 + +## 输出（都在 output/ 一个文件夹） + +- `<人名>.md` —— 每个人的 llmwiki 档案 +- `manifest.json` —— 执行清单（状态/耗时/run_id） +- `index.md` —— 可读清单表（谁完成/失败，点开看 md） + +**断点续跑**：已完成的人自动跳过；中途往名单里加人，下个执行窗会带上。 diff --git a/batch-distill/distill_people.py b/batch-distill/distill_people.py new file mode 100644 index 00000000..3e422cc5 --- /dev/null +++ b/batch-distill/distill_people.py @@ -0,0 +1,504 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +批量人物蒸馏 → llmwiki markdown（完全独立的单文件客户端 + 内置定时器） + +特点 +---- +- 串行（一个一个来）调用正在运行的 DeerFlow 服务问答接口 +- 复用线上「女娲 / huashu-nuwa」技能做提炼；**不联网**，只用已配置的 + 知识库查询技能（knowledge-search-v2）检索资料 +- agent 只负责产出内容，**markdown 由本脚本抓取最终回复后自己保存** +- 内置时间窗：每天 23:00 ~ 次日 08:00 才执行，到点自动暂停、次日继续 +- 断点续跑：已完成的人自动跳过；中途加名单也会被带上 + +你只需要改一处：下面的 BASE_URL（线上服务地址）。其余都不用填。 + +名单 +---- +脚本同目录放 people.xlsx 或 people.csv，**第一列 = 人名**，其余列=可选提示。 + +运行 +---- + python distill_people.py # 常驻，按 23:00~08:00 窗口自动执行 + python distill_people.py --now # 忽略时间窗，立刻把名单跑完（测试用） + python distill_people.py --redo # 忽略已完成，全部重做 +""" +from __future__ import annotations + +import argparse +import csv +import json +import re +import sys +import time +from datetime import datetime, time as dtime +from pathlib import Path + +try: + import requests +except ImportError: + sys.exit("缺少依赖 requests：pip install requests") + + +# ════════════════════════════════════════════════════════════════════════ +# 只改这一行：你的线上服务地址（不带结尾斜杠） +# ════════════════════════════════════════════════════════════════════════ +BASE_URL = "http://localhost:8001" +# ════════════════════════════════════════════════════════════════════════ + + +# ---- 以下都已内置，正常无需改动 ---- +USERNAME = "distiller" # 免密自动注册 +PASSWORD = "123ewq" # 共享口令（你们开了口令门） +NUWA_SKILL = "huashu-nuwa" # 女娲技能 +KB_SKILL = "knowledge-search-v2" # 知识库查询技能（不联网，用它检索） +DEPTH = "full" # full 完整 / fast 精简 + +HERE = Path(__file__).resolve().parent +OUTPUT_DIR = HERE / "output" +PEOPLE_FILE_DEFAULT = HERE / "people.xlsx" + +# 执行时间窗：23:00 ~ 次日 08:00（跨午夜） +WIN_START = dtime(23, 0) +WIN_END = dtime(8, 0) + +PER_PERSON_TIMEOUT = 1800 # 单人最长等待（秒） +POLL_INTERVAL = 5 # 轮询间隔（秒） +HTTP_TIMEOUT = 60 +TERMINAL_STATUS = {"success", "error", "timeout", "interrupted"} + + +# ================================ llmwiki 模板 ================================ +LLMWIKI_TEMPLATE = """\ +--- +name: <中文名/常用名> +aliases: [<别名/英文名...>] +type: person +era: <在世 | 历史人物> +domains: [<领域1>, <领域2>] +distilled_at: +sources_count: <检索到的来源条数> +primary_source_ratio: <一手来源占比, 0~1 的小数> +confidence: +--- + +# <人物名> + +> **一句话定位**：<此人看世界的独特方式，一句话> + +## 身份卡 +- **活跃年代**： +- **核心身份**： +- **自述(50字, 以其语气)**： + +## 核心观点（真信念 · 反复出现的主张） +1. ... + +## 心智模型（3–7 个；每个含：一句话 / 证据场景≥2 / 应用 / 局限 / 来源） +### 1. <模型名> +- **一句话**： +- **证据场景**： +- **应用方式**： +- **局限**： +- **来源**： + +## 决策启发式（5–10 条 · 如果 X 则 Y，带案例） +- 若 → + +## 表达 DNA +| 维度 | 特征 | +|------|------| +| 句式 | | +| 高频词 | | +| 幽默 | | +| 确定性 | | +| 引用习惯 | | + +## 价值观与反模式 +- **价值观排序**： +- **明确反对**： + +## 内在张力（≥2 对矛盾） +- ↔ + +## 经历时间线 +| 年份 | 事件 | 思想意义 | +|------|------|---------| +| | | | + +## 代表语录 +- “...” —（出处/年份，标一手/二手） + +## 智识谱系 +- **受影响于**： +- **影响了**： + +## 外部评价与争议 +- + +## 诚实边界 +- <至少 3 条具体局限，含信息截止日期> + +## 调研来源 +**一手(N)**： +**二手(N)**： +""" + + +def build_prompt(name: str, hints: str, depth: str) -> str: + depth_note = ( + "本次走【完整模式】：尽量覆盖著作/对话/表达/他者评价/决策/时间线 6 个维度。" + if depth == "full" + else "本次走【精简模式】：重点覆盖核心观点、心智模型、时间线三块即可。" + ) + hint_block = f"\n补充提示（来自名单其它列，可作为检索关键词）：{hints}\n" if hints.strip() else "" + return f"""\ +你的任务：把人物【{name}】蒸馏成一份 llmwiki 风格的人物档案 markdown。 + +请复用本服务中已启用的「女娲 / {NUWA_SKILL}」技能的提炼方法论 +（心智模型三重验证、表达DNA分析、诚实边界）。 + +⚠️ 本环境**无法联网，严禁使用 WebSearch 等任何联网工具**。 +所有资料**只能通过已配置的知识库查询技能「{KB_SKILL}」检索获取**。 +请充分调用该知识库技能，围绕该人物多角度检索（生平、观点、著作、语录、决策、评价等）， +基于检索到的内容做提炼。知识库里查不到的部分，如实在「诚实边界」标注，不要编造。 + +{depth_note} +{hint_block} +输出要求： +1. **不要写任何文件，不要调用 write_file / present_files**。 +2. 直接在你的回复正文里输出**完整中文 markdown 全文**，严格按下面模板结构 + （保留各级标题，逐项填实，不要保留尖括号占位符），**不要任何额外解说、前言或结语**： + +------------------ 模板开始 ------------------ +{LLMWIKI_TEMPLATE} +------------------ 模板结束 ------------------ + +现在开始，独立完成，不要向我提问。""" + + +# ================================ HTTP 客户端 ================================ +class DeerFlowClient: + def __init__(self, base_url: str): + self.base = base_url.rstrip("/") + self.s = requests.Session() + self.token: str | None = None + + def _headers(self) -> dict: + h = {"Content-Type": "application/json"} + if self.token: + h["Authorization"] = f"Bearer {self.token}" # 只用 Bearer，不带 cookie → 跳过 CSRF + return h + + def login(self) -> bool: + url = f"{self.base}/api/v1/auth/login/username" + params = {"password": PASSWORD} if PASSWORD else None + try: + r = self.s.post(url, params=params, json={"username": USERNAME}, timeout=HTTP_TIMEOUT) + except requests.RequestException as e: + print(f" [登录] 连接失败：{e}") + return False + if r.status_code == 200: + self.token = r.json().get("access_token") + self.s.cookies.clear() + print(f" [登录] 成功，用户={USERNAME}") + return True + print(f" [登录] 返回 {r.status_code}：{r.text[:200]}") + return False + + def ensure_skill_enabled(self, name: str) -> None: + try: + r = self.s.get(f"{self.base}/api/skills", headers=self._headers(), timeout=HTTP_TIMEOUT) + if r.status_code == 200: + skills = r.json().get("skills", []) + hit = next((x for x in skills if x.get("name") == name), None) + if hit is None: + print(f" [技能] 警告：服务里没找到 '{name}'") + return + if hit.get("enabled"): + print(f" [技能] '{name}' 已启用") + return + pr = self.s.put(f"{self.base}/api/skills/{name}", headers=self._headers(), + json={"enabled": True}, timeout=HTTP_TIMEOUT) + print(f" [技能] 启用 '{name}' → {pr.status_code}") + except requests.RequestException as e: + print(f" [技能] 检查 '{name}' 失败（忽略）：{e}") + + def create_thread(self) -> str: + r = self.s.post(f"{self.base}/api/threads", headers=self._headers(), json={}, timeout=HTTP_TIMEOUT) + r.raise_for_status() + return r.json()["thread_id"] + + def start_run(self, thread_id: str, prompt: str) -> str: + body = { + "input": {"messages": [{"role": "user", "content": prompt}]}, + "excluded_tools": ["ask_clarification"], # 禁澄清，无人值守不卡住 + } + r = self.s.post(f"{self.base}/api/threads/{thread_id}/runs", + headers=self._headers(), json=body, timeout=HTTP_TIMEOUT) + r.raise_for_status() + return r.json()["run_id"] + + def poll_run(self, thread_id: str, run_id: str) -> str: + deadline = time.monotonic() + PER_PERSON_TIMEOUT + last = "" + while time.monotonic() < deadline: + try: + r = self.s.get(f"{self.base}/api/threads/{thread_id}/runs/{run_id}", + headers=self._headers(), timeout=HTTP_TIMEOUT) + if r.status_code == 200: + st = r.json().get("status", "") + if st != last: + print(f" ...状态：{st}") + last = st + if st in TERMINAL_STATUS: + return st + except requests.RequestException as e: + print(f" 轮询出错（重试）：{e}") + time.sleep(POLL_INTERVAL) + return "timeout" + + def fetch_result_text(self, thread_id: str, run_id: str) -> str | None: + """抓取该 run 的最终 AI 回复文本（由 py 保存为 md）。""" + try: + r = self.s.get(f"{self.base}/api/threads/{thread_id}/runs/{run_id}/messages", + headers=self._headers(), params={"limit": 200}, timeout=HTTP_TIMEOUT) + if r.status_code == 200: + data = r.json().get("data", []) + for msg in reversed(data): + if msg.get("event_type") == "ai_message" or msg.get("type") == "ai": + txt = _extract_text(msg) + if txt and txt.strip(): + return txt + except requests.RequestException as e: + print(f" 取回复出错：{e}") + return None + + +def _extract_text(msg: dict) -> str: + c = msg.get("content") or msg.get("text") or "" + if isinstance(c, str): + return c + if isinstance(c, list): + parts = [] + for b in c: + if isinstance(b, dict): + parts.append(b.get("text") or b.get("content") or "") + else: + parts.append(str(b)) + return "\n".join(p for p in parts if p) + return str(c) + + +def _clean_markdown(text: str) -> str: + """去掉 agent 可能套上的 ```markdown ... ``` 围栏。""" + t = text.strip() + m = re.match(r"^```(?:markdown|md)?\s*\n(.*)\n```$", t, re.S) + if m: + t = m.group(1).strip() + return t + + +# ================================ 名单读取 ================================ +def slugify(name: str) -> str: + s = re.sub(r'[\\/:*?"<>|]+', "", name).strip() + s = re.sub(r"\s+", "_", s) + return s or "unnamed" + + +def read_people(path: Path) -> list[dict]: + if not path.exists(): + return [] + rows: list[list[str]] = [] + if path.suffix.lower() in (".xlsx", ".xlsm"): + try: + import openpyxl + except ImportError: + sys.exit("读取 .xlsx 需要 openpyxl：pip install openpyxl（或把名单存成 .csv）") + wb = openpyxl.load_workbook(path, read_only=True, data_only=True) + for row in wb.active.iter_rows(values_only=True): + rows.append(["" if c is None else str(c).strip() for c in row]) + elif path.suffix.lower() == ".csv": + with open(path, encoding="utf-8-sig", newline="") as f: + rows = [[c.strip() for c in r] for r in csv.reader(f)] + else: + sys.exit(f"不支持的名单格式：{path.suffix}（用 .xlsx 或 .csv）") + + if rows and re.search(r"(姓名|名字|人物|name)", rows[0][0] if rows[0] else "", re.I): + rows = rows[1:] + people, seen = [], set() + for r in rows: + if not r or not r[0].strip(): + continue + name = r[0].strip() + if name in seen: + continue + seen.add(name) + hints = " | ".join(x for x in r[1:] if x.strip()) + people.append({"name": name, "hints": hints}) + return people + + +# ================================ 时间窗 ================================ +def in_window(now: datetime | None = None) -> bool: + now = now or datetime.now() + t = now.time() + return t >= WIN_START or t < WIN_END # 跨午夜 + +def seconds_until_window(now: datetime | None = None) -> int: + now = now or datetime.now() + if in_window(now): + return 0 + target = now.replace(hour=WIN_START.hour, minute=WIN_START.minute, second=0, microsecond=0) + return max(1, int((target - now).total_seconds())) + +def sleep_interruptible(total: int) -> None: + end = time.monotonic() + total + while time.monotonic() < end: + time.sleep(min(30, max(1, int(end - time.monotonic())))) + + +# ================================ 蒸馏单人 ================================ +def distill_one(client: DeerFlowClient, person: dict, out_dir: Path, + manifest: dict, save_manifest, depth: str) -> None: + name, hints = person["name"], person["hints"] + slug = slugify(name) + md_path = out_dir / f"{slug}.md" + t0 = time.time() + + rec = {"name": name, "slug": slug, "status": "running", + "started_at": _now(), "file": f"{slug}.md"} + manifest[name] = rec + save_manifest() + + try: + tid = client.create_thread() + run_id = client.start_run(tid, build_prompt(name, hints, depth)) + status = client.poll_run(tid, run_id) + rec.update(thread_id=tid, run_id=run_id, run_status=status) + + text = client.fetch_result_text(tid, run_id) if status == "success" else None + if status == "success" and text: + md = _clean_markdown(text) + md_path.write_text(md, encoding="utf-8") + rec["status"] = "succeeded" + rec["chars"] = len(md) + print(f" ✅ 完成 → {md_path.name}（{len(md)} 字，耗时 {int(time.time()-t0)}s）") + else: + rec["status"] = "failed" + rec["error"] = f"run={status}, text={'有' if text else '无'}" + if text: + (out_dir / f"{slug}.partial.md").write_text(text, encoding="utf-8") + print(f" ❌ 失败（run={status}）") + except Exception as e: + rec["status"] = "failed" + rec["error"] = str(e) + print(f" ❌ 异常：{e}") + + rec["finished_at"] = _now() + rec["seconds"] = int(time.time() - t0) + manifest[name] = rec + save_manifest() + + +# ================================ 主流程 ================================ +def main(): + ap = argparse.ArgumentParser(description="批量人物蒸馏 → llmwiki（定时窗 23:00~08:00）") + ap.add_argument("excel", nargs="?", default=str(PEOPLE_FILE_DEFAULT), + help="名单文件（.xlsx 或 .csv），默认同目录 people.xlsx") + ap.add_argument("--now", action="store_true", help="忽略时间窗，立刻跑完整批（测试用）") + ap.add_argument("--redo", action="store_true", help="忽略已完成，全部重做") + args = ap.parse_args() + + excel = Path(args.excel) + OUTPUT_DIR.mkdir(parents=True, exist_ok=True) + manifest_path = OUTPUT_DIR / "manifest.json" + + manifest: dict = {} + if manifest_path.exists(): + try: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + except Exception: + manifest = {} + + # 校验名单存在 + if not read_people(excel): + sys.exit(f"名单为空或不存在：{excel}\n请放 people.xlsx（第一列写人名）。") + + client = DeerFlowClient(BASE_URL) + if not client.login(): + sys.exit("登录失败。请检查 BASE_URL 是否指向正在运行的服务、口令是否正确。") + client.ensure_skill_enabled(NUWA_SKILL) + client.ensure_skill_enabled(KB_SKILL) + + def save_manifest(): + manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") + _write_index(OUTPUT_DIR, read_people(excel), manifest) + + def pending() -> list[dict]: + people = read_people(excel) + out = [] + for p in people: + r = manifest.get(p["name"], {}) + done = (not args.redo) and r.get("status") == "succeeded" and (OUTPUT_DIR / f"{slugify(p['name'])}.md").exists() + if not done: + out.append(p) + return out + + print(f"启动 | 服务={BASE_URL} | 输出={OUTPUT_DIR}") + print(f"模式={'立即跑完' if args.now else '定时窗 23:00~08:00'} | 名单={excel}") + + try: + while True: + todo = pending() + if not todo: + print(f"[{_now()}] 全部已完成，待命中（每小时复查名单）...") + if args.now: + break + sleep_interruptible(3600) + continue + + if not args.now and not in_window(): + wait = seconds_until_window() + nxt = datetime.now().replace(hour=WIN_START.hour, minute=0, second=0, microsecond=0) + print(f"[{_now()}] 不在执行窗口，剩 {len(todo)} 人待蒸；" + f"休眠到 {nxt.strftime('%H:%M')}（约 {wait//60} 分钟）...") + sleep_interruptible(wait) + continue + + # 在窗口内（或 --now）：处理一个人，然后回到循环重新判断窗口/名单 + p = todo[0] + idx_total = len(read_people(excel)) + done_n = sum(1 for v in manifest.values() if v.get("status") == "succeeded") + print(f"[{_now()}] ({done_n}/{idx_total} 已完成) 蒸馏：{p['name']}") + distill_one(client, p, OUTPUT_DIR, manifest, save_manifest, DEPTH) + except KeyboardInterrupt: + print("\n已手动停止。重跑脚本会从未完成处继续。") + + ok = sum(1 for r in manifest.values() if r.get("status") == "succeeded") + fail = sum(1 for r in manifest.values() if r.get("status") == "failed") + print(f"\n结束：成功 {ok} / 失败 {fail}") + print(f"结果目录：{OUTPUT_DIR} | 清单：{manifest_path} | 索引：{OUTPUT_DIR/'index.md'}") + + +def _now() -> str: + return time.strftime("%Y-%m-%d %H:%M:%S") + + +def _write_index(out_dir: Path, people: list[dict], manifest: dict) -> None: + lines = ["# 蒸馏执行清单", "", f"更新时间：{_now()}", "", + "| # | 人物 | 状态 | 字数 | 耗时 | 文件 |", + "|---|------|------|------|------|------|"] + badge = {"succeeded": "✅完成", "failed": "❌失败", "running": "⏳进行中"} + for i, p in enumerate(people, 1): + r = manifest.get(p["name"], {}) + st = badge.get(r.get("status", ""), "⌛排队") + f = r.get("file", "") + link = f"[{f}]({f})" if r.get("status") == "succeeded" and f else "-" + lines.append(f"| {i} | {p['name']} | {st} | {r.get('chars','-')} | {r.get('seconds','-')} | {link} |") + (out_dir / "index.md").write_text("\n".join(lines) + "\n", encoding="utf-8") + + +if __name__ == "__main__": + main() diff --git a/batch-distill/people.example.csv b/batch-distill/people.example.csv new file mode 100644 index 00000000..bed69453 --- /dev/null +++ b/batch-distill/people.example.csv @@ -0,0 +1,4 @@ +姓名,领域(可选提示),资料链接或备注(可选) +查理·芒格,投资/多元思维, +理查德·费曼,物理/教学, +纳瓦尔·拉维坎特,创业/杠杆,https://nav.al/ diff --git a/frontend-web/src/core/admin/api.ts b/frontend-web/src/core/admin/api.ts index 62f1be45..4e06f4d7 100644 --- a/frontend-web/src/core/admin/api.ts +++ b/frontend-web/src/core/admin/api.ts @@ -93,15 +93,6 @@ export interface AdminLeaderboardDimension { last_seen_at: string; } -export interface AdminLeaderboardQuestion { - text: string; - user_id: string; - email: string; - thread_id: string; - run_id: string; - asked_at: string; -} - export interface AdminLeaderboardTrendPoint { date: string; runs: number; @@ -138,8 +129,6 @@ export interface AdminLeaderboardResponse { models: AdminLeaderboardDimension[]; agents: AdminLeaderboardDimension[]; run_statuses: AdminLeaderboardDimension[]; - recent_questions: AdminLeaderboardQuestion[]; - frequent_questions: AdminLeaderboardQuestion[]; trends: AdminLeaderboardTrendPoint[]; } @@ -154,7 +143,6 @@ export interface AdminUserAnalyticsResponse { models: AdminLeaderboardDimension[]; agents: AdminLeaderboardDimension[]; run_statuses: AdminLeaderboardDimension[]; - recent_questions: AdminLeaderboardQuestion[]; trends: AdminLeaderboardTrendPoint[]; } diff --git a/frontend-web/src/pages/AdminUserLeaderboardPage.tsx b/frontend-web/src/pages/AdminUserLeaderboardPage.tsx index de98095f..6310a806 100644 --- a/frontend-web/src/pages/AdminUserLeaderboardPage.tsx +++ b/frontend-web/src/pages/AdminUserLeaderboardPage.tsx @@ -49,7 +49,6 @@ import { MarkdownContent } from "@/components/workspace/messages/markdown-conten import type { AdminLeaderboardDimension, AdminLeaderboardOverview, - AdminLeaderboardQuestion, AdminLeaderboardResponse, AdminLeaderboardSkill, AdminLeaderboardTrendPoint, @@ -66,7 +65,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 LEADERBOARD_PAGE_SIZE = 10; const accentClasses: Record = { crimson: { @@ -132,11 +131,6 @@ function fmtTime(value?: string | null): string { return Number.isNaN(date.getTime()) ? value : date.toLocaleString(); } -function clampText(value: string, max = 120): string { - const text = value.trim(); - return text.length > max ? `${text.slice(0, max)}...` : text; -} - function userScore(user: AdminLeaderboardUser, tab: UserRankTab): number { if (tab === "questions") return user.qa_count; if (tab === "tokens") return user.total_tokens; @@ -207,12 +201,6 @@ function markdownList(items: string[]): string { function buildReportFocusMarkdown(data: AdminLeaderboardResponse | undefined, range: RangeState): string { if (!data) return ""; - const recentQuestions = data.recent_questions - .slice(0, 8) - .map((item) => `${clampText(item.text, 120)}（${fmtTime(item.asked_at)}）`); - const frequentQuestions = data.frequent_questions - .slice(0, 8) - .map((item) => clampText(item.text, 120)); const frequentUsage = [ ...data.skills.slice(0, 5).map((item) => `技能：${item.name}，${fmtNumber(item.call_count)} 次调用`), ...data.tools.slice(0, 5).map((item) => `工具：${item.name}，${fmtNumber(item.call_count)} 次调用`), @@ -223,12 +211,6 @@ function buildReportFocusMarkdown(data: AdminLeaderboardResponse | undefined, ra "", `统计周期：${range.since} 至 ${range.until}`, "", - "## 最近关注", - markdownList(recentQuestions), - "", - "## 高频问题", - markdownList(frequentQuestions), - "", "## 使用频率重点", markdownList(frequentUsage), ].join("\n"); @@ -905,43 +887,6 @@ function DimensionRanking({ ); } -function QuestionPanel({ - title, - items, - showUser = true, -}: { - title: string; - items: AdminLeaderboardQuestion[]; - showUser?: boolean; -}) { - return ( -
-
-
- -

{title}

-
- {items.length} -
- -
- {items.length === 0 &&
暂无问题记录
} - {items.map((item, index) => ( -
-
{clampText(item.text, 180)}
-
- {showUser && {item.email || item.user_id || "未知用户"}} - {fmtTime(item.asked_at)} - {item.thread_id && 会话 {item.thread_id.slice(0, 8)}} -
-
- ))} -
- -
- ); -} - function Metric({ label, value }: { label: string; value: number }) { return (
@@ -1006,16 +951,12 @@ function UserDetailPanel({
- + - 提问 趋势 技能 维度 - - -
@@ -1354,36 +1295,7 @@ export default function AdminUserLeaderboardPage() {
{data && reportStatus === "idle" && !reportText && ( -
-
-
- - 最近关注 -
-
- {data.recent_questions.slice(0, 3).map((item, index) => ( -
- {fmtTime(item.asked_at)} - {clampText(item.text, 58)} -
- ))} - {data.recent_questions.length === 0 &&
暂无数据
} -
-
-
-
- - 高频问题 -
-
- {data.frequent_questions.slice(0, 3).map((item, index) => ( -
- {index + 1}. {clampText(item.text, 72)} -
- ))} - {data.frequent_questions.length === 0 &&
暂无数据
} -
-
+
@@ -1645,12 +1557,6 @@ export default function AdminUserLeaderboardPage() {
- - -
- - -
)}
diff --git a/frontend-web/src/pages/PublicLeaderboardPage.tsx b/frontend-web/src/pages/PublicLeaderboardPage.tsx index df783113..8513f106 100644 --- a/frontend-web/src/pages/PublicLeaderboardPage.tsx +++ b/frontend-web/src/pages/PublicLeaderboardPage.tsx @@ -34,12 +34,10 @@ import { } from "@/components/ui/dialog"; import { Input } from "@/components/ui/input"; import { Pagination } from "@/components/ui/pagination"; -import { ScrollArea } from "@/components/ui/scroll-area"; import { Tabs, TabsList, TabsTrigger } from "@/components/ui/tabs"; import type { AdminLeaderboardDimension, AdminLeaderboardOverview, - AdminLeaderboardQuestion, AdminLeaderboardResponse, AdminLeaderboardSkill, AdminLeaderboardTrendPoint, @@ -52,8 +50,10 @@ type UserRankTab = "activity" | "questions" | "tokens" | "skills" | "efficiency" type LeaderboardAccent = "crimson" | "rose" | "amber" | "teal"; const RANK_COLORS = ["#C1121F", "#E11D48", "#F59E0B", "#991B1B", "#B45309"] as const; -const LEADERBOARD_PAGE_SIZE = 20; -const RANGE_DAYS = 30; +const LEADERBOARD_PAGE_SIZE = 10; +// 公开页只统计近 7 天:跨日快照在每次请求时合并,天数越多合并越慢(尤其常问问题的 +// 模糊聚合是 O(n^2))。7 天足够展示近期热度,同时把请求耗时压到最低。 +const RANGE_DAYS = 7; const accentClasses: Record = { crimson: { @@ -98,11 +98,6 @@ function fmtDate(value?: string | null): string { return Number.isNaN(date.getTime()) ? "" : date.toLocaleString("zh-CN", { hour12: false }); } -function clampText(value: string, max = 120): string { - const text = value.trim(); - return text.length > max ? `${text.slice(0, max)}...` : text; -} - function userScore(user: AdminLeaderboardUser, tab: UserRankTab): number { if (tab === "questions") return user.qa_count; if (tab === "tokens") return user.total_tokens; @@ -367,12 +362,6 @@ export default function PublicLeaderboardPage() {
- - {/* 问题 */} -
- - -
)} @@ -688,40 +677,6 @@ function DimensionRanking({ ); } -function QuestionPanel({ - title, - items, -}: { - title: string; - items: AdminLeaderboardQuestion[]; -}) { - return ( -
-
-
- -

{title}

-
- {items.length} -
- -
- {items.length === 0 &&
暂无问题记录
} - {items.map((item, index) => ( -
-
{clampText(item.text, 180)}
-
- {item.email || item.user_id || "未知用户"} - {fmtTime(item.asked_at)} -
-
- ))} -
- -
- ); -} - function TrendBars({ points }: { points: AdminLeaderboardTrendPoint[] }) { type TrendMetric = keyof Pick; const [metric, setMetric] = useState<"all" | TrendMetric>("all"); 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 ead24625..8d6b5711 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/admin_users.py +++ b/offline-backend-20260512/backend/app/gateway/routers/admin_users.py @@ -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, ) diff --git a/offline-backend-20260512/backend/app/gateway/routers/public_leaderboard.py b/offline-backend-20260512/backend/app/gateway/routers/public_leaderboard.py index b856d6e2..7d0d588c 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/public_leaderboard.py +++ b/offline-backend-20260512/backend/app/gateway/routers/public_leaderboard.py @@ -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