From 8d1669adad4a8b758a988b01ead40c182528a5de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Tue, 16 Jun 2026 14:17:30 +0800 Subject: [PATCH] =?UTF-8?q?feat(roundtable):=20=E5=90=8E=E5=8F=B0=E4=BD=9C?= =?UTF-8?q?=E4=B8=9A=E6=8C=89=E7=94=A8=E6=88=B7=E5=B9=B6=E5=8F=91=E9=99=90?= =?UTF-8?q?=E6=B5=81=20+=20=E8=8D=89=E7=A8=BF=E6=8C=81=E4=B9=85=E5=8C=96?= =?UTF-8?q?=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 后台挂起作业每个 agent 轮次（leader/seat/report/summary/dashboard）开跑前 先取该用户的后台并发额度（默认 2，ROUNDTABLE_BACKGROUND_SLOTS_PER_USER 可调）， 跑完释放；同一用户多个后台作业共享额度，避免把单连接 SQLite checkpointer 和 离线模型端点打满拖垮前台实时会商。前台实时路径不经过此额度，不受限。 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../hooks/useDraftPersistence.ts | 51 +++++++ .../pages/RoundtablePlanningPage.tsx | 125 +++++++++++++++--- .../gateway/roundtable_inprocess_gateway.py | 48 ++++++- 3 files changed, 204 insertions(+), 20 deletions(-) diff --git a/frontend-web/src/roundtable-planning/hooks/useDraftPersistence.ts b/frontend-web/src/roundtable-planning/hooks/useDraftPersistence.ts index 56b4cce5..599b62cb 100644 --- a/frontend-web/src/roundtable-planning/hooks/useDraftPersistence.ts +++ b/frontend-web/src/roundtable-planning/hooks/useDraftPersistence.ts @@ -120,6 +120,11 @@ export interface UseDraftPersistenceReturn { * 发起前 / 点查看前各存一次，保住当时的聊天记录不丢。返回 draft id。 */ saveDraftNow: (furthestStep?: 1 | 2 | 3) => Promise; + /** + * 用**显式传入的 step2 快照**强制存一次（不读 live 状态）。「后台挂起直接跳转」用： + * 跳转会清空 step2，调用方在跳转前冻结快照传进来，保证落库的是挂起那一刻的研讨记录。 + */ + saveStep2SnapshotNow: (step2: DraftStep2Snapshot, furthestStep?: 1 | 2 | 3) => Promise; /** 强制刷新草稿列表。 */ refresh: () => Promise; } @@ -287,6 +292,51 @@ export function useDraftPersistence( [currentStep, enqueueSave, refreshList], ); + /** + * 用**显式传入的 step2 快照**强制存一次（不读 live 状态）。 + * + * 「后台挂起改直接跳转」用：点挂起后页面立刻 resetToStep1 会清空 step2，而存稿是 + * fire-and-forget 在跳转之后才真正落库——若仍走 `saveDraftNow`（其 `getStep2Snapshot` + * 是延迟读取）就会存到被重置后的空 step2。调用方在跳转前同步 `getStep2Snapshot()` 冻结 + * 一份快照传进来，这里只认这份，确保存的是挂起那一刻的研讨记录。 + * + * 有草稿 id → update（保护重命名，不带 title）；无 id → 用当前 step1 + 传入 step2 建档。 + */ + const saveStep2SnapshotNow = useCallback( + (step2: DraftStep2Snapshot, furthestStep: 1 | 2 | 3 = 2): Promise => { + const run = async (): Promise => { + const id = currentDraftIdRef.current; + if (id) { + const updated = await updateDraft(id, { furthestStep, step2 }); + if (updated) { + currentDraftIdRef.current = updated.id; + return updated.id; + } + // 服务端已不存在 → 落回新建。 + } + const step1 = getStep1Snapshot(); + if (!hasSavableContent(step1, step2)) return null; + const created = await createDraft({ + ...(taskIdRef.current ? { taskId: taskIdRef.current } : {}), + title: deriveDraftName({ + intentReady: step1.intentReady, + interactiveMessages: step1.interactiveMessages, + }), + furthestStep, + step1, + step2, + }); + currentDraftIdRef.current = created.id; + setCurrentDraftId(created.id); + return created.id; + }; + const next = saveQueueRef.current.then(run, run); + saveQueueRef.current = next.catch(() => undefined); + return next as Promise; + }, + [getStep1Snapshot, hasSavableContent, setCurrentDraftId], + ); + const handleManualSave = useCallback(() => { const furthest = (currentStep as 1 | 2 | 3) || 1; void (async () => { @@ -535,6 +585,7 @@ export function useDraftPersistence( handleCancelRename, ensureDraft, saveDraftNow, + saveStep2SnapshotNow, /** 强制刷新草稿列表（如后台挂起后让「进行中」状态立刻出现）。 */ refresh: refreshList, }; diff --git a/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx b/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx index c0b01eb6..7ad4ee81 100644 --- a/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx +++ b/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx @@ -1143,6 +1143,26 @@ export default function RoundtablePlanningPage() { }); }, [resetToStep1]); + // 「挂起中」本地标记：点「后台挂起」后立刻跳转，建作业是 fire-and-forget——从点击到作业 + // 真正建好的窗口里，用它让历史记录显示「挂起中」（作业建好后转 queued/running，由后端状态接管）。 + // resetToStep1 不卸载本页，故该 state 跨「跳转」存活。 + const [pendingSuspendIds, setPendingSuspendIds] = useState>(() => new Set()); + const markPendingSuspend = useCallback((draftId: string) => { + setPendingSuspendIds((prev) => { + const next = new Set(prev); + next.add(draftId); + return next; + }); + }, []); + const clearPendingSuspend = useCallback((draftId: string) => { + setPendingSuspendIds((prev) => { + if (!prev.has(draftId)) return prev; + const next = new Set(prev); + next.delete(draftId); + return next; + }); + }, []); + // 「后台挂起」核心（手动按钮 / 离开页面自动挂起共用）：幂等保证草稿存在 → // 同草稿已有在跑作业则跳过 → 启动后台作业。后端另有「同草稿未终态作业幂等返回」 // 兜底（覆盖跨标签页 / beforeunload 与本函数并发触发），绝不重复建作业。 @@ -1191,27 +1211,87 @@ export default function RoundtablePlanningPage() { buildInitialIntent, ]); - // 「后台挂起」按钮：把当前研讨交给后端自动跑完 Step2 研讨 + Step3（总结报告 + 结果绘制）。 - // 挂起后页面像「新建任务」一样重置回第一步（后台作业继续在历史记录里跑）。 - const handleBackgroundSuspend = useCallback(async () => { - if (step2.selectedAgents.length === 0) { + // 「后台挂起」按钮（2026-06-16 改「直接跳转」）：点击后**立刻**跳回第一步并提示，挂起动作 + // （存稿 → 去重 → 建后台作业）改为 fire-and-forget 在后台完成，不再等接口返回才跳转。 + // 历史记录在作业真正建好前显示「挂起中」（pendingSuspendIds），建好后由后端状态（queued→ + // running）接管。⚠️ resetToStep1 会清空 step2 等 live 状态，故挂起入参 + step2 快照必须在 + // 跳转前同步冻结（getStep2SnapshotWithRuns 返回的是不可变快照对象，reset 替换 state 不影响它）。 + const handleBackgroundSuspend = useCallback(() => { + const selected = step2.selectedAgents; + if (selected.length === 0) { triggerToast("⚠️ 请先选择研讨席位再后台挂起"); return; } - let jobId: string | null = null; - try { - jobId = await startBackgroundJob(); - } catch (err) { - triggerToast(`❌ 后台挂起失败：${(err as Error).message}`); - return; - } - // 注册到全工作区通知器，作业完成时**跨页**弹「达成共识同款」提示。 - if (jobId) watchJob(jobId); - // 重置回第一步（同新建任务）；后台作业在历史记录里继续跑、可点开看进度。 + // —— 跳转前同步冻结所有需要的输入（reset 之后这些 live 状态都会被清空）—— + const step2Snapshot = getStep2SnapshotWithRuns(); + const agents = selected.map((a) => ({ agentId: a.agent_id, name: a.name })); + const intent = buildInitialIntent(); + const model = selectedModel || undefined; + const seatModeSnap = seatMode; + const mode = orchestrationMode; + const chainSnap = chainMeta; + const planSnap = mode === "dag" ? orchestrationPlan : null; + const draftId = drafts.currentDraftId; + + // 立刻：标记「挂起中」+ 跳回第一步 + 提示（不等接口）。 + if (draftId) markPendingSuspend(draftId); resetToStep1(); triggerToast("🚀 已转入后台，任务会自动跑完研讨、总结报告与结果绘制；可在「历史记录」查看进度"); - void drafts.refresh(); - }, [step2, drafts, startBackgroundJob, resetToStep1, triggerToast, watchJob]); + + // 后台 fire-and-forget：用冻结快照存稿 → 去重 → 建作业 → 注册完成通知。失败只提示、不挡跳转。 + void (async () => { + let watchId: string | null = null; + try { + const savedId = (await drafts.saveStep2SnapshotNow(step2Snapshot, 2)) ?? draftId ?? null; + // 防重复：该草稿已有未终态作业就不再新建（同 startBackgroundJob 的客户端去重， + // 后端另有「同草稿未终态作业幂等返回」兜底）。 + const existing = savedId ? await getJobByDraft(savedId).catch(() => null) : null; + const alreadyRunning = + existing && + (existing.status === "running" || + existing.status === "awaiting_input" || + existing.status === "queued"); + if (alreadyRunning) { + watchId = existing?.id ?? null; + } else { + const job = await startJob({ + draftId: savedId ?? undefined, + intent, + agents, + model, + seatMode: seatModeSnap, + orchestrationMode: mode, + chain: chainSnap, + orchestrationPlan: planSnap, + }); + watchId = job.id; + } + } catch (err) { + triggerToast(`❌ 后台挂起失败：${(err as Error).message}`); + } finally { + // 注册到全工作区通知器，作业完成时**跨页**弹「达成共识同款」提示。 + if (watchId) watchJob(watchId); + // 清「挂起中」本地标记：此后历史记录的状态由后端作业（queued→running）权威接管。 + if (draftId) clearPendingSuspend(draftId); + void drafts.refresh(); + } + })(); + }, [ + step2, + drafts, + getStep2SnapshotWithRuns, + buildInitialIntent, + selectedModel, + seatMode, + orchestrationMode, + orchestrationPlan, + chainMeta, + resetToStep1, + triggerToast, + watchJob, + markPendingSuspend, + clearPendingSuspend, + ]); // ── 自动后台挂起：研讨进行中离开本页面（路由卸载）或刷新 / 关标签页时，任务自动转后台 ── // 最新状态进 ref：卸载 cleanup 与 beforeunload 都是「拍快照时刻」的同步回调， @@ -1427,6 +1507,7 @@ export default function RoundtablePlanningPage() { setIsOpen={drafts.setIsHistoryOpen} drafts={drafts.drafts} currentDraftId={drafts.currentDraftId} + pendingSuspendIds={pendingSuspendIds} renamingDraftId={drafts.renamingDraftId} renameDraftValue={drafts.renameDraftValue} setRenameDraftValue={drafts.setRenameDraftValue} @@ -1752,6 +1833,8 @@ interface HistoryDropdownProps { setIsOpen: React.Dispatch>; drafts: ReturnType["drafts"]; currentDraftId: string | null; + /** 正在「挂起中」（已点后台挂起、作业尚未建好）的草稿 id 集合 —— 历史记录显示「挂起中」。 */ + pendingSuspendIds: Set; renamingDraftId: string | null; renameDraftValue: string; setRenameDraftValue: React.Dispatch>; @@ -1774,6 +1857,7 @@ function HistoryDropdown(props: HistoryDropdownProps) { setIsOpen, drafts, currentDraftId, + pendingSuspendIds, renamingDraftId, renameDraftValue, setRenameDraftValue, @@ -1844,6 +1928,8 @@ function HistoryDropdown(props: HistoryDropdownProps) { {drafts.map((d) => { const isCurrent = d.id === currentDraftId; const isRenaming = d.id === renamingDraftId; + // 「挂起中」：已点后台挂起、作业还没建好（本地标记），或作业刚建好仍排队（queued）。 + const isSuspending = pendingSuspendIds.has(d.id) || d.status === "queued"; // 「进行中 / 待澄清」且有 jobId → 点击打开流程图弹窗，而非加载草稿。 const isActiveJob = (d.status === "running" || d.status === "awaiting_input") && !!d.jobId; @@ -1881,7 +1967,12 @@ function HistoryDropdown(props: HistoryDropdownProps) { {d.name} - {d.status === "running" ? ( + {isSuspending ? ( + + + 挂起中 + + ) : d.status === "running" ? ( 进行中 diff --git a/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py b/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py index b0c494d4..5eee3c38 100644 --- a/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py +++ b/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py @@ -14,8 +14,10 @@ from __future__ import annotations +import asyncio import json import logging +import os import uuid from datetime import UTC, datetime from types import SimpleNamespace @@ -51,6 +53,33 @@ _SUMMARY_FILE = "方案总结报告.md" # 这里漏禁了 web_search，导致后台席位 / 报告能联网搜索而前端不能，现由 policy 统一。 +# ── 后台作业「按用户限流」（2026-06-16）───────────────────────────────────────── +# 痛点：进 2+ 个后台挂起后，整个网关变慢——建立智能体连接、等待模型响应都慢。 +# 根因：① checkpointer 用单连接 aiosqlite（AsyncSqliteSaver，内部一把 asyncio.Lock 串 +# 行全进程的 checkpoint 读写）；② 后台作业无并发上限——每个作业进程内驱动真实 leader/ +# seat/report run，DAG/recommend 的 stage 还用 asyncio.gather 把席位真并行铺开。多个后台 +# 作业同时把几十个 agent run 压到同一把 SQLite 锁 + 同一个离线模型端点上，前台实时会商 +# 的 checkpoint 读写 / 模型调用只能排队 → 「干啥都慢」。 +# +# 解法：给**每个用户**一个并发额度（默认 2），后台每个 agent 轮次（leader/seat/report/ +# summary/dashboard）真正开跑前先取一个额度，跑完释放。同一用户的所有后台作业共享这 2 个 +# 额度——随便挂多少个作业都能立刻建好，但同一瞬间真正在打模型/写 checkpoint 的后台轮次 +# 至多 2 个，不会把共享资源打满。**前台实时路径（multi_agent._leader_run/_special_run、 +# 普通聊天）不经过本文件的 `_run_agent_turn`，永不取这个额度，故前台不受限。** +_BACKGROUND_SLOTS_PER_USER = max(1, int(os.getenv("ROUNDTABLE_BACKGROUND_SLOTS_PER_USER", "2") or "2")) +# 按 user_id 缓存 Semaphore。网关全程跑在同一个事件循环上，普通 dict 取/建即可（无跨线程）。 +_user_background_sems: dict[str, asyncio.Semaphore] = {} + + +def _user_background_semaphore(user_id: str) -> asyncio.Semaphore: + """取该用户的后台并发额度信号量（懒建，每用户一把）。""" + sem = _user_background_sems.get(user_id) + if sem is None: + sem = asyncio.Semaphore(_BACKGROUND_SLOTS_PER_USER) + _user_background_sems[user_id] = sem + return sem + + # ── 进程内底层操作 ─────────────────────────────────────────────────────────── @@ -156,6 +185,7 @@ def _tool_calls_since_last_human(messages: list[dict[str, Any]]) -> list[dict[st async def _run_agent_turn( app, *, + user_id: str, thread_id: str, agent_name: str, message: str, @@ -170,6 +200,10 @@ async def _run_agent_turn( 镜像 multi_agent ``_run_payload`` 的 body 形状：agent 通过 ``context.agent_name`` 选， assistant_id 固定 lead_agent；关闭 memory recall / subagent / plan mode（临时研讨任务）。 + + 并发受**该用户的后台额度**约束（``_user_background_semaphore``）：真正开跑（start_run + + await run task）前先取额度，跑完释放。同一用户的所有后台作业共享这几个额度，避免 2+ + 后台挂起把单连接 SQLite checkpointer / 离线模型打满、拖垮前台实时会商。 """ from app.gateway.routers.thread_runs import RunCreateRequest from app.gateway.services import start_run @@ -201,9 +235,12 @@ async def _run_agent_turn( # 虚拟 request：带上空 ``state``（auth 中间件在真请求上会写 state.user / state.auth； # 进程内没有这层，下游凡是读 request.state.X 的代码必须能拿到 None 而不是 AttributeError）。 request = SimpleNamespace(app=app, state=SimpleNamespace(user=None, auth=None)) - record = await start_run(body, thread_id, request) # type: ignore[arg-type] - if getattr(record, "task", None) is not None: - await record.task # 阻塞到这轮 agent 跑完，checkpoint 已落盘 + # 取该用户的后台额度后再开跑：start_run + await task 是最重的部分（建 agent、调模型、 + # 落 checkpoint）。额度满时本轮在此 await 排队，等同用户某个在跑轮次释放额度后再开。 + async with _user_background_semaphore(user_id): + record = await start_run(body, thread_id, request) # type: ignore[arg-type] + if getattr(record, "task", None) is not None: + await record.task # 阻塞到这轮 agent 跑完，checkpoint 已落盘 return await _read_thread_messages(app, thread_id) @@ -385,6 +422,7 @@ class InProcessRoundtableGateway: policy = roundtable_run_policy("leader") messages = await _run_agent_turn( self._app, + user_id=self._user_id, thread_id=thread_ids[coordinator_name], agent_name=coordinator_name, message=msg, @@ -426,6 +464,7 @@ class InProcessRoundtableGateway: # 席位模式覆盖 policy 的 thinking/reasoning（None → policy 默认），与前端 _special_run 一致。 messages = await _run_agent_turn( self._app, + user_id=self._user_id, thread_id=thread_ids[agent_id], agent_name=agent_id, message=task, @@ -460,6 +499,7 @@ class InProcessRoundtableGateway: for attempt in range(1 + _ARTIFACT_RETRY_LIMIT): messages = await _run_agent_turn( self._app, + user_id=self._user_id, thread_id=report_thread_id, agent_name=REPORT_AGENT_ID, message=message, @@ -503,6 +543,7 @@ class InProcessRoundtableGateway: for attempt in range(1 + _ARTIFACT_RETRY_LIMIT): messages = await _run_agent_turn( self._app, + user_id=self._user_id, thread_id=summary_thread_id, agent_name=SUMMARY_AGENT_ID, message=message, @@ -538,6 +579,7 @@ class InProcessRoundtableGateway: for attempt in range(1 + _ARTIFACT_RETRY_LIMIT): messages = await _run_agent_turn( self._app, + user_id=self._user_id, thread_id=thread_id, agent_name=REPORT_AGENT_ID, message=message,