1944 lines
78 KiB
Python
1944 lines
78 KiB
Python
"""Admin-only window into another user's activity.
|
||
|
||
Lets an administrator inspect any user's conversation history (threads + the
|
||
Q&A messages inside them) and their accumulated memory (global + per-agent
|
||
buckets). All endpoints require ``system_role == "admin"``; they are strictly
|
||
read-only.
|
||
|
||
The underlying stores normally auto-scope every query to the *calling* user
|
||
via a contextvar. These endpoints pass an explicit ``user_id`` so the admin
|
||
reads the *target* user's data instead — which is exactly what those store
|
||
methods' ``user_id`` parameter is designed for.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
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 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 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
|
||
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.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.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_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:
|
||
"""Reject non-admin callers. In no-auth mode there is no user → allow."""
|
||
try:
|
||
user = await get_current_user_from_request(request)
|
||
except HTTPException:
|
||
raise
|
||
if user is None:
|
||
return
|
||
if getattr(user, "system_role", None) != "admin":
|
||
raise HTTPException(status_code=403, detail="Admin only")
|
||
|
||
|
||
# ── Leaderboard / analytics ─────────────────────────────────────────────────
|
||
|
||
|
||
class AdminLeaderboardUser(BaseModel):
|
||
user_id: str
|
||
email: str = ""
|
||
system_role: str = "user"
|
||
thread_count: int = 0
|
||
run_count: int = 0
|
||
qa_count: int = 0
|
||
llm_call_count: int = 0
|
||
tool_call_count: int = 0
|
||
skill_call_count: int = 0
|
||
total_tokens: int = 0
|
||
input_tokens: int = 0
|
||
output_tokens: int = 0
|
||
last_active_at: str = ""
|
||
|
||
|
||
class AdminLeaderboardSkill(BaseModel):
|
||
name: str
|
||
call_count: int = 0
|
||
user_count: int = 0
|
||
success_count: int = 0
|
||
error_count: int = 0
|
||
avg_duration_ms: int = 0
|
||
last_used_at: str = ""
|
||
|
||
|
||
class AdminLeaderboardDimension(BaseModel):
|
||
name: str
|
||
count: int = 0
|
||
user_count: int = 0
|
||
success_count: int = 0
|
||
error_count: int = 0
|
||
total_tokens: int = 0
|
||
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
|
||
questions: int = 0
|
||
tokens: int = 0
|
||
users: int = 0
|
||
|
||
|
||
class AdminLeaderboardOverview(BaseModel):
|
||
total_users: int = 0
|
||
active_users: int = 0
|
||
total_threads: int = 0
|
||
total_runs: int = 0
|
||
total_questions: int = 0
|
||
total_tokens: int = 0
|
||
total_skill_calls: int = 0
|
||
total_tool_calls: int = 0
|
||
error_runs: int = 0
|
||
|
||
|
||
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)
|
||
users_by_tokens: list[AdminLeaderboardUser] = Field(default_factory=list)
|
||
skills: list[AdminLeaderboardSkill] = Field(default_factory=list)
|
||
tools: list[AdminLeaderboardSkill] = Field(default_factory=list)
|
||
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)
|
||
|
||
|
||
class AdminUserAnalyticsResponse(BaseModel):
|
||
range_days: int
|
||
since: str
|
||
until: str
|
||
user: AdminLeaderboardUser
|
||
overview: AdminLeaderboardOverview
|
||
skills: list[AdminLeaderboardSkill] = Field(default_factory=list)
|
||
tools: list[AdminLeaderboardSkill] = Field(default_factory=list)
|
||
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)
|
||
|
||
|
||
class AdminLeaderboardReportResponse(BaseModel):
|
||
report: str
|
||
generated_at: str
|
||
cached: bool = False
|
||
model_name: str | None = None
|
||
|
||
|
||
def _now() -> datetime:
|
||
return datetime.now()
|
||
|
||
|
||
def _iso(value) -> str:
|
||
return value.isoformat() if isinstance(value, datetime) else str(value or "")
|
||
|
||
|
||
def _parse_datetime(value: str | None, fallback: datetime) -> datetime:
|
||
if not value:
|
||
return fallback
|
||
text = value.strip()
|
||
if not text:
|
||
return fallback
|
||
if text.endswith("Z"):
|
||
text = text[:-1] + "+00:00"
|
||
try:
|
||
return datetime.fromisoformat(text)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=400, detail=f"Invalid datetime: {value}") from exc
|
||
|
||
|
||
def _resolve_range(days: int, since: str | None, until: str | None) -> tuple[datetime, datetime, int]:
|
||
end = _parse_datetime(until, _now())
|
||
start = _parse_datetime(since, end - timedelta(days=max(1, min(days, 365))))
|
||
if start > end:
|
||
raise HTTPException(status_code=400, detail="since must be before until")
|
||
span_days = max(1, min(365, (end - start).days + 1))
|
||
return start, end, span_days
|
||
|
||
|
||
def _question_count_expr():
|
||
return func.sum(case((RunRow.first_human_message.is_not(None), 1), else_=0))
|
||
|
||
|
||
_SYSTEM_MARKER_RE = re.compile(r"<(?:uploaded_files|attachments|files|system|metadata)>.*?</(?:uploaded_files|attachments|files|system|metadata)>", re.I | re.S)
|
||
_TAG_RE = re.compile(r"</?[^>]+>")
|
||
|
||
|
||
def _clean_question_text(value: str | None) -> str:
|
||
text = str(value or "").strip()
|
||
if not text:
|
||
return ""
|
||
text = _SYSTEM_MARKER_RE.sub(" ", text)
|
||
text = _TAG_RE.sub(" ", text)
|
||
text = re.sub(r"\s+", " ", text).strip()
|
||
return text
|
||
|
||
|
||
def _is_displayable_question(value: str | None) -> bool:
|
||
text = _clean_question_text(value)
|
||
if not text:
|
||
return False
|
||
lowered = text.lower()
|
||
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 _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": 5,
|
||
"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)
|
||
|
||
|
||
def _non_scheduled_run_filter():
|
||
"""Exclude scheduler-generated runs that predate scheduled_task_runs links."""
|
||
return or_(
|
||
RunRow.first_human_message.is_(None),
|
||
not_(RunRow.first_human_message.like("【定时任务触发】%")),
|
||
)
|
||
|
||
|
||
def _non_scheduled_tool_filter():
|
||
return or_(
|
||
ToolCallMetricRow.run_id.is_(None),
|
||
RunRow.run_id.is_(None),
|
||
_non_scheduled_run_filter(),
|
||
)
|
||
|
||
|
||
def _tool_time_filter(since: datetime, until: datetime):
|
||
return and_(ToolCallMetricRow.created_at >= since, ToolCallMetricRow.created_at <= until)
|
||
|
||
|
||
def _llm_time_filter(since: datetime, until: datetime):
|
||
return and_(LlmCallMetricRow.created_at >= since, LlmCallMetricRow.created_at <= until)
|
||
|
||
|
||
def _exclude_user_filter(column, exclude_user_ids: set[str]):
|
||
if not exclude_user_ids:
|
||
return None
|
||
return or_(column.is_(None), column.not_in(list(exclude_user_ids)))
|
||
|
||
|
||
def _apply_exclude(stmt, column, exclude_user_ids: set[str]):
|
||
condition = _exclude_user_filter(column, exclude_user_ids)
|
||
return stmt.where(condition) if condition is not None else stmt
|
||
|
||
|
||
def _admin_user_ids_stmt():
|
||
return select(UserRow.id).where(UserRow.system_role == "admin")
|
||
|
||
|
||
def _scheduled_run_ids_stmt():
|
||
return select(ScheduledTaskRunRow.agent_run_id).where(ScheduledTaskRunRow.agent_run_id.is_not(None))
|
||
|
||
|
||
def _apply_user_scope(stmt, column, *, exclude_user_ids: set[str], include_admins: bool):
|
||
stmt = _apply_exclude(stmt, column, exclude_user_ids)
|
||
if not include_admins:
|
||
stmt = stmt.where(or_(column.is_(None), column.not_in(_admin_user_ids_stmt())))
|
||
return stmt
|
||
|
||
|
||
def _apply_run_scope(
|
||
stmt,
|
||
*,
|
||
user_column,
|
||
run_id_column,
|
||
exclude_user_ids: set[str],
|
||
include_admins: bool,
|
||
include_failed: bool,
|
||
include_scheduled: bool,
|
||
):
|
||
stmt = _apply_user_scope(stmt, user_column, exclude_user_ids=exclude_user_ids, include_admins=include_admins)
|
||
if not include_failed:
|
||
stmt = stmt.where(RunRow.status != "error")
|
||
if not include_scheduled:
|
||
stmt = stmt.where(or_(run_id_column.is_(None), run_id_column.not_in(_scheduled_run_ids_stmt())))
|
||
stmt = stmt.where(_non_scheduled_run_filter())
|
||
return stmt
|
||
|
||
|
||
def _apply_tool_scope(
|
||
stmt,
|
||
*,
|
||
exclude_user_ids: set[str],
|
||
include_admins: bool,
|
||
include_scheduled: bool,
|
||
):
|
||
stmt = _apply_user_scope(stmt, ToolCallMetricRow.user_id, exclude_user_ids=exclude_user_ids, include_admins=include_admins)
|
||
if not include_scheduled:
|
||
stmt = stmt.where(or_(ToolCallMetricRow.run_id.is_(None), ToolCallMetricRow.run_id.not_in(_scheduled_run_ids_stmt())))
|
||
stmt = stmt.outerjoin(RunRow, RunRow.run_id == ToolCallMetricRow.run_id).where(_non_scheduled_tool_filter())
|
||
return stmt
|
||
|
||
|
||
async def _query_user_leaderboard(
|
||
session,
|
||
since: datetime,
|
||
until: datetime,
|
||
user_id: str | None = None,
|
||
exclude_user_ids: set[str] | None = None,
|
||
include_scheduled: bool = True,
|
||
include_admins: bool = True,
|
||
include_failed: bool = True,
|
||
) -> list[AdminLeaderboardUser]:
|
||
exclude_user_ids = exclude_user_ids or set()
|
||
run_agg = (
|
||
select(
|
||
RunRow.user_id.label("user_id"),
|
||
func.count(RunRow.run_id).label("run_count"),
|
||
_question_count_expr().label("qa_count"),
|
||
func.coalesce(func.sum(RunRow.llm_call_count), 0).label("llm_call_count"),
|
||
func.coalesce(func.sum(RunRow.total_tokens), 0).label("total_tokens"),
|
||
func.coalesce(func.sum(RunRow.total_input_tokens), 0).label("input_tokens"),
|
||
func.coalesce(func.sum(RunRow.total_output_tokens), 0).label("output_tokens"),
|
||
func.max(RunRow.updated_at).label("last_active_at"),
|
||
)
|
||
.where(_run_time_filter(since, until))
|
||
.group_by(RunRow.user_id)
|
||
)
|
||
run_agg = _apply_run_scope(
|
||
run_agg,
|
||
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,
|
||
).subquery()
|
||
thread_agg = (
|
||
select(ThreadMetaRow.user_id.label("user_id"), func.count(ThreadMetaRow.thread_id).label("thread_count"))
|
||
.where(and_(ThreadMetaRow.created_at >= since, ThreadMetaRow.created_at <= until))
|
||
.group_by(ThreadMetaRow.user_id)
|
||
)
|
||
thread_agg = _apply_user_scope(
|
||
thread_agg,
|
||
ThreadMetaRow.user_id,
|
||
exclude_user_ids=exclude_user_ids,
|
||
include_admins=include_admins,
|
||
).subquery()
|
||
tool_agg = (
|
||
select(
|
||
ToolCallMetricRow.user_id.label("user_id"),
|
||
func.count(ToolCallMetricRow.id).label("tool_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)
|
||
)
|
||
tool_agg = _apply_tool_scope(
|
||
tool_agg,
|
||
exclude_user_ids=exclude_user_ids,
|
||
include_admins=include_admins,
|
||
include_scheduled=include_scheduled,
|
||
).subquery()
|
||
|
||
stmt = (
|
||
select(
|
||
UserRow.id,
|
||
UserRow.email,
|
||
UserRow.system_role,
|
||
func.coalesce(thread_agg.c.thread_count, 0),
|
||
func.coalesce(run_agg.c.run_count, 0),
|
||
func.coalesce(run_agg.c.qa_count, 0),
|
||
func.coalesce(run_agg.c.llm_call_count, 0),
|
||
func.coalesce(tool_agg.c.tool_call_count, 0),
|
||
func.coalesce(tool_agg.c.skill_call_count, 0),
|
||
func.coalesce(run_agg.c.total_tokens, 0),
|
||
func.coalesce(run_agg.c.input_tokens, 0),
|
||
func.coalesce(run_agg.c.output_tokens, 0),
|
||
run_agg.c.last_active_at,
|
||
)
|
||
.outerjoin(run_agg, run_agg.c.user_id == UserRow.id)
|
||
.outerjoin(thread_agg, thread_agg.c.user_id == UserRow.id)
|
||
.outerjoin(tool_agg, tool_agg.c.user_id == UserRow.id)
|
||
)
|
||
if user_id:
|
||
stmt = stmt.where(UserRow.id == user_id)
|
||
stmt = _apply_user_scope(stmt, UserRow.id, exclude_user_ids=exclude_user_ids, include_admins=include_admins)
|
||
stmt = stmt.order_by(
|
||
desc(func.coalesce(run_agg.c.run_count, 0) + func.coalesce(tool_agg.c.tool_call_count, 0)),
|
||
desc(func.coalesce(run_agg.c.total_tokens, 0)),
|
||
)
|
||
rows = (await session.execute(stmt)).all()
|
||
return [
|
||
AdminLeaderboardUser(
|
||
user_id=r[0],
|
||
email=r[1] or "",
|
||
system_role=r[2] or "user",
|
||
thread_count=int(r[3] or 0),
|
||
run_count=int(r[4] or 0),
|
||
qa_count=int(r[5] or 0),
|
||
llm_call_count=int(r[6] or 0),
|
||
tool_call_count=int(r[7] or 0),
|
||
skill_call_count=int(r[8] or 0),
|
||
total_tokens=int(r[9] or 0),
|
||
input_tokens=int(r[10] or 0),
|
||
output_tokens=int(r[11] or 0),
|
||
last_active_at=_iso(r[12]),
|
||
)
|
||
for r in rows
|
||
]
|
||
|
||
|
||
async def _query_call_rankings(
|
||
session,
|
||
since: datetime,
|
||
until: datetime,
|
||
*,
|
||
user_id: str | None,
|
||
field: str,
|
||
limit: int,
|
||
exclude_user_ids: set[str] | None = None,
|
||
include_scheduled: bool = True,
|
||
include_admins: bool = True,
|
||
) -> list[AdminLeaderboardSkill]:
|
||
exclude_user_ids = exclude_user_ids or set()
|
||
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 = (
|
||
select(
|
||
name_col.label("name"),
|
||
call_count,
|
||
func.count(func.distinct(ToolCallMetricRow.user_id)).label("user_count"),
|
||
func.sum(case((ToolCallMetricRow.status == "success", 1), else_=0)).label("success_count"),
|
||
func.sum(case((ToolCallMetricRow.status != "success", 1), else_=0)).label("error_count"),
|
||
func.coalesce(func.avg(ToolCallMetricRow.duration_ms), 0).label("avg_duration_ms"),
|
||
last_used_at,
|
||
)
|
||
.where(_tool_time_filter(since, until), name_col.is_not(None), name_col != "")
|
||
.group_by(name_col)
|
||
.order_by(call_count.desc(), last_used_at.desc())
|
||
.limit(limit)
|
||
)
|
||
if user_id:
|
||
stmt = stmt.where(ToolCallMetricRow.user_id == user_id)
|
||
stmt = _apply_tool_scope(
|
||
stmt,
|
||
exclude_user_ids=exclude_user_ids,
|
||
include_admins=include_admins,
|
||
include_scheduled=include_scheduled,
|
||
)
|
||
rows = (await session.execute(stmt)).all()
|
||
return [
|
||
AdminLeaderboardSkill(
|
||
name=r[0] or "",
|
||
call_count=int(r[1] or 0),
|
||
user_count=int(r[2] or 0),
|
||
success_count=int(r[3] or 0),
|
||
error_count=int(r[4] or 0),
|
||
avg_duration_ms=int(r[5] or 0),
|
||
last_used_at=_iso(r[6]),
|
||
)
|
||
for r in rows
|
||
]
|
||
|
||
|
||
async def _query_run_dimensions(
|
||
session,
|
||
since: datetime,
|
||
until: datetime,
|
||
*,
|
||
user_id: str | None,
|
||
field: str,
|
||
limit: int,
|
||
exclude_user_ids: set[str] | None = None,
|
||
include_scheduled: bool = True,
|
||
include_admins: bool = True,
|
||
include_failed: bool = True,
|
||
) -> list[AdminLeaderboardDimension]:
|
||
exclude_user_ids = exclude_user_ids or set()
|
||
if field == "model":
|
||
name_col = RunRow.model_name
|
||
elif field == "agent":
|
||
name_col = RunRow.assistant_id
|
||
elif field == "status":
|
||
name_col = RunRow.status
|
||
else:
|
||
raise ValueError(f"Unknown run dimension: {field}")
|
||
|
||
count = func.count(RunRow.run_id).label("count")
|
||
last_seen_at = func.max(RunRow.updated_at).label("last_seen_at")
|
||
stmt = (
|
||
select(
|
||
name_col.label("name"),
|
||
count,
|
||
func.count(func.distinct(RunRow.user_id)).label("user_count"),
|
||
func.sum(case((RunRow.status == "success", 1), else_=0)).label("success_count"),
|
||
func.sum(case((RunRow.status == "error", 1), else_=0)).label("error_count"),
|
||
func.coalesce(func.sum(RunRow.total_tokens), 0).label("total_tokens"),
|
||
last_seen_at,
|
||
)
|
||
.where(_run_time_filter(since, until), name_col.is_not(None), name_col != "")
|
||
.group_by(name_col)
|
||
.order_by(count.desc(), last_seen_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 session.execute(stmt)).all()
|
||
return [
|
||
AdminLeaderboardDimension(
|
||
name=r[0] or "",
|
||
count=int(r[1] or 0),
|
||
user_count=int(r[2] or 0),
|
||
success_count=int(r[3] or 0),
|
||
error_count=int(r[4] or 0),
|
||
total_tokens=int(r[5] or 0),
|
||
last_seen_at=_iso(r[6]),
|
||
)
|
||
for r in rows
|
||
]
|
||
|
||
|
||
async def _query_model_rankings(
|
||
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,
|
||
) -> list[AdminLeaderboardDimension]:
|
||
exclude_user_ids = exclude_user_ids or set()
|
||
count = func.count(LlmCallMetricRow.id).label("count")
|
||
last_seen_at = func.max(LlmCallMetricRow.created_at).label("last_seen_at")
|
||
stmt = (
|
||
select(
|
||
LlmCallMetricRow.model_name.label("name"),
|
||
count,
|
||
func.count(func.distinct(LlmCallMetricRow.user_id)).label("user_count"),
|
||
func.sum(case((LlmCallMetricRow.status == "success", 1), else_=0)).label("success_count"),
|
||
func.sum(case((LlmCallMetricRow.status == "error", 1), else_=0)).label("error_count"),
|
||
func.coalesce(func.sum(LlmCallMetricRow.total_tokens), 0).label("total_tokens"),
|
||
last_seen_at,
|
||
)
|
||
.where(_llm_time_filter(since, until), LlmCallMetricRow.model_name.is_not(None), LlmCallMetricRow.model_name != "")
|
||
.group_by(LlmCallMetricRow.model_name)
|
||
.order_by(count.desc(), last_seen_at.desc())
|
||
.limit(limit)
|
||
)
|
||
if user_id:
|
||
stmt = stmt.where(LlmCallMetricRow.user_id == user_id)
|
||
stmt = _apply_user_scope(
|
||
stmt,
|
||
LlmCallMetricRow.user_id,
|
||
exclude_user_ids=exclude_user_ids,
|
||
include_admins=include_admins,
|
||
)
|
||
if not include_failed:
|
||
stmt = stmt.where(LlmCallMetricRow.status != "error")
|
||
if not include_scheduled:
|
||
stmt = stmt.where(or_(LlmCallMetricRow.run_id.is_(None), LlmCallMetricRow.run_id.not_in(_scheduled_run_ids_stmt())))
|
||
rows = (await session.execute(stmt)).all()
|
||
return [
|
||
AdminLeaderboardDimension(
|
||
name=r[0] or "",
|
||
count=int(r[1] or 0),
|
||
user_count=int(r[2] or 0),
|
||
success_count=int(r[3] or 0),
|
||
error_count=int(r[4] or 0),
|
||
total_tokens=int(r[5] or 0),
|
||
last_seen_at=_iso(r[6]),
|
||
)
|
||
for r in rows
|
||
]
|
||
|
||
|
||
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 session.execute(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 session.execute(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,
|
||
until: datetime,
|
||
*,
|
||
user_id: str | None = None,
|
||
exclude_user_ids: set[str] | None = None,
|
||
include_scheduled: bool = True,
|
||
include_admins: bool = True,
|
||
include_failed: bool = True,
|
||
) -> list[AdminLeaderboardTrendPoint]:
|
||
exclude_user_ids = exclude_user_ids or set()
|
||
bucket = func.date(RunRow.created_at)
|
||
stmt = (
|
||
select(
|
||
bucket.label("date"),
|
||
func.count(RunRow.run_id).label("runs"),
|
||
_question_count_expr().label("questions"),
|
||
func.coalesce(func.sum(RunRow.total_tokens), 0).label("tokens"),
|
||
func.count(func.distinct(RunRow.user_id)).label("users"),
|
||
)
|
||
.where(_run_time_filter(since, until))
|
||
.group_by(bucket)
|
||
.order_by(bucket.asc())
|
||
)
|
||
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 session.execute(stmt)).all()
|
||
return [
|
||
AdminLeaderboardTrendPoint(
|
||
date=str(r[0]),
|
||
runs=int(r[1] or 0),
|
||
questions=int(r[2] or 0),
|
||
tokens=int(r[3] or 0),
|
||
users=int(r[4] or 0),
|
||
)
|
||
for r in rows
|
||
]
|
||
|
||
|
||
async def _query_overview(
|
||
session,
|
||
since: datetime,
|
||
until: datetime,
|
||
*,
|
||
user_id: str | None = None,
|
||
exclude_user_ids: set[str] | None = None,
|
||
include_scheduled: bool = True,
|
||
include_admins: bool = True,
|
||
include_failed: bool = True,
|
||
) -> AdminLeaderboardOverview:
|
||
exclude_user_ids = exclude_user_ids or set()
|
||
run_where = [_run_time_filter(since, until)]
|
||
thread_where = [and_(ThreadMetaRow.created_at >= since, ThreadMetaRow.created_at <= until)]
|
||
tool_where = [_tool_time_filter(since, until)]
|
||
if user_id:
|
||
run_where.append(RunRow.user_id == user_id)
|
||
thread_where.append(ThreadMetaRow.user_id == user_id)
|
||
tool_where.append(ToolCallMetricRow.user_id == user_id)
|
||
if exclude_user_ids:
|
||
run_where.append(_exclude_user_filter(RunRow.user_id, exclude_user_ids))
|
||
thread_where.append(_exclude_user_filter(ThreadMetaRow.user_id, exclude_user_ids))
|
||
tool_where.append(_exclude_user_filter(ToolCallMetricRow.user_id, exclude_user_ids))
|
||
if not include_admins:
|
||
run_where.append(or_(RunRow.user_id.is_(None), RunRow.user_id.not_in(_admin_user_ids_stmt())))
|
||
thread_where.append(or_(ThreadMetaRow.user_id.is_(None), ThreadMetaRow.user_id.not_in(_admin_user_ids_stmt())))
|
||
tool_where.append(or_(ToolCallMetricRow.user_id.is_(None), ToolCallMetricRow.user_id.not_in(_admin_user_ids_stmt())))
|
||
if not include_failed:
|
||
run_where.append(RunRow.status != "error")
|
||
if not include_scheduled:
|
||
run_where.append(or_(RunRow.run_id.is_(None), RunRow.run_id.not_in(_scheduled_run_ids_stmt())))
|
||
tool_where.append(or_(ToolCallMetricRow.run_id.is_(None), ToolCallMetricRow.run_id.not_in(_scheduled_run_ids_stmt())))
|
||
tool_where.append(_non_scheduled_tool_filter())
|
||
|
||
total_users_stmt = select(func.count(UserRow.id))
|
||
total_users_stmt = _apply_user_scope(total_users_stmt, UserRow.id, exclude_user_ids=exclude_user_ids, include_admins=include_admins)
|
||
total_users = 1 if user_id else (await session.execute(total_users_stmt)).scalar_one()
|
||
run_row = (await session.execute(
|
||
select(
|
||
func.count(RunRow.run_id),
|
||
_question_count_expr(),
|
||
func.coalesce(func.sum(RunRow.total_tokens), 0),
|
||
func.count(func.distinct(RunRow.user_id)),
|
||
func.sum(case((RunRow.status == "error", 1), else_=0)),
|
||
).where(*run_where)
|
||
)).one()
|
||
thread_count = (await session.execute(select(func.count(ThreadMetaRow.thread_id)).where(*thread_where))).scalar_one()
|
||
tool_row = (await session.execute(
|
||
select(
|
||
func.count(ToolCallMetricRow.id),
|
||
func.sum(case((func.nullif(ToolCallMetricRow.skill_name, "").is_not(None), 1), else_=0)),
|
||
)
|
||
.outerjoin(RunRow, RunRow.run_id == ToolCallMetricRow.run_id)
|
||
.where(*tool_where)
|
||
)).one()
|
||
|
||
return AdminLeaderboardOverview(
|
||
total_users=int(total_users or 0),
|
||
active_users=int(run_row[3] or 0),
|
||
total_threads=int(thread_count or 0),
|
||
total_runs=int(run_row[0] or 0),
|
||
total_questions=int(run_row[1] or 0),
|
||
total_tokens=int(run_row[2] or 0),
|
||
total_tool_calls=int(tool_row[0] or 0),
|
||
total_skill_calls=int(tool_row[1] or 0),
|
||
error_runs=int(run_row[4] or 0),
|
||
)
|
||
|
||
|
||
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,
|
||
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"),
|
||
exclude_user_ids: list[str] | None = Query(default=None, description="User ids excluded from this leaderboard aggregation"),
|
||
include_scheduled: bool | None = Query(default=None, description="Include runs created by scheduled tasks"),
|
||
include_admins: bool | None = Query(default=None, description="Include admin accounts in leaderboard aggregation"),
|
||
include_failed: bool | None = Query(default=None, description="Include failed runs and LLM calls"),
|
||
limit: int = Query(default=20, ge=5, le=100),
|
||
) -> AdminLeaderboardResponse:
|
||
"""Return admin analytics for user activity, Q&A, skills, questions, and trends."""
|
||
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}
|
||
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(),
|
||
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)
|
||
|
||
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(
|
||
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,
|
||
"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)
|
||
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]
|
||
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]],
|
||
"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 (
|
||
"请基于以下管理员排行榜统计数据，生成一份中文用户行为分析报告。"
|
||
"报告需要有结论，不要只罗列数字；请使用 Markdown 标题、列表和表格，让前端可以直接渲染。"
|
||
"重点要求："
|
||
"1) 你是专门负责分析用户行为的专家，不要自称企业 AI 应用运营分析师；"
|
||
"2) 优先突出用户最近常用问题、最近关注的实时问题、事件、主题和业务方向，必须引用 recent_questions 中的近期问题作为依据；"
|
||
"3) 单独分析 frequent_questions，归纳高频关注簇，说明这些高频问题反映出的需求强度和重复频率；"
|
||
"4) 结合 top_users、top_skills、top_tools、top_models 和 trends，说明使用频率、使用增长、常用能力和重点用户行为画像；"
|
||
"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_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="你是专门负责分析用户行为的专家，擅长从用户最近常用问题、关注主题、使用频率和调用指标中生成管理报告；报告聚焦用户行为，不讨论系统本身问题。"),
|
||
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
|
||
|
||
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=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="你是专门负责分析用户行为的专家，擅长从用户最近常用问题、关注主题、使用频率和调用指标中生成管理报告；报告聚焦用户行为，不讨论系统本身问题。"),
|
||
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"},
|
||
)
|
||
|
||
|
||
@router.get("/{user_id}/analytics", response_model=AdminUserAnalyticsResponse)
|
||
async def get_user_analytics(
|
||
user_id: str,
|
||
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=20, ge=5, le=100),
|
||
) -> AdminUserAnalyticsResponse:
|
||
"""Return one user's admin analytics profile."""
|
||
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
|
||
common = {
|
||
"include_scheduled": settings.include_scheduled,
|
||
"include_admins": settings.include_admins,
|
||
"include_failed": settings.include_failed,
|
||
}
|
||
tool_common = {
|
||
"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:
|
||
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")
|
||
overview = await _query_overview(session, since_dt, until_dt, user_id=user_id, **common)
|
||
skills = await _query_call_rankings(session, since_dt, until_dt, user_id=user_id, field="skill", limit=limit, **tool_common)
|
||
tools = await _query_call_rankings(session, since_dt, until_dt, user_id=user_id, field="tool", limit=limit, **tool_common)
|
||
models = await _query_model_rankings(session, since_dt, until_dt, user_id=user_id, limit=limit, **common)
|
||
if not models:
|
||
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(
|
||
range_days=range_days,
|
||
since=_iso(since_dt),
|
||
until=_iso(until_dt),
|
||
user=users[0],
|
||
overview=overview,
|
||
skills=skills,
|
||
tools=tools,
|
||
models=models,
|
||
agents=agents,
|
||
run_statuses=run_statuses,
|
||
recent_questions=recent_questions,
|
||
trends=trends,
|
||
)
|
||
|
||
|
||
# ── Threads ────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class AdminThreadItem(BaseModel):
|
||
thread_id: str
|
||
display_name: str = ""
|
||
status: str = "idle"
|
||
created_at: str = ""
|
||
updated_at: str = ""
|
||
|
||
|
||
@router.get("/{user_id}/threads", response_model=list[AdminThreadItem])
|
||
async def list_user_threads(
|
||
user_id: str,
|
||
request: Request,
|
||
limit: int = Query(default=100, ge=1, le=500),
|
||
offset: int = Query(default=0, ge=0),
|
||
) -> list[AdminThreadItem]:
|
||
"""List one user's conversation threads, newest first."""
|
||
await _require_admin(request)
|
||
repo = get_thread_store(request)
|
||
rows = await repo.search(limit=limit, offset=offset, user_id=user_id)
|
||
return [
|
||
AdminThreadItem(
|
||
thread_id=r["thread_id"],
|
||
display_name=r.get("display_name") or "",
|
||
status=r.get("status", "idle"),
|
||
created_at=str(r.get("created_at") or ""),
|
||
updated_at=str(r.get("updated_at") or ""),
|
||
)
|
||
for r in rows
|
||
]
|
||
|
||
|
||
@router.get("/{user_id}/threads/{thread_id}/messages", response_model=list[dict])
|
||
async def list_user_thread_messages(
|
||
user_id: str,
|
||
thread_id: str,
|
||
request: Request,
|
||
limit: int = Query(default=400, ge=1, le=1000),
|
||
) -> list[dict]:
|
||
"""Return the conversation messages of one of the user's threads.
|
||
|
||
Read straight from the LangGraph checkpoint — the same source the chat UI
|
||
renders — rather than the optional ``run_events`` audit log, which may be
|
||
unconfigured (``run_events.backend=memory``) or empty for older threads.
|
||
"""
|
||
await _require_admin(request)
|
||
|
||
repo = get_thread_store(request)
|
||
# ``user_id=None`` bypasses the owner filter so the admin can fetch any row;
|
||
# we then assert the thread really belongs to the target user.
|
||
thread = await repo.get(thread_id, user_id=None)
|
||
if thread is None:
|
||
raise HTTPException(status_code=404, detail="Thread not found")
|
||
owner = thread.get("user_id")
|
||
if owner is not None and owner != user_id:
|
||
raise HTTPException(status_code=404, detail="Thread does not belong to this user")
|
||
|
||
checkpointer = get_checkpointer(request)
|
||
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
|
||
try:
|
||
checkpoint_tuple = await checkpointer.aget_tuple(config)
|
||
except Exception as exc: # noqa: BLE001 - surface a clean error
|
||
logger.error("Failed to read checkpoint for thread %s: %s", thread_id, exc, exc_info=True)
|
||
raise HTTPException(status_code=500, detail="Failed to read thread state") from exc
|
||
if checkpoint_tuple is None:
|
||
return []
|
||
|
||
checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {}
|
||
channel_values = checkpoint.get("channel_values", {}) if isinstance(checkpoint, dict) else {}
|
||
values = serialize_channel_values(channel_values)
|
||
messages = values.get("messages", [])
|
||
if not isinstance(messages, list):
|
||
return []
|
||
# Keep the most recent ``limit`` messages — long threads stay bounded.
|
||
return messages[-limit:]
|
||
|
||
|
||
# ── Memory ─────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class AdminMemoryBucket(BaseModel):
|
||
name: str = Field(..., description="Bucket name: 'global' or the agent id (used for CRUD)")
|
||
display_name: str = Field("", description="Human-readable name shown in the UI")
|
||
data: dict = Field(default_factory=dict, description="Raw memory data for the bucket")
|
||
|
||
|
||
class AdminMemoryEntryAddRequest(BaseModel):
|
||
content: str = Field(..., min_length=1, description="Entry content to add")
|
||
agent: str | None = Field(None, description="Agent name; None = global USER.md bucket")
|
||
|
||
|
||
class AdminMemoryEntryReplaceRequest(BaseModel):
|
||
old_text: str = Field(..., min_length=1, description="Short unique substring to locate the entry")
|
||
content: str = Field(..., min_length=1, description="New content to replace with")
|
||
agent: str | None = Field(None, description="Agent name; None = global USER.md bucket")
|
||
|
||
|
||
class AdminMemoryResponse(BaseModel):
|
||
user_id: str
|
||
version: str = Field(default="v2", description="记忆版本：v1 或 v2")
|
||
buckets: list[AdminMemoryBucket] = Field(default_factory=list)
|
||
|
||
|
||
def _read_memory_bucket(user_id: str, agent_id: str | None) -> dict:
|
||
"""Read one Memory V2 file into a JSON-able ``{"entries": [...]}`` dict."""
|
||
provider = BuiltinFileProvider()
|
||
provider.initialize(user_id=user_id, agent_id=agent_id)
|
||
target = "user" if agent_id is None else "memory"
|
||
return {"entries": provider.read(target)}
|
||
|
||
|
||
def _read_v1_memory_bucket(user_id: str, agent_id: str | None) -> dict:
|
||
"""Read one V1 memory.json into its raw dict (facts / user / history)."""
|
||
provider = _build_v1_provider(user_id, agent_id)
|
||
return provider.read_v1_data()
|
||
|
||
|
||
@router.get("/{user_id}/memory", response_model=AdminMemoryResponse)
|
||
async def get_user_memory(user_id: str, request: Request) -> AdminMemoryResponse:
|
||
"""Return every memory bucket for one user (V1 or V2 depending on config)."""
|
||
await _require_admin(request)
|
||
|
||
# Source agents from the database: only agents owned by this specific user.
|
||
agent_store = get_agent_store(request)
|
||
all_agents = await agent_store.list_visible(user_id)
|
||
owned_agents = sorted(
|
||
[a for a in all_agents if a.get("user_id") == user_id],
|
||
key=lambda a: (a.get("name") or a["id"]).lower(),
|
||
)
|
||
|
||
if _is_v1(user_id):
|
||
# V1 模式：每个桶读 memory.json，内容为 {facts, user, history}
|
||
buckets: list[AdminMemoryBucket] = []
|
||
try:
|
||
buckets.append(AdminMemoryBucket(name="default", display_name="默认助手", data=_read_v1_memory_bucket(user_id, None)))
|
||
except Exception:
|
||
logger.exception("Failed to read V1 global memory for user %s", user_id)
|
||
buckets.append(AdminMemoryBucket(name="default", display_name="默认助手", data={}))
|
||
|
||
for agent in owned_agents:
|
||
agent_id: str = agent["id"]
|
||
display_name: str = agent.get("name") or agent_id
|
||
try:
|
||
data = _read_v1_memory_bucket(user_id, agent_id)
|
||
except Exception:
|
||
data = {}
|
||
buckets.append(AdminMemoryBucket(name=agent_id, display_name=display_name, data=data))
|
||
|
||
return AdminMemoryResponse(user_id=user_id, version="v1", buckets=buckets)
|
||
|
||
# V2 模式：读 USER.md / MEMORY.md，内容为 {entries: [...]}
|
||
buckets = [
|
||
AdminMemoryBucket(name="default", display_name="默认助手", data=_read_memory_bucket(user_id, None))
|
||
]
|
||
for agent in owned_agents:
|
||
agent_id = agent["id"]
|
||
display_name = agent.get("name") or agent_id
|
||
try:
|
||
data = _read_memory_bucket(user_id, agent_id)
|
||
except Exception:
|
||
data = {"entries": []}
|
||
buckets.append(AdminMemoryBucket(name=agent_id, display_name=display_name, data=data))
|
||
|
||
return AdminMemoryResponse(user_id=user_id, version="v2", buckets=buckets)
|
||
|
||
|
||
@router.post("/{user_id}/memory/entries", response_model=AdminMemoryBucket)
|
||
async def admin_add_memory_entry(
|
||
user_id: str,
|
||
body: AdminMemoryEntryAddRequest,
|
||
request: Request,
|
||
) -> AdminMemoryBucket:
|
||
"""Add a memory entry for a user (admin only)."""
|
||
await _require_admin(request)
|
||
target = "user" if body.agent is None else "memory"
|
||
provider = BuiltinFileProvider()
|
||
provider.initialize(user_id=user_id, agent_id=body.agent)
|
||
try:
|
||
result = provider.add(target, body.content)
|
||
if result is False:
|
||
raise HTTPException(status_code=400, detail="Failed to add entry (quota exceeded or write error)")
|
||
except HTTPException:
|
||
raise
|
||
except Exception as exc:
|
||
logger.error("admin_add_memory_entry failed for user %s: %s", user_id, exc, exc_info=True)
|
||
raise HTTPException(status_code=500, detail="Failed to write memory entry") from exc
|
||
bucket_name = "global" if body.agent is None else body.agent
|
||
return AdminMemoryBucket(name=bucket_name, data=_read_memory_bucket(user_id, body.agent))
|
||
|
||
|
||
@router.put("/{user_id}/memory/entries", response_model=AdminMemoryBucket)
|
||
async def admin_replace_memory_entry(
|
||
user_id: str,
|
||
body: AdminMemoryEntryReplaceRequest,
|
||
request: Request,
|
||
) -> AdminMemoryBucket:
|
||
"""Replace a memory entry for a user (admin only)."""
|
||
await _require_admin(request)
|
||
target = "user" if body.agent is None else "memory"
|
||
provider = BuiltinFileProvider()
|
||
provider.initialize(user_id=user_id, agent_id=body.agent)
|
||
try:
|
||
result = provider.replace(target, body.old_text, body.content)
|
||
if result is False:
|
||
raise HTTPException(status_code=404, detail="Entry not found or write failed")
|
||
except HTTPException:
|
||
raise
|
||
except Exception as exc:
|
||
logger.error("admin_replace_memory_entry failed for user %s: %s", user_id, exc, exc_info=True)
|
||
raise HTTPException(status_code=500, detail="Failed to replace memory entry") from exc
|
||
bucket_name = "global" if body.agent is None else body.agent
|
||
return AdminMemoryBucket(name=bucket_name, data=_read_memory_bucket(user_id, body.agent))
|
||
|
||
|
||
@router.delete("/{user_id}/memory/entries", response_model=AdminMemoryBucket)
|
||
async def admin_delete_memory_entry(
|
||
user_id: str,
|
||
request: Request,
|
||
old_text: str = Query(..., min_length=1, description="Short unique substring to locate the entry"),
|
||
agent: str | None = Query(None, description="Agent name; None = global USER.md bucket"),
|
||
) -> AdminMemoryBucket:
|
||
"""Delete a memory entry for a user (admin only)."""
|
||
await _require_admin(request)
|
||
target = "user" if agent is None else "memory"
|
||
provider = BuiltinFileProvider()
|
||
provider.initialize(user_id=user_id, agent_id=agent)
|
||
try:
|
||
result = provider.remove(target, old_text)
|
||
if result is False:
|
||
raise HTTPException(status_code=404, detail="Entry not found")
|
||
except HTTPException:
|
||
raise
|
||
except Exception as exc:
|
||
logger.error("admin_delete_memory_entry failed for user %s: %s", user_id, exc, exc_info=True)
|
||
raise HTTPException(status_code=500, detail="Failed to delete memory entry") from exc
|
||
bucket_name = "global" if agent is None else agent
|
||
return AdminMemoryBucket(name=bucket_name, data=_read_memory_bucket(user_id, agent))
|
||
|
||
|
||
# ── V1 Memory CRUD (admin) ─────────────────────────────────────────────────
|
||
|
||
_V1_USER_FIELDS = {"workContext", "personalContext", "topOfMind"}
|
||
_V1_HISTORY_FIELDS = {"recentMonths", "earlierContext", "longTermBackground"}
|
||
|
||
|
||
class AdminV1SectionUpdateRequest(BaseModel):
|
||
section_type: str = Field(..., description="'user' 或 'history'")
|
||
field: str = Field(..., description="字段名，如 workContext / recentMonths")
|
||
summary: str = Field(..., description="新的摘要文本")
|
||
agent: str | None = Field(None, description="Agent id；None 表示用户级全局记忆")
|
||
|
||
|
||
class AdminV1FactAddRequest(BaseModel):
|
||
content: str = Field(..., min_length=1)
|
||
category: str = Field(default="context")
|
||
confidence: float = Field(default=0.8, ge=0.0, le=1.0)
|
||
agent: str | None = Field(None)
|
||
|
||
|
||
class AdminV1FactUpdateRequest(BaseModel):
|
||
content: str | None = Field(None)
|
||
category: str | None = Field(None)
|
||
confidence: float | None = Field(None, ge=0.0, le=1.0)
|
||
agent: str | None = Field(None)
|
||
|
||
|
||
@router.put("/{user_id}/memory/v1/sections", response_model=AdminMemoryBucket)
|
||
async def admin_update_v1_section(
|
||
user_id: str,
|
||
body: AdminV1SectionUpdateRequest,
|
||
request: Request,
|
||
) -> AdminMemoryBucket:
|
||
"""更新 V1 记忆的某个摘要字段（用户画像 / 历史）。"""
|
||
await _require_admin(request)
|
||
if body.section_type == "user" and body.field not in _V1_USER_FIELDS:
|
||
raise HTTPException(status_code=400, detail=f"无效 user 字段：{body.field}")
|
||
if body.section_type == "history" and body.field not in _V1_HISTORY_FIELDS:
|
||
raise HTTPException(status_code=400, detail=f"无效 history 字段：{body.field}")
|
||
if body.section_type not in ("user", "history"):
|
||
raise HTTPException(status_code=400, detail="section_type 须为 'user' 或 'history'")
|
||
|
||
from deerflow.agents.memory.v1.storage import get_memory_storage, utc_now_iso_z
|
||
storage = get_memory_storage()
|
||
data = storage.load(body.agent, user_id=user_id)
|
||
data.setdefault(body.section_type, {})[body.field] = {
|
||
"summary": body.summary,
|
||
"updatedAt": utc_now_iso_z(),
|
||
}
|
||
if not storage.save(data, body.agent, user_id=user_id):
|
||
raise HTTPException(status_code=500, detail="写入失败")
|
||
|
||
bucket_name = "global" if body.agent is None else body.agent
|
||
return AdminMemoryBucket(name=bucket_name, data=_read_v1_memory_bucket(user_id, body.agent))
|
||
|
||
|
||
@router.post("/{user_id}/memory/v1/facts", response_model=AdminMemoryBucket)
|
||
async def admin_add_v1_fact(
|
||
user_id: str,
|
||
body: AdminV1FactAddRequest,
|
||
request: Request,
|
||
) -> AdminMemoryBucket:
|
||
"""新增一条 V1 事实记录。"""
|
||
await _require_admin(request)
|
||
try:
|
||
from deerflow.agents.memory.v1.updater import create_memory_fact
|
||
create_memory_fact(
|
||
body.content,
|
||
category=body.category,
|
||
confidence=body.confidence,
|
||
agent_name=body.agent,
|
||
user_id=user_id,
|
||
)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||
except Exception as exc:
|
||
raise HTTPException(status_code=500, detail="新增失败") from exc
|
||
bucket_name = "global" if body.agent is None else body.agent
|
||
return AdminMemoryBucket(name=bucket_name, data=_read_v1_memory_bucket(user_id, body.agent))
|
||
|
||
|
||
@router.put("/{user_id}/memory/v1/facts/{fact_id}", response_model=AdminMemoryBucket)
|
||
async def admin_update_v1_fact(
|
||
user_id: str,
|
||
fact_id: str,
|
||
body: AdminV1FactUpdateRequest,
|
||
request: Request,
|
||
) -> AdminMemoryBucket:
|
||
"""更新一条 V1 事实记录（内容 / 类别 / 置信度）。"""
|
||
await _require_admin(request)
|
||
try:
|
||
from deerflow.agents.memory.v1.updater import update_memory_fact
|
||
update_memory_fact(
|
||
fact_id,
|
||
content=body.content,
|
||
category=body.category,
|
||
confidence=body.confidence,
|
||
agent_name=body.agent,
|
||
user_id=user_id,
|
||
)
|
||
except KeyError:
|
||
raise HTTPException(status_code=404, detail=f"事实 '{fact_id}' 不存在")
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||
except Exception as exc:
|
||
raise HTTPException(status_code=500, detail="更新失败") from exc
|
||
bucket_name = "global" if body.agent is None else body.agent
|
||
return AdminMemoryBucket(name=bucket_name, data=_read_v1_memory_bucket(user_id, body.agent))
|
||
|
||
|
||
@router.delete("/{user_id}/memory/v1/facts/{fact_id}", response_model=AdminMemoryBucket)
|
||
async def admin_delete_v1_fact(
|
||
user_id: str,
|
||
fact_id: str,
|
||
request: Request,
|
||
agent: str | None = Query(None, description="Agent id；None 表示用户级全局记忆"),
|
||
) -> AdminMemoryBucket:
|
||
"""删除一条 V1 事实记录。"""
|
||
await _require_admin(request)
|
||
try:
|
||
from deerflow.agents.memory.v1.updater import delete_memory_fact
|
||
delete_memory_fact(fact_id, agent_name=agent, user_id=user_id)
|
||
except KeyError:
|
||
raise HTTPException(status_code=404, detail=f"事实 '{fact_id}' 不存在")
|
||
except Exception as exc:
|
||
raise HTTPException(status_code=500, detail="删除失败") from exc
|
||
bucket_name = "global" if agent is None else agent
|
||
return AdminMemoryBucket(name=bucket_name, data=_read_v1_memory_bucket(user_id, agent))
|