feat(roundtable): 后台作业按用户并发限流 + 草稿持久化优化

后台挂起作业每个 agent 轮次(leader/seat/report/summary/dashboard)开跑前
先取该用户的后台并发额度(默认 2,ROUNDTABLE_BACKGROUND_SLOTS_PER_USER 可调),
跑完释放;同一用户多个后台作业共享额度,避免把单连接 SQLite checkpointer 和
离线模型端点打满拖垮前台实时会商。前台实时路径不经过此额度,不受限。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
董宏凯andClaude Opus 4.8 committed 2026-06-16 14:17:30 +08:00
1 parent da3d38013e
commit 8d1669adad
3 files changed
+204 -20

No files matched your search

@@ -120,6 +120,11 @@ export interface UseDraftPersistenceReturn {
* 发起前 / 点查看前各存一次,保住当时的聊天记录不丢。返回 draft id。 * 发起前 / 点查看前各存一次,保住当时的聊天记录不丢。返回 draft id。
*/ */
saveDraftNow: (furthestStep?: 1 | 2 | 3) => Promise<string | null>; saveDraftNow: (furthestStep?: 1 | 2 | 3) => Promise<string | null>;
/**
* 用**显式传入的 step2 快照**强制存一次(不读 live 状态)。「后台挂起直接跳转」用:
* 跳转会清空 step2,调用方在跳转前冻结快照传进来,保证落库的是挂起那一刻的研讨记录。
*/
saveStep2SnapshotNow: (step2: DraftStep2Snapshot, furthestStep?: 1 | 2 | 3) => Promise<string | null>;
/** 强制刷新草稿列表。 */ /** 强制刷新草稿列表。 */
refresh: () => Promise<void>; refresh: () => Promise<void>;
} }
@@ -287,6 +292,51 @@ export function useDraftPersistence(
[currentStep, enqueueSave, refreshList], [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<string | null> => {
const run = async (): Promise<string | null> => {
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<string | null>;
},
[getStep1Snapshot, hasSavableContent, setCurrentDraftId],
);
const handleManualSave = useCallback(() => { const handleManualSave = useCallback(() => {
const furthest = (currentStep as 1 | 2 | 3) || 1; const furthest = (currentStep as 1 | 2 | 3) || 1;
void (async () => { void (async () => {
@@ -535,6 +585,7 @@ export function useDraftPersistence(
handleCancelRename, handleCancelRename,
ensureDraft, ensureDraft,
saveDraftNow, saveDraftNow,
saveStep2SnapshotNow,
/** 强制刷新草稿列表(如后台挂起后让「进行中」状态立刻出现)。 */ /** 强制刷新草稿列表(如后台挂起后让「进行中」状态立刻出现)。 */
refresh: refreshList, refresh: refreshList,
}; };
@@ -1143,6 +1143,26 @@ export default function RoundtablePlanningPage() {
}); });
}, [resetToStep1]); }, [resetToStep1]);
// 「挂起中」本地标记:点「后台挂起」后立刻跳转,建作业是 fire-and-forget——从点击到作业
// 真正建好的窗口里,用它让历史记录显示「挂起中」(作业建好后转 queued/running,由后端状态接管)。
// resetToStep1 不卸载本页,故该 state 跨「跳转」存活。
const [pendingSuspendIds, setPendingSuspendIds] = useState<Set<string>>(() => 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 与本函数并发触发),绝不重复建作业。 // 兜底(覆盖跨标签页 / beforeunload 与本函数并发触发),绝不重复建作业。
@@ -1191,27 +1211,87 @@ export default function RoundtablePlanningPage() {
buildInitialIntent, buildInitialIntent,
]); ]);
// 「后台挂起」按钮:把当前研讨交给后端自动跑完 Step2 研讨 + Step3(总结报告 + 结果绘制)。 // 「后台挂起」按钮(2026-06-16 改「直接跳转」):点击后**立刻**跳回第一步并提示,挂起动作
// 挂起后页面像「新建任务」一样重置回第一步(后台作业继续在历史记录里跑)。 // (存稿 → 去重 → 建后台作业)改为 fire-and-forget 在后台完成,不再等接口返回才跳转。
const handleBackgroundSuspend = useCallback(async () => { // 历史记录在作业真正建好前显示「挂起中」(pendingSuspendIds),建好后由后端状态(queued→
if (step2.selectedAgents.length === 0) { // running)接管。⚠️ resetToStep1 会清空 step2 等 live 状态,故挂起入参 + step2 快照必须在
// 跳转前同步冻结(getStep2SnapshotWithRuns 返回的是不可变快照对象,reset 替换 state 不影响它)。
const handleBackgroundSuspend = useCallback(() => {
const selected = step2.selectedAgents;
if (selected.length === 0) {
triggerToast("⚠️ 请先选择研讨席位再后台挂起"); triggerToast("⚠️ 请先选择研讨席位再后台挂起");
return; return;
} }
let jobId: string | null = null; // —— 跳转前同步冻结所有需要的输入(reset 之后这些 live 状态都会被清空)——
try { const step2Snapshot = getStep2SnapshotWithRuns();
jobId = await startBackgroundJob(); const agents = selected.map((a) => ({ agentId: a.agent_id, name: a.name }));
} catch (err) { const intent = buildInitialIntent();
triggerToast(`❌ 后台挂起失败:${(err as Error).message}`); const model = selectedModel || undefined;
return; const seatModeSnap = seatMode;
} const mode = orchestrationMode;
// 注册到全工作区通知器,作业完成时**跨页**弹「达成共识同款」提示。 const chainSnap = chainMeta;
if (jobId) watchJob(jobId); const planSnap = mode === "dag" ? orchestrationPlan : null;
// 重置回第一步(同新建任务);后台作业在历史记录里继续跑、可点开看进度。 const draftId = drafts.currentDraftId;
// 立刻:标记「挂起中」+ 跳回第一步 + 提示(不等接口)。
if (draftId) markPendingSuspend(draftId);
resetToStep1(); resetToStep1();
triggerToast("🚀 已转入后台,任务会自动跑完研讨、总结报告与结果绘制;可在「历史记录」查看进度"); 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 都是「拍快照时刻」的同步回调, // 最新状态进 ref:卸载 cleanup 与 beforeunload 都是「拍快照时刻」的同步回调,
@@ -1427,6 +1507,7 @@ export default function RoundtablePlanningPage() {
setIsOpen={drafts.setIsHistoryOpen} setIsOpen={drafts.setIsHistoryOpen}
drafts={drafts.drafts} drafts={drafts.drafts}
currentDraftId={drafts.currentDraftId} currentDraftId={drafts.currentDraftId}
pendingSuspendIds={pendingSuspendIds}
renamingDraftId={drafts.renamingDraftId} renamingDraftId={drafts.renamingDraftId}
renameDraftValue={drafts.renameDraftValue} renameDraftValue={drafts.renameDraftValue}
setRenameDraftValue={drafts.setRenameDraftValue} setRenameDraftValue={drafts.setRenameDraftValue}
@@ -1752,6 +1833,8 @@ interface HistoryDropdownProps {
setIsOpen: React.Dispatch<React.SetStateAction<boolean>>; setIsOpen: React.Dispatch<React.SetStateAction<boolean>>;
drafts: ReturnType<typeof useDraftPersistence>["drafts"]; drafts: ReturnType<typeof useDraftPersistence>["drafts"];
currentDraftId: string | null; currentDraftId: string | null;
/** 正在「挂起中」(已点后台挂起、作业尚未建好)的草稿 id 集合 —— 历史记录显示「挂起中」。 */
pendingSuspendIds: Set<string>;
renamingDraftId: string | null; renamingDraftId: string | null;
renameDraftValue: string; renameDraftValue: string;
setRenameDraftValue: React.Dispatch<React.SetStateAction<string>>; setRenameDraftValue: React.Dispatch<React.SetStateAction<string>>;
@@ -1774,6 +1857,7 @@ function HistoryDropdown(props: HistoryDropdownProps) {
setIsOpen, setIsOpen,
drafts, drafts,
currentDraftId, currentDraftId,
pendingSuspendIds,
renamingDraftId, renamingDraftId,
renameDraftValue, renameDraftValue,
setRenameDraftValue, setRenameDraftValue,
@@ -1844,6 +1928,8 @@ function HistoryDropdown(props: HistoryDropdownProps) {
{drafts.map((d) => { {drafts.map((d) => {
const isCurrent = d.id === currentDraftId; const isCurrent = d.id === currentDraftId;
const isRenaming = d.id === renamingDraftId; const isRenaming = d.id === renamingDraftId;
// 「挂起中」:已点后台挂起、作业还没建好(本地标记),或作业刚建好仍排队(queued)。
const isSuspending = pendingSuspendIds.has(d.id) || d.status === "queued";
// 「进行中 / 待澄清」且有 jobId → 点击打开流程图弹窗,而非加载草稿。 // 「进行中 / 待澄清」且有 jobId → 点击打开流程图弹窗,而非加载草稿。
const isActiveJob = const isActiveJob =
(d.status === "running" || d.status === "awaiting_input") && !!d.jobId; (d.status === "running" || d.status === "awaiting_input") && !!d.jobId;
@@ -1881,7 +1967,12 @@ function HistoryDropdown(props: HistoryDropdownProps) {
{d.name} {d.name}
</span> </span>
<span className="text-[10px] text-slate-400 mt-0.5 flex items-center gap-1 font-mono"> <span className="text-[10px] text-slate-400 mt-0.5 flex items-center gap-1 font-mono">
{d.status === "running" ? ( {isSuspending ? (
<span className="inline-flex items-center gap-1 text-sky-500 font-semibold">
<Loader2 className="w-3 h-3 animate-spin" />
挂起中
</span>
) : d.status === "running" ? (
<span className="inline-flex items-center gap-1 text-primary font-semibold"> <span className="inline-flex items-center gap-1 text-primary font-semibold">
<Loader2 className="w-3 h-3 animate-spin" /> <Loader2 className="w-3 h-3 animate-spin" />
进行中 进行中
@@ -14,8 +14,10 @@
from __future__ import annotations from __future__ import annotations
import asyncio
import json import json
import logging import logging
import os
import uuid import uuid
from datetime import UTC, datetime from datetime import UTC, datetime
from types import SimpleNamespace from types import SimpleNamespace
@@ -51,6 +53,33 @@ _SUMMARY_FILE = "方案总结报告.md"
# 这里漏禁了 web_search,导致后台席位 / 报告能联网搜索而前端不能,现由 policy 统一。 # 这里漏禁了 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( async def _run_agent_turn(
app, app,
*, *,
user_id: str,
thread_id: str, thread_id: str,
agent_name: str, agent_name: str,
message: str, message: str,
@@ -170,6 +200,10 @@ async def _run_agent_turn(
镜像 multi_agent ``_run_payload`` 的 body 形状:agent 通过 ``context.agent_name`` 选, 镜像 multi_agent ``_run_payload`` 的 body 形状:agent 通过 ``context.agent_name`` 选,
assistant_id 固定 lead_agent;关闭 memory recall / subagent / plan mode(临时研讨任务)。 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.routers.thread_runs import RunCreateRequest
from app.gateway.services import start_run 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``(auth 中间件在真请求上会写 state.user / state.auth;
# 进程内没有这层,下游凡是读 request.state.X 的代码必须能拿到 None 而不是 AttributeError)。 # 进程内没有这层,下游凡是读 request.state.X 的代码必须能拿到 None 而不是 AttributeError)。
request = SimpleNamespace(app=app, state=SimpleNamespace(user=None, auth=None)) request = SimpleNamespace(app=app, state=SimpleNamespace(user=None, auth=None))
record = await start_run(body, thread_id, request) # type: ignore[arg-type] # 取该用户的后台额度后再开跑:start_run + await task 是最重的部分(建 agent、调模型、
if getattr(record, "task", None) is not None: # 落 checkpoint)。额度满时本轮在此 await 排队,等同用户某个在跑轮次释放额度后再开。
await record.task # 阻塞到这轮 agent 跑完,checkpoint 已落盘 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) return await _read_thread_messages(app, thread_id)
@@ -385,6 +422,7 @@ class InProcessRoundtableGateway:
policy = roundtable_run_policy("leader") policy = roundtable_run_policy("leader")
messages = await _run_agent_turn( messages = await _run_agent_turn(
self._app, self._app,
user_id=self._user_id,
thread_id=thread_ids[coordinator_name], thread_id=thread_ids[coordinator_name],
agent_name=coordinator_name, agent_name=coordinator_name,
message=msg, message=msg,
@@ -426,6 +464,7 @@ class InProcessRoundtableGateway:
# 席位模式覆盖 policy 的 thinking/reasoning(None → policy 默认),与前端 _special_run 一致。 # 席位模式覆盖 policy 的 thinking/reasoning(None → policy 默认),与前端 _special_run 一致。
messages = await _run_agent_turn( messages = await _run_agent_turn(
self._app, self._app,
user_id=self._user_id,
thread_id=thread_ids[agent_id], thread_id=thread_ids[agent_id],
agent_name=agent_id, agent_name=agent_id,
message=task, message=task,
@@ -460,6 +499,7 @@ class InProcessRoundtableGateway:
for attempt in range(1 + _ARTIFACT_RETRY_LIMIT): for attempt in range(1 + _ARTIFACT_RETRY_LIMIT):
messages = await _run_agent_turn( messages = await _run_agent_turn(
self._app, self._app,
user_id=self._user_id,
thread_id=report_thread_id, thread_id=report_thread_id,
agent_name=REPORT_AGENT_ID, agent_name=REPORT_AGENT_ID,
message=message, message=message,
@@ -503,6 +543,7 @@ class InProcessRoundtableGateway:
for attempt in range(1 + _ARTIFACT_RETRY_LIMIT): for attempt in range(1 + _ARTIFACT_RETRY_LIMIT):
messages = await _run_agent_turn( messages = await _run_agent_turn(
self._app, self._app,
user_id=self._user_id,
thread_id=summary_thread_id, thread_id=summary_thread_id,
agent_name=SUMMARY_AGENT_ID, agent_name=SUMMARY_AGENT_ID,
message=message, message=message,
@@ -538,6 +579,7 @@ class InProcessRoundtableGateway:
for attempt in range(1 + _ARTIFACT_RETRY_LIMIT): for attempt in range(1 + _ARTIFACT_RETRY_LIMIT):
messages = await _run_agent_turn( messages = await _run_agent_turn(
self._app, self._app,
user_id=self._user_id,
thread_id=thread_id, thread_id=thread_id,
agent_name=REPORT_AGENT_ID, agent_name=REPORT_AGENT_ID,
message=message, message=message,