feat(roundtable): Step2 提速四件套 + 方案总结报告意图对齐
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) <noreply@anthropic.com>
This commit is contained in:
1 parent
9a1b674950
commit
3ab674e438
25 files changed
+985
-44
No files matched your search
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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 跳过)。
|
||||
|
||||
@@ -285,6 +285,20 @@ export interface UseStep2OrchestrationOptions {
|
||||
* OR。recommend / 未选链条 → 空 {}(各席位仅靠其自身开关)。
|
||||
*/
|
||||
seatSkillDirectives?: Record<string, boolean>;
|
||||
/**
|
||||
* 每席位「推理深度覆盖」映射 {agentId: SeatMode | null}(沿用所选业务链条里每个席位的配置)。
|
||||
* 派活该 special 席位时按 agentId 查表,**席位覆盖优先、否则跟随链条级 `selectedMode`**:
|
||||
* `seatModeToRequestFields(seatModes[agentId] ?? selectedMode)`。模型卡顿时把次要席位降到
|
||||
* `flash` 即可大幅提速。recommend / 未选链条 / 未覆盖 → 跟随 `selectedMode`。
|
||||
*/
|
||||
seatModes?: Record<string, SeatMode | null>;
|
||||
/**
|
||||
* 「研讨前并行取数」开关(来源业务链条的 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<SeatMode>(selectedMode);
|
||||
// 每席位技能识别强化映射(所选业务链条各席位开关)同样打到 ref,席位调用按 agentId 读最新值。
|
||||
const seatSkillDirectivesRef = useRef<Record<string, boolean>>(seatSkillDirectives);
|
||||
// 每席位推理深度覆盖映射(所选业务链条各席位 seatMode)打到 ref,席位调用按 agentId 读最新值;
|
||||
// 覆盖优先、否则跟随 selectedModeRef。
|
||||
const seatModesRef = useRef<Record<string, SeatMode | null>>(seatModes);
|
||||
// 「研讨前并行取数」开关 + 本轮取数产物({name,content}[])+ 本轮是否已取过的闸门。取数产物作为
|
||||
// peerDeliveries 注入研讨轮席位;闸门保证一轮研讨只取一次(resetInitGate / resetStep2State 复位)。
|
||||
const gatherFirstRef = useRef<boolean>(gatherFirst);
|
||||
const gatheredDeliveriesRef = useRef<Array<{ name: string; content: string }>>([]);
|
||||
const gatherPhaseDoneRef = useRef<boolean>(false);
|
||||
const agentConfigsRef = useRef<Record<number, AgentConfig>>(agentConfigs);
|
||||
const agentNameToRoleRef = useRef(agentNameToRole);
|
||||
const threadIdsRef = useRef<ThreadIdMap | null>(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<void> => {
|
||||
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]);
|
||||
|
||||
@@ -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(
|
||||
"请严格遵循你的职责设定:全程简体中文。" +
|
||||
|
||||
@@ -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 };
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -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 点击加入默认行为)。 */
|
||||
|
||||
@@ -137,6 +137,16 @@ export interface StartJobPayload {
|
||||
* 后台席位 run 按 agentId 查表(与各席位智能体自身的同名开关取 OR)决定是否注入。缺省 {}。
|
||||
*/
|
||||
seatSkillDirectives?: Record<string, boolean>;
|
||||
/**
|
||||
* 每席位「推理深度覆盖」映射 {agentId: SeatMode | null}(沿用所选业务链条里每个席位单独配的
|
||||
* 推理深度)。后台席位 run 按 agentId 查表,覆盖优先、否则跟随链条级 `seatMode`。缺省 {}。
|
||||
*/
|
||||
seatModes?: Record<string, SeatMode | null>;
|
||||
/**
|
||||
* 「研讨前并行取数」开关(来源链条的 gatherFirst)。true = 后台作业研讨前先并行取数再研讨。
|
||||
* 在 `toStartBody` 里下发为后端 `gather_first`。缺省 false。
|
||||
*/
|
||||
gatherFirst?: boolean;
|
||||
}
|
||||
|
||||
/** 终态集合——SSE/轮询判定作业是否结束。 */
|
||||
|
||||
@@ -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<SeatMode>(DEFAULT_SEAT_MODE);
|
||||
// 链条级「研讨前并行取数」开关(逐条 opt-in,默认关):开 = 选入研讨时研讨前先并行取数再研讨。
|
||||
const [gatherFirst, setGatherFirst] = useState<boolean>(false);
|
||||
// 分层编排是唯一数据源;落库时展平为 seats + 折叠为 stages。
|
||||
const [stageGroups, setStageGroups] = useState<StageGroups>([]);
|
||||
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<number | null>(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() {
|
||||
</p>
|
||||
</div>
|
||||
|
||||
{/* 研讨前并行取数:逐条 opt-in 的提速开关(默认关)。 */}
|
||||
<div className={cn("flex items-start justify-between gap-3 border border-border bg-muted/30 p-3", R_CARD)}>
|
||||
<div className="space-y-1">
|
||||
<label className="flex items-center gap-1 text-xs font-semibold text-foreground">
|
||||
<Gauge className="size-3.5 text-primary" />
|
||||
研讨前并行取数(提速)
|
||||
</label>
|
||||
<p className="text-[11px] leading-relaxed text-muted-foreground">
|
||||
开启后,选入研讨时会**先并行跑一轮「只取数」**:各席位同时用技能把相关信息收集回来(不研讨),
|
||||
产出作为共享资料注入后续研讨轮。把最慢的技能检索前置、并行做一次,给(尤其按链条顺序串行的)
|
||||
研讨段大幅提速;研讨阶段席位仍可按需再调技能。
|
||||
</p>
|
||||
</div>
|
||||
<Switch checked={gatherFirst} onCheckedChange={setGatherFirst} className="mt-0.5 shrink-0" />
|
||||
</div>
|
||||
|
||||
{/* 总控提示词:label + AI 优化(配色沿用现有 primary)+ 大号等宽编辑区 */}
|
||||
<div className="space-y-2">
|
||||
<div className="flex items-center justify-between gap-2">
|
||||
@@ -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({
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{/* 控制:技能识别强化 / 查看详情 / 并入上一阶段 / 独立成阶段 / 并入下一阶段 / 移除 */}
|
||||
{/* 控制:推理深度 / 技能识别强化 / 查看详情 / 并入上一阶段 / 独立成阶段 / 并入下一阶段 / 移除 */}
|
||||
<div className="flex shrink-0 items-center gap-0.5 border-l border-border px-2">
|
||||
{/* 席位推理深度覆盖:小图标 → 下拉框(跟随链条默认 + 四档)。设了覆盖 = 高亮。
|
||||
「卡顿提速」用:把信息收集/边角席位降到「快速」,只给关键席位留「专业/多智能体」。 */}
|
||||
<DropdownMenu>
|
||||
<DropdownMenuTrigger asChild>
|
||||
<button
|
||||
type="button"
|
||||
title="席位推理深度(覆盖链条默认;降到「快速」可在模型卡顿时大幅提速)"
|
||||
className={cn(
|
||||
"p-1 transition",
|
||||
R_CTRL,
|
||||
seat.seat_mode
|
||||
? "bg-primary/10 text-primary"
|
||||
: "text-muted-foreground hover:bg-muted hover:text-foreground",
|
||||
)}
|
||||
>
|
||||
<Gauge className="size-3.5" />
|
||||
</button>
|
||||
</DropdownMenuTrigger>
|
||||
<DropdownMenuContent align="end" className="w-60">
|
||||
<DropdownMenuLabel className="text-[11px] font-semibold">
|
||||
席位推理深度
|
||||
<span className="mt-0.5 block text-[10px] font-normal text-muted-foreground">
|
||||
覆盖链条级默认,仅本席位生效。模型卡顿时把次要席位降到「快速」可大幅提速。
|
||||
</span>
|
||||
</DropdownMenuLabel>
|
||||
<DropdownMenuSeparator />
|
||||
<DropdownMenuItem
|
||||
className={cn("text-xs", !seat.seat_mode && "font-semibold text-primary")}
|
||||
onClick={() => onSeatModeChange(null)}
|
||||
>
|
||||
跟随链条默认
|
||||
</DropdownMenuItem>
|
||||
{SEAT_MODE_OPTIONS.map((opt) => (
|
||||
<DropdownMenuItem
|
||||
key={opt.value}
|
||||
className={cn("flex-col items-start gap-0 text-xs", seat.seat_mode === opt.value && "font-semibold text-primary")}
|
||||
onClick={() => onSeatModeChange(opt.value)}
|
||||
>
|
||||
<span>{opt.label}</span>
|
||||
<span className="text-[10px] font-normal leading-tight text-muted-foreground">{opt.description}</span>
|
||||
</DropdownMenuItem>
|
||||
))}
|
||||
</DropdownMenuContent>
|
||||
</DropdownMenu>
|
||||
{/* 会商技能识别强化:小图标 → 下拉框(开关 + 说明)。开启 = 高亮。 */}
|
||||
<DropdownMenu>
|
||||
<DropdownMenuTrigger asChild>
|
||||
|
||||
@@ -315,6 +315,13 @@ export default function RoundtablePlanningPage() {
|
||||
// Step 2 每席位「技能识别强化」开关映射 {agentId: bool}:由所选业务链条里**每个席位**的同名
|
||||
// 配置决定(recommend 无链条 → 空 {})。透传给每个席位 run,与各席位智能体自身的开关取 OR。
|
||||
const [seatSkillDirectives, setSeatSkillDirectives] = useState<Record<string, boolean>>({});
|
||||
// Step 2 每席位「推理深度覆盖」映射 {agentId: SeatMode|null}:由所选业务链条里**每个席位**单独配
|
||||
// 的推理深度决定(null/未配 → 跟随链条级 seatMode)。透传给席位 run(覆盖优先、否则跟随 seatMode)。
|
||||
// 「卡顿提速」用:把次要席位降到 flash 即可大幅缩短第二步耗时。
|
||||
const [seatModes, setSeatModes] = useState<Record<string, SeatMode | null>>({});
|
||||
// Step 2「研讨前并行取数」开关:由所选业务链条的 gatherFirst 决定(recommend / 无链条 → false)。
|
||||
// true = 研讨前先并行跑一轮全席位『取数』,产出作为共享资料注入研讨轮,给(尤其 chain 串行)研讨段提速。
|
||||
const [gatherFirst, setGatherFirst] = useState<boolean>(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,
|
||||
|
||||
@@ -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 `<skill_system>` 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 再按其流程取信息/执行") toLine truncated
|
||||
| **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 `<skill_system>` 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 再按其流程取信息/执行") toLine truncated
|
||||
| **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_<sha1>` ? 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/<code>`. 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"`. |
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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",
|
||||
)
|
||||
|
||||
@@ -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(
|
||||
|
||||
+16
@@ -22,6 +22,19 @@
|
||||
|
||||
请通读全部原料:提炼每个席位的核心观点 / 交付,串成一份完整的方案总结。
|
||||
|
||||
## 动笔前:意图对齐与缺口核对(质量关键,每轮首次生成报告都要做)
|
||||
各席位可能由能力较弱的模型产出,交付未必充分、未必都扣住了用户意图。你是这份报告的**质量兜底**——不能只是把各席位说的拼接起来,而要**对齐用户意图、核对覆盖、补强缺口**。动笔前先在脑内完成(这张核对表不必输出):
|
||||
|
||||
1. **拆解意图**:把「任务意图」(目标 / 约束 / 关键假设)拆成若干条**用户真正想要的诉求**;业务链总控编排提示里对最终产出的要求也并入诉求清单。
|
||||
2. **逐条核对覆盖**:对每条诉求,检查各席位交付 + 总控共识**是否覆盖、覆盖得够不够、有没有跑题或自相矛盾**。
|
||||
3. **据此组织报告**,并按下面的「缺口处理护栏」补强——让最终报告明显**比"各席位交付的简单拼接"更扣用户意图**:意图里要的、席位没说清的,你补强或标缺口;意图里没要的、席位跑题的,不要喧宾夺主。
|
||||
|
||||
### 缺口处理护栏(避免弱模型把幻觉带进报告)
|
||||
- **能从已有原料重组 / 推理补上的** → 直接补全、讲透,正常写进报告。这是你最该做的增值:把零散、跑题、含糊的席位产物,按用户意图重新梳理成扣题的方案。
|
||||
- **超出席位原料、属于你自己的合理推断 / 建议** → 可以写,但**显式标注「(综合推断)」**,让用户能区分哪些有席位依据、哪些是你的补充判断。
|
||||
- **缺真实信息、靠现有原料补不了的**(需要最新数据 / 外部事实,而席位没给)→ **绝不凭空编**。在报告对应位置用一行醒目标注:`> ⚠️ 待补充:<具体缺什么>。如需补全,可在下方对话里告诉我(例如「补一下 X 的最新情况」),我会据此完善。` 把球交给用户——你本身不联网、不调研,真补由用户在对话里发起。
|
||||
- 始终:**宁可标注缺口,也不要假装完整**;绝不编造与原料矛盾、或无依据的"事实"。
|
||||
|
||||
## 产出要求(最重要)
|
||||
|
||||
### ⓪ 先判断模式:用户是否要求了结构化输出 / 流程图
|
||||
@@ -122,5 +135,8 @@
|
||||
- ❌ 修订流程图时只输出改动的节点(必须输出全量 flow-json,前端整体重绘)
|
||||
- ❌ 在报告任何位置添加「AI 自动生成」「AI generated」之类的标签、徽章或水印
|
||||
- ❌ 编造与原料明显矛盾的内容
|
||||
- ❌ 只把各席位交付简单拼接、不做「意图对齐与缺口核对」(席位可能弱,你要兜底扣住用户意图)
|
||||
- ❌ 缺真实信息时凭空编造来"填满"报告(应按护栏标注「> ⚠️ 待补充:…」交给用户在对话里补)
|
||||
- ❌ 自己的推断不加「(综合推断)」标注,让用户误以为有席位依据
|
||||
- ❌ 忽视「业务链总控编排提示」的口径要求(它优先级最高)
|
||||
- ❌ 编排提示已指定输出结构 / 重点席位,却仍按默认章节平均用力(结构要完全跟用户走,重点席位要明显加权)
|
||||
@@ -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])
|
||||
|
||||
@@ -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)
|
||||
|
||||
+142
-18
@@ -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 / 图表)、全程简体中文。"
|
||||
"报告结构:默认用叙事型结构(一句话结论 → 背景目标 → 各席位观点 → 综合分析 → 最终共识方案 → 执行路线);"
|
||||
|
||||
+54
@@ -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")
|
||||
+6
@@ -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(
|
||||
|
||||
+6
@@ -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)
|
||||
|
||||
@@ -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 自动建,无需补列。
|
||||
-- 以下是给【已存在的表】补的列:
|
||||
|
||||
@@ -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 # 取数阶段三席位真并行
|
||||
@@ -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"
|
||||
@@ -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
|
||||
Reference in new issue
Block a user