diff --git a/CLAUDE.md b/CLAUDE.md index ba906966..d37524d5 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -86,6 +86,8 @@ The frontend (`frontend-web/`) is a React 19 + Vite SPA: - **敏感词管理 (Sensitive Word Management)**: configures the **展示层脱敏映射** (原始词 → 替换词). **Gated on the username `lqq`** (not the admin role) — `getAccountDisplayName().toLowerCase() === "lqq"` (username = email `@` prefix, else `user_id`, same as the profile page); the dropdown entry, the route guard, and the page-level guard all use this check. `src/lib/desensitize.ts` ships a frozen seed map `SEED_DESENSITIZE_MAP` (the deployment seed) plus a live mutable `DESENSITIZE_MAP` (initialized from the seed so `desensitize()` works before any fetch). `DesensitizeWordsLoader` (mounted in `App.tsx` `AppProviders`) fetches `/api/sensitive-words` on startup and, if non-empty, calls `replaceDesensitizeMap(...)` to make the backend config the source of truth (empty/failed → keep seed). Page `SensitiveWordsPage.tsx` (route `/page/strategy/admin/sensitive-words`, opened from the 「敏感词管理」 item in the **bottom-left settings dropdown** of `components/page-sidebar.tsx`, present in **both** full + collapsed sidebar modes) lists/adds/edits/启停/deletes words and saves the **whole set atomically** (replace-all); a 「导入种子词」 button merges any missing `SEED_DESENSITIZE_MAP` entries into the editor. Client + hooks in `strategy-components/api/sensitive-words.ts` (`wordsToMap` folds enabled rows into the runtime map). Backend: `/api/sensitive-words` (global/shared; open read, write allowed for admin **or** the `lqq` account), mirroring 菜单管理's store shape. **侧边栏 + 导航栏 menu labels are desensitized** at the data source: `useSidebarMenuItems`/`useActiveThirdLevel` (`core/page-layout/use-sidebar-menu.ts`) map every node's `label` through a desensitizer built **directly from the `/api/sensitive-words` query** (`makeDesensitizer(wordsToMap(words))`, falling back to the global seed `desensitize` while empty) — this sidesteps the `replaceDesensitizeMap` effect-timing race. Only `label` is rewritten; `path`/`id`/`icon` are untouched so routing/highlight are unaffected. All nav surfaces (`page-sidebar.tsx`, `page-sidebar-dark.tsx`, `page-sidebar-v1.tsx`, `strategy-components/Header.tsx`, `PageRoutes.tsx`) consume this hook and are covered; `MenuManagementPage` uses the raw tree so admins still edit real labels. - **技能地址管理 (Skill Address Management, admin-only)**: 扫描运行中的技能（`skills/{public,custom}` 下所有 `.md`/`.py`）里的 **http(s) URL 与 IP/IP:端口**，按唯一地址聚合全部出现位置（跨 .md/.py），支持**批量替换**（如把 UAT 地址一次性改成生产）。主体抽成可复用组件 `SkillAddressesPanel`（`pages/SkillAddressesPage.tsx` 导出，默认导出 `SkillAddressesPage` 仍保留旧路由 `/page/strategy/admin/skill-addresses` 作直链回退），**作为「技能地址管理」标签页嵌入管理员「技能管理」页 `pages/AdminSkillsPage.tsx`**（该页加 Tabs：「技能列表」+「技能地址管理」；入口在「设置和更多」下拉的「技能管理」`workspace-nav-menu.tsx` → `/page/workspace/admin/skills`，不再单列于 `components/page-sidebar.tsx` 左下角设置下拉）：打开自动扫描并带**进度条**（扫描逐技能 / 应用逐文件 X/N，应对 150+ 技能），URL/IP 类型过滤 + 搜索 + 「片段批量套用」（如所有 `uat.4.cn` → `prod.4.cn`）、每地址「替换为」输入框（可展开看 `file:line` 片段）、内联**预览**、**应用**、**多关键词检索**（空格分隔全部命中），以及**修改记录**面板——每次应用生成一条记录（显示时间/改了哪些地址 from→to/文件数/处数），可**针对单次记录回退**（二次确认）。client + NDJSON 流式解析在 `strategy-components/api/skill-addresses.ts`。后端 `/api/skill-addresses`（**仅管理员**）：所有文件 IO 走 `asyncio.to_thread` 不阻塞事件循环；`GET /scan`(+`/scan/stream`)、`POST /preview`、`POST /apply`(+`/apply/stream`)、`GET /backups`、`POST /rollback/{id}`；**子串安全替换**（基于 token span，不会误伤子串）、`expected_total` 乐观锁（409）、写前自动磁盘备份到 `{skills_root}/.address-edits/`（同时写一份 `{id}.manifest.json` 记录本次替换明细，供修改记录展示）、应用后刷新技能系统提示缓存。纯引擎 `deerflow/skills/address_scan.py`，router `app/gateway/routers/skill_addresses.py`，tests `tests/test_skill_addresses.py`。范围仅运行中技能，不动 `deploy/minimal/skills/`。 - **按钮管理 (Task Button Management, admin-only)**: configures the **可配置跳转按钮** rendered in the left task panel of the task deep-link workspaces (rwfx / 3qfx / xdfx, see `TaskFxWorkspace`). Each button = `{business, label, linkType, target, appendTaskId, idParam, idKind, openMode, loginParams, appendAuth, authTokenParam, authNameParam, enabled, sortOrder}`; `linkType` is `url` (外链 `window.open`, optional append of `?{idParam}={id}` — **`idParam` is the editable query-param name**, e.g. `id`/`task_id`), `business` (内部切换到另一业务, reuses `resolveTaskDeeplinkTarget`, always task id), or `purpose` (弹出当前任务的「任务目的」表格 via the shared `TaskPurposeDialog`, no target — configurable for any page incl. xdfx). **`url` buttons support two independent, coexisting login-append mechanisms**: (1) **`loginParams` (追加登录参数, like 应用管理/Light App Center)** — a list of `{key, source, customValue}` resolved at click time: `runTaskButton` reads `userInfo` (via `readUserInfo` from `strategy-components/lib/light-app`, embed-isolated through `authStorage`) and appends `?{key}={value}` for each (`source` = a userInfo field such as `accessToken`/`yUserId`, or `other` → literal `customValue`); empty list = nothing appended. Reuses `LOGIN_SOURCE_OPTIONS` from light-app; stored as JSON column `task_buttons.login_params_json` (`PortableJSON`, migration `20260625_01`). (2) **`appendAuth` (追加登录态)** — appends `?{authTokenParam}={token1}&{authNameParam}={username}` (default param names `token`/`username`) so the跳转目标外部系统 can identify the current user — e.g. `…/action/sentiment?token=&username=lizhaoxing`. **Gated on token login**: `token1` = the `?authToken=` used by token-exchange login; `username` = the raw login name the backend resolved from `getTokenInfo`. Both are stashed at login by `persistLoginCredentials` into `login.token1`/`login.username` (`core/auth/login-credentials.ts`, scoped via [[scoped-storage]] so嵌入/独立会话隔离 + 随独立会话迁移). Only token login writes them → a non-token-login session has no `token1`, so `runTaskButton` silently skips the auth append (the jump still works, just without those two params). The frontend reads `username` straight from the new `UsernameLoginResponse.username` field (the stored email local-part is lossy — lowercased/normalized — so not used), **not** by decoding the JWT. Stored as columns `task_buttons.{append_auth, auth_token_param, auth_name_param}` (migration `20260625_02`). The admin page renders both editors for `url` buttons (loginParams 虚线框 + 追加登录态 switch). **`idKind` (`task`/`action`) only matters for xdfx**: its deep-link `?taskId=` is an 行动id and the 任务id is resolved at login (`fetchTaskIdFromAction`) — `TaskContext` now also stashes the raw `actionId`, so a url button can append either (default `task`). Page `TaskButtonsPage.tsx` (route `/page/strategy/admin/task-buttons`, **admin-gated**, entry「按钮管理」in the bottom-left settings dropdown of `components/page-sidebar.tsx`, both full + collapsed modes) has per-business tabs, edits name/type/target/启停/排序/增删, saves the **whole set atomically** (replace-all), plus 「导入默认」 to merge code-seed defaults. **Seed/fallback** (like 敏感词管理): only when the whole table is empty (fresh deploy, never saved) do the workspace + page fall back to `defaultTaskButtons(cfg)` — rwfx's url buttons (编辑/上一步/下一步/3q详情) built from the **live** `/api/public/config/task-deeplink` plus cross-business switch buttons for **all three** businesses (each switches to the other two — rwfx→三情/行动, 3qfx→任务/行动, xdfx→任务/三情). Once saved, the store is the sole source of truth. **Bottom 快捷跳转 bar groups by prev/next**: `TaskJumpBar` renders only「上一步/下一步」-type buttons in the bottom bar — a `prev` group (id 含 `prev` 或文案「上一步」, 描边次要样式 + ←图标) rendered **左侧**, then a `next` group (id 含 `next` 或文案「下一步」, 实心主色) on the **右侧** (container `justify-end`); everything else goes to the right-edge floating menu. The default 上一步 button (`rwfx-prev`, `url`, target = `DASHBOARD_AGENT_ROUTE` `/page/strategy/dashboard-agent`, `appendTaskId` + `idParam=taskId` → `?taskId=任务id`, `openMode=self`) 默认指向**应用内**「大屏绘制」页，在 **both** 单智能体 (`TaskFxWorkspace`) 与 多智能体会商 (`RoundtablePlanningPage` 深链 rwfx-A) 都显示（共用 `TaskJumpBar`）。**`runTaskButton` 的 `url` 类型支持应用内路由**：target 以单个 `/` 开头时按内部路由处理——`self` 走 react-router 同窗导航（不刷新），`blank` 拼成当前源的 hash 路由新标签打开；外链行为不变。 The old hardcoded jump buttons were migrated out: `TaskFxWorkspace` no longer hardcodes 切换到三情分析, `TaskActionsPanel` keeps only the non-jump 清除记录, and `TaskPurposePanel` dropped its 3q详情 button (now configurable) **and defaults the 目的树 table to collapsed** (chevron toggle). Helpers `core/auth/task-buttons.ts` (`TaskButton`, `defaultTaskButtons`, `resolveTaskButtons`, `buttonsForBusiness`, `runTaskButton`), client+hooks `strategy-components/api/task-buttons.ts`. Backend: `/api/task-buttons` (global/shared; open read, admin replace-all), store `deerflow.persistence.task_buttons` (table `task_buttons`, auto-created by `create_all`), wired `app.state.task_button_store`. Tests: `tests/test_task_buttons.py`. +- **实时并发监控 (Concurrency / System-Pressure Monitor, admin-only)**: a live panel mounted at the **top of the admin 排行榜 page** (`components/workspace/admin/concurrency-monitor.tsx` `ConcurrencyMonitorPanel`, rendered in `pages/AdminUserLeaderboardPage.tsx`) showing **当前并发的大模型调用数 + 列表（用户/智能体/类型/已运行时长）**, each row with a 「停止」button (and a 「全部停止」) that **truly aborts** that run's in-flight model stream via `RunManager.cancel` (`POST /api/admin/active-runs/{run_id}/cancel`) — to relieve pressure when 使用人数过多. Two line charts (minute/hour toggle): **并发量折线图** (peak/avg, `GET /api/admin/active-runs/concurrency`, fed by a lightweight background sampler into `concurrency_samples`) and **大模型出字速度折线图** (tokens/sec, `GET /api/admin/active-runs/token-speed`, aggregated from `llm_call_metrics`) so a dip pinpoints where the bottleneck is. A **「后台采样」Switch** (`system_settings.json → concurrency_monitor.enabled`, **default off / opt-in**, `GET|PUT /api/admin/active-runs/settings`) turns the background sampling on/off **at runtime, no restart** (the live concurrency count + 「停止」 work even while sampling is off — only the 并发量 history chart needs it on); everything is designed to **never impact production** (one tiny INSERT/15s, retention-purged, best-effort). Client/hooks: `core/admin/active-runs.ts` + `core/admin/hooks.ts` (`useActiveRuns`/`useConcurrencySeries`/`useTokenSpeedSeries`/`useCancel*`/`useMonitorSettings`). See the backend CLAUDE.md「Admin Concurrency Monitor」. +- **会话冲突 409 中文化**: when a user submits a new question while the previous turn is still streaming, the backend's `RunManager.create_or_reject` rejects with HTTP 409 (`services.py` now returns a **Chinese** detail「当前对话还在生成回答…」). `core/threads/hooks.ts` `onError` detects the conflict (`isActiveRunConflictError`: 409 / "already has an active run" / the Chinese text) and shows a friendly `toast.info` instead of a raw error. - **State**: TanStack Query for server state, context providers (`SubtasksProvider`, `ArtifactsProvider`, `PromptInputProvider`) scoped to `WorkspaceRoutes` - **LLM streaming**: `@langchain/langgraph-sdk` streams messages from the backend via SSE - **UI**: Radix UI primitives + TailwindCSS v4 + shadcn-style components in `src/components/ui/` diff --git a/frontend-web/src/components/workspace/admin/concurrency-monitor.tsx b/frontend-web/src/components/workspace/admin/concurrency-monitor.tsx new file mode 100644 index 00000000..51d6e6b8 --- /dev/null +++ b/frontend-web/src/components/workspace/admin/concurrency-monitor.tsx @@ -0,0 +1,472 @@ +"use client"; + +import { + Activity, + Gauge, + Loader2, + OctagonX, + RefreshCw, + Square, + Zap, +} from "lucide-react"; +import { useMemo, useState } from "react"; +import { toast } from "sonner"; + +import { Badge } from "@/components/ui/badge"; +import { Button } from "@/components/ui/button"; +import { Switch } from "@/components/ui/switch"; +import type { + ActiveRunItem, + ConcurrencyPoint, + Granularity, + TokenSpeedPoint, +} from "@/core/admin/active-runs"; +import { + useActiveRuns, + useCancelActiveRun, + useCancelAllActiveRuns, + useConcurrencySeries, + useMonitorSettings, + useTokenSpeedSeries, + useUpdateMonitorSettings, +} from "@/core/admin/hooks"; +import { cn } from "@/lib/utils"; + +/** lookback window (hours) per granularity. */ +const LOOKBACK: Record = { minute: 3, hour: 48 }; + +function fmtBucket(iso: string, granularity: Granularity): string { + const date = new Date(iso); + if (Number.isNaN(date.getTime())) return iso; + const mm = String(date.getMonth() + 1).padStart(2, "0"); + const dd = String(date.getDate()).padStart(2, "0"); + const hh = String(date.getHours()).padStart(2, "0"); + const mi = String(date.getMinutes()).padStart(2, "0"); + return granularity === "hour" ? `${mm}-${dd} ${hh}:00` : `${hh}:${mi}`; +} + +function fmtElapsed(seconds: number): string { + if (seconds < 60) return `${seconds}s`; + const m = Math.floor(seconds / 60); + const s = seconds % 60; + if (m < 60) return `${m}m${s.toString().padStart(2, "0")}s`; + const h = Math.floor(m / 60); + return `${h}h${(m % 60).toString().padStart(2, "0")}m`; +} + +interface LineSeries { + name: string; + color: string; + values: (number | null)[]; +} + +/** Minimal dependency-free SVG line chart (1–2 series + area under the first). */ +function LineChart({ + labels, + series, + unit, + emptyHint, +}: { + labels: string[]; + series: LineSeries[]; + unit: string; + emptyHint: string; +}) { + const width = 760; + const height = 220; + const pad = { top: 18, right: 16, bottom: 26, left: 40 }; + const cw = width - pad.left - pad.right; + const ch = height - pad.top - pad.bottom; + + const allValues = series.flatMap((s) => s.values.filter((v): v is number => v != null)); + const max = Math.max(1, ...allValues); + const n = labels.length; + + if (n === 0) { + return ( +
+ {emptyHint} +
+ ); + } + + const xFor = (i: number) => pad.left + (n <= 1 ? cw / 2 : (i / (n - 1)) * cw); + const yFor = (v: number) => pad.top + ch - (v / max) * ch; + const grid = [0, 0.25, 0.5, 0.75, 1]; + + const linePath = (vals: (number | null)[]) => { + let d = ""; + let started = false; + vals.forEach((v, i) => { + if (v == null) { + started = false; + return; + } + d += `${started ? "L" : "M"} ${xFor(i).toFixed(1)} ${yFor(v).toFixed(1)} `; + started = true; + }); + return d.trim(); + }; + + const first = series[0]; + const areaPath = (() => { + if (!first) return ""; + const line = linePath(first.values); + if (!line) return ""; + return `${line} L ${xFor(n - 1).toFixed(1)} ${(pad.top + ch).toFixed(1)} L ${xFor(0).toFixed(1)} ${(pad.top + ch).toFixed(1)} Z`; + })(); + + const labelStep = Math.max(1, Math.ceil(n / 7)); + + return ( + + + + + + + + {grid.map((r) => { + const y = pad.top + ch * r; + return ( + + + + {Math.round(max * (1 - r))} + + + ); + })} + {areaPath && } + {series.map((s) => ( + + ))} + {labels.map((label, i) => { + if (i !== 0 && i !== n - 1 && i % labelStep !== 0) return null; + return ( + + {label} + + ); + })} + {/* hover dots with native tooltip */} + {series.map((s) => + s.values.map((v, i) => + v == null ? null : ( + + {`${labels[i]} · ${s.name}: ${v}${unit}`} + + ), + ), + )} + + ); +} + +function ChartCard({ + icon: Icon, + title, + subtitle, + children, +}: { + icon: typeof Activity; + title: string; + subtitle: string; + children: React.ReactNode; +}) { + return ( +
+
+ +

{title}

+ {subtitle} +
+ {children} +
+ ); +} + +function ActiveRunRow({ + run, + onStop, + stopping, +}: { + run: ActiveRunItem; + onStop: (runId: string) => void; + stopping: boolean; +}) { + return ( +
+
+
{run.user_name || "未知用户"}
+
+ {run.thread_title || run.thread_id} +
+
+
{run.agent_name}
+ + {run.kind} + + + {run.status === "running" ? "运行中" : "排队中"} + + + {fmtElapsed(run.elapsed_seconds)} + + +
+ ); +} + +export function ConcurrencyMonitorPanel() { + const [open, setOpen] = useState(true); + const [granularity, setGranularity] = useState("minute"); + + const { data: live, isLoading: liveLoading, refetch, isRefetching } = useActiveRuns(open); + const concurrency = useConcurrencySeries(granularity, LOOKBACK[granularity], open); + const tokenSpeed = useTokenSpeedSeries(granularity, LOOKBACK[granularity], open); + const { data: settings } = useMonitorSettings(); + const updateSettings = useUpdateMonitorSettings(); + const cancelOne = useCancelActiveRun(); + const cancelAll = useCancelAllActiveRuns(); + + const [stoppingId, setStoppingId] = useState(null); + + const count = live?.count ?? 0; + const runs = live?.runs ?? []; + const samplerOn = settings?.enabled ?? false; + + const concurrencyChart = useMemo(() => { + const points: ConcurrencyPoint[] = concurrency.data?.points ?? []; + return { + labels: points.map((p) => fmtBucket(p.bucket, granularity)), + series: [ + { name: "峰值并发", color: "#C1121F", values: points.map((p) => p.peak) }, + { name: "平均并发", color: "#F59E0B", values: points.map((p) => p.avg) }, + ] as LineSeries[], + }; + }, [concurrency.data, granularity]); + + const tokenChart = useMemo(() => { + const points: TokenSpeedPoint[] = tokenSpeed.data?.points ?? []; + return { + labels: points.map((p) => fmtBucket(p.bucket, granularity)), + series: [ + { name: "平均速度", color: "#2563EB", values: points.map((p) => p.avg_tokens_per_sec) }, + { name: "最慢", color: "#9333EA", values: points.map((p) => p.min_tokens_per_sec) }, + ] as LineSeries[], + }; + }, [tokenSpeed.data, granularity]); + + const handleStop = async (runId: string) => { + setStoppingId(runId); + try { + const res = await cancelOne.mutateAsync(runId); + toast[res.cancelled ? "success" : "info"](res.cancelled ? "已停止该对话调用" : "该调用已结束或不存在"); + } catch (e) { + toast.error(e instanceof Error ? e.message : "停止失败"); + } finally { + setStoppingId(null); + } + }; + + const handleStopAll = async () => { + if (runs.length === 0) return; + if (!window.confirm(`确定要停止当前全部 ${count} 个并发调用吗？这会中断这些用户正在生成的回答。`)) return; + try { + const res = await cancelAll.mutateAsync(); + toast.success(`已停止 ${res.cancelled_count} 个调用`); + } catch (e) { + toast.error(e instanceof Error ? e.message : "全部停止失败"); + } + }; + + const handleToggleSampler = async (next: boolean) => { + try { + await updateSettings.mutateAsync(next); + toast.success(next ? "已开启并发采样" : "已关闭并发采样（折线图不再新增历史点）"); + } catch (e) { + toast.error(e instanceof Error ? e.message : "切换失败"); + } + }; + + return ( +
+ {/* header */} +
+
+ + + +
+
实时并发监控
+
+ 监控大模型并发压力，可直接停止任意对话调用 +
+
+
+ +
+
+ 当前并发 + 0 ? "text-[#C1121F] dark:text-[#7FACF2]" : "text-[#202124] dark:text-white", + )} + > + {count} + +
+
+ 后台采样 + +
+ + + +
+
+ + {open && ( + <> + {!samplerOn && ( +
+ 后台采样默认关闭：实时并发数与「停止」功能始终可用。需要并发量历史折线图时，打开右上角「后台采样」开关即可开始记录（对线上服务影响极小，可随时关闭）。 +
+ )} + + {/* granularity toggle */} +
+ 统计粒度 + {(["minute", "hour"] as Granularity[]).map((g) => ( + + ))} + + {granularity === "minute" ? "近 3 小时" : "近 48 小时"} + +
+ + {/* charts */} +
+ + {concurrency.isLoading ? ( +
+ 加载中 +
+ ) : ( + + )} + + + {tokenSpeed.isLoading ? ( +
+ 加载中 +
+ ) : ( + + )} + +
+ + {/* active run list */} +
+
+ 正在进行的调用 + {count} + {liveLoading && } +
+ {runs.length === 0 ? ( +
+ 当前没有正在进行的大模型调用 +
+ ) : ( +
+
+ 用户 / 会话 + 智能体 + 类型 + 状态 + 已运行 + 操作 +
+ {runs.map((run) => ( + + ))} +
+ )} +
+ + )} +
+ ); +} + +export default ConcurrencyMonitorPanel; diff --git a/frontend-web/src/core/admin/active-runs.ts b/frontend-web/src/core/admin/active-runs.ts new file mode 100644 index 00000000..31f4c14b --- /dev/null +++ b/frontend-web/src/core/admin/active-runs.ts @@ -0,0 +1,159 @@ +import { apiFetch } from "@/core/api"; +import { getBackendBaseURL } from "@/core/config"; +import { OptType, writeYLog } from "@/core/log/y-log"; + +/** One in-flight model run as seen by the admin concurrency monitor. */ +export interface ActiveRunItem { + run_id: string; + thread_id: string; + status: string; + created_at: string; + elapsed_seconds: number; + user_id: string; + user_name: string; + agent_id: string; + agent_name: string; + thread_title: string; + kind: string; + multitask_strategy: string; +} + +export interface ActiveRunsResponse { + count: number; + server_time: string; + runs: ActiveRunItem[]; +} + +export interface ConcurrencyPoint { + bucket: string; + peak: number; + avg: number; + samples: number; +} + +export interface ConcurrencySeriesResponse { + granularity: string; + since: string; + until: string; + enabled: boolean; + points: ConcurrencyPoint[]; +} + +export interface TokenSpeedPoint { + bucket: string; + avg_tokens_per_sec: number; + min_tokens_per_sec: number; + max_tokens_per_sec: number; + samples: number; +} + +export interface TokenSpeedSeriesResponse { + granularity: string; + since: string; + until: string; + points: TokenSpeedPoint[]; +} + +export type Granularity = "minute" | "hour"; + +const base = () => `${getBackendBaseURL()}/api/admin/active-runs`; + +async function parse(res: Response, what: string): Promise { + if (!res.ok) { + const err = (await res.json().catch(() => ({}))) as { detail?: string }; + throw new Error(err.detail ?? `${what} (HTTP ${res.status})`); + } + return res.json() as Promise; +} + +/** Live snapshot of all in-flight runs. Polled — intentionally NOT y-logged. */ +export async function loadActiveRuns(): Promise { + return parse(await apiFetch(base()), "Failed to load active runs"); +} + +/** Concurrency time-series (peak/avg per minute|hour). Polled — not y-logged. */ +export async function loadConcurrencySeries( + granularity: Granularity, + hours?: number, +): Promise { + const params = new URLSearchParams({ granularity }); + if (hours) params.set("hours", String(hours)); + return parse( + await apiFetch(`${base()}/concurrency?${params}`), + "Failed to load concurrency series", + ); +} + +/** Token-speed time-series (tokens/sec per minute|hour). Polled — not y-logged. */ +export async function loadTokenSpeedSeries( + granularity: Granularity, + hours?: number, +): Promise { + const params = new URLSearchParams({ granularity }); + if (hours) params.set("hours", String(hours)); + return parse( + await apiFetch(`${base()}/token-speed?${params}`), + "Failed to load token speed series", + ); +} + +/** Truly stop one in-flight run (cancels the live model stream). */ +export async function cancelActiveRun(runId: string): Promise<{ run_id: string; cancelled: boolean }> { + const url = `${base()}/${encodeURIComponent(runId)}/cancel`; + const result = await parse<{ run_id: string; cancelled: boolean }>( + await apiFetch(url, { method: "POST" }), + "Failed to cancel run", + ); + writeYLog({ + optType: OptType.other, + optDetail: "CM助手-全量-并发监控-停止对话调用", + url, + reqParam: { runId }, + returnParam: result, + }); + return result; +} + +/** Stop every in-flight run at once. */ +export async function cancelAllActiveRuns(): Promise<{ cancelled_count: number }> { + const url = `${base()}/cancel-all`; + const result = await parse<{ cancelled_count: number }>( + await apiFetch(url, { method: "POST" }), + "Failed to cancel all runs", + ); + writeYLog({ + optType: OptType.other, + optDetail: "CM助手-全量-并发监控-全部停止", + url, + returnParam: result, + }); + return result; +} + +/** Read the background-sampler on/off switch. */ +export async function getMonitorSettings(): Promise<{ enabled: boolean }> { + return parse<{ enabled: boolean }>( + await apiFetch(`${base()}/settings`), + "Failed to load monitor settings", + ); +} + +/** Toggle the background concurrency sampler (admin only, no restart). */ +export async function updateMonitorSettings(enabled: boolean): Promise<{ enabled: boolean }> { + const url = `${base()}/settings`; + const result = await parse<{ enabled: boolean }>( + await apiFetch(url, { + method: "PUT", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ enabled }), + }), + "Failed to update monitor settings", + ); + writeYLog({ + optType: enabled ? OptType.enabled : OptType.disabled, + optDetail: `CM助手-全量-并发监控-${enabled ? "开启" : "关闭"}采样`, + url, + reqParam: { enabled }, + }); + return result; +} diff --git a/frontend-web/src/core/admin/hooks.ts b/frontend-web/src/core/admin/hooks.ts index 228dad1c..fb25393b 100644 --- a/frontend-web/src/core/admin/hooks.ts +++ b/frontend-web/src/core/admin/hooks.ts @@ -17,6 +17,96 @@ import { updateAdminV1Fact, updateAdminV1Section, } from "./api"; +import { + cancelActiveRun, + cancelAllActiveRuns, + getMonitorSettings, + loadActiveRuns, + loadConcurrencySeries, + loadTokenSpeedSeries, + updateMonitorSettings, + type Granularity, +} from "./active-runs"; + +// ── 实时并发监控 (admin) ───────────────────────────────────────────────────── + +/** Live snapshot of in-flight runs. Polls every `intervalMs` while `enabled`. */ +export function useActiveRuns(enabled = true, intervalMs = 4000) { + return useQuery({ + queryKey: ["admin-active-runs", "live"], + queryFn: loadActiveRuns, + enabled, + refetchInterval: enabled ? intervalMs : false, + refetchOnWindowFocus: false, + }); +} + +/** 并发量时序 (peak/avg per minute|hour). */ +export function useConcurrencySeries( + granularity: Granularity, + hours: number | undefined, + enabled = true, + intervalMs = 15000, +) { + return useQuery({ + queryKey: ["admin-active-runs", "concurrency", granularity, hours ?? null], + queryFn: () => loadConcurrencySeries(granularity, hours), + enabled, + refetchInterval: enabled ? intervalMs : false, + refetchOnWindowFocus: false, + }); +} + +/** 大模型 token 出字速度时序 (tokens/sec per minute|hour). */ +export function useTokenSpeedSeries( + granularity: Granularity, + hours: number | undefined, + enabled = true, + intervalMs = 15000, +) { + return useQuery({ + queryKey: ["admin-active-runs", "token-speed", granularity, hours ?? null], + queryFn: () => loadTokenSpeedSeries(granularity, hours), + enabled, + refetchInterval: enabled ? intervalMs : false, + refetchOnWindowFocus: false, + }); +} + +export function useCancelActiveRun() { + const qc = useQueryClient(); + return useMutation({ + mutationFn: (runId: string) => cancelActiveRun(runId), + onSuccess: () => qc.invalidateQueries({ queryKey: ["admin-active-runs", "live"] }), + }); +} + +export function useCancelAllActiveRuns() { + const qc = useQueryClient(); + return useMutation({ + mutationFn: () => cancelAllActiveRuns(), + onSuccess: () => qc.invalidateQueries({ queryKey: ["admin-active-runs", "live"] }), + }); +} + +export function useMonitorSettings(enabled = true) { + return useQuery({ + queryKey: ["admin-active-runs", "settings"], + queryFn: getMonitorSettings, + enabled, + refetchOnWindowFocus: false, + }); +} + +export function useUpdateMonitorSettings() { + const qc = useQueryClient(); + return useMutation({ + mutationFn: (nextEnabled: boolean) => updateMonitorSettings(nextEnabled), + onSuccess: (data) => { + qc.setQueryData(["admin-active-runs", "settings"], data); + }, + }); +} export function useUserLeaderboard( range: { diff --git a/frontend-web/src/core/threads/hooks.ts b/frontend-web/src/core/threads/hooks.ts index 9c694496..e3f725d8 100644 --- a/frontend-web/src/core/threads/hooks.ts +++ b/frontend-web/src/core/threads/hooks.ts @@ -219,6 +219,19 @@ function isRunNotFoundError(message: string): boolean { return /run\s+[\w-]{8,}\s+not\s+found/i.test(message); } +/** + * 同一会话上一轮还没生成完，又提交了新问题 → 后端返回 409 + * 「already has an active run」。匹配 HTTP 409、英文兜底文案，或后端已中文化 + * 的提示，统一映射成一句友好的中文。 + */ +function isActiveRunConflictError(message: string): boolean { + return ( + /\b409\b/.test(message) || + /already has an active run/i.test(message) || + /还在生成回答/.test(message) + ); +} + function buildCurrentUserLabel(): string { const auth = getStoredAuth(); if (!auth) { @@ -555,6 +568,12 @@ export function useThreadStream({ } return; } + // 上一轮还没结束又发了新问题（后端 409）：给一句明确的中文提示，而不是 + // 把原始的英文/HTTP 错误抛给用户。 + if (isActiveRunConflictError(message)) { + toast.info("当前对话还在生成回答，请等待本轮回复结束，或先点「停止」再发送新的问题。"); + return; + } toast.error(message); }, onFinish(state) { diff --git a/frontend-web/src/pages/AdminUserLeaderboardPage.tsx b/frontend-web/src/pages/AdminUserLeaderboardPage.tsx index 6310a806..06dd8e25 100644 --- a/frontend-web/src/pages/AdminUserLeaderboardPage.tsx +++ b/frontend-web/src/pages/AdminUserLeaderboardPage.tsx @@ -46,6 +46,7 @@ import { ScrollArea } from "@/components/ui/scroll-area"; import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select"; import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs"; import { MarkdownContent } from "@/components/workspace/messages/markdown-content"; +import { ConcurrencyMonitorPanel } from "@/components/workspace/admin/concurrency-monitor"; import type { AdminLeaderboardDimension, AdminLeaderboardOverview, @@ -1389,6 +1390,8 @@ export default function AdminUserLeaderboardPage() {
+ + {isLoading && (
diff --git a/offline-backend-20260512/backend/CLAUDE.md b/offline-backend-20260512/backend/CLAUDE.md index eab65988..1d8b2e1d 100644 --- a/offline-backend-20260512/backend/CLAUDE.md +++ b/offline-backend-20260512/backend/CLAUDE.md @@ -297,6 +297,15 @@ Two on-demand/automatic refresh paths layer on top: - **Manual "????"** ? `POST /leaderboard/refresh-today` force-rebuilds *today's* snapshot **synchronously** (bounded by the same semaphore/timeout the scheduler uses), then returns the merged range. Lets an admin pull the latest real-time records into the stats without waiting for the TTL. Frontend: a "????" button on `AdminUserLeaderboardPage` ? `useRefreshTodayLeaderboard` (`core/admin/hooks.ts`), which primes + invalidates the `["admin-activity","leaderboard"]` query. - **Daily finalize of the previous day** ? once per day, on/after `ADMIN_LEADERBOARD_FINALIZE_HOUR` (local Beijing, default `1` = ?? 1 ?), the scheduler tick force-recomputes *yesterday* a single time so runs/tool-calls that only landed near midnight are fully captured. The decision is the pure `deerflow.persistence.admin_stats.finalize.should_finalize_previous_day` (status/generated_at/now/hour) ? the rebuild stamps `generated_at` to today, which self-debounces it for the rest of the day (no extra state file). Tests: `tests/test_leaderboard_finalize.py`. +### Admin Concurrency Monitor (实时并发监控 / 系统压力) + +A live admin monitor embedded **at the top of `AdminUserLeaderboardPage`** (`components/workspace/admin/concurrency-monitor.tsx`, `ConcurrencyMonitorPanel`) that answers "当前有几个并发的大模型调用、是谁、能不能直接停掉" and charts system pressure over time. Built on the existing in-memory `RunManager` (no new run plumbing). + +- **Live concurrency + 真正停止** — `RunManager.list_active()` returns every run whose status is `pending`/`running` (the registry keeps finished records around until `cleanup`, so it filters by status). `len(list_active())` is the current concurrent-call count. Router `app/gateway/routers/admin_active_runs.py` (`/api/admin/active-runs`, **admin only** via the shared `admin_users._require_admin`): `GET /` returns `{count, server_time, runs:[{run_id, thread_id, status, created_at, elapsed_seconds, user_id, user_name, agent_id, agent_name, thread_title, kind, multitask_strategy}]}` — each run best-effort enriched (batched DB) with the owning user's email (`threads_meta.user_id`→`users.email`) and the agent's display name (`agents.name`); `kind` is derived from thread metadata (圆桌会商 / 系统/后台 / 任务对话 / 对话). `POST /{run_id}/cancel` calls `RunManager.cancel(run_id, action="interrupt")`, which sets the abort event **and cancels the run's asyncio task** — i.e. it truly aborts the in-flight model stream, freeing capacity. `POST /cancel-all` stops every in-flight run. +- **并发量折线图 (按分钟/小时)** — a lightweight background **sampler** (`_concurrency_sampler_loop` in `app/gateway/app.py`, singleton file-locked, started in the lifespan) records `len(list_active())` every `CONCURRENCY_SAMPLE_INTERVAL_SECONDS` (default 15s) into the new `concurrency_samples` table (`deerflow.persistence.concurrency`, `ConcurrencySampleRow{sampled_at, active_count}`, auto-created by `create_all`, **no migration**; store wired as `app.state.concurrency_sample_store`). ~Hourly it purges rows older than `CONCURRENCY_SAMPLE_RETENTION_HOURS` (default 168h). `GET /concurrency?granularity=minute|hour&hours=N` aggregates samples into per-bucket **peak + avg** (`ConcurrencySampleStore.aggregate`, bucketed in Python so it's identical across sqlite/mysql/postgres; fetch capped at 60k rows). Default lookback: minute→3h, hour→48h. +- **大模型出字速度折线图 (tokens/sec, 按分钟/小时)** — `GET /token-speed?granularity=&hours=` aggregates the persisted `llm_call_metrics` (`status="success"`) into per-bucket avg/min/max tokens/sec (deriving tokens/sec from `output_tokens÷duration_ms` when the stored value is null). A dip pinpoints the model/provider as the bottleneck for that window — "知道具体原因出现在哪里". +- **On/off + 不影响线上** — the whole feature is designed to never impact production: sampling is one tiny INSERT, retention-purged, all paths best-effort/guarded, queries bounded. A **runtime switch** lives in `system_settings.json → concurrency_monitor.enabled` (`ConcurrencyMonitorSettings`, **default false / opt-in**, mtime-hot-reloaded): the background loop always runs (just sleeps) but only samples when an admin turns it on, and it re-reads the flag **every tick** so toggling it on/off **takes effect immediately without a restart** (the on-demand live snapshot + 停止 stay available regardless; only the historical chart needs sampling on). `GET|PUT /api/admin/active-runs/settings` reads/writes it (admin only); the panel's 「后台采样」 Switch drives it. Env `CONCURRENCY_SAMPLE_ENABLED=0` is a hard master kill (never starts the loop). Frontend client/hooks: `core/admin/active-runs.ts` + `core/admin/hooks.ts` (`useActiveRuns` polls 4s while the panel is open, `useConcurrencySeries`/`useTokenSpeedSeries` poll 15s, `useCancelActiveRun`/`useCancelAllActiveRuns`, `useMonitorSettings`/`useUpdateMonitorSettings`). Tests: `tests/test_concurrency_monitor.py`. + ### Scheduled Task Runtime `ScheduledTaskService` (`deerflow/runtime/scheduler/service.py`) polls due tasks every `poll_interval_seconds` and executes each in its own throwaway thread, serialized by a shared `_agent_run_lock`. diff --git a/offline-backend-20260512/backend/app/gateway/app.py b/offline-backend-20260512/backend/app/gateway/app.py index 4e0fe6ea..fe20b1ce 100644 --- a/offline-backend-20260512/backend/app/gateway/app.py +++ b/offline-backend-20260512/backend/app/gateway/app.py @@ -27,6 +27,7 @@ from app.gateway.csrf_middleware import CSRFMiddleware from app.gateway.deps import langgraph_runtime from app.gateway.routers import ( ai_writing, + admin_active_runs, admin_users, agents, artifacts, @@ -488,6 +489,18 @@ _LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS = _env_float( "ADMIN_LEADERBOARD_BACKGROUND_STARTUP_DELAY_SECONDS", 60.0, ) +# Concurrency (system-pressure) sampler: how often to snapshot the in-flight run +# count, and how long to keep samples (drives the admin 并发量 line chart, which +# aggregates per minute/hour). Sampling is in-memory cheap (one tiny INSERT). +_CONCURRENCY_SAMPLE_ENABLED = _env_bool("CONCURRENCY_SAMPLE_ENABLED", True) +_CONCURRENCY_SAMPLE_INTERVAL_SECONDS = _env_float( + "CONCURRENCY_SAMPLE_INTERVAL_SECONDS", + 15.0, +) +_CONCURRENCY_SAMPLE_RETENTION_HOURS = _env_float( + "CONCURRENCY_SAMPLE_RETENTION_HOURS", + 168.0, # 7 days +) # Hour (local Beijing time, 0-23) after which the scheduler force-recomputes the # *previous* day's snapshot once, so records that landed near midnight are fully # captured. Default 1 ? "???? 1 ????????". @@ -828,6 +841,58 @@ async def _leaderboard_snapshot_scheduler_loop() -> None: _release_singleton_lock(lock_handle) +async def _concurrency_sampler_loop(app: FastAPI) -> None: + """Periodically record the in-flight run count for the admin pressure chart. + + Runs in the gateway process, singleton-locked so only one worker samples. + Each tick records ``len(run_manager.list_active())`` and, roughly hourly, + purges samples older than the retention window. Best-effort throughout — a + failed tick is logged and the loop continues. + """ + if not _CONCURRENCY_SAMPLE_ENABLED: + logger.info("Concurrency sampler disabled by CONCURRENCY_SAMPLE_ENABLED") + return + + store = getattr(app.state, "concurrency_sample_store", None) + run_mgr = getattr(app.state, "run_manager", None) + if store is None or run_mgr is None: + logger.info("Concurrency sampler not started (no store / run manager — memory backend?)") + return + + from deerflow.config.runtime_paths import runtime_home + + lock_handle = _acquire_singleton_lock(runtime_home() / ".concurrency_sampler.lock") + if lock_handle is None: + logger.info("Concurrency sampler: another gateway worker holds the lock; this worker skips") + return + + interval = max(1.0, _CONCURRENCY_SAMPLE_INTERVAL_SECONDS) + purge_every_ticks = max(1, int(3600 / interval)) # ~once per hour + logger.info("Concurrency sampler started: interval=%.1fs retention=%.1fh", interval, _CONCURRENCY_SAMPLE_RETENTION_HOURS) + ticks = 0 + try: + while True: + try: + # Runtime toggle (admin can turn sampling off without a restart). + # mtime-cached read, so this is cheap every tick. + from deerflow.config.system_settings import get_concurrency_monitor_settings + + if get_concurrency_monitor_settings().enabled: + active = await run_mgr.list_active() + await store.record(len(active)) + ticks += 1 + if ticks % purge_every_ticks == 0: + from datetime import UTC, datetime, timedelta + + cutoff = datetime.now(UTC) - timedelta(hours=_CONCURRENCY_SAMPLE_RETENTION_HOURS) + await store.purge_older_than(cutoff) + except Exception: + logger.exception("Concurrency sampler tick failed (non-fatal)") + await asyncio.sleep(interval) + finally: + _release_singleton_lock(lock_handle) + + def _warn_if_multi_worker() -> None: """?? uvicorn / gunicorn ? worker ???? ERROR ??? @@ -1044,6 +1109,13 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: except Exception: logger.exception("Leaderboard snapshot scheduler failed to start (non-fatal)") + # Concurrency (system-pressure) sampler — lightweight, singleton-locked, + # runtime-toggleable via system_settings.concurrency_monitor.enabled. + try: + app.state.concurrency_sampler_task = asyncio.create_task(_concurrency_sampler_loop(app)) + except Exception: + logger.exception("Concurrency sampler failed to start (non-fatal)") + # Start IM channel service if any channels are configured try: from app.channels.service import start_channel_service @@ -1102,6 +1174,16 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: except Exception: logger.exception("Leaderboard snapshot scheduler shutdown error") + concurrency_sampler_task = getattr(app.state, "concurrency_sampler_task", None) + if concurrency_sampler_task is not None: + concurrency_sampler_task.cancel() + try: + await concurrency_sampler_task + except asyncio.CancelledError: + pass + except Exception: + logger.exception("Concurrency sampler shutdown error") + # ?? 7 ????AI ?? graph ???? checkpointer / in-flight task? # thread ??? LangGraph Server ? checkpointer ???lifespan ??? # ``langgraph_runtime()`` ? contextmanager ??????????????? @@ -1434,6 +1516,8 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an app.include_router(admin_users.router) app.include_router(admin_users.self_router) + app.include_router(admin_active_runs.router) + # Admin tool / skill call metrics audit + export app.include_router(tool_metrics.router) diff --git a/offline-backend-20260512/backend/app/gateway/deps.py b/offline-backend-20260512/backend/app/gateway/deps.py index d8dd17e6..49db1366 100644 --- a/offline-backend-20260512/backend/app/gateway/deps.py +++ b/offline-backend-20260512/backend/app/gateway/deps.py @@ -116,6 +116,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]: app.state.llm_metrics_store = make_llm_metrics_store(sf) + from deerflow.persistence.concurrency import make_concurrency_sample_store + + app.state.concurrency_sample_store = make_concurrency_sample_store(sf) + from deerflow.persistence.tool_metrics import make_tool_metrics_store app.state.tool_metrics_store = make_tool_metrics_store(sf) @@ -433,6 +437,11 @@ def get_llm_metrics_store(request: Request): return getattr(request.app.state, "llm_metrics_store", None) +def get_concurrency_sample_store(request: Request): + """Return the concurrency-sample store, or None on the memory backend.""" + return getattr(request.app.state, "concurrency_sample_store", None) + + def get_tool_metrics_store(request: Request): """Return the tool-call metrics store, or None when not configured.""" return getattr(request.app.state, "tool_metrics_store", None) diff --git a/offline-backend-20260512/backend/app/gateway/routers/admin_active_runs.py b/offline-backend-20260512/backend/app/gateway/routers/admin_active_runs.py new file mode 100644 index 00000000..52937acf --- /dev/null +++ b/offline-backend-20260512/backend/app/gateway/routers/admin_active_runs.py @@ -0,0 +1,406 @@ +"""Admin-only live concurrency monitor for in-flight model runs. + +Surfaces every *currently running* call into the agent runtime (interactive +chat, agent chat, roundtable seats, …) so an administrator can see how much +concurrent pressure the system is putting on the LLM and **truly stop** any of +them — ``RunManager.cancel`` aborts the run's asyncio task, which cancels the +in-flight model stream, immediately freeing that capacity. + +All endpoints require ``system_role == "admin"`` (shared ``_require_admin``). +The data source is the in-memory :class:`RunManager` registry (``list_active``), +enriched best-effort with the owning user's email + the agent's display name +from the DB. Enrichment failures degrade gracefully (the run still shows, just +with rawer labels) — the monitor must never break because a lookup failed. +""" + +from __future__ import annotations + +import logging +from datetime import UTC, datetime, timedelta + +from fastapi import APIRouter, Query, Request +from pydantic import BaseModel +from sqlalchemy import select + +from app.gateway.deps import get_concurrency_sample_store, get_run_manager +from app.gateway.routers.admin_users import _require_admin +from deerflow.config.system_settings import load_system_settings, save_system_settings +from deerflow.persistence.agents.model import AgentRow +from deerflow.persistence.engine import get_session_factory +from deerflow.persistence.llm_metrics.model import LlmCallMetricRow +from deerflow.persistence.thread_meta.model import ThreadMetaRow +from deerflow.persistence.user.model import UserRow + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/api/admin/active-runs", tags=["admin-active-runs"]) + +# Lookback windows are clamped so an aggregation never scans an unbounded range. +_MAX_LOOKBACK_HOURS = 24 * 14 # 14 days + +# Per-granularity default lookback (hours): minute view = recent, hour view = wider. +_DEFAULT_LOOKBACK_HOURS = {"minute": 3, "hour": 48} + +# assistant_id values that are not real custom-agent ids (the default lead agent). +_DEFAULT_ASSISTANT_IDS = {"", "default", "lead_agent", "agent"} + + +class ActiveRunItem(BaseModel): + run_id: str + thread_id: str + status: str + created_at: str = "" + elapsed_seconds: int = 0 + user_id: str = "" + user_name: str = "" + agent_id: str = "" + agent_name: str = "" + thread_title: str = "" + kind: str = "对话" + multitask_strategy: str = "reject" + + +class ActiveRunsResponse(BaseModel): + count: int + server_time: str + runs: list[ActiveRunItem] + + +class CancelRunResponse(BaseModel): + run_id: str + cancelled: bool + + +class CancelAllResponse(BaseModel): + cancelled_count: int + + +class ConcurrencyPoint(BaseModel): + bucket: str + peak: int + avg: float + samples: int + + +class ConcurrencySeriesResponse(BaseModel): + granularity: str + since: str + until: str + enabled: bool + points: list[ConcurrencyPoint] + + +class TokenSpeedPoint(BaseModel): + bucket: str + avg_tokens_per_sec: float + min_tokens_per_sec: float + max_tokens_per_sec: float + samples: int + + +class TokenSpeedSeriesResponse(BaseModel): + granularity: str + since: str + until: str + points: list[TokenSpeedPoint] + + +class MonitorSettingsResponse(BaseModel): + enabled: bool + + +def _resolve_window(granularity: str, hours: float | None) -> tuple[str, datetime, datetime]: + """Clamp the granularity + lookback window into (gran, since, until).""" + gran = "hour" if granularity == "hour" else "minute" + lookback = hours if hours and hours > 0 else _DEFAULT_LOOKBACK_HOURS[gran] + lookback = min(float(lookback), _MAX_LOOKBACK_HOURS) + until = datetime.now(UTC) + since = until - timedelta(hours=lookback) + return gran, since, until + + +def _bucket_start(dt: datetime, granularity: str) -> datetime: + if granularity == "hour": + return dt.replace(minute=0, second=0, microsecond=0) + return dt.replace(second=0, microsecond=0) + + +def _derive_tokens_per_sec(tps, output_tokens, total_tokens, duration_ms) -> float | None: + """Use the stored tokens/sec, else derive from tokens ÷ duration.""" + if tps not in (None, 0): + return float(tps) + numerator = output_tokens if output_tokens else total_tokens + if not numerator or not duration_ms or duration_ms <= 0: + return None + return round((float(numerator) / float(duration_ms)) * 1000.0, 3) + + +def _elapsed_seconds(created_at: str, now: datetime) -> int: + """Seconds since the (UTC ISO) ``created_at`` stamp; 0 on any parse error.""" + if not created_at: + return 0 + try: + started = datetime.fromisoformat(created_at) + except (ValueError, TypeError): + return 0 + if started.tzinfo is None: + started = started.replace(tzinfo=UTC) + delta = (now - started).total_seconds() + return max(0, int(delta)) + + +def _kind_for(metadata: dict | None) -> str: + """Coarse human label for the run's origin, from its thread metadata.""" + meta = metadata or {} + thread_type = str(meta.get("thread_type") or "") + if thread_type == "roundtable": + return "圆桌会商" + if meta.get("system"): + return "系统/后台" + if meta.get("taskId"): + return "任务对话" + return "对话" + + +async def _enrich(records) -> dict: + """Batch-resolve thread metadata, user emails and agent names. + + Returns a dict of lookup maps; every query is wrapped so a failure just + yields empty maps (the runs still render with raw ids). + """ + thread_ids = {r.thread_id for r in records if r.thread_id} + assistant_ids = { + r.assistant_id + for r in records + if r.assistant_id and r.assistant_id not in _DEFAULT_ASSISTANT_IDS + } + + thread_meta: dict[str, ThreadMetaRow] = {} + user_emails: dict[str, str] = {} + agent_names: dict[str, str] = {} + + try: + session_factory = get_session_factory() + async with session_factory() as session: + if thread_ids: + rows = ( + await session.execute( + select(ThreadMetaRow).where(ThreadMetaRow.thread_id.in_(thread_ids)) + ) + ).scalars().all() + thread_meta = {row.thread_id: row for row in rows} + + user_ids = {row.user_id for row in thread_meta.values() if row.user_id} + if user_ids: + rows = ( + await session.execute( + select(UserRow.id, UserRow.email).where(UserRow.id.in_(user_ids)) + ) + ).all() + user_emails = {r[0]: (r[1] or "") for r in rows} + + if assistant_ids: + rows = ( + await session.execute( + select(AgentRow.id, AgentRow.name).where(AgentRow.id.in_(assistant_ids)) + ) + ).all() + agent_names = {r[0]: (r[1] or "") for r in rows} + except Exception: # pragma: no cover - enrichment is best-effort + logger.warning("Failed to enrich active runs (non-fatal)", exc_info=True) + + return {"thread_meta": thread_meta, "user_emails": user_emails, "agent_names": agent_names} + + +@router.get("", response_model=ActiveRunsResponse) +@router.get("/", response_model=ActiveRunsResponse) +async def list_active_runs(request: Request) -> ActiveRunsResponse: + """List every in-flight run with who/what is driving it (admin only).""" + await _require_admin(request) + run_mgr = get_run_manager(request) + records = await run_mgr.list_active() + now = datetime.now(UTC) + maps = await _enrich(records) + thread_meta = maps["thread_meta"] + user_emails = maps["user_emails"] + agent_names = maps["agent_names"] + + items: list[ActiveRunItem] = [] + for r in records: + meta_row = thread_meta.get(r.thread_id) + user_id = (meta_row.user_id if meta_row else "") or "" + metadata = (meta_row.metadata_json if meta_row else None) or r.metadata or {} + assistant_id = r.assistant_id or "" + if assistant_id in _DEFAULT_ASSISTANT_IDS: + agent_name = "默认助手" + else: + agent_name = agent_names.get(assistant_id) or assistant_id + items.append( + ActiveRunItem( + run_id=r.run_id, + thread_id=r.thread_id, + status=r.status.value, + created_at=r.created_at, + elapsed_seconds=_elapsed_seconds(r.created_at, now), + user_id=user_id, + user_name=user_emails.get(user_id) or user_id or "未知用户", + agent_id=assistant_id, + agent_name=agent_name, + thread_title=(meta_row.display_name if meta_row else "") or "", + kind=_kind_for(metadata), + multitask_strategy=r.multitask_strategy, + ) + ) + + return ActiveRunsResponse(count=len(items), server_time=now.isoformat(), runs=items) + + +@router.post("/{run_id}/cancel", response_model=CancelRunResponse) +async def cancel_active_run( + run_id: str, + request: Request, + action: str = Query(default="interrupt", description="interrupt 保留检查点 / rollback 回滚"), +) -> CancelRunResponse: + """Truly stop one in-flight run — cancels its task and the live model stream.""" + await _require_admin(request) + run_mgr = get_run_manager(request) + normalized = "rollback" if action == "rollback" else "interrupt" + cancelled = await run_mgr.cancel(run_id, action=normalized) + return CancelRunResponse(run_id=run_id, cancelled=cancelled) + + +@router.post("/cancel-all", response_model=CancelAllResponse) +async def cancel_all_active_runs(request: Request) -> CancelAllResponse: + """Stop every in-flight run at once (admin emergency relief valve).""" + await _require_admin(request) + run_mgr = get_run_manager(request) + records = await run_mgr.list_active() + cancelled = 0 + for r in records: + if await run_mgr.cancel(r.run_id, action="interrupt"): + cancelled += 1 + return CancelAllResponse(cancelled_count=cancelled) + + +@router.get("/concurrency", response_model=ConcurrencySeriesResponse) +async def concurrency_series( + request: Request, + granularity: str = Query(default="minute", description="minute | hour"), + hours: float | None = Query(default=None, description="回看小时数(默认按粒度取)"), +) -> ConcurrencySeriesResponse: + """Per-minute / per-hour 并发量 series (peak + avg) for the pressure chart.""" + await _require_admin(request) + gran, since, until = _resolve_window(granularity, hours) + enabled = load_system_settings().concurrency_monitor.enabled + store = get_concurrency_sample_store(request) + points: list[ConcurrencyPoint] = [] + if store is not None: + buckets = await store.aggregate(since=since, until=until, granularity=gran) + points = [ + ConcurrencyPoint(bucket=b.bucket, peak=b.peak, avg=b.avg, samples=b.samples) + for b in buckets + ] + return ConcurrencySeriesResponse( + granularity=gran, + since=since.isoformat(), + until=until.isoformat(), + enabled=enabled, + points=points, + ) + + +@router.get("/token-speed", response_model=TokenSpeedSeriesResponse) +async def token_speed_series( + request: Request, + granularity: str = Query(default="minute", description="minute | hour"), + hours: float | None = Query(default=None, description="回看小时数(默认按粒度取)"), +) -> TokenSpeedSeriesResponse: + """Per-minute / per-hour 大模型出字速度 (tokens/sec) series. + + Aggregated from the persisted LLM-call metrics: a dip in tokens/sec for a + bucket points at the model/provider being the bottleneck for that window. + """ + await _require_admin(request) + gran, since, until = _resolve_window(granularity, hours) + + buckets: dict[datetime, dict[str, float]] = {} + try: + session_factory = get_session_factory() + if session_factory is not None: + async with session_factory() as session: + rows = ( + await session.execute( + select( + LlmCallMetricRow.created_at, + LlmCallMetricRow.tokens_per_sec, + LlmCallMetricRow.output_tokens, + LlmCallMetricRow.total_tokens, + LlmCallMetricRow.duration_ms, + ) + .where( + LlmCallMetricRow.created_at >= since, + LlmCallMetricRow.created_at <= until, + LlmCallMetricRow.status == "success", + ) + .order_by(LlmCallMetricRow.created_at) + .limit(100_000) + ) + ).all() + for created_at, tps_raw, output_tokens, total_tokens, duration_ms in rows: + tps = _derive_tokens_per_sec(tps_raw, output_tokens, total_tokens, duration_ms) + if tps is None or tps <= 0 or created_at is None: + continue + if created_at.tzinfo is None: + created_at = created_at.replace(tzinfo=UTC) + key = _bucket_start(created_at, gran) + entry = buckets.get(key) + if entry is None: + buckets[key] = {"sum": tps, "min": tps, "max": tps, "count": 1} + else: + entry["sum"] += tps + entry["min"] = min(entry["min"], tps) + entry["max"] = max(entry["max"], tps) + entry["count"] += 1 + except Exception: # pragma: no cover - chart must degrade gracefully + logger.warning("Failed to aggregate token-speed series (non-fatal)", exc_info=True) + + points: list[TokenSpeedPoint] = [] + for key in sorted(buckets): + entry = buckets[key] + count = max(1, int(entry["count"])) + points.append( + TokenSpeedPoint( + bucket=key.isoformat(), + avg_tokens_per_sec=round(entry["sum"] / count, 2), + min_tokens_per_sec=round(entry["min"], 2), + max_tokens_per_sec=round(entry["max"], 2), + samples=count, + ) + ) + return TokenSpeedSeriesResponse( + granularity=gran, + since=since.isoformat(), + until=until.isoformat(), + points=points, + ) + + +@router.get("/settings", response_model=MonitorSettingsResponse) +async def get_monitor_settings(request: Request) -> MonitorSettingsResponse: + """Read the concurrency-monitor on/off switch (admin only).""" + await _require_admin(request) + return MonitorSettingsResponse(enabled=load_system_settings().concurrency_monitor.enabled) + + +@router.put("/settings", response_model=MonitorSettingsResponse) +async def update_monitor_settings( + request: Request, + body: MonitorSettingsResponse, +) -> MonitorSettingsResponse: + """Toggle the background concurrency sampler on/off (admin only, no restart).""" + await _require_admin(request) + settings = load_system_settings() + updated = settings.model_copy( + update={"concurrency_monitor": settings.concurrency_monitor.model_copy(update={"enabled": bool(body.enabled)})} + ) + save_system_settings(updated) + return MonitorSettingsResponse(enabled=updated.concurrency_monitor.enabled) diff --git a/offline-backend-20260512/backend/app/gateway/services.py b/offline-backend-20260512/backend/app/gateway/services.py index 00905ef6..406850ab 100644 --- a/offline-backend-20260512/backend/app/gateway/services.py +++ b/offline-backend-20260512/backend/app/gateway/services.py @@ -581,7 +581,12 @@ async def start_run( multitask_strategy=body.multitask_strategy, ) except ConflictError as exc: - raise HTTPException(status_code=409, detail=str(exc)) from exc + # 同一会话上一轮回答还没结束，又提交了新问题 → 拒绝。给用户中文提示， + # 而不是把内部英文「Thread ... already has an active run」直接抛到前端。 + raise HTTPException( + status_code=409, + detail="当前对话还在生成回答，请等待本轮回复结束，或先点「停止」再发送新的问题。", + ) from exc except UnsupportedStrategyError as exc: raise HTTPException(status_code=501, detail=str(exc)) from exc t_post_create = time.perf_counter() diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/config/system_settings.py b/offline-backend-20260512/backend/packages/harness/deerflow/config/system_settings.py index b92fac9b..0bb40ac3 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/config/system_settings.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/config/system_settings.py @@ -234,6 +234,21 @@ class LoginWhitelistSettings(BaseModel): model_config = ConfigDict(extra="ignore") +class ConcurrencyMonitorSettings(BaseModel): + """管理员「实时并发监控」开关(排行榜页的系统压力监控)。 + + ``enabled`` 控制**后台并发采样器**是否持续把当前并发调用数写库(供并发量 + 折线图按分钟/小时聚合)。**默认关**(opt-in,由管理员在排行榜页手动开启), + 以对线上服务零影响为先;开启后采样器才开始写库,关闭则立即停止(每 tick 都 + 重读该开关,免重启),实时快照查询与「停止」始终可用、零常驻成本。设计上即便 + 开启也对线上零影响:采样仅一条极小 INSERT、有保留期清理、聚合查询有行数上限、 + 全程 best-effort。 + """ + + enabled: bool = Field(default=False, description="是否启用后台并发采样(默认关;开→开始写库,关→停止写库、折线图不再新增历史点)") + model_config = ConfigDict(extra="ignore") + + class SystemSettings(BaseModel): """Top-level system settings document.""" @@ -246,6 +261,7 @@ class SystemSettings(BaseModel): leaderboard: LeaderboardSettings = Field(default_factory=LeaderboardSettings) skill_compression: SkillCompressionSettings = Field(default_factory=SkillCompressionSettings) agent_modules: AgentModuleSettings = Field(default_factory=AgentModuleSettings) + concurrency_monitor: ConcurrencyMonitorSettings = Field(default_factory=ConcurrencyMonitorSettings) model_config = ConfigDict(extra="ignore") @@ -280,6 +296,15 @@ def get_skill_compression_settings() -> SkillCompressionSettings: return SkillCompressionSettings() +def get_concurrency_monitor_settings() -> ConcurrencyMonitorSettings: + """Convenience accessor for the concurrency-monitor block (safe defaults on error).""" + try: + return load_system_settings().concurrency_monitor + except Exception: + logger.debug("Failed to read concurrency monitor settings; using defaults", exc_info=True) + return ConcurrencyMonitorSettings() + + def save_system_settings(settings: SystemSettings) -> SystemSettings: """Persist the settings document atomically, returning the saved value.""" path = _settings_path() diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/concurrency/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/concurrency/__init__.py new file mode 100644 index 00000000..f1d4a190 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/concurrency/__init__.py @@ -0,0 +1,15 @@ +"""Concurrency (system-pressure) sampling persistence.""" + +from deerflow.persistence.concurrency.model import ConcurrencySampleRow +from deerflow.persistence.concurrency.store import ( + ConcurrencyBucket, + ConcurrencySampleStore, + make_concurrency_sample_store, +) + +__all__ = [ + "ConcurrencyBucket", + "ConcurrencySampleRow", + "ConcurrencySampleStore", + "make_concurrency_sample_store", +] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/concurrency/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/concurrency/model.py new file mode 100644 index 00000000..aecde237 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/concurrency/model.py @@ -0,0 +1,31 @@ +"""ORM model for periodic concurrency (system-pressure) samples. + +A background sampler writes one row every few seconds holding the number of +in-flight model runs at that instant. The admin monitor aggregates these into +per-minute / per-hour buckets to draw the 并发量 pressure line chart, even for +time windows before the admin opened the page. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from sqlalchemy import Integer +from sqlalchemy.orm import Mapped, mapped_column + +from deerflow.persistence.base import Base +from deerflow.persistence.types import BeijingDateTime + + +class ConcurrencySampleRow(Base): + """One sampled snapshot of concurrent in-flight model runs.""" + + __tablename__ = "concurrency_samples" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + # Wall-clock time the sample was taken (Beijing time on disk, tz-aware in Python). + sampled_at: Mapped[datetime] = mapped_column( + BeijingDateTime(), index=True, default=lambda: datetime.now(UTC) + ) + # Number of runs whose status was pending/running at ``sampled_at``. + active_count: Mapped[int] = mapped_column(Integer, default=0) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/concurrency/store.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/concurrency/store.py new file mode 100644 index 00000000..46eda554 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/concurrency/store.py @@ -0,0 +1,142 @@ +"""Async store for concurrency samples + minute/hour aggregation.""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta + +from sqlalchemy import delete, select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from deerflow.persistence.concurrency.model import ConcurrencySampleRow + +logger = logging.getLogger(__name__) + +# How far back the aggregation will fetch raw rows before bucketing. A safety +# cap so a degenerate range never tries to load the whole table into memory. +_MAX_ROWS = 60_000 + + +@dataclass +class ConcurrencyBucket: + """One minute/hour bucket of the concurrency series.""" + + bucket: str # ISO start of the bucket + peak: int # max concurrent runs observed in the bucket + avg: float # mean concurrent runs in the bucket + samples: int # how many raw samples fell into the bucket + + +def _bucket_start(dt: datetime, granularity: str) -> datetime: + """Truncate a timestamp to the start of its minute/hour bucket.""" + if granularity == "hour": + return dt.replace(minute=0, second=0, microsecond=0) + return dt.replace(second=0, microsecond=0) + + +class ConcurrencySampleStore: + """Persists periodic concurrency snapshots and aggregates them.""" + + def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: + self._sf = session_factory + + async def record(self, active_count: int, *, sampled_at: datetime | None = None) -> None: + """Insert one sample. Best-effort — never raises into the caller.""" + try: + async with self._sf() as session: + session.add( + ConcurrencySampleRow( + active_count=max(0, int(active_count)), + sampled_at=sampled_at or datetime.now(UTC), + ) + ) + await session.commit() + except Exception: # pragma: no cover - sampling must never break the app + logger.warning("Failed to record concurrency sample", exc_info=True) + + async def aggregate( + self, + *, + since: datetime, + until: datetime, + granularity: str = "minute", + ) -> list[ConcurrencyBucket]: + """Return per-minute / per-hour buckets of peak + average concurrency. + + Bucketing is done in Python (not SQL ``date_trunc``/``strftime``) so the + logic is identical across the sqlite/mysql/postgres backends. + """ + gran = "hour" if granularity == "hour" else "minute" + try: + async with self._sf() as session: + rows = ( + await session.execute( + select(ConcurrencySampleRow.sampled_at, ConcurrencySampleRow.active_count) + .where( + ConcurrencySampleRow.sampled_at >= since, + ConcurrencySampleRow.sampled_at <= until, + ) + .order_by(ConcurrencySampleRow.sampled_at) + .limit(_MAX_ROWS) + ) + ).all() + except Exception: + logger.warning("Failed to aggregate concurrency samples", exc_info=True) + return [] + + buckets: dict[datetime, dict[str, float]] = {} + for sampled_at, active_count in rows: + if sampled_at is None: + continue + if sampled_at.tzinfo is None: + sampled_at = sampled_at.replace(tzinfo=UTC) + key = _bucket_start(sampled_at, gran) + value = int(active_count or 0) + entry = buckets.get(key) + if entry is None: + buckets[key] = {"peak": value, "sum": value, "count": 1} + else: + entry["peak"] = max(entry["peak"], value) + entry["sum"] += value + entry["count"] += 1 + + result: list[ConcurrencyBucket] = [] + for key in sorted(buckets): + entry = buckets[key] + count = max(1, int(entry["count"])) + result.append( + ConcurrencyBucket( + bucket=key.isoformat(), + peak=int(entry["peak"]), + avg=round(entry["sum"] / count, 2), + samples=count, + ) + ) + return result + + async def purge_older_than(self, cutoff: datetime) -> int: + """Delete samples older than ``cutoff``. Returns rows removed.""" + try: + async with self._sf() as session: + result = await session.execute( + delete(ConcurrencySampleRow).where(ConcurrencySampleRow.sampled_at < cutoff) + ) + await session.commit() + return int(result.rowcount or 0) + except Exception: # pragma: no cover + logger.warning("Failed to purge concurrency samples", exc_info=True) + return 0 + + +def make_concurrency_sample_store( + session_factory: async_sessionmaker[AsyncSession] | None, +) -> ConcurrencySampleStore | None: + """Factory mirroring the other ``make_*_store`` helpers. + + Returns ``None`` on the memory backend (no SQL session factory), in which + case the monitor simply reports a live snapshot with no historical chart. + """ + if session_factory is None: + return None + return ConcurrencySampleStore(session_factory) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py index 0de34faf..884be67d 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py @@ -23,6 +23,7 @@ from deerflow.persistence.agents.model import ( ) from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow from deerflow.persistence.business_mapping.model import BusinessMappingRow +from deerflow.persistence.concurrency.model import ConcurrencySampleRow from deerflow.persistence.dashboard_sessions.model import DashboardSessionRow from deerflow.persistence.memory_config.model import UserMemoryConfigRow from deerflow.persistence.feedback.model import FeedbackRow @@ -85,6 +86,7 @@ __all__ = [ "AgentSkillRow", "AdminLeaderboardDailyStatRow", "BusinessMappingRow", + "ConcurrencySampleRow", "DashboardSessionRow", "KnowledgeEmbeddingRow", "KnowledgeEntityRow", diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py index 64deed00..a0783d87 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py @@ -279,6 +279,20 @@ class RunManager: async with self._lock: return any(r.thread_id == thread_id and r.status in (RunStatus.pending, RunStatus.running) for r in self._runs.values()) + async def list_active(self) -> list[RunRecord]: + """Return every in-flight run (status pending/running), oldest first. + + Powers the admin concurrency monitor: each record represents one live + call into the model runtime (interactive chat, agent chat, roundtable + seat, …), so ``len(list_active())`` is the current concurrent-call count + and each record can be cancelled to relieve model pressure. Completed + records linger in ``_runs`` until :meth:`cleanup`, so we filter by status + rather than returning the whole registry. + """ + async with self._lock: + # Dict insertion order == creation order, so this is oldest-first. + return [r for r in self._runs.values() if r.status in (RunStatus.pending, RunStatus.running)] + async def cleanup(self, run_id: str, *, delay: float = 300) -> None: """Remove a run record after an optional delay.""" if delay > 0: