From 3ab674e438957472b7758ea711bd8ae8298dbbb3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Sat, 20 Jun 2026 16:28:02 +0800 Subject: [PATCH] =?UTF-8?q?feat(roundtable):=20Step2=20=E6=8F=90=E9=80=9F?= =?UTF-8?q?=E5=9B=9B=E4=BB=B6=E5=A5=97=20+=20=E6=96=B9=E6=A1=88=E6=80=BB?= =?UTF-8?q?=E7=BB=93=E6=8A=A5=E5=91=8A=E6=84=8F=E5=9B=BE=E5=AF=B9=E9=BD=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step2 提速: - 每席位推理深度覆盖(seat_mode): 业务链条逐席位可配 flash/thinking/pro/ultra, 覆盖链条级 seatMode;前台三处 dispatch + 后台 job 透传, run_seat 用 _SEAT_MODE_PARAMS 解析 - recommend 后端批内并行: engine._run_recommend 由串行 for 改 asyncio.gather(对齐前端) - 研讨前并行取数两段式(gather_first, 业务链条 opt-in): 研讨前全席位并行只取数, 产出作为共享资料注入研讨轮(_seat_message gathered + 前台 runGatherPhase); 新增 gather_first 列(migration 20260620_02 + idempotent SQL) - 削减重复广播: suppressPreBroadcast 扩到 recommend(round>=2)/chain(i>0)/chain收口, 保留首轮背景打底,省 N 次读写大 thread 方案总结报告效果: - summary 加"意图对齐+缺口核对+防幻觉护栏": SOUL + 后端 build_summary_prompt + 前端 buildSummaryPrompt 三处对齐; 推断标注(综合推断)、缺真实信息标注待补充交给用户、绝不凭空编 Tests: test_roundtable_seat_modes / test_roundtable_gather_phase / test_roundtable_summary_prompt Co-Authored-By: Claude Opus 4.8 (1M context) --- .../src/roundtable-planning/api/chains.ts | 8 + .../api/roundtable-jobs.ts | 7 + .../components/BusinessChainPicker.tsx | 11 +- .../hooks/useStep2Orchestration.ts | 189 +++++++++++++++++- .../hooks/useStep3Summary.ts | 10 + .../roundtable-planning/lib/business-chain.ts | 34 +++- .../roundtable-planning/lib/chain-stages.ts | 16 ++ .../src/roundtable-planning/lib/job-types.ts | 10 + .../pages/BusinessChainEditorPage.tsx | 91 ++++++++- .../pages/RoundtablePlanningPage.tsx | 28 +++ offline-backend-20260512/backend/CLAUDE.md | 2 +- .../backend/app/gateway/deps.py | 2 + .../gateway/roundtable_inprocess_gateway.py | 39 +++- .../app/gateway/roundtable_job_executor.py | 8 + .../roundtable-summary/SOUL.md | 16 ++ .../app/gateway/routers/roundtable_chains.py | 11 + .../app/gateway/routers/roundtable_jobs.py | 8 + .../agents/roundtable_orchestrator/engine.py | 160 +++++++++++++-- .../20260620_02_chain_gather_first.py | 54 +++++ .../persistence/roundtable_chains/model.py | 6 + .../persistence/roundtable_chains/sql.py | 6 + .../scripts/sql/column_additions_mysql.sql | 5 + .../tests/test_roundtable_gather_phase.py | 140 +++++++++++++ .../tests/test_roundtable_seat_modes.py | 118 +++++++++++ .../tests/test_roundtable_summary_prompt.py | 50 +++++ 25 files changed, 985 insertions(+), 44 deletions(-) create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260620_02_chain_gather_first.py create mode 100644 offline-backend-20260512/backend/tests/test_roundtable_gather_phase.py create mode 100644 offline-backend-20260512/backend/tests/test_roundtable_seat_modes.py create mode 100644 offline-backend-20260512/backend/tests/test_roundtable_summary_prompt.py 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; + /** + * 每席位「推理深度覆盖」映射 {agentId: SeatMode | null}（沿用所选业务链条里每个席位的配置）。 + * 派活该 special 席位时按 agentId 查表，**席位覆盖优先、否则跟随链条级 `selectedMode`**： + * `seatModeToRequestFields(seatModes[agentId] ?? selectedMode)`。模型卡顿时把次要席位降到 + * `flash` 即可大幅提速。recommend / 未选链条 / 未覆盖 → 跟随 `selectedMode`。 + */ + seatModes?: Record; + /** + * 「研讨前并行取数」开关（来源业务链条的 gatherFirst，逐条 opt-in，缺省 false）。true = 正式研讨 + * 前先对**全席位并行**跑一轮「只取数」（各席位只调技能取数、不研讨），产出作为共享资料（peerDeliveries） + * 注入后续研讨轮，把最慢的技能检索前置并行、给（尤其 chain 串行的）研讨段提速。每轮研讨只在**本轮 + * 研讨真正启动时**取数一次（resetInitGate 复位后下一轮再取），澄清续跑/干预不重复取数。 + */ + gatherFirst?: boolean; /** * 编排模式: * - `recommend`(默认)—— 总控自由派活循环(`runOrchestration`); @@ -441,6 +455,8 @@ export function useStep2Orchestration( selectedModel, selectedMode, seatSkillDirectives = {}, + seatModes = {}, + gatherFirst = false, orchestrationMode, orchestrationPlan, dagExecMode = "fast", @@ -576,6 +592,14 @@ export function useStep2Orchestration( const selectedModeRef = useRef(selectedMode); // 每席位技能识别强化映射（所选业务链条各席位开关）同样打到 ref，席位调用按 agentId 读最新值。 const seatSkillDirectivesRef = useRef>(seatSkillDirectives); + // 每席位推理深度覆盖映射（所选业务链条各席位 seatMode）打到 ref，席位调用按 agentId 读最新值； + // 覆盖优先、否则跟随 selectedModeRef。 + const seatModesRef = useRef>(seatModes); + // 「研讨前并行取数」开关 + 本轮取数产物（{name,content}[]）+ 本轮是否已取过的闸门。取数产物作为 + // peerDeliveries 注入研讨轮席位；闸门保证一轮研讨只取一次（resetInitGate / resetStep2State 复位）。 + const gatherFirstRef = useRef(gatherFirst); + const gatheredDeliveriesRef = useRef>([]); + const gatherPhaseDoneRef = useRef(false); const agentConfigsRef = useRef>(agentConfigs); const agentNameToRoleRef = useRef(agentNameToRole); const threadIdsRef = useRef(threadIds); @@ -618,6 +642,12 @@ export function useStep2Orchestration( useEffect(() => { seatSkillDirectivesRef.current = seatSkillDirectives; }, [seatSkillDirectives]); + useEffect(() => { + seatModesRef.current = seatModes; + }, [seatModes]); + useEffect(() => { + gatherFirstRef.current = gatherFirst; + }, [gatherFirst]); useEffect(() => { agentConfigsRef.current = agentConfigs; }, [agentConfigs]); @@ -1585,7 +1615,8 @@ export function useStep2Orchestration( "③ 风险 / 注意事项（如有）；④ 与其它席位的依赖 / 承接关系（如有）。"; // 优化4：跨轮累积各席位**完整交付**，作为 peerDeliveries 传给后续席位的 run // （后端注入 context，席位用 read_peer_delivery 按需取全文，不进 prompt 不耗 token）。 - const priorDeliveries: Array<{ name: string; content: string }> = []; + // 研讨前并行取数（gatherFirst）的产物作为初始 peerDeliveries 注入：研讨席位开局即带共享资料。 + const priorDeliveries: Array<{ name: string; content: string }> = [...gatheredDeliveriesRef.current]; // 共享的「单席位交付执行」——串行/并行分支共用。含**失败自动重试一次**（优化5， // 对齐 Step3 自愈：网络错误 / 502 重试，abort 不重试）。返回交付正文（aborted → 空）。 @@ -1613,7 +1644,8 @@ export function useStep2Orchestration( peerDeliveries: priorDeliveries, model: perSeatModel || selectedModelRef.current || undefined, // 席位执行模式（快速/思考/专业问答/多智能体问答）→ thinking/reasoning/subagent。 - ...seatModeToRequestFields(selectedModeRef.current), + // 席位覆盖优先（业务链条里给该席位单独配的推理深度），否则跟随链条级 selectedMode。 + ...seatModeToRequestFields(seatModesRef.current[a.agentId] ?? selectedModeRef.current), // 会商技能识别强化（该席位在所选业务链条里的开关，按 agentId 查表）；与该席位 // agent 自身开关取 OR。 seatSkillDirective: seatSkillDirectivesRef.current[a.agentId] ?? false, @@ -1698,6 +1730,11 @@ export function useStep2Orchestration( // 第 2 轮起(已有席位交付)才是「综合轮」:让后端注入各席位完整交付供总控收口。 // 首轮纯派活、无交付,传 false 保持上下文精简。 synthesisMode: safetyCounter > 1, + // 削减重复广播(2026-06-20):首轮(safetyCounter==1)保留预广播——它把任务背景播进各席位 + // thread,为席位打底;第 2 轮起的续轮/综合广播是「总控收到:<续轮 prompt>」纯噪声 + N 次 + // 读写巨大席位 thread 的纯浪费(还会让弱模型从历史模仿派活),故 suppress。席位的前序交付 + // 另经「席位交付广播」(broadcastSeatDeliveryToOthers)送达,不受影响。 + suppressPreBroadcast: safetyCounter > 1, // 用户随干预上传的参考文件(consume-once:仅首轮 leader 带,之后清空)。 files: takeOnce(pendingInjectFilesRef), model: selectedModelRef.current || undefined, @@ -1969,6 +2006,10 @@ export function useStep2Orchestration( dAgentThreadId: ids, agentName: coord, newMessage: leaderTask, + // 削减重复广播(2026-06-20):首个席位(i===0)保留预广播为其打底任务背景;从第 2 个席位起 + // suppress——后续席位的前序交付已由「席位交付广播」送达其 thread,总控派活 prompt 再播一遍 + // 是 N 次读写巨大 thread 的纯浪费(还会让弱模型模仿派活)。 + suppressPreBroadcast: i > 0, // 用户随干预上传的参考文件(consume-once:仅首轮 leader 带,之后清空)。 files: takeOnce(pendingInjectFilesRef), model: selectedModelRef.current || undefined, @@ -2043,9 +2084,13 @@ export function useStep2Orchestration( agentName: seat.agent_id, displayName: seatName, newMessage: task, + // 研讨前并行取数（gatherFirst）的产物作为共享资料注入：chain 席位也能直接引用、少重复检索。 + // chain 串行靠 thread 广播看前序交付；取数资料这里另经 peerDeliveries 一并带上。 + peerDeliveries: gatheredDeliveriesRef.current, model: perSeatModel || selectedModelRef.current || undefined, // 席位执行模式 → thinking/reasoning/subagent（与推荐/DAG 路径一致）。 - ...seatModeToRequestFields(selectedModeRef.current), + // 席位覆盖优先（业务链条里给该席位单独配的推理深度），否则跟随链条级 selectedMode。 + ...seatModeToRequestFields(seatModesRef.current[seat.agent_id] ?? selectedModeRef.current), // 会商技能识别强化（该席位在所选业务链条里的开关，按 agentId 查表）；与该席位 // agent 自身开关取 OR。 seatSkillDirective: seatSkillDirectivesRef.current[seat.agent_id] ?? false, @@ -2109,6 +2154,8 @@ export function useStep2Orchestration( "各链条席位已按顺序完成发言,请你综合所有席位的交付,输出最终方案与结论。不要再派活、不要再向任何席位分配任务。", // 链条收口就是综合轮:让后端注入各席位完整交付正文。 synthesisMode: true, + // 削减重复广播:收口轮不派活、研讨已结束,把总控收口 prompt 播进席位 thread 纯浪费,suppress。 + suppressPreBroadcast: true, model: selectedModelRef.current || undefined, signal: closeoutAborter.signal, onTextDelta: (text) => appendToDialogue(closeoutBubbleId, text), @@ -2207,7 +2254,8 @@ export function useStep2Orchestration( let finishedAllStages = true; // 跨 stage 累积的**完整交付** {name, content}[] —— 注入下一 stage 席位的 run context // (peerDeliveries),供 read_peer_delivery 工具按需取全文(§12.9 传摘要按需传全文)。 - const priorDeliveries: Array<{ name: string; content: string }> = []; + // 研讨前并行取数（gatherFirst）的产物作为初始注入：首个 stage 席位开局即带共享资料。 + const priorDeliveries: Array<{ name: string; content: string }> = [...gatheredDeliveriesRef.current]; // Phase 7：初始化 DAG 流程图节点状态 —— 计划内所有席位置 pending,执行过程中逐个点亮。 const markDagNode = (agentId: string, state: ChainNodeState) => @@ -2388,7 +2436,8 @@ export function useStep2Orchestration( peerDeliveries: priorDeliveries, model: perSeatModel || selectedModelRef.current || undefined, // 席位执行模式 → thinking/reasoning/subagent（与推荐/链条路径一致）。 - ...seatModeToRequestFields(selectedModeRef.current), + // 席位覆盖优先（业务链条里给该席位单独配的推理深度），否则跟随链条级 selectedMode。 + ...seatModeToRequestFields(seatModesRef.current[agentId] ?? selectedModeRef.current), // 会商技能识别强化（该席位在所选业务链条里的开关，按 agentId 查表）；与该 // 席位 agent 自身开关取 OR。 seatSkillDirective: seatSkillDirectivesRef.current[agentId] ?? false, @@ -2516,10 +2565,131 @@ export function useStep2Orchestration( ], ); + // 研讨前「并行取数」阶段：对**全席位并行**跑一轮「只取数、不研讨」，产出存进 gatheredDeliveriesRef + // 作为共享资料注入后续研讨轮（recommend/dag 经 priorDeliveries、chain 经 peerDeliveries）。复用 + // recommend 同款并行批显示。每轮研讨只取一次（由 runActiveOrchestration 的 gatherPhaseDoneRef 闸门控制）。 + const runGatherPhase = useCallback( + async (ids: ThreadIdMap): Promise => { + const agents = selectedAgentsRef.current; + if (agents.length === 0) return; + // 与后端 engine._GATHER_TASK 同义：只取数、不研讨、不下结论。 + const GATHER_TASK = + "【资料收集阶段——只取数、不研讨】\n现在是正式研讨**之前**的并行资料收集阶段。请你**只负责把与本次任务、" + + "与你这个席位职责相关的真实信息/数据收集、整理出来**（优先用你配置的技能去检索/获取真实信息，没有技能可凭专业知识梳理），" + + "为后续正式研讨打底。\n**这一步不要做分析评估、不要给方案、不要下结论**——把判断与方案留到研讨阶段。" + + "请把你收集到的关键信息整理成条理清晰的事实/要点清单输出，**不要向用户提问或等待确认**。"; + const seatPrep = agents.map((a) => { + const meta = agentNameToRoleRef.current[a.agent_id.toLowerCase()]; + const seatName = meta?.displayName ?? a.name; + const avatarType = meta?.avatarType ?? "blue"; + const seatThreadId = resolveSeatThreadId(ids, a.agent_id); + const bubbleId = appendStreamingDialogue({ + sender: seatName, + avatarType, + initialContent: "", + confidence: "并行取数中", + agentThreadId: seatThreadId, + }); + return { agentId: a.agent_id, seatName, avatarType, seatThreadId, bubbleId, meta }; + }); + markParallelSpeakers( + seatPrep.map((p) => p.meta?.speakerId).filter((v): v is number => typeof v === "number"), + ); + setParallelBatch( + seatPrep.map((p) => ({ + agentId: p.agentId, + displayName: p.seatName, + avatarType: p.avatarType, + status: "running" as const, + bubbleId: p.bubbleId, + })), + ); + setSelectedParallelSeats(seatPrep.slice(0, 1).map((p) => p.agentId)); + const markBatchStatus = (agentId: string, status: ParallelSeatView["status"]) => + setParallelBatch((prev) => prev.map((p) => (p.agentId === agentId ? { ...p, status } : p))); + + const results = await Promise.all( + seatPrep.map(async ({ agentId, seatName, seatThreadId, bubbleId, meta }) => { + const perSeatModel = meta ? agentConfigsRef.current[meta.speakerId]?.model : undefined; + const subAborter = new AbortController(); + activeAbortersRef.current.add(subAborter); + try { + const specRes = await streamMultiAgent({ + agentType: "special", + dAgentThreadId: ids, + agentName: agentId, + displayName: seatName, + newMessage: GATHER_TASK, + model: perSeatModel || selectedModelRef.current || undefined, + ...seatModeToRequestFields(seatModesRef.current[agentId] ?? selectedModeRef.current), + seatSkillDirective: seatSkillDirectivesRef.current[agentId] ?? false, + signal: subAborter.signal, + onTextDelta: (text) => ingestParallelDelta(bubbleId, text), + onPhaseChange: (phase, toolName, args) => { + const label = phaseLabel(phase, "special", toolName); + setDialogueConfidence(bubbleId, label); + appendStepIfNew(bubbleId, label, `${phase}:${toolName ?? ""}`, args); + handleArtifactToolPhase(phase, toolName, args, seatThreadId); + }, + onUpstreamFrame: (frame) => { + handleUpstreamFrame(frame, seatThreadId); + ingestSubagentTaskFrame(frame, bubbleId); + }, + }); + activeAbortersRef.current.delete(subAborter); + finalizeDialogue(bubbleId, specRes.content, "资料收集"); + markBatchStatus(agentId, "done"); + if (meta) markSpeakerIdle(meta.speakerId); + parallelStreamRef.current.delete(bubbleId); + return specRes.content?.trim() ? { name: seatName, content: specRes.content } : null; + } catch (err) { + activeAbortersRef.current.delete(subAborter); + // 取数失败不阻断研讨：该席位无资料即可，研讨阶段它仍可自行检索。 + finalizeDialogue(bubbleId, `(取数失败) ${(err as Error).message}`, "取数失败"); + markBatchStatus(agentId, "done"); + parallelStreamRef.current.delete(bubbleId); + return null; + } + }), + ); + + setParallelBatch([]); + setSelectedParallelSeats([]); + resetParallelStreams(); + gatheredDeliveriesRef.current = results.filter( + (r): r is { name: string; content: string } => !!r, + ); + }, + [ + appendStreamingDialogue, + finalizeDialogue, + ingestParallelDelta, + resetParallelStreams, + markParallelSpeakers, + markSpeakerIdle, + resolveSeatThreadId, + handleArtifactToolPhase, + handleUpstreamFrame, + ingestSubagentTaskFrame, + setDialogueConfidence, + appendStepIfNew, + ], + ); + // 按当前模式分发到对应编排函数。init / 恢复 / 干预 / 澄清回复统一走这里。 const runActiveOrchestration = useCallback( - (seedMessage: string, ids: ThreadIdMap, coord: string) => { + async (seedMessage: string, ids: ThreadIdMap, coord: string) => { const mode = orchestrationModeRef.current; + // 研讨前并行取数（opt-in，每轮一次）：先把取数产物准备好，再分发到具体编排（研讨轮注入资料）。 + // gatherPhaseDoneRef 闸门保证澄清续跑 / 干预重入不重复取数；置位在 await 前，避免并发重入双跑。 + if (gatherFirstRef.current && !gatherPhaseDoneRef.current) { + gatherPhaseDoneRef.current = true; + try { + await runGatherPhase(ids); + } catch (err) { + console.warn("[roundtable] 研讨前并行取数阶段失败，跳过取数继续研讨:", err); + } + } // dag:**只在计划尚未跑过时**执行整张 DAG;一旦跑过(用户干预 / 恢复 / 澄清回复), // 改走 leader 自由派活循环,由总控接着派活,而不是把 DAG 从头重跑一遍。 if ( @@ -2534,7 +2704,7 @@ export function useStep2Orchestration( } return runOrchestration(seedMessage, ids, coord); }, - [runOrchestration, runChainOrchestration, runDagOrchestration], + [runGatherPhase, runOrchestration, runChainOrchestration, runDagOrchestration], ); // ─── Effects ───────────────────────────────────────────────────── @@ -2850,6 +3020,9 @@ export function useStep2Orchestration( step2InitRef.current = false; // 新一轮研讨:允许 DAG 计划重新执行一遍。 dagPlanExecutedRef.current = false; + // 新一轮研讨:允许「研讨前并行取数」再取一次（清掉上一轮取数产物 + 闸门）。 + gatherPhaseDoneRef.current = false; + gatheredDeliveriesRef.current = []; setInitNonce((n) => n + 1); }, []); @@ -2968,6 +3141,8 @@ export function useStep2Orchestration( hasAutoOpenedRef.current = new Set(); pendingInjectPromptRef.current = null; dagPlanExecutedRef.current = false; + gatherPhaseDoneRef.current = false; + gatheredDeliveriesRef.current = []; deselectArtifact(); step2InitRef.current = false; }, [abortAllInFlight, deselectArtifact, setArtifacts, resetParallelStreams]); diff --git a/frontend-web/src/roundtable-planning/hooks/useStep3Summary.ts b/frontend-web/src/roundtable-planning/hooks/useStep3Summary.ts index b4f9cb5b..1a55216d 100644 --- a/frontend-web/src/roundtable-planning/hooks/useStep3Summary.ts +++ b/frontend-web/src/roundtable-planning/hooks/useStep3Summary.ts @@ -315,6 +315,16 @@ function buildSummaryPrompt(opts: { } parts.push(""); parts.push(buildMaterials(opts)); + // 意图对齐 + 缺口核对 + 防幻觉护栏（弱席位场景的质量兜底）——两种模式通用，与后端 build_summary_prompt 同义。 + parts.push(""); + parts.push( + "【动笔前必做：意图对齐与缺口核对】各席位可能由能力较弱的模型产出，未必扣住用户意图、未必充分。" + + "你是这份总结的质量兜底，不能只把各席位交付简单拼接。请先把上面的「任务意图」（目标/约束/关键假设）拆成" + + "用户核心诉求清单（编排提示对最终产出的要求也并入），逐条核对各席位交付+共识是否覆盖、够不够、有无跑题或矛盾；" + + "再据此组织并按护栏补强：① 能从已有原料重组/推理补上的→直接补全讲透；② 属于你自己的推断/建议→显式标注「（综合推断）」；" + + "③ 缺真实信息、靠现有原料补不了的→绝不凭空编，用一行『> ⚠️ 待补充：<具体缺什么>。可在对话里让我补全』标出，交给用户。" + + "最终产出要明显比各席位交付的简单拼接更扣用户意图。", + ); if (opts.flowMode) { parts.push( "请严格遵循你的职责设定：全程简体中文。" + diff --git a/frontend-web/src/roundtable-planning/lib/business-chain.ts b/frontend-web/src/roundtable-planning/lib/business-chain.ts index dfd68c94..fc3918c8 100644 --- a/frontend-web/src/roundtable-planning/lib/business-chain.ts +++ b/frontend-web/src/roundtable-planning/lib/business-chain.ts @@ -36,6 +36,15 @@ export interface ChainSeat { * （无需 DB 列），在链条编辑器里逐席位开关。 */ seat_skill_directive?: boolean; + /** + * 可选「席位推理深度覆盖」：覆盖链条级 `seatMode`，只对该席位生效。值是同一套 + * `SeatMode`（flash/thinking/pro/ultra）；null / 缺省 = 跟随链条级默认。用于「卡顿提速」： + * 把信息收集/边角席位单独降到 `flash`（不思考、最快），只给关键席位留 `pro`/`ultra`。 + * 随 `seats` JSON 落库（无需 DB 列，但需在后端 `ChainSeatSchema` 显式声明，否则 model_dump + * 会剥掉）。**字段名用 snake_case `seat_mode`** 与 `model`/`seat_skill_directive` 一致（seats + * 数组在 `api/chains.ts` 不做 camel↔snake 转换，键名须与后端 schema 逐字匹配）。 + */ + seat_mode?: SeatMode | null; } /** 一条可复用的业务链条（与后端 `/api/roundtable-chains` 记录对应）。 */ @@ -68,6 +77,12 @@ export interface BusinessChain { * 缺省 = 未设，调用方回退 `DEFAULT_SEAT_MODE`(pro)。 */ seatMode?: SeatMode; + /** + * 可选「研讨前并行取数」开关（逐条 opt-in，缺省 false）。true = 选入研讨时，研讨前先跑一轮 + * **全席位并行『取数』**（各席位只调技能取数、不研讨），产出作为共享资料注入后续研讨轮，把最慢 + * 的技能延迟前置并行、给（尤其 chain 串行的）研讨段提速。三种编排通用。后端列 `gather_first`。 + */ + gatherFirst?: boolean; /** 是否已发布为公共（所有账号可在「公共」筛选查看 / 选入研讨）。 */ isPublic?: boolean; /** 是否当前用户所有（仅本人可编辑 / 删除 / 发布；公共筛选里别人的链条为 false）。 */ @@ -89,7 +104,7 @@ export interface SeatSelectionResult { * `model` 为该席位在来源链条里配置的模型覆盖（空 / 缺省 = 跟随全局模型）。 * `seat_skill_directive` 为该席位在来源链条里配置的「技能识别强化」开关（缺省 = 关）。 */ - agents: Array<{ agent_id: string; name: string; model?: string | null; seat_skill_directive?: boolean }>; + agents: Array<{ agent_id: string; name: string; model?: string | null; seat_skill_directive?: boolean; seatMode?: SeatMode | null }>; /** * 决定 Step 2 走哪种编排： * - `recommend` 总控自由派活； @@ -115,6 +130,11 @@ export interface SeatSelectionResult { * 缺省由调用方按 `DEFAULT_SEAT_MODE`（pro）处理。 */ seatMode?: SeatMode; + /** + * 「研讨前并行取数」开关（来源链条的 `gatherFirst`）。true = 研讨前先并行取数再研讨。 + * 三种编排通用，缺省 false。recommend（无链条）= false。 + */ + gatherFirst?: boolean; } /** 链条席位数硬约束，与圆桌推荐弹窗的 `MIN_SELECTED`/`MAX_SELECTED` 对齐。 */ @@ -146,6 +166,10 @@ export function buildChainSeatSelection( // 席位「技能识别强化」开关（链条配置）：按 agent_id 取，缺省 = 关。 const skillById = new Map(chain.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。 + // 输出到 SeatSelectionResult.agents[].seatMode（camel，喂给编排/作业的 seatModes map）。 + const seatModeById = new Map(chain.seats.map((s) => [s.agent_id, s.seat_mode ?? null])); + const seatModeOf = (id: string) => seatModeById.get(id) ?? null; const survivingSeats = chain.seats.filter((s) => existingIds.has(s.agent_id)); if (survivingSeats.length < CHAIN_MIN_SEATS) return null; @@ -167,13 +191,13 @@ export function buildChainSeatSelection( if (hasParallel || hasGoal || hasCoord) { const agents = plan.stages .flatMap((st) => st.agentIds) - .map((id) => ({ agent_id: id, name: nameOf(id), model: modelOf(id), seat_skill_directive: skillOf(id) })); - return { agents, mode: "dag", chain: chainMeta, plan, dagExecMode: "fast", seatMode: chain.seatMode }; + .map((id) => ({ agent_id: id, name: nameOf(id), model: modelOf(id), seat_skill_directive: skillOf(id), seatMode: seatModeOf(id) })); + return { agents, mode: "dag", chain: chainMeta, plan, dagExecMode: "fast", seatMode: chain.seatMode, gatherFirst: chain.gatherFirst ?? false }; } } - 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) })); - return { agents, mode: "chain", chain: chainMeta, seatMode: chain.seatMode }; + 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) })); + return { agents, mode: "chain", chain: chainMeta, seatMode: chain.seatMode, gatherFirst: chain.gatherFirst ?? false }; } /** diff --git a/frontend-web/src/roundtable-planning/lib/chain-stages.ts b/frontend-web/src/roundtable-planning/lib/chain-stages.ts index 0fc1c20f..b5a85676 100644 --- a/frontend-web/src/roundtable-planning/lib/chain-stages.ts +++ b/frontend-web/src/roundtable-planning/lib/chain-stages.ts @@ -13,6 +13,7 @@ * 对应需求文档 `docs/roundtable-dag-orchestration.md` §2（分层编辑器，竖列分组）。 */ import type { BusinessChain, ChainSeat } from "@/roundtable-planning/lib/business-chain"; +import type { SeatMode } from "@/roundtable-planning/lib/seat-mode"; /** 一个并行 stage：含其并行席位与可选「本阶段目标」（注入给总控/席位，仅 dag 模式生效）。 */ export interface StageGroup { @@ -140,6 +141,21 @@ export function setSeatSkillDirective(groups: StageGroups, agentId: string, enab }); } +/** + * 设置某个席位的「推理深度覆盖」（不可变，供编辑器席位下拉 onChange）。 + * `mode` 为 null = 清除覆盖（跟随链条级 seatMode）。按 agent_id 定位，跨 stage 适用。 + * 随 `seats` JSON 落库。 + */ +export function setSeatMode(groups: StageGroups, agentId: string, mode: SeatMode | null): StageGroups { + return groups.map((g) => { + if (!g.seats.some((s) => s.agent_id === agentId)) return g; + return { + ...g, + seats: g.seats.map((s) => (s.agent_id === agentId ? { ...s, seat_mode: mode } : s)), + }; + }); +} + // ── 变换（均返回新结构，不可变）────────────────────────────────────────────── /** 末尾追加一个**新单席位 stage**（线性追加，pool 点击加入默认行为）。 */ diff --git a/frontend-web/src/roundtable-planning/lib/job-types.ts b/frontend-web/src/roundtable-planning/lib/job-types.ts index 6c3a9f4e..9d5ee618 100644 --- a/frontend-web/src/roundtable-planning/lib/job-types.ts +++ b/frontend-web/src/roundtable-planning/lib/job-types.ts @@ -137,6 +137,16 @@ export interface StartJobPayload { * 后台席位 run 按 agentId 查表（与各席位智能体自身的同名开关取 OR）决定是否注入。缺省 {}。 */ seatSkillDirectives?: Record; + /** + * 每席位「推理深度覆盖」映射 {agentId: SeatMode | null}（沿用所选业务链条里每个席位单独配的 + * 推理深度）。后台席位 run 按 agentId 查表，覆盖优先、否则跟随链条级 `seatMode`。缺省 {}。 + */ + seatModes?: Record; + /** + * 「研讨前并行取数」开关（来源链条的 gatherFirst）。true = 后台作业研讨前先并行取数再研讨。 + * 在 `toStartBody` 里下发为后端 `gather_first`。缺省 false。 + */ + gatherFirst?: boolean; } /** 终态集合——SSE/轮询判定作业是否结束。 */ diff --git a/frontend-web/src/roundtable-planning/pages/BusinessChainEditorPage.tsx b/frontend-web/src/roundtable-planning/pages/BusinessChainEditorPage.tsx index 5278e721..192e4f07 100644 --- a/frontend-web/src/roundtable-planning/pages/BusinessChainEditorPage.tsx +++ b/frontend-web/src/roundtable-planning/pages/BusinessChainEditorPage.tsx @@ -27,6 +27,7 @@ import { Bot, ChevronDown, ChevronUp, + Gauge, GitBranch, Globe, GripVertical, @@ -67,6 +68,9 @@ import { cn } from "@/lib/utils"; import { DropdownMenu, DropdownMenuContent, + DropdownMenuItem, + DropdownMenuLabel, + DropdownMenuSeparator, DropdownMenuTrigger, } from "@/components/ui/dropdown-menu"; import { Switch } from "@/components/ui/switch"; @@ -111,6 +115,7 @@ import { moveStage, removeSeat as removeSeatFromGroups, seatExistsInGroups, + setSeatMode as setSeatModeOnGroups, setSeatModel, setSeatSkillDirective, setStageGoal, @@ -173,6 +178,8 @@ export default function BusinessChainEditorPage() { const [coordinatorPrompt, setCoordinatorPrompt] = useState(""); // 默认席位执行模式（快速/思考/专业问答/多智能体）—— 选入研讨时弹窗默认用它。 const [seatMode, setSeatMode] = useState(DEFAULT_SEAT_MODE); + // 链条级「研讨前并行取数」开关（逐条 opt-in，默认关）：开 = 选入研讨时研讨前先并行取数再研讨。 + const [gatherFirst, setGatherFirst] = useState(false); // 分层编排是唯一数据源；落库时展平为 seats + 折叠为 stages。 const [stageGroups, setStageGroups] = useState([]); const [saving, setSaving] = useState(false); @@ -298,6 +305,7 @@ export default function BusinessChainEditorPage() { setDescription(chain.description ?? ""); setCoordinatorPrompt(chain.coordinatorPrompt ?? ""); setSeatMode(chain.seatMode ?? DEFAULT_SEAT_MODE); + setGatherFirst(chain.gatherFirst ?? false); setStageGroups(chainToStageGroups(chain)); setStep(1); }, []); @@ -310,6 +318,7 @@ export default function BusinessChainEditorPage() { setDescription(""); setCoordinatorPrompt(""); setSeatMode(DEFAULT_SEAT_MODE); + setGatherFirst(false); setStageGroups([]); setStep(1); }, []); @@ -379,6 +388,11 @@ export default function BusinessChainEditorPage() { setStageGroups((prev) => setSeatSkillDirective(prev, agentId, enabled)); }, []); + // 席位「推理深度覆盖」：编辑器席位小图标下拉 onChange（null = 清除，跟随链条级 seatMode）。 + const setSeatModeCb = useCallback((agentId: string, mode: SeatMode | null) => { + setStageGroups((prev) => setSeatModeOnGroups(prev, agentId, mode)); + }, []); + // ── 提示词 AI 优化（总控提示词 / 各 Stage 阶段目标）──────────────────────── const [optimizingCoordinator, setOptimizingCoordinator] = useState(false); const [goalOptimizingIdx, setGoalOptimizingIdx] = useState(null); @@ -579,6 +593,8 @@ export default function BusinessChainEditorPage() { model: s.model ?? null, // 席位「会商技能识别强化」开关：随席位一起落库（缺省 = false 关）。 seat_skill_directive: s.seat_skill_directive ?? false, + // 席位「推理深度覆盖」：随席位一起落库（null = 跟随链条级 seatMode）。 + seat_mode: s.seat_mode ?? null, }; }); // 总控编排提示：trim 后空串落 null（清空），与后端「空=不注入」一致。 @@ -597,6 +613,7 @@ export default function BusinessChainEditorPage() { stageGoals, coordinatorPrompt: coordPrompt, seatMode, + gatherFirst, }); if (!updated) { toast.error("链条已不存在，可能被删除"); @@ -616,6 +633,7 @@ export default function BusinessChainEditorPage() { stageGoals, coordinatorPrompt: coordPrompt, seatMode, + gatherFirst, }); setCurrentChainId(created.id); toast.success("已创建业务链条"); @@ -631,6 +649,7 @@ export default function BusinessChainEditorPage() { description, coordinatorPrompt, seatMode, + gatherFirst, stageGroups, seatCountInvalid, currentChainId, @@ -769,6 +788,22 @@ export default function BusinessChainEditorPage() {

+ {/* 研讨前并行取数：逐条 opt-in 的提速开关（默认关）。 */} +
+
+ +

+ 开启后，选入研讨时会**先并行跑一轮「只取数」**：各席位同时用技能把相关信息收集回来（不研讨）， + 产出作为共享资料注入后续研讨轮。把最慢的技能检索前置、并行做一次，给（尤其按链条顺序串行的） + 研讨段大幅提速；研讨阶段席位仍可按需再调技能。 +

+
+ +
+ {/* 总控提示词：label + AI 优化（配色沿用现有 primary）+ 大号等宽编辑区 */}
@@ -1147,6 +1182,7 @@ export default function BusinessChainEditorPage() { onSplitSeat={splitSeat} onSeatModelChange={setSeatModelCb} onSeatSkillDirectiveChange={setSeatSkillDirectiveCb} + onSeatModeChange={setSeatModeCb} onStageGoalChange={setGoal} goalOptimizingIdx={goalOptimizingIdx} onOptimizeGoal={handleOptimizeStageGoal} @@ -1326,6 +1362,7 @@ function StagesColumn({ onSplitSeat, onSeatModelChange, onSeatSkillDirectiveChange, + onSeatModeChange, onStageGoalChange, goalOptimizingIdx, onOptimizeGoal, @@ -1343,6 +1380,7 @@ function StagesColumn({ onSplitSeat: (agentId: string) => void; onSeatModelChange: (agentId: string, model: string | null) => void; onSeatSkillDirectiveChange: (agentId: string, enabled: boolean) => void; + onSeatModeChange: (agentId: string, mode: SeatMode | null) => void; onStageGoalChange: (stageIdx: number, goal: string) => void; /** 正在 AI 优化阶段目标的 stage 下标（null = 没有进行中的优化）。 */ goalOptimizingIdx: number | null; @@ -1442,6 +1480,7 @@ function StagesColumn({ onSplitSeat={onSplitSeat} onSeatModelChange={onSeatModelChange} onSeatSkillDirectiveChange={onSeatSkillDirectiveChange} + onSeatModeChange={onSeatModeChange} onGoalChange={(goal) => onStageGoalChange(stageIdx, goal)} goalOptimizing={goalOptimizingIdx === stageIdx} anyGoalOptimizing={goalOptimizingIdx !== null} @@ -1472,6 +1511,7 @@ function StageSection({ onSplitSeat, onSeatModelChange, onSeatSkillDirectiveChange, + onSeatModeChange, onGoalChange, goalOptimizing, anyGoalOptimizing, @@ -1491,6 +1531,7 @@ function StageSection({ onSplitSeat: (agentId: string) => void; onSeatModelChange: (agentId: string, model: string | null) => void; onSeatSkillDirectiveChange: (agentId: string, enabled: boolean) => void; + onSeatModeChange: (agentId: string, mode: SeatMode | null) => void; onGoalChange: (goal: string) => void; /** 本 Stage 的阶段目标正在 AI 优化中。 */ goalOptimizing: boolean; @@ -1653,6 +1694,7 @@ function StageSection({ onSplit={() => onSplitSeat(seat.agent_id)} onModelChange={(model) => onSeatModelChange(seat.agent_id, model)} onSkillDirectiveChange={(enabled) => onSeatSkillDirectiveChange(seat.agent_id, enabled)} + onSeatModeChange={(mode) => onSeatModeChange(seat.agent_id, mode)} onDetail={() => onSeatDetail(seat.agent_id, seat.name)} /> ))} @@ -1677,6 +1719,7 @@ function SeatStrip({ onSplit, onModelChange, onSkillDirectiveChange, + onSeatModeChange, onDetail, }: { seat: ChainSeat; @@ -1693,6 +1736,8 @@ function SeatStrip({ onModelChange: (model: string | null) => void; /** 席位「会商技能识别强化」开关变更。 */ onSkillDirectiveChange: (enabled: boolean) => void; + /** 席位「推理深度覆盖」变更（null = 跟随链条级 seatMode）。 */ + onSeatModeChange: (mode: SeatMode | null) => void; onDetail: () => void; }) { const { attributes, listeners, setNodeRef, transform, transition, isDragging } = @@ -1764,8 +1809,52 @@ function SeatStrip({
- {/* 控制：技能识别强化 / 查看详情 / 并入上一阶段 / 独立成阶段 / 并入下一阶段 / 移除 */} + {/* 控制：推理深度 / 技能识别强化 / 查看详情 / 并入上一阶段 / 独立成阶段 / 并入下一阶段 / 移除 */}
+ {/* 席位推理深度覆盖：小图标 → 下拉框（跟随链条默认 + 四档）。设了覆盖 = 高亮。 + 「卡顿提速」用：把信息收集/边角席位降到「快速」，只给关键席位留「专业/多智能体」。 */} + + + + + + + 席位推理深度 + + 覆盖链条级默认，仅本席位生效。模型卡顿时把次要席位降到「快速」可大幅提速。 + + + + onSeatModeChange(null)} + > + 跟随链条默认 + + {SEAT_MODE_OPTIONS.map((opt) => ( + onSeatModeChange(opt.value)} + > + {opt.label} + {opt.description} + + ))} + + {/* 会商技能识别强化：小图标 → 下拉框（开关 + 说明）。开启 = 高亮。 */} diff --git a/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx b/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx index a82cd0c6..4813c6ce 100644 --- a/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx +++ b/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx @@ -315,6 +315,13 @@ export default function RoundtablePlanningPage() { // Step 2 每席位「技能识别强化」开关映射 {agentId: bool}：由所选业务链条里**每个席位**的同名 // 配置决定（recommend 无链条 → 空 {}）。透传给每个席位 run，与各席位智能体自身的开关取 OR。 const [seatSkillDirectives, setSeatSkillDirectives] = useState>({}); + // Step 2 每席位「推理深度覆盖」映射 {agentId: SeatMode|null}：由所选业务链条里**每个席位**单独配 + // 的推理深度决定（null/未配 → 跟随链条级 seatMode）。透传给席位 run（覆盖优先、否则跟随 seatMode）。 + // 「卡顿提速」用：把次要席位降到 flash 即可大幅缩短第二步耗时。 + const [seatModes, setSeatModes] = useState>({}); + // Step 2「研讨前并行取数」开关：由所选业务链条的 gatherFirst 决定（recommend / 无链条 → false）。 + // true = 研讨前先并行跑一轮全席位『取数』，产出作为共享资料注入研讨轮，给（尤其 chain 串行）研讨段提速。 + const [gatherFirst, setGatherFirst] = useState(false); // Step 2 编排模式：recommend = 总控自由派活；chain = 总控按业务链条顺序派活； // dag = 分层 DAG（stage 内并行、stage 间串行，配合 orchestrationPlan）。 // 由推荐弹窗确认时的 SeatSelectionResult.mode 决定（见 handleConfirmAgents）。 @@ -471,6 +478,8 @@ export default function RoundtablePlanningPage() { selectedModel, selectedMode: seatMode, seatSkillDirectives, + seatModes, + gatherFirst, orchestrationMode, orchestrationPlan, dagExecMode, @@ -953,6 +962,13 @@ export default function RoundtablePlanningPage() { setSeatSkillDirectives( Object.fromEntries(result.agents.map((a) => [a.agent_id, a.seat_skill_directive ?? false])), ); + // 每席位推理深度覆盖：取所选业务链条里**每席位**单独配的 seatMode → {agentId: SeatMode|null} + // 映射（recommend / 未配 → null，跟随链条级 seatMode）。 + setSeatModes( + Object.fromEntries(result.agents.map((a) => [a.agent_id, a.seatMode ?? null])), + ); + // 研讨前并行取数：取所选业务链条的开关（recommend / 旧调用未带 → false）。 + setGatherFirst(result.gatherFirst ?? false); setChainMeta(result.mode === "recommend" ? null : result.chain ?? null); setOrchestrationPlan(result.mode === "dag" ? result.plan ?? null : null); setDagExecMode(result.mode === "dag" ? result.dagExecMode ?? "fast" : "fast"); @@ -1528,6 +1544,8 @@ export default function RoundtablePlanningPage() { model: selectedModel || undefined, // 模型沿用第二步选择 seatMode, // 席位执行模式沿用第二步选择（后台续跑同样按此模式跑席位） seatSkillDirectives, // 每席位技能识别强化沿用所选业务链条里各席位的开关 + seatModes, // 每席位推理深度覆盖沿用所选业务链条里各席位的配置 + gatherFirst, // 研讨前并行取数沿用所选业务链条的开关 // 后台作业引擎支持 dag（后端 _run_dag：stage 串行 / stage 内并行）—— // 直接把当前编排模式透传，业务链的分层 / 并行 / 阶段目标 / 总控提示词全部保留。 orchestrationMode, @@ -1545,6 +1563,8 @@ export default function RoundtablePlanningPage() { selectedModel, seatMode, seatSkillDirectives, + seatModes, + gatherFirst, orchestrationMode, orchestrationPlan, chainMeta, @@ -1569,6 +1589,8 @@ export default function RoundtablePlanningPage() { const model = selectedModel || undefined; const seatModeSnap = seatMode; const seatSkillDirectivesSnap = seatSkillDirectives; + const seatModesSnap = seatModes; + const gatherFirstSnap = gatherFirst; const mode = orchestrationMode; const chainSnap = chainMeta; const planSnap = mode === "dag" ? orchestrationPlan : null; @@ -1604,6 +1626,8 @@ export default function RoundtablePlanningPage() { model, seatMode: seatModeSnap, seatSkillDirectives: seatSkillDirectivesSnap, + seatModes: seatModesSnap, + gatherFirst: gatherFirstSnap, orchestrationMode: mode, chain: chainSnap, orchestrationPlan: planSnap, @@ -1628,6 +1652,8 @@ export default function RoundtablePlanningPage() { selectedModel, seatMode, seatSkillDirectives, + seatModes, + gatherFirst, orchestrationMode, orchestrationPlan, chainMeta, @@ -1669,6 +1695,8 @@ export default function RoundtablePlanningPage() { model: selectedModel || undefined, seatMode, seatSkillDirectives, + seatModes, + gatherFirst, orchestrationMode, chain: chainMeta, orchestrationPlan: orchestrationMode === "dag" ? orchestrationPlan : null, diff --git a/offline-backend-20260512/backend/CLAUDE.md b/offline-backend-20260512/backend/CLAUDE.md index 53a399e4..30b9d866 100644 --- a/offline-backend-20260512/backend/CLAUDE.md +++ b/offline-backend-20260512/backend/CLAUDE.md @@ -247,7 +247,7 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA | **Public Shares** (`/api/public/shares`) | **Unauthenticated** (the `/api/public/` prefix is whitelisted in `auth_middleware._PUBLIC_PATH_PREFIXES`). `GET /{code}` - read-only conversation snapshot (title + serialized messages + artifacts) for a `view`-mode share; messages are read straight from the latest checkpoint. `GET /{code}/files/{path}` - serves a file referenced by that conversation (images, Markdown, text, PDF, ?) so the shared page's links are openable; path resolution is confined to the owner's thread directory. Active content (`text/html`, `application/xhtml+xml`, `image/svg+xml`) is forced to download (`_public_share_disposition`); everything else is served inline. Missing/revoked/non-`view` codes all return 404. | | **Roundtable Drafts** (`/api/roundtable-drafts`) | **Strictly per-user** persistence for the roundtable-planning page (replaces the old browser `localStorage`). `GET /` - list drafts (lightweight meta); `POST /` - create; `GET/PUT/DELETE /{id}` - full read / partial update / delete (all owner-scoped). A draft = one full session (Step 1 intent + Step 2 roundtable), step snapshots stored as JSON in `PortableLongText`. Recommendation history bound to a draft: `GET /{id}/recommendations` - list newest-first; `POST /{id}/recommendations` - append one run (objective/status/rationale/picks/candidates). taskId 深链会商记录 are **not** here — they live in the separate `/api/roundtable-task-drafts` store (see below) so personal history never mixes with task records. Backed by `deerflow.persistence.roundtable_drafts` (tables `roundtable_drafts`, `roundtable_recommend_history`); store wired in `deps.py` as `app.state.roundtable_draft_store`. Tests: `tests/test_roundtable_task_drafts.py`. | | **Roundtable Task Drafts** (`/api/roundtable-task-drafts`) | **Task-scoped, unpartitioned** store for taskId-deeplink roundtable 聊天记录 (无界嵌入抽屉). Keyed **only by `task_id`** (one task → many records), **never partitioned by user** — any authenticated caller may read/write/delete a task's records (`created_by` is audit-only). Physically separate from `roundtable_drafts` so the two never mix. `GET /by-task/{task_id}` - all drafts for a task, newest-first (history dropdown), enriched with the latest background-job status per draft via `RoundtableJobRepository.status_map_by_drafts`; `GET /by-task/{task_id}/latest` - newest draft (回显最新一条); `POST /` - create (`task_id` required); `GET/PUT/DELETE /{id}` - full read / partial update / delete (all user-agnostic). Backed by `deerflow.persistence.roundtable_task_drafts` (table `roundtable_task_drafts`, auto-created by `create_all`; migration `20260618_01`); store wired in `deps.py` as `app.state.roundtable_task_draft_store`. Existing `task_id`-bearing `roundtable_drafts` rows are migrated over by `scripts/migrate_roundtable_task_drafts.py`. Tests: `tests/test_roundtable_task_drafts.py`, `tests/test_roundtable_drafts_by_task.py`. | -| **Roundtable Diagnostics** (`/api/roundtable-diagnostics`) | 会商全流程诊断日志 (报错 + 关键事件). One row per event in `roundtable_diagnostics`: `{scope (background = 后台挂起 / foreground = 前台服务端观测 / client = 前端上报), stage (step1_intent/step2_init/step2_leader/step2_seat/step2_recommend/step2_orchestration/step3_summary/step3_report/step3_dashboard/step3_structure/upstream/job), level (info/warning/error), event, message, detail (JSON), job_id/draft_id/task_id/user_id/agent_id/agent_name/cycle, created_at}`. **Reads are admin-only**: `GET /` - filter (scope/stage/level/event/job_id/draft_id/task_id/user_id/q/since/until; `scope` accepts a comma-separated set so 前端「前台调用」= `foreground,client`) + paginate (newest-first), each item enriched with `userName` (user_id → 邮箱 `@`-prefix via `get_local_provider().get_user`, best-effort batched cache); `GET /facets` - distinct values for filter dropdowns; `POST /admin/cleanup` - retention purge. **`POST /report` is NOT admin-gated** — any authenticated user's frontend posts the errors *the user actually hit* (HTTP 502 before any SSE frame / network drop / SSE error frame) so they're collected too; stored as `scope="client"`. Backed by `deerflow.persistence.roundtable_diagnostics` (table `roundtable_diagnostics`, auto-created by `create_all`; migration `20260619_01`); store wired in `deps.py` as `app.state.roundtable_diagnostic_store` + a process-level default store (`register_default_store`) so the harness-side engine/gateway can write across the app/harness boundary. **Writers** (all best-effort — diagnostics never break a run): background path uses `DiagnosticsRecorder` (bound to job ctx) threaded through `RoundtableJobExecutor` → `JobProgressWriter.diag(...)` (engine transitions/step3 failures/empty-delivery) + `InProcessRoundtableGateway` (per-turn timeout/report·summary·dashboard truncation+missing); foreground path uses `app/gateway/roundtable_diag.py::record_foreground_diag` in `multi_agent.py`/`intent.py`/`recommend.py` error frames. **Bug-fix shipped alongside**: `InProcessRoundtableGateway._run_agent_turn` guards each agent run with an **inactivity (idle) timeout** (`_await_run_with_idle_guard`, `ROUNDTABLE_SEAT_IDLE_TIMEOUT_SECONDS` default 1200s = 20min, tuned high for unstable intranet models) — **not** a total-duration cap, so a slow-but-alive run that keeps streaming tokens runs as long as it needs; only a run that emits **zero StreamBridge events for `idle_timeout` seconds straight** (a true freeze: 内网模型连接池挂死 / sqlite 锁在挂载卷上挂死) is cancelled + raises (顶层把 job 标 error，而非永久冻结) + records a `seat_turn_timeout` diagnostic. Activity is observed via `app.state.stream_bridge` (the worker publishes every chunk there). Optional absolute cap `ROUNDTABLE_SEAT_MAX_TOTAL_SECONDS` (default 0 = off). Frontend: admin page `pages/RoundtableDiagnosticsPage.tsx` (route `/page/strategy/admin/roundtable-diagnostics`), opened from the **会商页右上角设置按钮** (admin only — `handleSettingsClick` in `roundtable-planning/pages/RoundtablePlanningPage.tsx`); admin read client `strategy-components/api/roundtable-diagnostics.ts`. **Client error reporter** `roundtable-planning/api/diagnostics-report.ts` (`reportRoundtableError` fire-and-forget + `keepalive`, `setRoundtableErrorContext` for draft/task attribution set by the page, `stageForRoundtableAgent`) is called at every client-observable failure: `multi-agent.ts` (`streamMultiAgent` — covers Step2 leader/seat AND Step3 report/summary/dashboard/structure — reports HTTP error + SSE `{status:"error"}` frame + worker `event:error` frame [模型调用没成功] + fetch-reject [网络不可达] + mid-stream drop [连接重置/超时], deduped via a `reported` flag; `initMultiAgent` HTTP/error-frame/incomplete/mid-stream), `intent.ts` (Step1 HTTP+error-frame), `recommend.ts` (Step2 recommend HTTP+error-frame), `roundtable-jobs.ts` (job api/stream). The admin page's first filter is a **后台挂起 / 前台调用** toggle (frontend = `foreground,client`); the stage dropdown options adapt to it (background → step2+step3+job; frontend → step1+step2+step3) and each row shows the reporting user's name + 👤. Tests: `tests/test_roundtable_diagnostics.py`, `tests/test_roundtable_inprocess_timeout.py`. **Seat skill-recognition (弱模型补偿)**: a seat = one custom agent run whose configured skills (`agent config.yaml → skills`) are injected **only** via the system prompt's `` block (model must `read_file` the SKILL.md itself) — weak models (deepseek-chat 等) ignore it and never use their skills. `app/gateway/roundtable_seat_skills.py::append_seat_skill_directive(agent_id, task, *, chain_enabled=False)` appends an explicit Chinese directive (each configured-and-enabled skill's name + description + SKILL.md container path + "先 read_file 再按其流程取信息/执行") to the **seat's per-turn task (human turn)** — far stronger pull on weak models than the system prompt. Wired into **both** seat paths: foreground `multi_agent._special_run` (only `role in {seat, seat_parallel}`; Step3 singletons report/summary/dashboard/ingest excluded) + background `InProcessRoundtableGateway.run_seat`. No-op (returns task unchanged) when the seat has no configured skills. **2026-06-20: default-off + two opt-in switches (OR semantics)** — was global-on, now injects only when **either** the seat agent's own switch (`AgentConfig.seat_skill_directive`, edited in the agent edit dialog `agent-card.tsx` + `agent-detail-panel.tsx`) **or** the **per-seat** switch on that agent **within the selected business chain** is on. The per-seat chain switch lives **in the chain's `seats` JSON** (`ChainSeatSchema.seat_skill_directive`, no DB column — rides the existing JSON like the per-seat `model` override), edited via a small ✨ icon → dropdown (Switch + description) on each agent card in `BusinessChainEditorPage`. It is threaded **per-seat** to each seat run: foreground `multi_agent` per-call field `seat_skill_directive` (orchestration looks it up per agentId from a `{agentId: bool}` session map) → `_special_run` → `chain_enabled`; background as the job field `seat_skill_directives: dict[str,bool]` (StartJobRequest → StartParams → deps factory → `InProcessRoundtableGateway._seat_skill_directives`) → `run_seat` looks up `agent_id` → `chain_enabled`. `ROUNDTABLE_SEAT_SKILL_DIRECTIVE` is only a **master kill-switch** (default allow; set `0/false` to hard-disable regardless of the switches — never force-enables). Tests: `tests/test_roundtable_seat_skills.py`, `tests/test_agents_backend.py`. | +| **Roundtable Diagnostics** (`/api/roundtable-diagnostics`) | 会商全流程诊断日志 (报错 + 关键事件). One row per event in `roundtable_diagnostics`: `{scope (background = 后台挂起 / foreground = 前台服务端观测 / client = 前端上报), stage (step1_intent/step2_init/step2_leader/step2_seat/step2_recommend/step2_orchestration/step3_summary/step3_report/step3_dashboard/step3_structure/upstream/job), level (info/warning/error), event, message, detail (JSON), job_id/draft_id/task_id/user_id/agent_id/agent_name/cycle, created_at}`. **Reads are admin-only**: `GET /` - filter (scope/stage/level/event/job_id/draft_id/task_id/user_id/q/since/until; `scope` accepts a comma-separated set so 前端「前台调用」= `foreground,client`) + paginate (newest-first), each item enriched with `userName` (user_id → 邮箱 `@`-prefix via `get_local_provider().get_user`, best-effort batched cache); `GET /facets` - distinct values for filter dropdowns; `POST /admin/cleanup` - retention purge. **`POST /report` is NOT admin-gated** — any authenticated user's frontend posts the errors *the user actually hit* (HTTP 502 before any SSE frame / network drop / SSE error frame) so they're collected too; stored as `scope="client"`. Backed by `deerflow.persistence.roundtable_diagnostics` (table `roundtable_diagnostics`, auto-created by `create_all`; migration `20260619_01`); store wired in `deps.py` as `app.state.roundtable_diagnostic_store` + a process-level default store (`register_default_store`) so the harness-side engine/gateway can write across the app/harness boundary. **Writers** (all best-effort — diagnostics never break a run): background path uses `DiagnosticsRecorder` (bound to job ctx) threaded through `RoundtableJobExecutor` → `JobProgressWriter.diag(...)` (engine transitions/step3 failures/empty-delivery) + `InProcessRoundtableGateway` (per-turn timeout/report·summary·dashboard truncation+missing); foreground path uses `app/gateway/roundtable_diag.py::record_foreground_diag` in `multi_agent.py`/`intent.py`/`recommend.py` error frames. **Bug-fix shipped alongside**: `InProcessRoundtableGateway._run_agent_turn` guards each agent run with an **inactivity (idle) timeout** (`_await_run_with_idle_guard`, `ROUNDTABLE_SEAT_IDLE_TIMEOUT_SECONDS` default 1200s = 20min, tuned high for unstable intranet models) — **not** a total-duration cap, so a slow-but-alive run that keeps streaming tokens runs as long as it needs; only a run that emits **zero StreamBridge events for `idle_timeout` seconds straight** (a true freeze: 内网模型连接池挂死 / sqlite 锁在挂载卷上挂死) is cancelled + raises (顶层把 job 标 error，而非永久冻结) + records a `seat_turn_timeout` diagnostic. Activity is observed via `app.state.stream_bridge` (the worker publishes every chunk there). Optional absolute cap `ROUNDTABLE_SEAT_MAX_TOTAL_SECONDS` (default 0 = off). Frontend: admin page `pages/RoundtableDiagnosticsPage.tsx` (route `/page/strategy/admin/roundtable-diagnostics`), opened from the **会商页右上角设置按钮** (admin only — `handleSettingsClick` in `roundtable-planning/pages/RoundtablePlanningPage.tsx`); admin read client `strategy-components/api/roundtable-diagnostics.ts`. **Client error reporter** `roundtable-planning/api/diagnostics-report.ts` (`reportRoundtableError` fire-and-forget + `keepalive`, `setRoundtableErrorContext` for draft/task attribution set by the page, `stageForRoundtableAgent`) is called at every client-observable failure: `multi-agent.ts` (`streamMultiAgent` — covers Step2 leader/seat AND Step3 report/summary/dashboard/structure — reports HTTP error + SSE `{status:"error"}` frame + worker `event:error` frame [模型调用没成功] + fetch-reject [网络不可达] + mid-stream drop [连接重置/超时], deduped via a `reported` flag; `initMultiAgent` HTTP/error-frame/incomplete/mid-stream), `intent.ts` (Step1 HTTP+error-frame), `recommend.ts` (Step2 recommend HTTP+error-frame), `roundtable-jobs.ts` (job api/stream). The admin page's first filter is a **后台挂起 / 前台调用** toggle (frontend = `foreground,client`); the stage dropdown options adapt to it (background → step2+step3+job; frontend → step1+step2+step3) and each row shows the reporting user's name + 👤. Tests: `tests/test_roundtable_diagnostics.py`, `tests/test_roundtable_inprocess_timeout.py`. **Seat skill-recognition (弱模型补偿)**: a seat = one custom agent run whose configured skills (`agent config.yaml → skills`) are injected **only** via the system prompt's `` block (model must `read_file` the SKILL.md itself) — weak models (deepseek-chat 等) ignore it and never use their skills. `app/gateway/roundtable_seat_skills.py::append_seat_skill_directive(agent_id, task, *, chain_enabled=False)` appends an explicit Chinese directive (each configured-and-enabled skill's name + description + SKILL.md container path + "先 read_file 再按其流程取信息/执行") to the **seat's per-turn task (human turn)** — far stronger pull on weak models than the system prompt. Wired into **both** seat paths: foreground `multi_agent._special_run` (only `role in {seat, seat_parallel}`; Step3 singletons report/summary/dashboard/ingest excluded) + background `InProcessRoundtableGateway.run_seat`. No-op (returns task unchanged) when the seat has no configured skills. **2026-06-20: default-off + two opt-in switches (OR semantics)** — was global-on, now injects only when **either** the seat agent's own switch (`AgentConfig.seat_skill_directive`, edited in the agent edit dialog `agent-card.tsx` + `agent-detail-panel.tsx`) **or** the **per-seat** switch on that agent **within the selected business chain** is on. The per-seat chain switch lives **in the chain's `seats` JSON** (`ChainSeatSchema.seat_skill_directive`, no DB column — rides the existing JSON like the per-seat `model` override), edited via a small ✨ icon → dropdown (Switch + description) on each agent card in `BusinessChainEditorPage`. It is threaded **per-seat** to each seat run: foreground `multi_agent` per-call field `seat_skill_directive` (orchestration looks it up per agentId from a `{agentId: bool}` session map) → `_special_run` → `chain_enabled`; background as the job field `seat_skill_directives: dict[str,bool]` (StartJobRequest → StartParams → deps factory → `InProcessRoundtableGateway._seat_skill_directives`) → `run_seat` looks up `agent_id` → `chain_enabled`. `ROUNDTABLE_SEAT_SKILL_DIRECTIVE` is only a **master kill-switch** (default allow; set `0/false` to hard-disable regardless of the switches — never force-enables). Tests: `tests/test_roundtable_seat_skills.py`, `tests/test_agents_backend.py`. **2026-06-20 Step2 提速三件套**：(1) **每席位推理深度覆盖** —— 业务链条 `seats` JSON 里逐席位可配 `seat_mode`（flash/thinking/pro/ultra，覆盖链条级 `seatMode`；编辑器席位卡右侧 ⏱Gauge 图标下拉，null=跟随链条默认）。前台经 `useStep2Orchestration` 三处 dispatch 的 `seatModeToRequestFields(seatModes[id] ?? selectedMode)` 透传；后台经 job `seat_modes: dict[str,str]` → `StartParams` → `InProcessRoundtableGateway._seat_modes`，`run_seat` 用 `_SEAT_MODE_PARAMS`（与前端 `seatModeToRunParams` 逐字对齐）解析覆盖作业级 seat_*（未命中→跟随作业级→policy 默认）。无 DB 列（ride `seats` JSON，但 `ChainSeatSchema.seat_mode` 须显式声明否则 Pydantic 剥掉）。Tests: `tests/test_roundtable_seat_modes.py`。(2) **recommend 批内并行** —— 后端 `engine._run_recommend` 由 for-await 串行改 `asyncio.gather`（对齐前端 2026-06-13 已并行；总控一轮派多席位=判定相互独立→并发；批内用派活前 deliveries 快照、批后统一并入，语义不变）。(3) **研讨前并行『取数』两段式**（业务链条 `gather_first` 列，逐条 opt-in，migration `20260620_02`，editor「研讨前并行取数」开关）—— `gather_first=True` 时 `run_orchestration` 在分发前先 `engine._run_gather_phase`（全席位 `asyncio.gather` 跑 `_GATHER_TASK`「只取数不研讨」），产出 `gathered: dict` 经 `_seat_message(gathered=...)` 作为「已收集资料」块注入 recommend/chain/dag 各研讨轮席位任务（与「前序交付」块分开），把最慢的技能延迟前置并行、给（尤其 chain 串行的）研讨段提速；研讨阶段不限制再调技能。前台对称实现：`useStep2Orchestration.runGatherPhase` 在 `runActiveOrchestration` 分发前跑一次（`gatherPhaseDoneRef` 闸门，resetInitGate 复位），产出存 `gatheredDeliveriesRef` 经 `peerDeliveries`/初始 `priorDeliveries` 注入研讨席位。链条级 `gatherFirst`(camel)↔`gather_first`(snake) 经 `api/chains.ts` 映射；job 经 `StartJobRequest.gather_first`→`StartParams`→`run_orchestration(gather_first=…)`。Tests: `tests/test_roundtable_gather_phase.py`。(4) **削减重复广播**（前端 only，多智能体 foreground 路径）—— `suppress_pre_broadcast`（仅 leader：抑制「总控本轮输入→各席位 thread」预广播 + 「总控给X派活」→兄弟席位的 post 广播，二者一个 flag）此前只在 DAG 派活轮开；现扩到 recommend（`safetyCounter > 1`，即第 2 轮起的续轮/综合）+ chain 逐席位（`i > 0`，第 2 个席位起）+ chain 收口轮（恒 true）。**保留每种模式的首轮预广播**（它为席位打底任务背景），只 suppress 重复的续轮广播（N 次读写巨大席位 thread 的纯浪费 + 弱模型模仿派活）；席位的前序交付另经「席位交付广播」`broadcastSeatDeliveryToOthers` 送达，不受影响。后台引擎路径无此广播（`_seat_message` 直接拼上下文），故 #4 仅前端。 | | **Roundtable Draft Shares** (`/api/roundtable-draft-shares`) | Per-draft sharing for the roundtable-planning page, mirroring Thread Shares but the shared resource is a `roundtable_drafts` row. Two modes (`roundtable_draft_shares.mode`). **`import`**: `POST /` `{draft_id, mode}` - create (or reuse) a revocable code; `GET /{code}/preview` + `POST /{code}/import` - copy that draft (step1/2/3) into the caller's account under a deterministic per-(code,recipient) id (`imp_` ? re-import is a no-op). **`view`**: a public, read-only result page ? see Public Roundtable Shares. Both: `GET /` - list my codes; `DELETE /{code}` - revoke. Backed by `deerflow.persistence.roundtable_draft_shares` (table `roundtable_draft_shares`, migration `20260606_01`); store wired in `deps.py` as `app.state.roundtable_draft_share_store`. Tests: `tests/test_roundtable_draft_shares.py`. | | **Public Roundtable Shares** (`/api/public/roundtable-shares`) | **Unauthenticated** (`/api/public/` whitelisted). `GET /{code}` - read-only snapshot for a `view`-mode draft share: returns the Step 3??????report HTML (`step3.html`, a self-contained document) + title/summary/`has_report`. The frontend renders it in a sandboxed `iframe` (no `allow-scripts`) at `#/roundtable-share/`. 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