diff --git a/frontend-web/src/roundtable-planning/api/chains.ts b/frontend-web/src/roundtable-planning/api/chains.ts
index 57df4440..6138efef 100644
--- a/frontend-web/src/roundtable-planning/api/chains.ts
+++ b/frontend-web/src/roundtable-planning/api/chains.ts
@@ -31,6 +31,8 @@ export interface ChainCreatePayload {
coordinatorPrompt?: string | null;
/** 默认席位执行模式(flash/thinking/pro/ultra);null = 未设。 */
seatMode?: SeatMode | null;
+ /** 研讨前并行取数开关(逐条 opt-in);缺省 false。 */
+ gatherFirst?: boolean;
}
/** Partial update — send only what changed. */
@@ -46,6 +48,8 @@ export interface ChainUpdatePayload {
coordinatorPrompt?: string | null;
/** 默认席位执行模式;send `null` to clear, omit to leave. */
seatMode?: SeatMode | null;
+ /** 研讨前并行取数开关;send true/false 改,omit 不改。 */
+ gatherFirst?: boolean;
/** 发布开关:true 发布为公共 / false 取消发布;omit 不改。仅 owner 生效。 */
isPublic?: boolean;
}
@@ -76,6 +80,7 @@ interface ServerChain {
stage_goals?: (string | null)[] | null;
coordinator_prompt?: string | null;
seat_mode?: string | null;
+ gather_first?: boolean;
is_public?: boolean;
is_mine?: boolean;
owner_name?: string | null;
@@ -93,6 +98,7 @@ function toChain(s: ServerChain): BusinessChain {
stageGoals: Array.isArray(s.stage_goals) ? s.stage_goals : null,
coordinatorPrompt: s.coordinator_prompt ?? null,
seatMode: (s.seat_mode as SeatMode | null) ?? undefined,
+ gatherFirst: s.gather_first ?? false,
isPublic: s.is_public ?? false,
// 后端缺省时按「我的」处理(向后兼容旧响应)。
isMine: s.is_mine ?? true,
@@ -112,6 +118,7 @@ function fromCreate(p: ChainCreatePayload) {
...(p.stageGoals !== undefined ? { stage_goals: p.stageGoals } : {}),
...(p.coordinatorPrompt !== undefined ? { coordinator_prompt: p.coordinatorPrompt } : {}),
...(p.seatMode !== undefined ? { seat_mode: p.seatMode } : {}),
+ ...(p.gatherFirst !== undefined ? { gather_first: p.gatherFirst } : {}),
};
}
@@ -126,6 +133,7 @@ function fromUpdate(p: ChainUpdatePayload) {
if (p.stageGoals !== undefined) out.stage_goals = p.stageGoals;
if (p.coordinatorPrompt !== undefined) out.coordinator_prompt = p.coordinatorPrompt;
if (p.seatMode !== undefined) out.seat_mode = p.seatMode;
+ if (p.gatherFirst !== undefined) out.gather_first = p.gatherFirst;
if (p.isPublic !== undefined) out.is_public = p.isPublic;
return out;
}
diff --git a/frontend-web/src/roundtable-planning/api/roundtable-jobs.ts b/frontend-web/src/roundtable-planning/api/roundtable-jobs.ts
index befe38e1..8045b638 100644
--- a/frontend-web/src/roundtable-planning/api/roundtable-jobs.ts
+++ b/frontend-web/src/roundtable-planning/api/roundtable-jobs.ts
@@ -101,6 +101,13 @@ function toStartBody(p: StartJobPayload) {
// 每席位技能识别强化映射 {agentId: bool}:沿用所选业务链条里各席位的开关;缺省 {}。
// 后端 run_seat 按 agentId 查表,与各席位 agent 自身开关取 OR。
seat_skill_directives: p.seatSkillDirectives ?? {},
+ // 每席位推理深度覆盖映射 {agentId: SeatMode|null}:沿用所选业务链条里各席位单独配的推理深度;
+ // 后端 run_seat 按 agentId 查表,覆盖优先、否则跟随作业级 seatMode(thinking/reasoning/subagent)。
+ seat_modes: Object.fromEntries(
+ Object.entries(p.seatModes ?? {}).filter(([, v]) => v != null) as [string, string][],
+ ),
+ // 研讨前并行取数开关(来源链条 gatherFirst):true = 后台研讨前先并行取数再研讨。缺省 false。
+ gather_first: p.gatherFirst ?? false,
...(seat
? {
thinking_enabled: seat.thinking_enabled,
diff --git a/frontend-web/src/roundtable-planning/components/BusinessChainPicker.tsx b/frontend-web/src/roundtable-planning/components/BusinessChainPicker.tsx
index 9a1b5b36..88de95a1 100644
--- a/frontend-web/src/roundtable-planning/components/BusinessChainPicker.tsx
+++ b/frontend-web/src/roundtable-planning/components/BusinessChainPicker.tsx
@@ -153,19 +153,22 @@ export function BusinessChainPicker({
// 席位「会商技能识别强化」开关(链条配置):按 agent_id 取,缺省 = 关。
const skillById = new Map(selectedChain.seats.map((s) => [s.agent_id, s.seat_skill_directive ?? false]));
const skillOf = (id: string) => skillById.get(id) ?? false;
+ // 席位「推理深度覆盖」(链条配置 seat_mode):按 agent_id 取,null/缺省 = 跟随链条级 seatMode。
+ const seatModeById = new Map(selectedChain.seats.map((s) => [s.agent_id, s.seat_mode ?? null]));
+ const seatModeOf = (id: string) => seatModeById.get(id) ?? null;
// 链条带分层、过滤存活后仍并行 → dag 编排(携带快速/沙箱子模式);否则退回线性 chain。
if (dagPlan) {
// dag:席位顺序按 stage 展平(与 plan 一致)。
const agents = dagPlan.stages
.flatMap((st) => st.agentIds)
- .map((id) => ({ agent_id: id, name: nameOf(id), model: modelOf(id), seat_skill_directive: skillOf(id) }));
- onConfirm({ agents, mode: "dag", chain: chainMeta, plan: dagPlan, dagExecMode });
+ .map((id) => ({ agent_id: id, name: nameOf(id), model: modelOf(id), seat_skill_directive: skillOf(id), seatMode: seatModeOf(id) }));
+ onConfirm({ agents, mode: "dag", chain: chainMeta, plan: dagPlan, dagExecMode, gatherFirst: selectedChain.gatherFirst ?? false });
return;
}
- const agents = survivingSeats.map((s) => ({ agent_id: s.agent_id, name: nameOf(s.agent_id), model: modelOf(s.agent_id), seat_skill_directive: skillOf(s.agent_id) }));
- onConfirm({ agents, mode: "chain", chain: chainMeta });
+ const agents = survivingSeats.map((s) => ({ agent_id: s.agent_id, name: nameOf(s.agent_id), model: modelOf(s.agent_id), seat_skill_directive: skillOf(s.agent_id), seatMode: seatModeOf(s.agent_id) }));
+ onConfirm({ agents, mode: "chain", chain: chainMeta, gatherFirst: selectedChain.gatherFirst ?? false });
};
// 整片脱敏:链条卡片区(链条名/描述/席位智能体名)。输入框不受影响(Hook 跳过)。
diff --git a/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts b/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts
index bf8e86e1..37fc60cb 100644
--- a/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts
+++ b/frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts
@@ -285,6 +285,20 @@ export interface UseStep2OrchestrationOptions {
* OR。recommend / 未选链条 → 空 {}(各席位仅靠其自身开关)。
*/
seatSkillDirectives?: Record
+ 开启后,选入研讨时会**先并行跑一轮「只取数」**:各席位同时用技能把相关信息收集回来(不研讨), + 产出作为共享资料注入后续研讨轮。把最慢的技能检索前置、并行做一次,给(尤其按链条顺序串行的) + 研讨段大幅提速;研讨阶段席位仍可按需再调技能。 +
+`. Missing/revoked/non-`view` codes all return 404. |
| **Scheduled Tasks** (`/api/scheduled-tasks`) | Per-user cron-like agent tasks. `GET/POST /` - list/create owned tasks; `PATCH/DELETE /{id}` - update/delete (owner); `POST /{id}/{pause,resume,trigger}`; `GET /{id}/runs`, `GET /runs/{rid}`, run-file preview/edit/download; `GET /runs/{rid}/messages` - agent messages for the run's underlying agent run (used to inspect a stuck/running run; authorized via `_load_run`); `GET /runs/{rid}/html` - generated HTML document for an `html_page` run; `DELETE /runs/{rid}` - owner-scoped, with an admin fallback so admins can delete any user's run; publish/subscribe under `/{id}/{publish,unpublish,subscribe}` and `GET /published`. **Admin (cross-user):** `GET /admin/all` - every task; `PATCH/DELETE /admin/{id}` - edit/delete any task; `POST /admin/{id}/trigger`; `GET /admin/{id}/runs`. Admin routes require `system_role == "admin"`. |
diff --git a/offline-backend-20260512/backend/app/gateway/deps.py b/offline-backend-20260512/backend/app/gateway/deps.py
index 1f45f004..d8dd17e6 100644
--- a/offline-backend-20260512/backend/app/gateway/deps.py
+++ b/offline-backend-20260512/backend/app/gateway/deps.py
@@ -340,6 +340,8 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
seat_reasoning_effort=p.seat_reasoning_effort,
seat_subagent_enabled=p.seat_subagent_enabled,
seat_skill_directives=p.seat_skill_directives,
+ # 每席位推理深度覆盖(业务链条里各席位单独配):run_seat 命中即覆盖作业级 seatMode。
+ seat_modes=p.seat_modes,
# 诊断日志上下文:网关每轮 leader/seat/report/summary/dashboard 报错均带这些归属。
job_id=p.job_id,
draft_id=p.draft_id,
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 712082ff..b589bcee 100644
--- a/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py
+++ b/offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py
@@ -509,6 +509,18 @@ def _build_seat_intro(agents: list[SeatRef]) -> str:
)
+# 席位执行模式 → (thinking_enabled, reasoning_effort, subagent_enabled) 派生参数。
+# 与前端 `lib/seat-mode.ts::seatModeToRunParams` **逐字段对齐**(务必同步修改),供每席位
+# 推理深度覆盖(self._seat_modes)在 run_seat 里把 SeatMode 解析成具体运行参数。未知/缺省档
+# 落 None(→ 跟随作业级 seat_* / policy 默认)。
+_SEAT_MODE_PARAMS: dict[str, tuple[bool | None, str | None, bool]] = {
+ "flash": (False, "minimal", False),
+ "thinking": (True, "low", False),
+ "pro": (True, "medium", False),
+ "ultra": (True, "high", True),
+}
+
+
# ── 网关实现 ─────────────────────────────────────────────────────────────────
@@ -526,6 +538,7 @@ class InProcessRoundtableGateway:
seat_reasoning_effort: str | None = None,
seat_subagent_enabled: bool = False,
seat_skill_directives: dict[str, bool] | None = None,
+ seat_modes: dict[str, str] | None = None,
job_id: str | None = None,
draft_id: str | None = None,
task_id: str | None = None,
@@ -552,6 +565,9 @@ class InProcessRoundtableGateway:
# 的配置)。run_seat 按 agent_id 查表,与各席位 agent 自身的 config.yaml 开关取 OR,
# 传给 append_seat_skill_directive。默认空 dict(仅各 agent 自身开关生效)。
self._seat_skill_directives = dict(seat_skill_directives or {})
+ # 每席位「推理深度覆盖」映射 {agent_id: SeatMode}。run_seat 命中即用该档(_SEAT_MODE_PARAMS)
+ # 派生的 thinking/reasoning/subagent 覆盖作业级 seat_*;未命中则跟随作业级。默认空 dict。
+ self._seat_modes = dict(seat_modes or {})
self._seat_intro = ""
self._first_leader = True
@@ -626,7 +642,18 @@ class InProcessRoundtableGateway:
seat_task = append_seat_skill_directive(
agent_id, task, chain_enabled=bool(self._seat_skill_directives.get(agent_id, False))
)
- # 席位模式覆盖 policy 的 thinking/reasoning(None → policy 默认),与前端 _special_run 一致。
+ # 每席位「推理深度覆盖」:业务链条里给该席位单独配了 seatMode → 用该档派生的
+ # thinking/reasoning/subagent **覆盖作业级**(self._seat_*);未配则跟随作业级。
+ # 作业级仍为 None 时落 policy 默认。与前端 useStep2Orchestration「覆盖优先、否则跟随
+ # selectedMode」逐席位语义一致。
+ seat_thinking, seat_reasoning, seat_subagent = (
+ self._seat_thinking_enabled,
+ self._seat_reasoning_effort,
+ self._seat_subagent_enabled,
+ )
+ override = _SEAT_MODE_PARAMS.get(self._seat_modes.get(agent_id, ""))
+ if override is not None:
+ seat_thinking, seat_reasoning, seat_subagent = override
messages = await _run_agent_turn(
self._app,
user_id=self._user_id,
@@ -637,16 +664,12 @@ class InProcessRoundtableGateway:
excluded_tools=policy.excluded_tools,
skill_stop_names=policy.skill_stop_names,
thinking_enabled=(
- policy.thinking_enabled
- if self._seat_thinking_enabled is None
- else self._seat_thinking_enabled
+ policy.thinking_enabled if seat_thinking is None else seat_thinking
),
reasoning_effort=(
- policy.reasoning_effort
- if self._seat_reasoning_effort is None
- else self._seat_reasoning_effort
+ policy.reasoning_effort if seat_reasoning is None else seat_reasoning
),
- subagent_enabled=self._seat_subagent_enabled,
+ subagent_enabled=seat_subagent,
diag=self._diag,
stage="step2_seat",
)
diff --git a/offline-backend-20260512/backend/app/gateway/roundtable_job_executor.py b/offline-backend-20260512/backend/app/gateway/roundtable_job_executor.py
index 6a552be4..f0d83a43 100644
--- a/offline-backend-20260512/backend/app/gateway/roundtable_job_executor.py
+++ b/offline-backend-20260512/backend/app/gateway/roundtable_job_executor.py
@@ -193,6 +193,13 @@ class StartParams:
# 配置)。透传给进程内网关,run_seat 按 agent_id 查表,与各席位 agent 自身 config.yaml 的
# 同名开关取 OR。默认空 dict。
seat_skill_directives: dict[str, bool] = field(default_factory=dict)
+ # 每席位「推理深度覆盖」映射 {agent_id: SeatMode}(值 flash/thinking/pro/ultra,沿用 Step2 所选
+ # 业务链条里每个席位单独配的推理深度)。透传给进程内网关,run_seat 按 agent_id 查表:命中 → 用该
+ # 档派生的 thinking/reasoning/subagent 覆盖作业级 seatMode;未命中 → 跟随作业级。默认空 dict。
+ seat_modes: dict[str, str] = field(default_factory=dict)
+ # 是否在研讨前先跑一轮**全席位并行『取数』**(业务链条逐条 opt-in):各席位并行只调技能取数、
+ # 不研讨,产出作为共享资料注入后续研讨轮。透传给 run_orchestration 的 gather_first。默认 False。
+ gather_first: bool = False
# 网关工厂:给定一次启动参数,返回一个 RoundtableGateway 实例。
@@ -280,6 +287,7 @@ class RoundtableJobExecutor:
intent_text=params.intent_text,
enable_report=params.enable_report,
plan=params.orchestration_plan,
+ gather_first=params.gather_first,
)
# 终态:把最终对话 + step3 + jobStatus 落进同一个 run(含 furthest_step=3)。
await self._write_run_to_draft(
diff --git a/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-summary/SOUL.md b/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-summary/SOUL.md
index fdaa5e19..cda907f5 100644
--- a/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-summary/SOUL.md
+++ b/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-summary/SOUL.md
@@ -22,6 +22,19 @@
请通读全部原料:提炼每个席位的核心观点 / 交付,串成一份完整的方案总结。
+## 动笔前:意图对齐与缺口核对(质量关键,每轮首次生成报告都要做)
+各席位可能由能力较弱的模型产出,交付未必充分、未必都扣住了用户意图。你是这份报告的**质量兜底**——不能只是把各席位说的拼接起来,而要**对齐用户意图、核对覆盖、补强缺口**。动笔前先在脑内完成(这张核对表不必输出):
+
+1. **拆解意图**:把「任务意图」(目标 / 约束 / 关键假设)拆成若干条**用户真正想要的诉求**;业务链总控编排提示里对最终产出的要求也并入诉求清单。
+2. **逐条核对覆盖**:对每条诉求,检查各席位交付 + 总控共识**是否覆盖、覆盖得够不够、有没有跑题或自相矛盾**。
+3. **据此组织报告**,并按下面的「缺口处理护栏」补强——让最终报告明显**比"各席位交付的简单拼接"更扣用户意图**:意图里要的、席位没说清的,你补强或标缺口;意图里没要的、席位跑题的,不要喧宾夺主。
+
+### 缺口处理护栏(避免弱模型把幻觉带进报告)
+- **能从已有原料重组 / 推理补上的** → 直接补全、讲透,正常写进报告。这是你最该做的增值:把零散、跑题、含糊的席位产物,按用户意图重新梳理成扣题的方案。
+- **超出席位原料、属于你自己的合理推断 / 建议** → 可以写,但**显式标注「(综合推断)」**,让用户能区分哪些有席位依据、哪些是你的补充判断。
+- **缺真实信息、靠现有原料补不了的**(需要最新数据 / 外部事实,而席位没给)→ **绝不凭空编**。在报告对应位置用一行醒目标注:`> ⚠️ 待补充:<具体缺什么>。如需补全,可在下方对话里告诉我(例如「补一下 X 的最新情况」),我会据此完善。` 把球交给用户——你本身不联网、不调研,真补由用户在对话里发起。
+- 始终:**宁可标注缺口,也不要假装完整**;绝不编造与原料矛盾、或无依据的"事实"。
+
## 产出要求(最重要)
### ⓪ 先判断模式:用户是否要求了结构化输出 / 流程图
@@ -122,5 +135,8 @@
- ❌ 修订流程图时只输出改动的节点(必须输出全量 flow-json,前端整体重绘)
- ❌ 在报告任何位置添加「AI 自动生成」「AI generated」之类的标签、徽章或水印
- ❌ 编造与原料明显矛盾的内容
+- ❌ 只把各席位交付简单拼接、不做「意图对齐与缺口核对」(席位可能弱,你要兜底扣住用户意图)
+- ❌ 缺真实信息时凭空编造来"填满"报告(应按护栏标注「> ⚠️ 待补充:…」交给用户在对话里补)
+- ❌ 自己的推断不加「(综合推断)」标注,让用户误以为有席位依据
- ❌ 忽视「业务链总控编排提示」的口径要求(它优先级最高)
- ❌ 编排提示已指定输出结构 / 重点席位,却仍按默认章节平均用力(结构要完全跟用户走,重点席位要明显加权)
diff --git a/offline-backend-20260512/backend/app/gateway/routers/roundtable_chains.py b/offline-backend-20260512/backend/app/gateway/routers/roundtable_chains.py
index f476a9d7..b06926ee 100644
--- a/offline-backend-20260512/backend/app/gateway/routers/roundtable_chains.py
+++ b/offline-backend-20260512/backend/app/gateway/routers/roundtable_chains.py
@@ -53,6 +53,11 @@ class ChainSeatSchema(BaseModel):
# 与该席位 agent 自身 config.yaml 的同名开关取 OR。默认 False(关)。整条 seats 数组以
# JSON 透传持久化,故新增此字段无需迁移;仅需在此 schema 显式声明,否则 model_dump() 会剥掉它。
seat_skill_directive: bool = False
+ # 该席位「推理深度覆盖」(flash/thinking/pro/ultra):覆盖链条级 seat_mode,仅本席位生效。
+ # None = 跟随链条级默认。「卡顿提速」用:把次要席位降到 flash。整条 seats 数组以 JSON 透传
+ # 持久化,故无需迁移;仅需在此 schema 显式声明,否则 model_dump() 会剥掉它。非法值在运行期
+ # 被 _SEAT_MODE_PARAMS.get() 安全忽略(回退作业级),故此处只约束长度、不强校验取值。
+ seat_mode: str | None = Field(default=None, max_length=16)
class ChainResponse(BaseModel):
@@ -69,6 +74,8 @@ class ChainResponse(BaseModel):
coordinator_prompt: str | None = None
# 默认席位执行模式(flash/thinking/pro/ultra);None = 未设,选入研讨时回退默认(pro)。
seat_mode: str | None = None
+ # 研讨前并行取数开关(逐条 opt-in);True = 选入研讨时研讨前先并行取数。缺省 False。
+ gather_first: bool = False
# 发布状态:True = 公共(所有账号可在「公共」筛选查看 / 选用)。
is_public: bool = False
# 是否当前用户所有(仅本人可编辑 / 删除 / 发布;公共筛选里别人的链条 is_mine=False)。
@@ -96,6 +103,7 @@ class ChainCreateRequest(BaseModel):
stage_goals: list[str | None] | None = None
coordinator_prompt: str | None = Field(default=None, max_length=4000)
seat_mode: str | None = Field(default=None, max_length=16)
+ gather_first: bool = False
class ChainUpdateRequest(BaseModel):
@@ -107,6 +115,8 @@ class ChainUpdateRequest(BaseModel):
stage_goals: list[str | None] | None = None
coordinator_prompt: str | None = Field(default=None, max_length=4000)
seat_mode: str | None = Field(default=None, max_length=16)
+ # 研讨前并行取数开关;send 显式 true/false 改,omit 不改。
+ gather_first: bool | None = None
# 发布开关(True 发布为公共 / False 取消发布);仅 owner 可改(update 本就 owner 限定)。
is_public: bool | None = None
@@ -289,6 +299,7 @@ async def create_chain(request: Request, body: ChainCreateRequest) -> ChainRespo
"stage_goals": body.stage_goals,
"coordinator_prompt": body.coordinator_prompt,
"seat_mode": body.seat_mode,
+ "gather_first": body.gather_first,
},
)
await _enrich_owner_names([row])
diff --git a/offline-backend-20260512/backend/app/gateway/routers/roundtable_jobs.py b/offline-backend-20260512/backend/app/gateway/routers/roundtable_jobs.py
index 639ffce1..b06c59e2 100644
--- a/offline-backend-20260512/backend/app/gateway/routers/roundtable_jobs.py
+++ b/offline-backend-20260512/backend/app/gateway/routers/roundtable_jobs.py
@@ -81,6 +81,12 @@ class StartJobRequest(BaseModel):
# 同名配置后下发)。后台席位 run 据此(与各席位 agent 自身 config.yaml 的开关取 OR)决定
# 是否注入。缺省空 dict(全部按 False/关,仅各 agent 自身开关生效)。
seat_skill_directives: dict[str, bool] = Field(default_factory=dict)
+ # 每席位「推理深度覆盖」映射 {agent_id: SeatMode}(前端读所选业务链条里**每个席位**单独配的
+ # 推理深度后下发;值为 flash/thinking/pro/ultra)。后台席位 run 按 agent_id 查表:命中则用该档
+ # 派生的 thinking/reasoning/subagent 覆盖作业级 seatMode;未命中 → 跟随作业级。缺省空 dict。
+ seat_modes: dict[str, str] = Field(default_factory=dict)
+ # 是否研讨前先并行『取数』(业务链条逐条 opt-in;前端读所选链条的开关后下发)。默认 False。
+ gather_first: bool = False
class ResumeJobRequest(BaseModel):
@@ -240,6 +246,8 @@ async def start_job(request: Request, body: StartJobRequest) -> RoundtableJobRes
seat_reasoning_effort=body.reasoning_effort,
seat_subagent_enabled=body.subagent_enabled,
seat_skill_directives=dict(body.seat_skill_directives or {}),
+ seat_modes=dict(body.seat_modes or {}),
+ gather_first=bool(body.gather_first),
)
)
return _to_response(row)
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/engine.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/engine.py
index cdbcca68..b28d054b 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/engine.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/engine.py
@@ -80,6 +80,7 @@ async def run_orchestration(
intent_text: str = "",
enable_report: bool = True,
plan: dict | None = None,
+ gather_first: bool = False,
) -> OrchestrationResult:
"""运行一次完整编排(循环 → 共识 → 报告绘制)。
@@ -97,23 +98,32 @@ async def run_orchestration(
可解析出非空 stages 时走 ``_run_dag``;否则(dag 但 plan 缺失/为空)回退 recommend。
enable_report: 是否在收口后执行 Step3(方案总结报告 + 结果绘制)。False 时只跑完
Step2 研讨即收尾(step3=None)。两段产物各自故障隔离,详见 ``_run_step3``。
+ gather_first: 是否在研讨前先跑一轮**全席位并行『取数』**(见 ``_run_gather_phase``),结果作为
+ 共享资料注入各研讨轮席位任务。业务链条逐条 opt-in(默认 False)。三种模式通用——尤其让
+ chain(串行)的技能延迟前置并行、研讨段提速。
"""
intent = (intent_text or seed_message or "").strip()
try:
await progress.begin()
await progress.diag(
stage="step2_orchestration", level="info", event="begin",
- message=f"会商编排启动(mode={mode},{len(agents)} 个席位)",
- detail={"mode": mode, "agents": [a.agent_id for a in agents]},
+ message=f"会商编排启动(mode={mode},{len(agents)} 个席位{',研讨前并行取数' if gather_first else ''})",
+ detail={"mode": mode, "agents": [a.agent_id for a in agents], "gather_first": gather_first},
+ )
+ # 研讨前并行取数(opt-in):全席位并行只取数,产出作为共享资料注入后续研讨。失败/为空安全降级。
+ gathered = (
+ await _run_gather_phase(gateway, progress, agents, thread_ids, model, intent)
+ if gather_first
+ else {}
)
dag_plan = parse_dag_plan(plan) if mode == "dag" else None
if mode == "dag" and dag_plan is not None:
- outcome = await _run_dag(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message, intent, dag_plan)
+ outcome = await _run_dag(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message, intent, dag_plan, gathered=gathered)
elif mode == "chain":
- outcome = await _run_chain(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message, intent)
+ outcome = await _run_chain(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message, intent, gathered=gathered)
else:
# recommend,或 dag 但 plan 缺失/为空 → 回退总控自由派活(绝不静默跑错语义)。
- outcome = await _run_recommend(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message, intent)
+ outcome = await _run_recommend(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message, intent, gathered=gathered)
if outcome.kind == "clarification":
question = outcome.question or "总控需要补充信息,请补充后继续。"
@@ -199,11 +209,30 @@ def _deliveries_block(deliveries: dict[str, str], name_by_id: dict[str, str]) ->
return "\n\n".join(parts)
-def _seat_message(intent_text: str, deliveries: dict[str, str], name_by_id: dict[str, str], task: str) -> str:
- """把任务背景 + 前序交付 + 防提问约束拼进席位任务(替代 thread 广播)。"""
+def _seat_message(
+ intent_text: str,
+ deliveries: dict[str, str],
+ name_by_id: dict[str, str],
+ task: str,
+ *,
+ gathered: dict[str, str] | None = None,
+) -> str:
+ """把任务背景 + (可选)研讨前并行收集的资料 + 前序交付 + 防提问约束拼进席位任务(替代 thread 广播)。
+
+ ``gathered``(可选)= 研讨前「并行取数阶段」各席位收集回来的**原始资料/事实**(见
+ ``_run_gather_phase``),与 ``deliveries``(研讨阶段的结论性交付)分块、各自标题不同:
+ 资料供席位直接引用以**减少重复检索**,结论供席位承接。两者都为空则都不拼。
+ """
parts: list[str] = []
if intent_text.strip():
parts.append(f"【本次研讨任务背景】\n{intent_text.strip()}")
+ if gathered:
+ gblock = _deliveries_block(gathered, name_by_id)
+ if gblock:
+ parts.append(
+ "【研讨前各席位已并行收集的资料/事实,供你直接引用——相关内容已在此,"
+ "尽量避免重复检索;确有缺口可再补充检索】\n" + gblock
+ )
block = _deliveries_block(deliveries, name_by_id)
if block:
parts.append("【前序各席位已完成的交付,请在此基础上承接、补充、不要重复或推翻】\n" + block)
@@ -212,6 +241,72 @@ def _seat_message(intent_text: str, deliveries: dict[str, str], name_by_id: dict
return "\n\n---\n\n".join(parts)
+# ── 研讨前「并行取数」阶段(可选,业务链条逐条 opt-in)──────────────────────────
+#
+# 各席位**只调技能/凭专业把相关信息收集回来、不研讨**,全席位 ``asyncio.gather`` 并行跑一轮,
+# 结果作为「共享资料」注入后续研讨轮(``_seat_message(gathered=...)``)。这样最慢的技能延迟只在
+# 这一次、并行发生;后续研讨段(哪怕 chain 串行)每席位基本变成一次纯推理调用,大幅缩短第二步
+# 总耗时。研讨阶段是否还再调技能不作限制(按用户设定)。
+
+_GATHER_TASK = (
+ "【资料收集阶段——只取数、不研讨】\n"
+ "现在是正式研讨**之前**的并行资料收集阶段。请你**只负责把与本次任务、与你这个席位职责相关的"
+ "真实信息/数据收集、整理出来**(优先用你配置的技能去检索/获取真实信息,没有技能可凭专业知识梳理),"
+ "为后续正式研讨打底。\n"
+ "**这一步不要做分析评估、不要给方案、不要下结论**——把判断与方案留到研讨阶段。"
+ "请把你收集到的关键信息整理成条理清晰的事实/要点清单输出。"
+)
+
+
+async def _run_gather_phase(
+ gateway: RoundtableGateway,
+ progress: JobProgressWriter,
+ agents: list[SeatRef],
+ thread_ids: dict[str, str],
+ model: str | None,
+ intent_text: str,
+) -> dict[str, str]:
+ """研讨前并行『取数』:所有席位并行只取数(不研讨),返回 {agent_id: 收集到的资料}。
+
+ 复用 ``_run_dag`` 同款 ``asyncio.gather`` + ``gateway.run_seat``;各席位独立 thread,进程内并行
+ 安全。某席位取数失败/为空不阻塞其它(其 gathered 项缺省即不注入)。结果由调用方传给后续研讨轮的
+ ``_seat_message(gathered=...)``。
+ """
+ name_by_id = {a.agent_id: a.name for a in agents}
+ gathered: dict[str, str] = {}
+ await progress.diag(
+ stage="step2_orchestration", level="info", event="gather_begin",
+ message=f"研讨前并行资料收集阶段启动({len(agents)} 个席位并行取数)",
+ detail={"agents": [a.agent_id for a in agents]},
+ )
+
+ async def _gather_one(seat: SeatRef) -> None:
+ await progress.seat_active(seat.agent_id)
+ seat_msg = _seat_message(intent_text, {}, name_by_id, _GATHER_TASK)
+ try:
+ seat_turn = await gateway.run_seat(thread_ids=thread_ids, agent_id=seat.agent_id, task=seat_msg, model=model)
+ except Exception: # noqa: BLE001 — 单席位取数失败不拖垮整阶段;其 gathered 缺省即可
+ logger.warning("gather phase seat %s failed", seat.agent_id, exc_info=True)
+ await progress.seat_done(seat.agent_id, 0)
+ return
+ text = (seat_turn.final_text or "").strip()
+ if text:
+ gathered[seat.agent_id] = text
+ await progress.add_seat_dialogue(
+ seat.agent_id, name_by_id.get(seat.agent_id, seat.agent_id), seat_turn.final_text, 0,
+ tool_calls=seat_turn.tool_calls, thread_id=thread_ids.get(seat.agent_id),
+ )
+ await progress.seat_done(seat.agent_id, 0)
+
+ await progress.dispatching(0)
+ await asyncio.gather(*(_gather_one(seat) for seat in agents))
+ await progress.diag(
+ stage="step2_orchestration", level="info", event="gather_done",
+ message=f"资料收集阶段完成({len(gathered)}/{len(agents)} 个席位有产出)",
+ )
+ return gathered
+
+
async def _run_recommend(
gateway: RoundtableGateway,
progress: JobProgressWriter,
@@ -221,6 +316,8 @@ async def _run_recommend(
model: str | None,
seed_message: str,
intent_text: str,
+ *,
+ gathered: dict[str, str] | None = None,
) -> _LoopOutcome:
deliveries: dict[str, str] = {}
last_leader = ""
@@ -246,24 +343,35 @@ async def _run_recommend(
if turn.is_consensus:
return _LoopOutcome("consensus", last_leader_content=last_leader, deliveries=deliveries)
- # **顺序**派活 —— 不要并行;每个席位带上「背景 + 前序交付」上下文
+ # 批内**并行**派活(与前端 runOrchestration「总控一轮派多席位=判定相互独立→并行」一致,
+ # 2026-06-20 由串行改并行):总控一轮派出的多个席位彼此独立,并发跑可大幅缩短本轮耗时。
+ # 每个席位带上「背景 +(可选)研讨前并行取数资料 + 前序交付」上下文。各席位独立 thread,进程内
+ # gather 安全;后端用户级信号量(_BACKGROUND_SLOTS_PER_USER)仍限制实际并发,保护 sqlite 锁。
+ # 批内交付在本轮结束后统一并入 deliveries(同前端:批内不互相看,下一轮总控才看到全部)。
await progress.dispatching(cycle)
- for agent_id, task in turn.dispatched:
+ # 派活前对照快照(批内各席位看到的「前序交付」一致,不受批内执行顺序影响 → 并行无副作用)。
+ prior_snapshot = dict(deliveries)
+ batch: dict[str, str] = {}
+
+ async def _run_one(agent_id: str, task: str, cyc: int = cycle) -> None:
await progress.seat_active(agent_id)
- seat_msg = _seat_message(intent_text, deliveries, name_by_id, task)
+ seat_msg = _seat_message(intent_text, prior_snapshot, name_by_id, task, gathered=gathered)
seat = await gateway.run_seat(thread_ids=thread_ids, agent_id=agent_id, task=seat_msg, model=model)
- deliveries[agent_id] = seat.final_text
+ batch[agent_id] = seat.final_text
if not (seat.final_text or "").strip():
await progress.diag(
stage="step2_seat", level="warning", event="seat_empty_delivery",
message=f"席位「{name_by_id.get(agent_id, agent_id)}」本轮交付为空(可能模型未产出或被截断)",
- agent_id=agent_id, agent_name=name_by_id.get(agent_id, agent_id), cycle=cycle,
+ agent_id=agent_id, agent_name=name_by_id.get(agent_id, agent_id), cycle=cyc,
)
await progress.add_seat_dialogue(
- agent_id, name_by_id.get(agent_id, agent_id), seat.final_text, cycle,
+ agent_id, name_by_id.get(agent_id, agent_id), seat.final_text, cyc,
tool_calls=seat.tool_calls, thread_id=thread_ids.get(agent_id),
)
- await progress.seat_done(agent_id, cycle)
+ await progress.seat_done(agent_id, cyc)
+
+ await asyncio.gather(*(_run_one(agent_id, task) for agent_id, task in turn.dispatched))
+ deliveries.update(batch)
await progress.bump_consensus()
# 续轮:把各席位本轮交付喂给总控,让它据实判断(总控 SOUL 依赖看到交付)
@@ -287,6 +395,8 @@ async def _run_chain(
model: str | None,
seed_message: str,
intent_text: str,
+ *,
+ gathered: dict[str, str] | None = None,
) -> _LoopOutcome:
deliveries: dict[str, str] = {}
last_leader = ""
@@ -323,9 +433,10 @@ async def _run_chain(
task = pick_chain_task(turn.dispatched, turn.final_text, seat.agent_id, seat.name)
- # (b) 当前席位交付(顺序,不并行);带上「背景 + 前序交付 + 别提问」上下文
+ # (b) 当前席位交付(顺序,不并行);带上「背景 +(可选)研讨前并行取数资料 + 前序交付 + 别提问」上下文。
+ # chain 严格串行(seat B 看 seat A 交付),但有 gathered 时各席位已带共享资料,不必各自从零检索 → 提速。
await progress.seat_active(seat.agent_id)
- seat_msg = _seat_message(intent_text, deliveries, name_by_id, task)
+ seat_msg = _seat_message(intent_text, deliveries, name_by_id, task, gathered=gathered)
seat_turn = await gateway.run_seat(thread_ids=thread_ids, agent_id=seat.agent_id, task=seat_msg, model=model)
deliveries[seat.agent_id] = seat_turn.final_text
if not (seat_turn.final_text or "").strip():
@@ -376,6 +487,8 @@ async def _run_dag(
seed_message: str,
intent_text: str,
plan: DagPlan,
+ *,
+ gathered: dict[str, str] | None = None,
) -> _LoopOutcome:
"""分层 DAG 编排:每个 stage 一轮总控派活(钉死本批席位)→ 批内席位**并行**执行
(``asyncio.gather``)→ 下一 stage 总控再派;全部 stage 完成后由总控收口综合。
@@ -429,9 +542,9 @@ async def _run_dag(
for seat in stage_seats:
await progress.seat_active(seat.agent_id)
- async def _run_one(seat: SeatRef) -> None:
+ async def _run_one(seat: SeatRef, _gathered: dict[str, str] | None = gathered) -> None:
task = pick_dag_seat_task(turn.dispatched, seat.agent_id, seat.name, goal=stage.goal)
- seat_msg = _seat_message(intent_text, deliveries, name_by_id, task)
+ seat_msg = _seat_message(intent_text, deliveries, name_by_id, task, gathered=_gathered)
seat_turn = await gateway.run_seat(thread_ids=thread_ids, agent_id=seat.agent_id, task=seat_msg, model=model)
deliveries[seat.agent_id] = seat_turn.final_text
if not (seat_turn.final_text or "").strip():
@@ -693,6 +806,17 @@ def build_summary_prompt(
lines.append("# 三、总控最终共识结论")
lines.append((consensus_text or "").strip() or "(未提供)")
lines.append("")
+ # 意图对齐 + 缺口核对 + 防幻觉护栏(弱席位场景的质量兜底)。即便部署里的 SOUL 未重新 seed,
+ # 这段也保证后台总结会主动扣用户意图、标注推断/缺口,而非简单拼接各席位交付。
+ lines.append(
+ "【动笔前必做:意图对齐与缺口核对】各席位可能由能力较弱的模型产出,未必扣住用户意图、未必充分。"
+ "你是这份报告的质量兜底,不能只把各席位交付简单拼接。请先把上面的「任务意图」(目标/约束/关键假设)"
+ "拆成用户的核心诉求清单(编排提示对最终产出的要求也并入),逐条核对各席位交付+共识是否覆盖、够不够、"
+ "有无跑题或矛盾;再据此组织报告并按护栏补强:① 能从已有原料重组/推理补上的→直接补全讲透;"
+ "② 属于你自己的推断/建议→写,但显式标注「(综合推断)」;③ 缺真实信息、靠现有原料补不了的→**绝不凭空编**,"
+ "用一行 `> ⚠️ 待补充:<具体缺什么>。可在对话里让我补全` 标出,把球交给用户。"
+ "最终报告要明显比「各席位交付的简单拼接」更扣用户意图。"
+ )
lines.append(
"请严格遵循你的职责设定:纯 Markdown 文本(不画 HTML / 图表)、全程简体中文。"
"报告结构:默认用叙事型结构(一句话结论 → 背景目标 → 各席位观点 → 综合分析 → 最终共识方案 → 执行路线);"
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260620_02_chain_gather_first.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260620_02_chain_gather_first.py
new file mode 100644
index 00000000..bb035e45
--- /dev/null
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260620_02_chain_gather_first.py
@@ -0,0 +1,54 @@
+"""Add gather_first column to roundtable_chains.
+
+Per-chain「研讨前并行取数」opt-in switch: when true, selecting the chain into a roundtable
+discussion first runs one **parallel data-gathering round** over all seats (each seat only
+calls its skills to fetch facts, no deliberation); the collected material is injected as
+shared context into the discussion rounds, front-loading the slow skill latency in parallel
+so the (possibly serial) discussion phase runs faster. Default false → existing behavior
+unchanged. BOOLEAN NOT NULL DEFAULT 0, matching ``RoundtableChainRow.gather_first``.
+
+``create_all`` adds the column for brand-new DBs; the startup ``_ensure_orm_columns_sync``
+sweep adds it to already-existing tables (sqlite/mysql/postgres). This revision keeps the
+alembic graph consistent with the rest of the chain columns and documents the change.
+
+Revision ID: 20260620_02
+Revises: 20260619_01
+Create Date: 2026-06-20
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+
+
+revision = "20260620_02"
+down_revision = "20260619_01"
+branch_labels = None
+depends_on = None
+
+
+def _has_table(table_name: str) -> bool:
+ inspector = inspect(op.get_bind())
+ return table_name in inspector.get_table_names()
+
+
+def _has_column(table_name: str, column_name: str) -> bool:
+ if not _has_table(table_name):
+ return False
+ inspector = inspect(op.get_bind())
+ return any(col["name"] == column_name for col in inspector.get_columns(table_name))
+
+
+def upgrade() -> None:
+ if _has_table("roundtable_chains") and not _has_column("roundtable_chains", "gather_first"):
+ op.add_column(
+ "roundtable_chains",
+ sa.Column("gather_first", sa.Boolean(), nullable=False, server_default="0"),
+ )
+
+
+def downgrade() -> None:
+ if _has_column("roundtable_chains", "gather_first"):
+ op.drop_column("roundtable_chains", "gather_first")
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_chains/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_chains/model.py
index eb2d5baf..bba96994 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_chains/model.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_chains/model.py
@@ -52,6 +52,12 @@ class RoundtableChainRow(Base):
# 可选「默认席位执行模式」:flash/thinking/pro/ultra(见前端 lib/seat-mode.ts)。选入研讨时
# 弹窗默认用它(仍可临时改)。NULL = 未设,调用方回退默认(pro)。短串,VARCHAR(16)。
seat_mode: Mapped[str | None] = mapped_column(String(16), nullable=True)
+ # 可选「研讨前并行取数」开关:True = 选入研讨时,研讨前先跑一轮全席位并行『取数』(各席位只调
+ # 技能取数、不研讨),产出作为共享资料注入后续研讨轮,把最慢的技能延迟前置并行、给串行研讨段提速。
+ # 默认 False(不变更现有行为)。逐条 opt-in。BOOLEAN,server_default "0"。
+ gather_first: Mapped[bool] = mapped_column(
+ Boolean, nullable=False, default=False, server_default="0"
+ )
# 发布开关:True = 公共(所有账号可在「公共」筛选里查看 / 选用);False = 仅本人可见。
# 编辑 / 删除始终只限 owner;发布只是放开「查看 + 选入研讨」。索引以支持「公共」列表筛选。
is_public: Mapped[bool] = mapped_column(
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_chains/sql.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_chains/sql.py
index f3506589..99c3ef2e 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_chains/sql.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_chains/sql.py
@@ -58,6 +58,8 @@ class RoundtableChainRepository:
"coordinator_prompt": getattr(row, "coordinator_prompt", None),
# 默认席位执行模式(flash/thinking/pro/ultra);None = 未设。
"seat_mode": getattr(row, "seat_mode", None),
+ # 研讨前并行取数开关(逐条 opt-in);缺省 False。
+ "gather_first": bool(getattr(row, "gather_first", False)),
# 发布状态 + 是否本人所有(前端据此显示「公共」徽标、决定能否编辑 / 发布)。
"is_public": bool(getattr(row, "is_public", False)),
"is_mine": (current_user_id is None) or (row.user_id == current_user_id),
@@ -113,6 +115,7 @@ class RoundtableChainRepository:
stage_goals=_dumps(data.get("stage_goals")),
coordinator_prompt=(data.get("coordinator_prompt") or None),
seat_mode=(data.get("seat_mode") or None),
+ gather_first=bool(data.get("gather_first") or False),
is_public=bool(data.get("is_public") or False),
created_at=now,
updated_at=now,
@@ -135,6 +138,7 @@ class RoundtableChainRepository:
stage_goals=_UNSET,
coordinator_prompt=_UNSET,
seat_mode=_UNSET,
+ gather_first=_UNSET,
is_public=_UNSET,
) -> dict[str, Any] | None:
async with self._sf() as session:
@@ -155,6 +159,8 @@ class RoundtableChainRepository:
row.coordinator_prompt = coordinator_prompt or None
if seat_mode is not _UNSET:
row.seat_mode = seat_mode or None
+ if gather_first is not _UNSET and gather_first is not None:
+ row.gather_first = bool(gather_first)
if is_public is not _UNSET and is_public is not None:
row.is_public = bool(is_public)
row.updated_at = datetime.now(UTC)
diff --git a/offline-backend-20260512/backend/scripts/sql/column_additions_mysql.sql b/offline-backend-20260512/backend/scripts/sql/column_additions_mysql.sql
index 852d3d78..b240ad09 100644
--- a/offline-backend-20260512/backend/scripts/sql/column_additions_mysql.sql
+++ b/offline-backend-20260512/backend/scripts/sql/column_additions_mysql.sql
@@ -89,6 +89,11 @@ CALL _add_column_if_missing('roundtable_jobs', 'orchestration_plan', 'LONGTEXT N
-- 选入研讨时弹窗默认用它(仍可临时改)。ORM: RoundtableChainRow.seat_mode = String(16), nullable。
CALL _add_column_if_missing('roundtable_chains', 'seat_mode', 'VARCHAR(16) NULL DEFAULT NULL');
+-- gather_first: 业务链条「研讨前并行取数」开关,对应 migration 20260620_02。True=选入研讨时研讨前先
+-- 跑一轮全席位并行取数(只调技能、不研讨),产出作为共享资料注入研讨轮。逐条 opt-in。
+-- ORM: RoundtableChainRow.gather_first = Boolean, NOT NULL, server_default '0'。
+CALL _add_column_if_missing('roundtable_chains', 'gather_first', 'TINYINT(1) NOT NULL DEFAULT 0');
+
-- ---- 岗位 (positions) 订阅系统,对应 migration 20260616_01 -------------------
-- positions / position_tag_bindings 是【新表】,由 create_all 自动建,无需补列。
-- 以下是给【已存在的表】补的列:
diff --git a/offline-backend-20260512/backend/tests/test_roundtable_gather_phase.py b/offline-backend-20260512/backend/tests/test_roundtable_gather_phase.py
new file mode 100644
index 00000000..fa8de8ff
--- /dev/null
+++ b/offline-backend-20260512/backend/tests/test_roundtable_gather_phase.py
@@ -0,0 +1,140 @@
+"""会商 Step2 性能改造引擎测试:研讨前并行『取数』阶段(gather_first)+ recommend 批内并行。
+
+覆盖:
+ - gather_first=True:研讨前先对**全席位并行**跑一轮「只取数」,产出作为共享资料注入后续研讨轮席位任务;
+ - gather_first=False(默认):不跑取数阶段,行为不变;
+ - recommend 一轮派多席位 → 批内**并行**执行(_run_recommend 由串行改 asyncio.gather,峰值并发>1)。
+
+引擎纯逻辑,用 JobProgressWriter(store=None) 干跑 + 录制网关,不需要 DB / 真模型。
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from deerflow.agents.roundtable_orchestrator import (
+ JobProgressWriter,
+ SeatRef,
+ run_orchestration,
+)
+from deerflow.agents.roundtable_orchestrator.types import LeaderTurn, SeatTurn
+
+_AGENTS = [SeatRef("seat-a", "情报"), SeatRef("seat-b", "环境"), SeatRef("seat-c", "方案")]
+# _GATHER_TASK 里的稳定标记(取数阶段)/ _seat_message 里注入资料块的稳定标记(研讨阶段)。
+_GATHER_MARK = "资料收集阶段"
+_GATHERED_BLOCK_MARK = "研讨前各席位已并行收集"
+
+
+def _writer() -> JobProgressWriter:
+ return JobProgressWriter(None, job_id="job-x", agents=_AGENTS)
+
+
+def _thread_ids() -> dict[str, str]:
+ return {a.agent_id: f"tid-{a.agent_id}" for a in _AGENTS} | {"roundtable-coordinator": "tid-co"}
+
+
+class _RecGateway:
+ """录制网关:run_leader 按预设各轮派活;run_seat 记录 (agent_id, task) 并追踪峰值并发。"""
+
+ def __init__(self, dispatch_cycles: list[list[tuple[str, str]]]) -> None:
+ self._cycles = dispatch_cycles
+ self._leader_calls = 0
+ self.seat_tasks: list[tuple[str, str]] = []
+ self.active = 0
+ self.max_active = 0
+
+ async def run_leader(self, *, thread_ids, coordinator_name, message, model) -> LeaderTurn:
+ i = self._leader_calls
+ self._leader_calls += 1
+ disp = self._cycles[i] if i < len(self._cycles) else []
+ return LeaderTurn(final_text=f"leader-{i}", dispatched=disp)
+
+ async def run_seat(self, *, thread_ids, agent_id, task, model) -> SeatTurn:
+ self.active += 1
+ self.max_active = max(self.max_active, self.active)
+ self.seat_tasks.append((agent_id, task))
+ await asyncio.sleep(0.01) # 留出并发重叠窗口
+ self.active -= 1
+ return SeatTurn(final_text=f"{agent_id} 的产出")
+
+ # Step3 关掉,不应被调用
+ async def create_report_thread(self) -> str: # pragma: no cover - 不应触发
+ raise AssertionError("enable_report=False 不应绘制报告")
+
+
+async def _run(gw: _RecGateway, *, gather_first: bool):
+ return await run_orchestration(
+ gateway=gw,
+ progress=_writer(),
+ agents=_AGENTS,
+ thread_ids=_thread_ids(),
+ coordinator_name="roundtable-coordinator",
+ model=None,
+ seed_message="背景X",
+ mode="recommend",
+ intent_text="背景X",
+ enable_report=False,
+ gather_first=gather_first,
+ )
+
+
+# 一轮派全部 3 席位 → 下一轮共识收口。
+_DISPATCH_ALL_THEN_CONSENSUS = [[("seat-a", "taskA"), ("seat-b", "taskB"), ("seat-c", "taskC")], []]
+
+
+@pytest.mark.asyncio
+async def test_recommend_dispatch_runs_in_parallel():
+ """recommend 一轮派多席位 → 批内并行(峰值并发 = 席位数,证明已由串行改并行)。"""
+ gw = _RecGateway(_DISPATCH_ALL_THEN_CONSENSUS)
+ result = await _run(gw, gather_first=False)
+ assert result.status == "done"
+ assert gw.max_active == 3 # 三席位真并行;若仍串行则恒为 1
+
+
+@pytest.mark.asyncio
+async def test_gather_first_false_skips_gather_phase():
+ """默认 gather_first=False:不跑取数阶段,run_seat 只在研讨轮被调,无取数任务。"""
+ gw = _RecGateway(_DISPATCH_ALL_THEN_CONSENSUS)
+ await _run(gw, gather_first=False)
+ assert not any(_GATHER_MARK in task for _, task in gw.seat_tasks)
+ # 研讨轮任务也不应带「已收集资料」块(没有取数)
+ assert not any(_GATHERED_BLOCK_MARK in task for _, task in gw.seat_tasks)
+ # 仅 3 次席位调用(一轮研讨派活)
+ assert len(gw.seat_tasks) == 3
+
+
+@pytest.mark.asyncio
+async def test_gather_first_runs_parallel_gather_then_injects_material():
+ """gather_first=True:研讨前对全席位并行跑「只取数」一轮,产出注入后续研讨轮任务。"""
+ gw = _RecGateway(_DISPATCH_ALL_THEN_CONSENSUS)
+ result = await _run(gw, gather_first=True)
+ assert result.status == "done"
+
+ gather_calls = [(aid, t) for aid, t in gw.seat_tasks if _GATHER_MARK in t]
+ discuss_calls = [(aid, t) for aid, t in gw.seat_tasks if _GATHER_MARK not in t]
+
+ # ① 取数阶段覆盖全部席位(每席位一次「只取数」任务)
+ assert {aid for aid, _ in gather_calls} == {"seat-a", "seat-b", "seat-c"}
+ # ② 取数任务在研讨任务之前(取数是前置阶段)
+ first_discuss_idx = gw.seat_tasks.index(discuss_calls[0])
+ last_gather_idx = max(gw.seat_tasks.index(c) for c in gather_calls)
+ assert last_gather_idx < first_discuss_idx
+ # ③ 研讨轮任务带上「研讨前各席位已并行收集的资料」块,且包含各席位取数产出
+ assert discuss_calls, "应有研讨轮任务"
+ for _, dtask in discuss_calls:
+ assert _GATHERED_BLOCK_MARK in dtask
+ assert "seat-a 的产出" in dtask # 注入了取数阶段的产物
+
+
+@pytest.mark.asyncio
+async def test_gather_phase_is_parallel():
+ """取数阶段本身并行(全席位并发跑,峰值并发 = 席位数)。"""
+ # 首轮 leader 直接共识(dispatched=[]),让研讨轮不抢占并发,单独度量取数阶段峰值。
+ gw = _RecGateway([[]])
+ await _run(gw, gather_first=True)
+ # 只有取数阶段跑了 run_seat(研讨轮因首轮共识没派活)
+ assert all(_GATHER_MARK in t for _, t in gw.seat_tasks)
+ assert len(gw.seat_tasks) == 3
+ assert gw.max_active == 3 # 取数阶段三席位真并行
diff --git a/offline-backend-20260512/backend/tests/test_roundtable_seat_modes.py b/offline-backend-20260512/backend/tests/test_roundtable_seat_modes.py
new file mode 100644
index 00000000..954218d9
--- /dev/null
+++ b/offline-backend-20260512/backend/tests/test_roundtable_seat_modes.py
@@ -0,0 +1,118 @@
+"""每席位「推理深度覆盖」(seat_modes)在进程内网关 run_seat 的解析测试。
+
+语义(与前端 useStep2Orchestration「覆盖优先、否则跟随链条级 selectedMode」逐席位对齐):
+- 席位命中 seat_modes → 用该档(_SEAT_MODE_PARAMS)派生的 thinking/reasoning/subagent **覆盖**
+ 作业级 seat_*(不论作业级是什么);
+- 席位未命中 → 跟随作业级 seat_thinking_enabled / seat_reasoning_effort / seat_subagent_enabled;
+- 作业级仍为 None → 落 policy("seat") 默认。
+
+只验证传给 _run_agent_turn 的三个运行参数,故把 _run_agent_turn / policy / 文本解析全部打桩,
+不触发真实 LLM / 沙箱。
+"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+
+@pytest.fixture
+def _gw(monkeypatch):
+ """构造一个被打桩的 InProcessRoundtableGateway,返回 (build, captured)。
+
+ build(seat_modes=..., seat_thinking_enabled=..., seat_reasoning_effort=..., seat_subagent_enabled=...)
+ 返回 gateway 实例;captured 记录最近一次 _run_agent_turn 收到的 thinking/reasoning/subagent。
+ """
+ import app.gateway.roundtable_inprocess_gateway as gw
+
+ captured: dict[str, object] = {}
+
+ async def _fake_run_agent_turn(*args, **kwargs):
+ captured["thinking_enabled"] = kwargs.get("thinking_enabled")
+ captured["reasoning_effort"] = kwargs.get("reasoning_effort")
+ captured["subagent_enabled"] = kwargs.get("subagent_enabled")
+ return []
+
+ # policy("seat") 默认:thinking=True / reasoning="medium"(= 现有席位默认 pro 行为)。
+ monkeypatch.setattr(
+ gw,
+ "roundtable_run_policy",
+ lambda role: SimpleNamespace(excluded_tools=[], skill_stop_names=[], thinking_enabled=True, reasoning_effort="medium"),
+ )
+ monkeypatch.setattr(gw, "_run_agent_turn", _fake_run_agent_turn)
+ # 技能强化与本测试无关 → 原样返回任务,避免触发 load_agent_config / skill storage。
+ monkeypatch.setattr(gw, "append_seat_skill_directive", lambda agent_id, task, **k: task)
+ monkeypatch.setattr(gw, "_latest_ai_text", lambda messages: "")
+ monkeypatch.setattr(gw, "_tool_calls_since_last_human", lambda messages: [])
+
+ def _build(**kwargs):
+ return gw.InProcessRoundtableGateway(SimpleNamespace(state=SimpleNamespace()), user_id="u1", agents=[], **kwargs)
+
+ return _build, captured
+
+
+async def _run(gateway):
+ await gateway.run_seat(thread_ids={"seat-1": "tid-1"}, agent_id="seat-1", task="任务", model="m")
+
+
+@pytest.mark.asyncio
+async def test_seat_mode_override_wins_over_job_level(_gw):
+ """席位配了 flash → 覆盖作业级 pro(thinking off / minimal / 无子代理)。"""
+ build, captured = _gw
+ gateway = build(
+ seat_modes={"seat-1": "flash"},
+ seat_thinking_enabled=True,
+ seat_reasoning_effort="high",
+ seat_subagent_enabled=True,
+ )
+ await _run(gateway)
+ assert captured["thinking_enabled"] is False
+ assert captured["reasoning_effort"] == "minimal"
+ assert captured["subagent_enabled"] is False
+
+
+@pytest.mark.asyncio
+async def test_ultra_override_enables_subagent(_gw):
+ build, captured = _gw
+ gateway = build(seat_modes={"seat-1": "ultra"})
+ await _run(gateway)
+ assert captured["thinking_enabled"] is True
+ assert captured["reasoning_effort"] == "high"
+ assert captured["subagent_enabled"] is True
+
+
+@pytest.mark.asyncio
+async def test_no_override_follows_job_level(_gw):
+ """席位未命中 seat_modes → 跟随作业级 seat_*。"""
+ build, captured = _gw
+ gateway = build(
+ seat_modes={"other": "flash"},
+ seat_thinking_enabled=True,
+ seat_reasoning_effort="low",
+ seat_subagent_enabled=False,
+ )
+ await _run(gateway)
+ assert captured["thinking_enabled"] is True
+ assert captured["reasoning_effort"] == "low"
+ assert captured["subagent_enabled"] is False
+
+
+@pytest.mark.asyncio
+async def test_no_override_no_job_level_falls_back_to_policy(_gw):
+ """既无席位覆盖、作业级也为 None → 落 policy 默认(thinking=True / medium)。"""
+ build, captured = _gw
+ gateway = build(seat_modes={})
+ await _run(gateway)
+ assert captured["thinking_enabled"] is True
+ assert captured["reasoning_effort"] == "medium"
+ assert captured["subagent_enabled"] is False
+
+
+@pytest.mark.asyncio
+async def test_unknown_mode_value_ignored_falls_through(_gw):
+ """非法档位值(前端不会下发,防御)→ 安全忽略,回退作业级/policy。"""
+ build, captured = _gw
+ gateway = build(seat_modes={"seat-1": "bogus"}, seat_reasoning_effort="low", seat_thinking_enabled=True)
+ await _run(gateway)
+ assert captured["thinking_enabled"] is True
+ assert captured["reasoning_effort"] == "low"
diff --git a/offline-backend-20260512/backend/tests/test_roundtable_summary_prompt.py b/offline-backend-20260512/backend/tests/test_roundtable_summary_prompt.py
new file mode 100644
index 00000000..b2f92741
--- /dev/null
+++ b/offline-backend-20260512/backend/tests/test_roundtable_summary_prompt.py
@@ -0,0 +1,50 @@
+"""方案总结报告 prompt(build_summary_prompt)的意图对齐 + 缺口护栏内容测试。
+
+效果改造 B:弱席位场景下,总结智能体不能只拼接各席位交付,而要先对齐用户意图、核对覆盖、
+按护栏补强(推断标注「(综合推断)」、缺真实信息标注「待补充」交给用户、绝不凭空编)。
+prompt 是纯函数,直接断言其文本包含这些指令 + 仍带齐意图/交付/共识/编排提示原料。
+"""
+
+from __future__ import annotations
+
+from deerflow.agents.roundtable_orchestrator.engine import build_summary_prompt
+from deerflow.agents.roundtable_orchestrator.types import SeatRef
+
+_AGENTS = [SeatRef("seat-a", "情报"), SeatRef("seat-b", "方案")]
+_DELIVERIES = {"seat-a": "情报席位交付内容", "seat-b": "方案席位交付内容"}
+
+
+def test_summary_prompt_carries_intent_and_alignment_directive():
+ out = build_summary_prompt(
+ "目标:拿下华东市场\n约束条件:\n- 预算有限",
+ _AGENTS,
+ _DELIVERIES,
+ "总控共识:分两期推进",
+ coordinator_prompt="重点突出风险",
+ )
+ # 原料齐全
+ assert "拿下华东市场" in out
+ assert "预算有限" in out
+ assert "情报席位交付内容" in out and "方案席位交付内容" in out
+ assert "总控共识:分两期推进" in out
+ assert "重点突出风险" in out # 编排提示
+ # 意图对齐 + 缺口核对指令
+ assert "意图对齐与缺口核对" in out
+ assert "核心诉求清单" in out
+ # 防幻觉护栏三条
+ assert "综合推断" in out
+ assert "待补充" in out
+ assert "绝不凭空编" in out
+
+
+def test_summary_prompt_alignment_directive_present_even_without_coordinator_prompt():
+ out = build_summary_prompt("目标:X", _AGENTS, _DELIVERIES, "共识 Y")
+ assert "意图对齐与缺口核对" in out
+ assert "待补充" in out
+
+
+def test_summary_prompt_handles_empty_deliveries_gracefully():
+ out = build_summary_prompt("目标:X", _AGENTS, {}, "")
+ # 无交付时仍给出兜底说明,且意图对齐指令照常在
+ assert "暂无文字交付" in out
+ assert "意图对齐与缺口核对" in out