From 086c280e5db5402b621fbcece49d9091417e94a3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Fri, 12 Jun 2026 13:52:06 +0800 Subject: [PATCH] =?UTF-8?q?=E5=9C=86=E6=A1=8C=E7=AC=AC=E4=B8=89=E6=AD=A5?= =?UTF-8?q?=E6=B5=81=E7=A8=8B=E5=9B=BE=E6=A8=A1=E5=BC=8F=20+=20=E5=90=8E?= =?UTF-8?q?=E5=8F=B0=20DAG=20=E4=BB=BB=E5=8A=A1=E7=BC=96=E6=8E=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 方案总结报告：用户要求结构化输出/流程图时跳过 md 直出 flow-json 流程图， 报告用文字展示；报告与流程图均可多轮修改、互相切换（detectFlowIntent 识别意图） - 流程图画布放右侧沙箱位（sidePanel 插槽）不覆盖消息列表，可与 md 预览切换； 节点随容器宽度自适应缩放并 fitView 居中 - 后台挂起支持真 DAG 编排（分层并行/阶段目标/总控提示词），新增 orchestration_plan 持久化列 + 迁移；大弹窗流程图按 DAG 分层渲染 Co-Authored-By: Claude Fable 5 --- .../api/roundtable-jobs.ts | 2 + .../components/DispatchChainJobModal.tsx | 2 + .../components/DispatchChainModal.tsx | 126 +++++++++----- .../components/FlowGraphPanel.tsx | 65 +++++-- .../components/Step2Panel.tsx | 4 +- .../components/Step2SandboxLayout.tsx | 20 ++- .../components/Step3ReportMode.tsx | 2 +- .../components/Step3SummaryMode.tsx | 93 +++++++--- .../hooks/useStep3Summary.ts | 159 ++++++++++++------ .../lib/flow-graph.test.ts | 19 +++ .../src/roundtable-planning/lib/flow-graph.ts | 24 +++ .../src/roundtable-planning/lib/job-types.ts | 10 +- .../pages/RoundtablePlanningPage.tsx | 10 +- .../app/gateway/roundtable_job_executor.py | 6 + .../roundtable-summary/SOUL.md | 36 +++- .../app/gateway/routers/roundtable_jobs.py | 8 + .../roundtable_orchestrator/__init__.py | 10 ++ .../agents/roundtable_orchestrator/engine.py | 118 ++++++++++++- .../simulated_gateway.py | 7 +- .../agents/roundtable_orchestrator/types.py | 136 ++++++++++++++- ...2_01_roundtable_jobs_orchestration_plan.py | 47 ++++++ .../persistence/roundtable_jobs/model.py | 6 +- .../persistence/roundtable_jobs/sql.py | 4 +- .../scripts/probe_roundtable_dag_job.py | 110 ++++++++++++ .../scripts/sql/column_additions_mysql.sql | 5 + 25 files changed, 880 insertions(+), 149 deletions(-) create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260612_01_roundtable_jobs_orchestration_plan.py create mode 100644 offline-backend-20260512/backend/scripts/probe_roundtable_dag_job.py diff --git a/frontend-web/src/roundtable-planning/api/roundtable-jobs.ts b/frontend-web/src/roundtable-planning/api/roundtable-jobs.ts index 1a4a664a..eac85913 100644 --- a/frontend-web/src/roundtable-planning/api/roundtable-jobs.ts +++ b/frontend-web/src/roundtable-planning/api/roundtable-jobs.ts @@ -65,6 +65,7 @@ function mapJob(raw: Record): RoundtableJob { pendingClarification: (raw.pendingClarification as string | null) ?? null, orchestrationMode: (raw.orchestrationMode as RoundtableJob["orchestrationMode"]) ?? "recommend", chain: (raw.chain as RoundtableJob["chain"]) ?? null, + orchestrationPlan: (raw.orchestrationPlan as RoundtableJob["orchestrationPlan"]) ?? null, error: (raw.error as string | null) ?? null, dialogues: Array.isArray(raw.dialogues) ? (raw.dialogues as JobDialogue[]) : [], step3: (raw.step3 as RoundtableJob["step3"]) ?? null, @@ -81,6 +82,7 @@ function toStartBody(p: StartJobPayload) { model: p.model, orchestrationMode: p.orchestrationMode ?? "recommend", chain: p.chain ?? null, + orchestrationPlan: p.orchestrationPlan ?? null, seedMessage: p.seedMessage, threadIds: p.threadIds ?? null, coordinatorName: p.coordinatorName ?? "roundtable-coordinator", diff --git a/frontend-web/src/roundtable-planning/components/DispatchChainJobModal.tsx b/frontend-web/src/roundtable-planning/components/DispatchChainJobModal.tsx index 8da7ebbc..19f2eba2 100644 --- a/frontend-web/src/roundtable-planning/components/DispatchChainJobModal.tsx +++ b/frontend-web/src/roundtable-planning/components/DispatchChainJobModal.tsx @@ -68,6 +68,8 @@ export function DispatchChainJobModal({ onView={job && onView ? () => onView(job) : undefined} error={job?.error} dialogues={job?.dialogues} + orchestrationMode={job?.orchestrationMode} + stages={job?.orchestrationPlan?.stages ?? null} pendingClarification={job?.pendingClarification} onResume={ jobId diff --git a/frontend-web/src/roundtable-planning/components/DispatchChainModal.tsx b/frontend-web/src/roundtable-planning/components/DispatchChainModal.tsx index 400bc7e9..ff170679 100644 --- a/frontend-web/src/roundtable-planning/components/DispatchChainModal.tsx +++ b/frontend-web/src/roundtable-planning/components/DispatchChainModal.tsx @@ -29,8 +29,10 @@ import { import { cn } from "@/lib/utils"; import { getAvatar } from "./Avatars"; +import { DagFlowChart } from "./DagFlowChart"; import { MarkdownContent } from "./MessageBubble"; import { chainTagClasses } from "../lib/business-chain"; +import type { OrchestrationStage } from "../lib/dag-types"; import { getAvatarTypeFor } from "../lib/roundtable-constants"; import type { ChainNodeState, @@ -67,6 +69,10 @@ export interface DispatchChainModalProps { error?: string | null; /** 研讨对话记录（leader 说明 + 各席位交付）。 */ dialogues?: JobDialogue[]; + /** 编排模式；==="dag" 且 stages 非空时，流程图按 stage 分层（DAG）绘制，否则线性。 */ + orchestrationMode?: "recommend" | "chain" | "dag"; + /** dag 模式的分层 stages（按 stage 分列画流程图，stage 内并行席位纵向并列）。 */ + stages?: ReadonlyArray | null; } const _NON_TERMINAL: ReadonlySet = new Set(["queued", "running", "awaiting_input"]); @@ -105,6 +111,8 @@ export function DispatchChainModal({ onCancel, error, dialogues, + orchestrationMode, + stages, }: DispatchChainModalProps) { const [answer, setAnswer] = useState(""); const convoRef = useRef(null); @@ -139,6 +147,28 @@ export function DispatchChainModal({ const reportActive = phase === "report_drawing"; const reportDone = phase === "report_done" || status === "done"; + // ── DAG（分层）流程图：mode==="dag" 且有 stages 时按 stage 分列绘制 ────────────── + const showDag = orchestrationMode === "dag" && (stages?.length ?? 0) > 0; + // 节点状态来自 dispatch_chain（agentId 小写 → state）；resolveSeat 取名称 + 头像配色。 + const dagSeatStates: Record = {}; + const nodeByAgent = new Map(); + for (const n of nodes) { + dagSeatStates[n.agentId.toLowerCase()] = n.state; + nodeByAgent.set(n.agentId.toLowerCase(), n); + } + const resolveDagSeat = (agentId: string, order: number) => { + const node = nodeByAgent.get(agentId.toLowerCase()); + return { + name: node?.name || agentId, + avatarType: node?.avatarType || getAvatarTypeFor(agentId, order), + }; + }; + const dagConsensusState: ChainNodeState = consensusDone + ? "done" + : consensusActive + ? "active" + : "pending"; + return ( @@ -167,51 +197,65 @@ export function DispatchChainModal({ {PHASE_LABEL[phase]} - {/* 横向流程图（横向滚动） */} -
-
- {/* 总控节点 */} - + {/* 流程图：dag 模式按 stage 分层（DAG）；否则线性总控→席位→共识→报告 */} + {showDag ? ( + // 分层 DAG 流程图：总控 → ┌Stage┐(并行席位纵向并列) → … → 🤝共识。 + // 后台本期不绘报告(Step3 禁用) → 不显示「结果绘制」里程碑。 + + ) : ( +
+
+ {/* 总控节点 */} + - {/* 席位节点 */} - {nodes.map((node) => ( - - - - - ))} + {/* 席位节点 */} + {nodes.map((node) => ( + + + + + ))} - {/* 共识里程碑 */} - - + {/* 共识里程碑 */} + + - {/* 报告绘制里程碑 */} - - + {/* 报告绘制里程碑 */} + + +
-
+ )} {/* 研讨对话（leader 说明 + 各席位交付，真网关下即真模型真实对话） */} {dialogues && dialogues.length > 0 && ( diff --git a/frontend-web/src/roundtable-planning/components/FlowGraphPanel.tsx b/frontend-web/src/roundtable-planning/components/FlowGraphPanel.tsx index 25ec4815..bcad46e8 100644 --- a/frontend-web/src/roundtable-planning/components/FlowGraphPanel.tsx +++ b/frontend-web/src/roundtable-planning/components/FlowGraphPanel.tsx @@ -32,6 +32,7 @@ import { } from "@xyflow/react"; import "@xyflow/react/dist/style.css"; import { + FileText, Loader2, LocateFixed, Maximize2, @@ -56,21 +57,26 @@ interface FlowGraphPanelProps { /** 流式绘制中（决定蚂蚁线 + 自动 fitView 跟随）。 */ streaming: boolean; onClose: () => void; + /** 沙箱里有 md 报告时提供：工具栏显示「MD 报告」按钮，点击切回 Markdown 预览。 */ + onSwitchToReport?: () => void; } // ── 自定义节点（按 type 着色的卡片）──────────────────────────────────────── function FlowCardNode({ data }: NodeProps) { - const node = (data as { node?: FlowGraphNode }).node; + const node = (data as { node?: FlowGraphNode; scale?: number }).node; + const scale = (data as { node?: FlowGraphNode; scale?: number }).scale ?? 1; if (!node) return null; const meta = FLOW_NODE_TYPE_META[node.type]; + // 节点宽度随容器宽度缩放（宽时放大、窄时缩小）。 + const scaledWidth = FLOW_NODE_W * scale; return (
- + {meta.label}
-
{node.label}
+
{node.label}
); @@ -99,22 +105,48 @@ function FlowCanvasInner({ fullscreen, onToggleFullscreen, onClose, + onSwitchToReport, }: FlowGraphPanelProps & { fullscreen: boolean; onToggleFullscreen: () => void }) { const rf = useReactFlow(); const [selectedId, setSelectedId] = useState(null); + const [scale, setScale] = useState(1); + const containerRef = useRef(null); + const fitTimerRef = useRef | null>(null); // 用户手动拖过的节点位置 —— 数据增量更新重新布局时保留，不被布局覆盖。 const manualPosRef = useRef>(new Map()); + // 监听容器宽度：宽度变化时按比例缩放节点（变宽→放大、变窄→缩小），并重新 fitView 居中。 + // 700px 基准 1.0，clamp 到 [0.7, 1.4]：拖宽到 ~1000px 节点放大、窄屏自动缩小。 + useEffect(() => { + const el = containerRef.current; + if (!el) return; + const resizeObs = new ResizeObserver(([entry]) => { + const width = entry.contentRect.width; + if (width <= 0) return; + setScale(Math.min(1.4, Math.max(0.7, width / 700))); + if (fitTimerRef.current) clearTimeout(fitTimerRef.current); + fitTimerRef.current = setTimeout(() => { + void rf.fitView({ duration: 200, padding: 0.15 }); + }, 120); + }); + resizeObs.observe(el); + return () => { + resizeObs.disconnect(); + if (fitTimerRef.current) clearTimeout(fitTimerRef.current); + }; + }, [rf]); + const layoutNodes = useMemo(() => layoutFlowGraph(data), [data]); const derivedNodes: Node[] = useMemo( () => layoutNodes.map((n) => ({ id: n.id, type: "flowCard", - position: manualPosRef.current.get(n.id) ?? { x: n.x, y: n.y }, - data: { node: n }, + // 节点随 scale 放大 → 布局坐标同步乘 scale，间距一起拉开，避免同层重叠。 + position: manualPosRef.current.get(n.id) ?? { x: n.x * scale, y: n.y * scale }, + data: { node: n, scale }, })), - [layoutNodes], + [layoutNodes, scale], ); const [nodes, setNodes] = useState(derivedNodes); @@ -180,6 +212,17 @@ function FlowCanvasInner({ )}
+ {onSwitchToReport && ( + + )} void rf.zoomIn({ duration: 150 })}> @@ -202,7 +245,7 @@ function FlowCanvasInner({
{/* 画布 */} -
+
@@ -393,6 +437,7 @@ export function FlowGraphPanel({ data, streaming, onClose }: FlowGraphPanelProps fullscreen={fullscreen} onToggleFullscreen={() => setFullscreen((f) => !f)} onClose={onClose} + onSwitchToReport={onSwitchToReport} />
diff --git a/frontend-web/src/roundtable-planning/components/Step2Panel.tsx b/frontend-web/src/roundtable-planning/components/Step2Panel.tsx index cddfe6a4..943b1914 100644 --- a/frontend-web/src/roundtable-planning/components/Step2Panel.tsx +++ b/frontend-web/src/roundtable-planning/components/Step2Panel.tsx @@ -44,8 +44,8 @@ import { DialogTitle, } from "@/components/ui/dialog"; -/** 后台挂起入口 —— 暂时隐藏,恢复时改为 true。 */ -const SHOW_BACKGROUND_SUSPEND = false; +/** 后台挂起入口 —— 后端 _run_dag 已支持业务链分层并行,放出该入口。 */ +const SHOW_BACKGROUND_SUSPEND = true; import { PersonaDrawer } from "@/roundtable-planning/components/PersonaDrawer"; import { ScrollToBottomButton } from "@/roundtable-planning/components/ScrollToBottomButton"; diff --git a/frontend-web/src/roundtable-planning/components/Step2SandboxLayout.tsx b/frontend-web/src/roundtable-planning/components/Step2SandboxLayout.tsx index 75201cad..1d69dea8 100644 --- a/frontend-web/src/roundtable-planning/components/Step2SandboxLayout.tsx +++ b/frontend-web/src/roundtable-planning/components/Step2SandboxLayout.tsx @@ -43,6 +43,11 @@ interface Step2SandboxLayoutProps { * 对齐主聊天 chats/:threadId 的实时预览体验。 */ seatMessages?: unknown[]; + /** + * 右侧面板的替代内容（如 Step3 流程图画布）——非空时强制展开右栏并替换沙箱预览， + * 左侧消息列表保持可见；置空即恢复沙箱（文件预览/列表）。 + */ + sidePanel?: React.ReactNode; } export function Step2SandboxLayout({ @@ -52,6 +57,7 @@ export function Step2SandboxLayout({ seatMessages, artifactsFill = false, onArtifactsClose, + sidePanel, }: Step2SandboxLayoutProps) { const layoutRef = useRef(null); const { @@ -64,11 +70,13 @@ export function Step2SandboxLayout({ const artifacts = rawArtifacts ?? []; const artifactPanelOpen = artifactsOpen; + // sidePanel（流程图等）非空时即使沙箱没开也要把右栏撑开。 + const panelOpen = artifactPanelOpen || !!sidePanel; useEffect(() => { if (artifactsFill || !layoutRef.current) return; - layoutRef.current.setLayout(artifactPanelOpen ? OPEN_MODE : CLOSE_MODE); - }, [artifactPanelOpen, artifactsFill]); + layoutRef.current.setLayout(panelOpen ? OPEN_MODE : CLOSE_MODE); + }, [panelOpen, artifactsFill]); const handleArtifactsClose = () => { deselect(); @@ -151,7 +159,7 @@ export function Step2SandboxLayout({ if (artifactsFill) { return ( -
{artifactPanelBody}
+
{sidePanel ?? artifactPanelBody}
 ); } @@ -172,17 +180,17 @@ export function Step2SandboxLayout({ id="roundtable-step2-separator" className={cn( "opacity-33 hover:opacity-100", - !artifactPanelOpen && "pointer-events-none opacity-0", + !panelOpen && "pointer-events-none opacity-0", )} /> - {artifactPanelBody} + {sidePanel ?? artifactPanelBody} diff --git a/frontend-web/src/roundtable-planning/components/Step3ReportMode.tsx b/frontend-web/src/roundtable-planning/components/Step3ReportMode.tsx index 29e63db9..4c57d8c7 100644 --- a/frontend-web/src/roundtable-planning/components/Step3ReportMode.tsx +++ b/frontend-web/src/roundtable-planning/components/Step3ReportMode.tsx @@ -329,7 +329,7 @@ export function Step3ReportMode({ dialogues={dialogues} error={error} themeMismatch={themeMismatch} - onRetry={redraw} + onRetry={() => redraw()} onRevisionSubmit={interject} onArtifactStepClick={openArtifactFromStep} onArtifactPreview={openArtifactPreview} diff --git a/frontend-web/src/roundtable-planning/components/Step3SummaryMode.tsx b/frontend-web/src/roundtable-planning/components/Step3SummaryMode.tsx index 26e08aa3..6d22d527 100644 --- a/frontend-web/src/roundtable-planning/components/Step3SummaryMode.tsx +++ b/frontend-web/src/roundtable-planning/components/Step3SummaryMode.tsx @@ -42,6 +42,7 @@ import { ArtifactPreviewModal } from "@/roundtable-planning/components/ArtifactP import { ConsensusDocxDownloadButton } from "@/roundtable-planning/components/ConsensusDocxDownloadButton"; import { FlowGraphPanel } from "@/roundtable-planning/components/FlowGraphPanel"; import { + detectFlowIntent, extractFlowGraph, stripFlowFence, FLOW_FENCE_OPEN, @@ -169,6 +170,37 @@ export function Step3SummaryMode({ if (!artifactsOpen) setFlowView(true); }, [flowGraph, artifactsOpen]); + // 沙箱里已写出的 md 报告（流程图画布上提供「MD 报告」切回按钮的锚点）。 + const mdArtifact = useMemo(() => { + for (const [threadId, msgs] of Object.entries(seatStubMessages)) { + const msg = msgs?.[0]; + if (!msg) continue; + for (const tc of msg.tool_calls) { + if (tc.name !== "write_file") continue; + const path = + typeof (tc.args as { path?: unknown })?.path === "string" + ? ((tc.args as { path?: string }).path as string) + : ""; + const content = + typeof (tc.args as { content?: unknown })?.content === "string" + ? ((tc.args as { content?: string }).content as string) + : ""; + if (/\.md$/i.test(path) && content.trim()) return { threadId, path }; + } + } + return null; + }, [seatStubMessages]); + + // 用户在输入框里明确要求结构化输出/流程图 → 立即切到流程图画布（旧图先顶着， + // 新 flow-json 流出来就增量重绘；首次还没有图时画布在数据到达后自动渲染）。 + const handleAsk = (message: string, files?: File[]) => { + if (detectFlowIntent(message)) { + flowAutoOpenedRef.current = true; + setFlowView(true); + } + ask(message, files); + }; + // 气泡里不展示生 JSON：把 flow-json 块替换成一句引导语。 const displayDialogues = useMemo( () => @@ -180,17 +212,20 @@ export function Step3SummaryMode({ [dialogues], ); const [leftCollapsed, setLeftCollapsed] = useState(false); - const prevArtifactsOpenRef = useRef(artifactsOpen); + // 右侧面板（沙箱预览或流程图画布）露出时自动收起左栏，关闭时复原——边沿触发，手动切换不回弹。 + const rightPanelVisible = artifactsOpen || (!!flowGraph && flowView); + const prevRightPanelVisibleRef = useRef(rightPanelVisible); useEffect(() => { - if (artifactsOpen !== prevArtifactsOpenRef.current) { - prevArtifactsOpenRef.current = artifactsOpen; - setLeftCollapsed(artifactsOpen); + if (rightPanelVisible !== prevRightPanelVisibleRef.current) { + prevRightPanelVisibleRef.current = rightPanelVisible; + setLeftCollapsed(rightPanelVisible); } - }, [artifactsOpen]); + }, [rightPanelVisible]); - // 进入即自动开跑；草稿里已存结果则恢复展示。卸载中断在途请求（理由同 Step3ReportMode）。 + // 进入即自动开跑；草稿里已存结果则恢复展示（流程图模式的快照可能只有 closing 没有 md）。 + // 卸载中断在途请求（理由同 Step3ReportMode）。 useEffect(() => { - if (savedSummary?.md) { + if (savedSummary && (savedSummary.md.trim() || savedSummary.closing?.trim())) { hydrateSaved(savedSummary); } else { start(); @@ -337,20 +372,39 @@ export function Step3SummaryMode({ )} - {/* 右栏：智能体进度 + 沙箱（复用第 2 步）。流程图画布作为覆盖层叠在其上。 */} + {/* 右栏：智能体进度 + 沙箱（复用第 2 步）。流程图画布放在右侧沙箱位 + （sidePanel 插槽），消息列表保持可见，可与 md 沙箱预览互相切换。 */}
setFlowView(false)} + onSwitchToReport={ + mdArtifact + ? () => { + setFlowView(false); + openArtifactPreview(mdArtifact.threadId, mdArtifact.path); + } + : undefined + } + /> + ) : undefined + } > )} - {/* 流程图画布（覆盖右栏；全屏在组件内部切换） */} - {flowGraph && flowView && ( - setFlowView(false)} - /> - )}
@@ -498,6 +544,7 @@ function AgentProgressPanel({ dialogues, error, busy, + mdAvailable, onRetry, onAsk, onPause, @@ -512,6 +559,8 @@ function AgentProgressPanel({ dialogues: ReturnType["dialogues"]; error: string | null; busy: boolean; + /** 沙箱里是否已有 md 报告 —— 决定提示语（纯流程图轮没有 .md 可预览/下载）。 */ + mdAvailable: boolean; onRetry: () => void; onAsk: (message: string, files?: File[]) => void; onPause: () => void; @@ -616,7 +665,9 @@ function AgentProgressPanel({ {busy && (
- 报告正文实时输出，右侧沙箱同步预览 Markdown + {mdAvailable + ? "报告正文实时输出，右侧沙箱同步预览 Markdown" + : "报告正文实时输出；输出结构化数据时流程图画布同步绘制"}
)} @@ -638,7 +689,9 @@ function AgentProgressPanel({ {phase === "done" && (
- 已生成方案总结报告，右侧可预览 / 下载 .md；可在下方继续提修改意见或提问。 + {mdAvailable + ? "已生成方案总结报告，右侧可预览 / 下载 .md；可在下方继续修改报告 / 流程图或提问。" + : "已生成方案总结，流程图可在画布查看；可在下方继续修改总结 / 流程图或提问。"}
)} @@ -683,7 +736,7 @@ function AgentProgressPanel({ placeholder={ busy ? "生成中也可随时插话干预：发送将打断当前轮并据此继续…" - : "继续对话：提修改意见（如「执行路线写细些」）/ 提问 / 上传参考文件" + : "继续对话：改报告（如「执行路线写细些」）/ 改流程图（如「总结成结构化输出」「流程图加上验收环节」）/ 提问" } className="min-h-8 max-h-28 flex-1 resize-none bg-transparent py-1.5 text-xs leading-relaxed text-slate-800 placeholder:text-slate-400 focus:outline-none" /> diff --git a/frontend-web/src/roundtable-planning/hooks/useStep3Summary.ts b/frontend-web/src/roundtable-planning/hooks/useStep3Summary.ts index 8ae53352..61ccd2db 100644 --- a/frontend-web/src/roundtable-planning/hooks/useStep3Summary.ts +++ b/frontend-web/src/roundtable-planning/hooks/useStep3Summary.ts @@ -40,6 +40,7 @@ import type { Step2Dialogue, StubAiMessage, } from "@/roundtable-planning/hooks/useStep2Orchestration"; +import { detectFlowIntent } from "@/roundtable-planning/lib/flow-graph"; import { nowTime } from "@/roundtable-planning/lib/roundtable-constants"; /** 报告输出文件的固定虚拟路径（恢复历史快照时用它重建 stub）。 */ @@ -149,31 +150,60 @@ function buildMaterials(opts: { return parts.join("\n"); } -/** 首轮总结 prompt。 */ +/** 首轮总结 prompt。`flowMode`=编排提示里明确要求了结构化输出/流程图 → 跳过 md 文件，直出流程图。 */ function buildSummaryPrompt(opts: { intentReady: ReportIntent | null; seatDeliveries: ReportSeatDelivery[]; consensusText: string; coordinatorPrompt?: string | null; + flowMode?: boolean; }): string { const parts: string[] = []; - parts.push( - "请基于以下圆桌多智能体研讨的成果，撰写一份以文字叙述为主线的 Markdown「方案总结报告」，" + - "用 write_file 工具写入 /mnt/user-data/outputs/方案总结报告.md，" + - "并把报告完整正文同时输出到对话回复里供用户直接阅读。", - ); + if (opts.flowMode) { + parts.push( + "请基于以下圆桌多智能体研讨的成果，产出一份**结构化的方案总结**。" + + "用户已明确要求结构化输出/流程图，因此本轮：" + + "①**不要**调用 write_file，**不要**生成任何 .md 文件；" + + "②把精炼的文字版总结报告直接写在对话回复里（供用户直接阅读，结构遵循编排提示的要求）；" + + "③在回复最末尾**必须**输出一个完整的 ```flow-json 代码块（前端据此实时绘制可交互流程图）。", + ); + } else { + parts.push( + "请基于以下圆桌多智能体研讨的成果，撰写一份以文字叙述为主线的 Markdown「方案总结报告」，" + + "用 write_file 工具写入 /mnt/user-data/outputs/方案总结报告.md，" + + "并把报告完整正文同时输出到对话回复里供用户直接阅读。", + ); + } parts.push(""); parts.push(buildMaterials(opts)); - parts.push( - "请严格遵循你的职责设定：纯 Markdown 文本（不画 HTML / 图表）、全程简体中文。" + - "报告结构：默认用叙事型结构（一句话结论 → 背景目标 → 各席位观点 → 综合分析 → 最终共识方案 → 执行路线）；" + - "但若上面的「业务链总控编排提示」对最终总结提出了要求（输出结构/格式、重点席位、口径、详略），" + - "那些要求优先于默认结构——用户指定的结构可以完全替换默认章节。" + - "若用户要求了结构化输出/流程图，按你职责设定在对话回复末尾追加 ```flow-json 代码块（前端据此实时绘制流程图）。", - ); + if (opts.flowMode) { + parts.push( + "请严格遵循你的职责设定：全程简体中文。" + + "flow-json 按你职责设定的「结构化流程数据」契约输出：节点按流程先后排列、" + + "每个节点的 from 只引用前面已出现的节点、6–20 个节点为宜、类型用 start/end/stage/task/decision/risk/deliverable/role。" + + "正文文字报告以「业务链总控编排提示」的要求为骨架，简明、重点突出。", + ); + } else { + parts.push( + "请严格遵循你的职责设定：纯 Markdown 文本（不画 HTML / 图表）、全程简体中文。" + + "报告结构：默认用叙事型结构（一句话结论 → 背景目标 → 各席位观点 → 综合分析 → 最终共识方案 → 执行路线）；" + + "但若上面的「业务链总控编排提示」对最终总结提出了要求（输出结构/格式、重点席位、口径、详略），" + + "那些要求优先于默认结构——用户指定的结构可以完全替换默认章节。" + + "若用户要求了结构化输出/流程图，按你职责设定在对话回复末尾追加 ```flow-json 代码块（前端据此实时绘制流程图）。", + ); + } return parts.join("\n"); } +/** + * 用户消息命中结构化/流程图意图时追加给智能体的系统指令（live 线程续问轮直接拼在消息后）。 + * 强调「全量 flow-json + 不写文件」——画图数据必须完整自洽，否则前端增量解析会画出半张图。 + */ +const FLOW_ASK_INSTRUCTION = + "（系统指令：用户要求结构化输出/流程图——本轮不要写或修改任何 .md 报告文件，" + + "在回复正文给出简短的文字说明/总结即可，并在回复最末尾输出一个**完整**的 ```flow-json 代码块" + + "（按你职责设定的「结构化流程数据」契约；输出全量数据而非增量）。）"; + /** 恢复草稿后线程已不在内存（无 live thread）时，新建线程续问用的 prompt：带上旧报告做底。 */ function buildResumeAskPrompt(opts: { intentReady: ReportIntent | null; @@ -182,8 +212,10 @@ function buildResumeAskPrompt(opts: { coordinatorPrompt?: string | null; previousMd: string; userMessage: string; + /** 用户消息命中结构化/流程图意图 → 指示不写文件、直出 flow-json。 */ + flowIntent?: boolean; }): string { - const { previousMd, userMessage, ...materials } = opts; + const { previousMd, userMessage, flowIntent, ...materials } = opts; const parts: string[] = []; parts.push( "这是一次对已有「方案总结报告」的继续对话。下面先给出研讨原料与你之前已生成的报告，" + @@ -202,6 +234,10 @@ function buildResumeAskPrompt(opts: { } parts.push("# 用户的新消息"); parts.push(userMessage.trim()); + if (flowIntent) { + parts.push(""); + parts.push(FLOW_ASK_INSTRUCTION); + } return parts.join("\n"); } @@ -649,15 +685,19 @@ export function useStep3Summary( ); }, []); - /** 把当前报告 md 种进某线程 stub + 开沙箱（续问时给 reconcile/预览一个锚点）。 */ + /** + * 把当前报告 md 种进某线程 stub（续问时给 reconcile/预览一个锚点）。 + * `open`=false 时只种不开沙箱——流程图轮用户要看的是画布，不要弹 md 预览抢占。 + */ const seedStubWithCurrentMd = useCallback( - (threadId: string) => { + (threadId: string, opts?: { open?: boolean }) => { const md = currentMdRef.current; if (!md.trim()) return; const { messageId, toolCallId } = upsertStub(threadId, "write_file", { path: SUMMARY_OUTPUT_PATH, content: md, }); + if (opts?.open === false) return; const virtualUrl = buildWriteFileVirtualUrl(SUMMARY_OUTPUT_PATH, messageId, toolCallId); openArtifact(threadId, virtualUrl); }, @@ -676,8 +716,10 @@ export function useStep3Summary( seedThreadId?: string | null; forceReconcile?: boolean; files?: RunFileRef[]; + /** 流程图轮：不自动打开 md 沙箱预览（用户要看的是流程图画布）。 */ + flowRound?: boolean; }): Promise => { - const { bubbleId, prompt, aborter, seedThreadId, forceReconcile, files } = args; + const { bubbleId, prompt, aborter, seedThreadId, forceReconcile, files, flowRound } = args; let threadId = seedThreadId ?? null; let agentName = SUMMARY_AGENT_NAME; @@ -694,7 +736,8 @@ export function useStep3Summary( prev.map((d) => (d.id === bubbleId ? { ...d, agentThreadId: threadId! } : d)), ); // 续问/恢复时把当前报告种进本线程 stub（续问 str_replace 被禁、整文件重写，但仍给预览锚点）。 - if (seedThreadId) seedStubWithCurrentMd(threadId); + // 流程图轮只种不开沙箱——md 保留着供用户随时切回去看。 + if (seedThreadId) seedStubWithCurrentMd(threadId, { open: !flowRound }); setPhase("streaming"); const res = await streamMultiAgent({ @@ -718,7 +761,8 @@ export function useStep3Summary( if (aborter.signal.aborted) return; const hasStubFile = (stubRef.current[threadId]?.[0]?.tool_calls.length ?? 0) > 0; - if (!hasStubFile && !seedThreadId) await openRealArtifactFallback(threadId); + // 流程图轮本就不产文件，不要去沙箱兜底翻文件（也避免把别的文件弹出来）。 + if (!hasStubFile && !seedThreadId && !flowRound) await openRealArtifactFallback(threadId); // 以真实 md 回灌 stub，拿到完整报告（ask 强制对账，防增量改后被长度守卫挡掉）。 const md = await reconcileStubFromRealFile(threadId, { @@ -731,9 +775,10 @@ export function useStep3Summary( setPhase("done"); // 报告有更新就存草稿（问答轮不写文件 → md 不变也无妨，仍带最新 closing）。 + // 流程图模式可能全程没有 md——只要 closing 有内容（含 flow-json）也要存，恢复时才能重画流程图。 const finalMd = md.trim() || currentMdRef.current; - if (finalMd.trim()) { - currentMdRef.current = finalMd; + if (finalMd.trim()) currentMdRef.current = finalMd; + if (finalMd.trim() || (res.content ?? "").trim()) { onSummarySavedRef.current?.({ md: finalMd, threadId, @@ -778,13 +823,16 @@ export function useStep3Summary( }); try { + // 编排提示里明确要求了结构化输出/流程图 → 首轮就走流程图模式（不写 md，直出 flow-json）。 + const flowMode = detectFlowIntent(coordinatorPrompt); const prompt = buildSummaryPrompt({ intentReady, seatDeliveries, consensusText, coordinatorPrompt, + flowMode, }); - await streamOnce({ bubbleId, prompt, aborter }); + await streamOnce({ bubbleId, prompt, aborter, flowRound: flowMode }); } catch (err) { if (aborter.signal.aborted) return; const msg = err instanceof Error ? err.message : String(err); @@ -817,6 +865,9 @@ export function useStep3Summary( abort(); if (wasBusy) freezeStreaming("已打断"); + // 输入框消息命中结构化/流程图意图 → 本轮直出 flow-json、不写 md（md 留在沙箱可切回）。 + const flowIntent = detectFlowIntent(text); + runningRef.current = true; setError(null); const aborter = new AbortController(); @@ -860,8 +911,10 @@ export function useStep3Summary( if (hadLiveThread) { // 真线程仍在 → 续问，发原始消息，靠线程上下文 + SOUL 自判改/答。 - const prompt = + // 命中流程图意图时附上系统指令（不写文件、末尾输出全量 flow-json）。 + const base = text || "我上传了参考文件，请先阅读文件内容，再据此更新报告或回答。"; + const prompt = flowIntent ? `${base}\n\n${FLOW_ASK_INSTRUCTION}` : base; await streamOnce({ bubbleId, prompt, @@ -869,6 +922,7 @@ export function useStep3Summary( seedThreadId: liveThreadRef.current, forceReconcile: true, files: fileRefs, + flowRound: flowIntent, }); } else { // 无 live thread（恢复草稿 / 首轮没建好就被打断）→ prompt 带研讨原料 + 旧报告做底。 @@ -881,6 +935,7 @@ export function useStep3Summary( previousMd: currentMdRef.current, userMessage: text || "请阅读我上传的参考文件后，据此更新报告或回答。", + flowIntent, }); await streamOnce({ bubbleId, @@ -889,6 +944,7 @@ export function useStep3Summary( seedThreadId: threadId, forceReconcile: true, files: fileRefs, + flowRound: flowIntent, }); } } catch (err) { @@ -933,30 +989,35 @@ export function useStep3Summary( // 从草稿快照恢复：不跑 SSE，用保存的 md 重建一条「完成」气泡 + 合成 stub，并打开沙箱预览。 // 保留 threadId 到 liveThreadRef —— 若快照带真线程 id，ask() 能直接续问（后端线程仍在）。 + // 流程图模式的快照可能没有 md（只有含 flow-json 的 closing）：跳过 stub/沙箱，只重建气泡， + // 流程图由组件从气泡内容里重新解析绘制。 const hydrateSaved = useCallback( (snap: DraftStep3SummarySnapshot) => { abort(); runningRef.current = false; + const hasMd = snap.md.trim().length > 0; const realThreadId = snap.threadId?.trim() || null; const synthThreadId = realThreadId ?? "roundtable-summary-restored"; const messageId = `roundtable-summary-stub-${synthThreadId}`; const toolCallId = `tc-${synthThreadId}-0`; - const stub: Record = { - [synthThreadId]: [ - { - id: messageId, - type: "ai", - content: "", - tool_calls: [ + const stub: Record = hasMd + ? { + [synthThreadId]: [ { - id: toolCallId, - name: "write_file", - args: { path: SUMMARY_OUTPUT_PATH, content: snap.md }, + id: messageId, + type: "ai", + content: "", + tool_calls: [ + { + id: toolCallId, + name: "write_file", + args: { path: SUMMARY_OUTPUT_PATH, content: snap.md }, + }, + ], }, ], - }, - ], - }; + } + : {}; stubRef.current = stub; currentMdRef.current = snap.md; liveThreadRef.current = realThreadId; @@ -975,23 +1036,27 @@ export function useStep3Summary( time: nowTime(), confidence: "历史记录", streaming: false, - steps: [ - { - id: `${bubbleId}-0`, - key: "tool_calling:write_file", - label: "已撰写报告 Markdown", - ts: nowTime(), - args: { path: SUMMARY_OUTPUT_PATH }, - }, - ], + steps: hasMd + ? [ + { + id: `${bubbleId}-0`, + key: "tool_calling:write_file", + label: "已撰写报告 Markdown", + ts: nowTime(), + args: { path: SUMMARY_OUTPUT_PATH }, + }, + ] + : [], agentThreadId: synthThreadId, }, ]); setError(null); setPhase("done"); - const virtualUrl = buildWriteFileVirtualUrl(SUMMARY_OUTPUT_PATH, messageId, toolCallId); - openArtifact(synthThreadId, virtualUrl); + if (hasMd) { + const virtualUrl = buildWriteFileVirtualUrl(SUMMARY_OUTPUT_PATH, messageId, toolCallId); + openArtifact(synthThreadId, virtualUrl); + } }, [abort, buildWriteFileVirtualUrl, openArtifact], ); diff --git a/frontend-web/src/roundtable-planning/lib/flow-graph.test.ts b/frontend-web/src/roundtable-planning/lib/flow-graph.test.ts index 4886d5ec..7b49dc95 100644 --- a/frontend-web/src/roundtable-planning/lib/flow-graph.test.ts +++ b/frontend-web/src/roundtable-planning/lib/flow-graph.test.ts @@ -9,6 +9,7 @@ import assert from "node:assert/strict"; import test from "node:test"; import { + detectFlowIntent, extractFlowGraph, layoutFlowGraph, stripFlowFence, @@ -109,6 +110,24 @@ test("layoutFlowGraph：按前驱分层（根 0 层，子层递增），保持 assert.deepEqual(positioned.map((n) => n.id), ["n1", "n2", "n3"]); }); +test("detectFlowIntent：明确短语命中（编排提示 / 输入框消息）", () => { + assert.ok(detectFlowIntent("最终总结请用结构化输出")); + assert.ok(detectFlowIntent("总结成格式化输出")); + assert.ok(detectFlowIntent("给我画一张流程图")); + assert.ok(detectFlowIntent("输出可视化流程")); + assert.ok(detectFlowIntent("请按 flow-json 契约输出")); + assert.ok(detectFlowIntent("重点总结方案设计席位，最后给结构化数据")); +}); + +test("detectFlowIntent：普通措辞不误触发", () => { + assert.equal(detectFlowIntent("用表格把要点结构化呈现"), false); + assert.equal(detectFlowIntent("把执行路线写得更细一些"), false); + assert.equal(detectFlowIntent("面向高管汇报，控制在一页内"), false); + assert.equal(detectFlowIntent(""), false); + assert.equal(detectFlowIntent(null), false); + assert.equal(detectFlowIntent(undefined), false); +}); + test("layoutFlowGraph：环不死循环", () => { const text = [ "```flow-json", diff --git a/frontend-web/src/roundtable-planning/lib/flow-graph.ts b/frontend-web/src/roundtable-planning/lib/flow-graph.ts index b921134d..92d01af8 100644 --- a/frontend-web/src/roundtable-planning/lib/flow-graph.ts +++ b/frontend-web/src/roundtable-planning/lib/flow-graph.ts @@ -212,6 +212,30 @@ export function extractFlowGraph(text: string): FlowGraphData | null { return { title, nodes, edges: validEdges }; } +/** + * 识别文本（业务链编排提示 / 用户输入框消息）里是否明确要求了结构化输出 / 流程图。 + * 命中 → 总结走「流程图模式」：跳过 .md 文件，报告用文字展示 + 末尾输出 flow-json。 + * 只匹配较明确的短语，避免「用表格结构化呈现」这类普通措辞误触发。 + */ +const FLOW_INTENT_PATTERNS = [ + "流程图", + "可视化流程", + "结构化输出", + "结构化数据", + "结构化流程", + "结构化总结", + "格式化输出", + "格式化总结", + "flow-json", + "flowchart", +]; + +export function detectFlowIntent(text: string | null | undefined): boolean { + if (!text) return false; + const lower = text.toLowerCase(); + return FLOW_INTENT_PATTERNS.some((p) => lower.includes(p)); +} + /** * 把回复里的 flow-json 块替换成一句人话（气泡里不展示生 JSON）。 * 块未闭合（流式中）→「正在输出」措辞；已闭合 → 引导去看流程图画布。 diff --git a/frontend-web/src/roundtable-planning/lib/job-types.ts b/frontend-web/src/roundtable-planning/lib/job-types.ts index fc957fa1..7ff47719 100644 --- a/frontend-web/src/roundtable-planning/lib/job-types.ts +++ b/frontend-web/src/roundtable-planning/lib/job-types.ts @@ -8,6 +8,8 @@ * `DispatchChainNode`，其余类型为后续阶段预留，先一并定义好避免反复改动。 */ +import type { OrchestrationPlan } from "./dag-types"; + /** 作业状态机（§2.1）。 */ export type JobStatus = | "queued" // 已创建，等待 worker 拉起 @@ -66,8 +68,10 @@ export interface RoundtableJob { activeAgentId: string | null; consensusPercentage: number; pendingClarification: string | null; - orchestrationMode: "recommend" | "chain"; + orchestrationMode: "recommend" | "chain" | "dag"; chain: { id: string; title: string } | null; + /** dag 模式分层编排计划（流程图按 stage 分层绘制）；非 dag 为 null。 */ + orchestrationPlan: OrchestrationPlan | null; error: string | null; /** 研讨对话记录（leader 说明 + 各席位交付），跑完 / 进行中都可看真实内容。 */ dialogues: JobDialogue[]; @@ -85,8 +89,10 @@ export interface StartJobPayload { /** 有序席位（recommend=选择顺序 / chain=链条顺序）。 */ agents: Array<{ agentId: string; name: string }>; model?: string; - orchestrationMode?: "recommend" | "chain"; + orchestrationMode?: "recommend" | "chain" | "dag"; chain?: { id: string; title: string } | null; + /** dag 模式分层编排计划（stages / 阶段目标 / 总控提示词 / finalSynthesis）。非 dag 省略。 */ + orchestrationPlan?: OrchestrationPlan | null; /** 显式第一轮 leader prompt（覆盖 intent）。 */ seedMessage?: string; /** 预建好的 thread 映射（Phase 3b 真网关用；模拟网关忽略）。 */ diff --git a/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx b/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx index 1231e129..1ee88874 100644 --- a/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx +++ b/frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx @@ -923,10 +923,13 @@ export default function RoundtablePlanningPage() { intent: buildInitialIntent(), agents, model: selectedModel || undefined, // 模型沿用第二步选择 - // 后台作业引擎暂不支持 dag(分层 DAG 只在前端 runDagOrchestration 跑); - // 挂起到后台时降级为 recommend(总控自由派活)。 - orchestrationMode: orchestrationMode === "dag" ? "recommend" : orchestrationMode, + // 后台作业引擎现已支持 dag（后端 _run_dag：stage 串行 / stage 内并行）—— + // 直接把当前编排模式透传，业务链的分层 / 并行 / 阶段目标 / 总控提示词全部保留。 + orchestrationMode, chain: chainMeta, + // dag 模式：把整份分层计划带给后端（含 stages / 阶段目标 / 总控提示词 / finalSynthesis）。 + // 非 dag 时为 null（后端按 mode 走 chain/recommend）。 + orchestrationPlan: orchestrationMode === "dag" ? orchestrationPlan : null, // 不传 threadIds：后台作业由真网关自建全新线程，绝不复用前端本地研讨的线程 // （那些线程可能有活跃 run，复用会撞 409）。 }); @@ -944,6 +947,7 @@ export default function RoundtablePlanningPage() { drafts, selectedModel, orchestrationMode, + orchestrationPlan, chainMeta, buildInitialIntent, resetToStep1, diff --git a/offline-backend-20260512/backend/app/gateway/roundtable_job_executor.py b/offline-backend-20260512/backend/app/gateway/roundtable_job_executor.py index a4538c55..b4279fd5 100644 --- a/offline-backend-20260512/backend/app/gateway/roundtable_job_executor.py +++ b/offline-backend-20260512/backend/app/gateway/roundtable_job_executor.py @@ -173,6 +173,9 @@ class StartParams: is_resume: bool = False # chain 模式来源链条（写进草稿 run 用，展示）。 chain: dict[str, Any] | None = None + # dag 模式分层编排计划（前端 OrchestrationPlan dict）：驱动引擎 _run_dag + 写进草稿 run + # 供前端流程图渲染。None = 非 dag（引擎按 mode 回退 chain/recommend）。 + orchestration_plan: dict[str, Any] | None = None # 是否生成 Step3 报告。默认沿用全局开关（当前临时禁用，见 ROUNDTABLE_BACKGROUND_REPORT_ENABLED）。 # 测试可显式传 True 覆盖以验证报告链路。 enable_report: bool = ROUNDTABLE_BACKGROUND_REPORT_ENABLED @@ -242,6 +245,7 @@ class RoundtableJobExecutor: mode=params.mode, intent_text=params.intent_text, enable_report=params.enable_report, + plan=params.orchestration_plan, ) # 终态：把最终对话 + step3 + jobStatus 落进同一个 run（含 furthest_step=3）。 await self._write_run_to_draft( @@ -323,6 +327,8 @@ class RoundtableJobExecutor: "seatStubMessages": _seat_stub_messages(dialogues), "orchestrationMode": params.mode, "chain": params.chain, + # dag 模式分层计划：让前端加载草稿后能像手动跑一样按 stage 画流程图。 + "orchestrationPlan": params.orchestration_plan, "step3": effective_step3, "jobId": params.job_id, "jobStatus": status, diff --git a/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-summary/SOUL.md b/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-summary/SOUL.md index b9e3d4b5..9f4a3b4d 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-summary/SOUL.md +++ b/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-summary/SOUL.md @@ -24,7 +24,13 @@ ## 产出要求（最重要） -你每一轮都要做**两件事**，缺一不可： +### ⓪ 先判断模式：用户是否要求了结构化输出 / 流程图 +当用户**明确要求**「结构化输出 / 结构化数据 / 流程图 / 格式化输出 / 可视化流程」时（要求可能写在业务链总控编排提示里，也可能是对话中的新消息），本轮进入**流程图模式**： +- **跳过下面的 A**——不要调用 `write_file`、不要生成任何 .md 文件； +- 把**精炼的文字版总结报告**直接写在对话回复里（文字展示给用户读，结构遵循用户要求）； +- 在回复**最末尾必须**输出一个完整的 ```flow-json 代码块（契约见「结构化流程数据」节）——这是本轮的**主产物**，绝不能省略。 + +未进入流程图模式时，每一轮都要做 A + B **两件事**，缺一不可： ### A. 写一个 Markdown 文件 - 用 `write_file` **一次性**把完整报告写入 `/mnt/user-data/outputs/方案总结报告.md`； @@ -51,8 +57,8 @@ 善用 Markdown 表格、有序 / 无序列表，把要点、阶段、里程碑、风险项、优劣势等结构化呈现，提升可读性。 -## 结构化流程数据（按需输出 flow-json） -当用户要求「结构化输出 / 结构化数据 / 流程图 / 可视化流程」时——要求可能写在业务链总控编排提示里，也可能是对话中的新消息——除了正常报告（文件 + 对话正文）外，再在**对话回复的最末尾**追加一个 ```flow-json 代码块，前端会据此**实时绘制可交互流程图**（边流式输出边画）： +## 结构化流程数据（flow-json 契约） +进入「流程图模式」时（见产出要求第 ⓪ 条：用户在编排提示或对话里明确要求了结构化输出 / 流程图），**不写 .md 文件**，在对话回复里给出精炼文字总结后，在**回复最末尾**追加一个 ```flow-json 代码块，前端会据此**实时绘制可交互流程图**（边流式输出边画）： ```flow-json {"title":"方案执行流程","nodes":[ @@ -71,18 +77,27 @@ - 用户没有要求结构化 / 流程图时，**不要**输出该块。 ## 多轮交互（用户会在输入框里继续和你对话） -用户读完报告后，可能在输入框里继续给你发消息。你要**自己判断**意图，分两种情况处理： +用户读完报告 / 流程图后，可能在输入框里继续给你发消息。你要**自己判断**意图——报告和流程图都是可修改的，分四种情况处理： -1. **提出修改意见**（如「把执行路线写得更细」「补一节预算分析」「风险部分太长，精简」）： +1. **修改报告**（如「把执行路线写得更细」「补一节预算分析」「风险部分太长，精简」）： - 在**原报告基础上做增量修改**——只改用户提到的地方，其余内容保持不变； - 用 `write_file` **整文件重写**到同一路径 `/mnt/user-data/outputs/方案总结报告.md`（`str_replace` 已被禁用，整文件重写才能让右侧预览实时刷新）； - - 然后在对话回复里给出修改后的完整报告正文。 + - 然后在对话回复里给出修改后的完整报告正文； + - 若此前已经输出过流程图、且这次修改影响了流程结构，可在回复末尾顺带输出更新后的完整 ```flow-json 块。 -2. **就报告 / 研讨内容提问**（如「为什么没采用方案 B？」「风险最高的是哪一项？」）： +2. **修改流程图**（如「流程图里加上验收环节」「把风险节点拆成两个」「换成按时间阶段排」）： + - **不要**改写 .md 报告文件； + - 在对话里简短说明改了什么，并在回复**最末尾**输出**完整修订版** ```flow-json 块——必须是**全量数据**（所有节点 + 连线），不是只给改动部分，前端会整体重绘。 + +3. **要求结构化输出 / 流程图**（如「总结成格式化输出」「画个流程图」「给我结构化数据」）： + - 进入流程图模式：**不写 / 不改 .md 文件**，正文给一段简短的文字说明，末尾输出完整 ```flow-json 块； + - 已有的 .md 报告**不要动**——用户随时可以在前端切回去看它。 + +4. **就报告 / 研讨内容提问**（如「为什么没采用方案 B？」「风险最高的是哪一项？」）： - **只在对话里用文字回答**，依据已有的研讨原料和报告内容作答； - - **不要**改写报告文件、**不要**重新 `write_file`。 + - **不要**改写报告文件、**不要**重新 `write_file`、**不要**输出 flow-json。 -判断不准时，倾向于「先简短回答，再问用户是否需要据此更新报告」。 +判断不准时，倾向于「先简短回答，再问用户是否需要据此更新报告 / 流程图」。 ## 禁止事项 - ❌ 生成 HTML / 绘制可视化页面 / 画图表（只写 Markdown 文本） @@ -91,6 +106,9 @@ - ❌ 写 Python / shell / 任何脚本去生成文件，或用 bash 跑脚本产出文件 - ❌ 写完文件后只回一句「已写入文件」而不给出报告正文（用户要在对话里读到报告） - ❌ 用户只是提问时还去改写报告文件 +- ❌ 用户要求了结构化输出 / 流程图，却只写 .md 报告而不输出 flow-json 块 +- ❌ 流程图模式下仍调用 `write_file` 生成 / 改写 .md 文件 +- ❌ 修订流程图时只输出改动的节点（必须输出全量 flow-json，前端整体重绘） - ❌ 在报告任何位置添加「AI 自动生成」「AI generated」之类的标签、徽章或水印 - ❌ 编造与原料明显矛盾的内容 - ❌ 忽视「业务链总控编排提示」的口径要求（它优先级最高） diff --git a/offline-backend-20260512/backend/app/gateway/routers/roundtable_jobs.py b/offline-backend-20260512/backend/app/gateway/routers/roundtable_jobs.py index 6a6e0af4..9bcfb7ba 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/roundtable_jobs.py +++ b/offline-backend-20260512/backend/app/gateway/routers/roundtable_jobs.py @@ -61,6 +61,9 @@ class StartJobRequest(BaseModel): model: str | None = None orchestrationMode: str = "recommend" chain: dict[str, Any] | None = None + # dag 模式分层编排计划（前端 OrchestrationPlan：{mode,stages,finalSynthesis,coordinatorPrompt?}）。 + # 仅 orchestrationMode=="dag" 时携带；驱动后端 _run_dag。缺失 → 引擎回退 recommend。 + orchestrationPlan: dict[str, Any] | None = None # 第一轮 leader prompt；缺省用 intent。 seedMessage: str | None = None # 预建好的 thread 映射（Phase 3b 真网关用；模拟网关忽略）。 @@ -96,6 +99,7 @@ class RoundtableJobResponse(BaseModel): pendingClarification: str | None = None orchestrationMode: str = "recommend" chain: dict[str, Any] | None = None + orchestrationPlan: dict[str, Any] | None = None error: str | None = None dialogues: list[dict[str, Any]] = Field(default_factory=list) step3: dict[str, Any] | None = None @@ -148,6 +152,7 @@ def _to_response(row: dict[str, Any]) -> RoundtableJobResponse: pendingClarification=row.get("pending_clarification"), orchestrationMode=row.get("orchestration_mode") or "recommend", chain=row.get("chain"), + orchestrationPlan=row.get("orchestration_plan"), error=row.get("error"), dialogues=row.get("dialogues") or [], step3=row.get("step3"), @@ -187,6 +192,7 @@ async def start_job(request: Request, body: StartJobRequest) -> RoundtableJobRes coordinator_name=body.coordinatorName, thread_ids=body.threadIds, chain=body.chain, + orchestration_plan=body.orchestrationPlan, dispatch_chain=_pending_nodes(body.agents), status="queued", ) @@ -204,6 +210,7 @@ async def start_job(request: Request, body: StartJobRequest) -> RoundtableJobRes coordinator_name=body.coordinatorName, draft_id=body.draftId, chain=body.chain, + orchestration_plan=body.orchestrationPlan, ) ) return _to_response(row) @@ -289,6 +296,7 @@ async def resume_job(request: Request, job_id: str, body: ResumeJobRequest) -> R draft_id=row.get("draft_id"), is_resume=True, chain=row.get("chain"), + orchestration_plan=row.get("orchestration_plan"), ) ) return _to_response(updated or row) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/__init__.py index 6e4720ea..2979b5cf 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/__init__.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/__init__.py @@ -20,13 +20,18 @@ from .types import ( CHAIN_CLOSEOUT_PROMPT, CONTINUE_PROMPT, MAX_CYCLES, + DagPlan, + DagStage, LeaderTurn, ReportResult, RoundtableGateway, SeatRef, SeatTurn, build_chain_leader_task, + build_dag_stage_leader_task, + parse_dag_plan, pick_chain_task, + pick_dag_seat_task, ) __all__ = [ @@ -40,9 +45,14 @@ __all__ = [ "SeatTurn", "ReportResult", "SeatRef", + "DagStage", + "DagPlan", "MAX_CYCLES", "CONTINUE_PROMPT", "CHAIN_CLOSEOUT_PROMPT", "build_chain_leader_task", "pick_chain_task", + "build_dag_stage_leader_task", + "parse_dag_plan", + "pick_dag_seat_task", ] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/engine.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/engine.py index 30203932..5a5bc352 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/engine.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/engine.py @@ -18,6 +18,7 @@ from __future__ import annotations +import asyncio import logging from dataclasses import dataclass, field from datetime import UTC, datetime @@ -27,11 +28,15 @@ from .types import ( CHAIN_CLOSEOUT_PROMPT, CONTINUE_PROMPT, MAX_CYCLES, + DagPlan, ReportResult, RoundtableGateway, SeatRef, build_chain_leader_task, + build_dag_stage_leader_task, + parse_dag_plan, pick_chain_task, + pick_dag_seat_task, ) logger = logging.getLogger(__name__) @@ -73,6 +78,7 @@ async def run_orchestration( mode: str = "recommend", intent_text: str = "", enable_report: bool = True, + plan: dict | None = None, ) -> OrchestrationResult: """运行一次完整编排（循环 → 共识 → 报告绘制）。 @@ -84,8 +90,10 @@ async def run_orchestration( coordinator_name: 总控 agent_id（如 "roundtable-coordinator"）。 model: 选定模型（None/"" 走后端默认）。 seed_message: 第一轮 leader 的 prompt（启动=初始意图；resume=用户答案续轮）。 - mode: "recommend" | "chain"。 + mode: "recommend" | "chain" | "dag"。 intent_text: 任务意图，用于报告 prompt（缺省退回 seed_message）。 + plan: dag 模式的分层编排计划（前端 ``OrchestrationPlan`` dict）。mode=="dag" 且 + 可解析出非空 stages 时走 ``_run_dag``；否则（dag 但 plan 缺失/为空）回退 recommend。 enable_report: 是否在收口后执行 Step3 报告绘制。False 时只跑完 Step2 研讨即收尾、 不生成报告（step3=None）。当前「后台挂起」临时禁用报告（进程内驱动报告 agent 在真模型下仍无法稳定写出 HTML，详见 multi-agent-background-run-遗留问题与下次继续.md）。 @@ -93,9 +101,13 @@ async def run_orchestration( intent = (intent_text or seed_message or "").strip() try: await progress.begin() - if mode == "chain": + 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) + elif mode == "chain": outcome = await _run_chain(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message, intent) else: + # recommend，或 dag 但 plan 缺失/为空 → 回退总控自由派活（绝不静默跑错语义）。 outcome = await _run_recommend(gateway, progress, agents, thread_ids, coordinator_name, model, seed_message, intent) if outcome.kind == "clarification": @@ -314,6 +326,108 @@ async def _run_chain( return _LoopOutcome("consensus", last_leader_content=last_leader, deliveries=deliveries) +# ── dag 模式（分层 DAG：stage 间串行、stage 内并行）────────────────────────────── + + +async def _run_dag( + gateway: RoundtableGateway, + progress: JobProgressWriter, + agents: list[SeatRef], + thread_ids: dict[str, str], + coordinator_name: str, + model: str | None, + seed_message: str, + intent_text: str, + plan: DagPlan, +) -> _LoopOutcome: + """分层 DAG 编排：每个 stage 一轮总控派活（钉死本批席位）→ 批内席位**并行**执行 + （``asyncio.gather``）→ 下一 stage 总控再派；全部 stage 完成后由总控收口综合。 + + 与前端 ``runDagOrchestration``「总控驱动每 stage 派活」同构（§12.4）。并行席位各自独立 + thread（``run_seat`` 内 ``thread_ids[agent_id]``），故进程内 gather 安全、无沙箱串台。 + """ + deliveries: dict[str, str] = {} + last_leader = "" + total = len(plan.stages) + name_by_id = {a.agent_id: a.name for a in agents} + + def _seat_ref(agent_id: str) -> SeatRef: + return SeatRef(agent_id, name_by_id.get(agent_id, agent_id)) + + for i, stage in enumerate(plan.stages): + stage_seats = [_seat_ref(aid) for aid in stage.agent_ids] + if not stage_seats: + continue + + # (a) 总控「派活轮」：钉死本 stage 席位，带任务背景 / 本阶段目标 / 总控编排指引 + 此前交付。 + await progress.leader_thinking(i + 1) + leader_task = build_dag_stage_leader_task( + seed_message=seed_message, + stage_index=i, + total_stages=total, + seats=stage_seats, + is_first=(i == 0), + goal=stage.goal, + coordinator_prompt=plan.coordinator_prompt, + ) + block = _deliveries_block(deliveries, name_by_id) + if block: + leader_task += f"\n\n此前各阶段席位已完成的交付如下（供你承上启下、为本批席位派活）：\n{block}" + turn = await gateway.run_leader( + thread_ids=thread_ids, + coordinator_name=coordinator_name, + message=leader_task, + model=model, + ) + if turn.final_text: + last_leader = turn.final_text + await progress.add_leader_dialogue( + turn.final_text, i + 1, tool_calls=turn.tool_calls, thread_id=thread_ids.get(coordinator_name) + ) + if turn.is_clarification: + return _LoopOutcome("clarification", question=turn.clarification, last_leader_content=last_leader, deliveries=deliveries) + + # (b) 批内并行：每个钉死席位拿到总控派给它的 task（漏派则给干净本职任务），并行执行。 + await progress.dispatching(i + 1) + for seat in stage_seats: + await progress.seat_active(seat.agent_id) + + async def _run_one(seat: SeatRef) -> 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_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 + await progress.add_seat_dialogue( + seat.agent_id, seat.name, seat_turn.final_text, i + 1, + tool_calls=seat_turn.tool_calls, thread_id=thread_ids.get(seat.agent_id), + ) + await progress.seat_done(seat.agent_id, i + 1) + + await asyncio.gather(*(_run_one(seat) for seat in stage_seats)) + + if total > 0: + await progress.set_consensus_pct(round((i + 1) / total * 90)) + + # (c) 收口：finalSynthesis 时总控综合全部 stage 交付，产出最终结论。 + if plan.final_synthesis and total > 0: + await progress.leader_thinking(total + 1) + block = _deliveries_block(deliveries, name_by_id) + closeout_msg = CHAIN_CLOSEOUT_PROMPT + (f"\n\n各阶段席位完整交付如下：\n{block}" if block else "") + closeout = await gateway.run_leader( + thread_ids=thread_ids, + coordinator_name=coordinator_name, + message=closeout_msg, + model=model, + ) + if closeout.final_text: + last_leader = closeout.final_text + await progress.add_leader_dialogue( + closeout.final_text, total + 1, tool_calls=closeout.tool_calls, thread_id=thread_ids.get(coordinator_name) + ) + + return _LoopOutcome("consensus", last_leader_content=last_leader, deliveries=deliveries) + + # ── 报告阶段 ───────────────────────────────────────────────────────────────── diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/simulated_gateway.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/simulated_gateway.py index d9e27820..2960cd41 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/simulated_gateway.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/simulated_gateway.py @@ -37,9 +37,10 @@ class SimulatedRoundtableGateway: async def run_leader(self, *, thread_ids, coordinator_name, message, model) -> LeaderTurn: await self._tick() self._leader_calls += 1 - if self._mode == "chain": - # chain 模式由引擎按链条顺序强制跑席位，leader 只给说明、不派活。 - return LeaderTurn(final_text="（模拟）总控：请当前席位按职责给出分析。", dispatched=[]) + if self._mode in ("chain", "dag"): + # chain / dag 模式由引擎按计划钉死席位强制执行，leader 只给说明、不显式派活 + # （引擎的 pick_*_task 会回退给席位干净本职任务）。 + return LeaderTurn(final_text="（模拟）总控：请本阶段席位按职责给出分析。", dispatched=[]) # recommend：首轮派活全部席位，次轮共识。 if self._leader_calls == 1: return LeaderTurn( diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/types.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/types.py index 3525493f..2d591491 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/types.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/types.py @@ -14,7 +14,7 @@ from __future__ import annotations from dataclasses import dataclass, field -from typing import Protocol, runtime_checkable +from typing import Any, Protocol, runtime_checkable # 编排循环最大轮次，与前端 roundtable-constants.ts MAX_CYCLES 对齐。 MAX_CYCLES = 8 @@ -195,3 +195,137 @@ def pick_chain_task( f"请你作为「{seat_name}」，基于任务背景与此前各席位的交付，" "从你的专业职责出发给出你的分析与交付。" ) + + +# ── dag 模式：分层编排计划 + prompt 构造（移植自前端 dag-types.ts）───────────────── +# +# DAG = 有序的 stage 列表：stage 间串行、stage 内并行。线性链是其特例（每层 1 席位）。 +# 与前端 `runDagOrchestration`「总控驱动每 stage 派活」同构（§3.2 方案 A + 批内并行）： +# 每个 stage 一轮总控派活（钉死本批席位）→ 批内席位并行执行 → 下一 stage 总控再派。 + + +@dataclass(frozen=True) +class DagStage: + """一个并行批次：``agent_ids`` 内的席位并行执行；``goal`` = 用户写的本阶段目标（可空）。""" + + agent_ids: list[str] + goal: str | None = None + + @property + def is_parallel(self) -> bool: + return len(self.agent_ids) > 1 + + +@dataclass(frozen=True) +class DagPlan: + """用户编排出的执行计划。``stages`` 顺序即串行顺序。""" + + stages: list[DagStage] + final_synthesis: bool = True + coordinator_prompt: str | None = None + + +def parse_dag_plan(raw: Any) -> DagPlan | None: + """把前端 ``OrchestrationPlan`` 形状的 dict 解析成 ``DagPlan``（容错）。 + + 无 stages / 全空 stage → 返回 None（调用方据此回退 recommend）。只保留有 agentId 的 stage。 + """ + if not isinstance(raw, dict): + return None + stages_raw = raw.get("stages") + if not isinstance(stages_raw, list): + return None + stages: list[DagStage] = [] + for s in stages_raw: + if not isinstance(s, dict): + continue + ids_raw = s.get("agentIds") or s.get("agent_ids") or [] + ids = [str(a) for a in ids_raw if str(a or "").strip()] + if not ids: + continue + goal = s.get("goal") + goal_text = str(goal).strip() if isinstance(goal, str) and goal.strip() else None + stages.append(DagStage(agent_ids=ids, goal=goal_text)) + if not stages: + return None + final_synthesis = raw.get("finalSynthesis") + if final_synthesis is None: + final_synthesis = raw.get("final_synthesis", True) + cp = raw.get("coordinatorPrompt") or raw.get("coordinator_prompt") + cp_text = str(cp).strip() if isinstance(cp, str) and cp.strip() else None + return DagPlan(stages=stages, final_synthesis=bool(final_synthesis), coordinator_prompt=cp_text) + + +def build_dag_stage_leader_task( + *, + seed_message: str, + stage_index: int, + total_stages: int, + seats: list[SeatRef], + is_first: bool, + goal: str | None = None, + coordinator_prompt: str | None = None, +) -> str: + """构造 dag「每 stage 派活轮」leader prompt：约束总控只对本 stage 钉死席位派活。 + + 对齐前端 ``buildDagStageLeaderTask``：首 stage 带任务背景，后续 stage 靠后端注入的此前 + 交付摘要承上；带「本阶段目标」与「总控编排指引（用户预设，最高优先）」。 + """ + ordinal = f"{stage_index + 1}/{total_stages}" + seat_list = "、".join(f"「{s.name}」（agent_name: {s.agent_id}）" for s in seats) + is_parallel = len(seats) > 1 + goal_text = (goal or "").strip() + coord_text = (coordinator_prompt or "").strip() + lines: list[str] = [] + if is_first: + lines += ["这是一次【分层 DAG 编排】的多智能体圆桌研讨。任务背景如下：", seed_message, ""] + else: + lines += ["分层研讨继续。系统已把此前各阶段席位的交付摘要注入到你本轮输入里，请据此承上启下。", ""] + if coord_text: + lines += [ + "【总控编排指引（用户预设，最高优先）】以下是本次业务链条为你（总控）预设的整体派活原则，请在每一阶段派活时都严格遵循：", + coord_text, + "", + ] + lines += [ + f"本研讨严格按预设分层顺序进行，共 {total_stages} 个阶段，现在是第 {ordinal} 个阶段。", + f"本阶段需要执行的席位（已由编排计划钉死）是：{seat_list}。", + ] + if goal_text: + lines += [ + f"【本阶段目标】{goal_text}", + "请据此目标为每个席位拆解任务，并要求他们显式承接上一阶段的产出来完成本阶段目标。", + ] + lines += [ + f"请你作为总控，通过 agent_orchestration 技能**分别**向上述{'这些席位' if is_parallel else '席位'}派活，" + f"结合任务背景{'、本阶段目标' if goal_text else ''}与此前交付，为每个席位分配聚焦其职责的本阶段任务。", + "只派给上述席位、且每个都必须派到；不要派给清单外的任何席位，也不要自己直接作答或给出最终结论。", + ] + if is_parallel: + lines.append(f"注意：本阶段这 {len(seats)} 个席位会并行执行，请确保给每个席位的 task 都能独立完成，不要让它们互相等待。") + return "\n".join(lines) + + +def pick_dag_seat_task( + dispatched: list[tuple[str, str]], + seat_agent_id: str, + seat_name: str, + *, + goal: str | None = None, +) -> str: + """从总控「每 stage 派活轮」终态取派给某钉死席位的 task（移植自前端 pickDagSeatTask）。 + + 1) 命中 agent_name 完全一致的派活 → 用它的 task; + 2) 未命中 / 总控只写文字没派活 → **不**把总控派活叙述喂给席位（防席位模仿派活），改给干净 + 本职任务（任务背景 + 此前交付由 ``_seat_message`` 另行注入）。 + 两条分支都带「本阶段目标」兜底。 + """ + goal_rule = f"\n【本阶段目标】{goal.strip()}\n请确保你的交付服务于该目标，并显式承接上一阶段的产出。" if (goal or "").strip() else "" + wanted = seat_agent_id.lower() + for name, task in dispatched: + if name.lower() == wanted and (task or "").strip(): + return f"请你作为「{seat_name}」，完成总控派给你的本阶段任务：\n{task.strip()}{goal_rule}" + return ( + f"请你作为「{seat_name}」，从你的专业职责出发，独立完成本阶段属于你的分析与交付。" + f"此前各阶段席位的交付已在上下文里，请据此承接。{goal_rule}" + ) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260612_01_roundtable_jobs_orchestration_plan.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260612_01_roundtable_jobs_orchestration_plan.py new file mode 100644 index 00000000..f7a2cf90 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260612_01_roundtable_jobs_orchestration_plan.py @@ -0,0 +1,47 @@ +"""Add orchestration_plan column to roundtable_jobs. + +后台挂起支持分层 DAG 编排（业务链并行）：把前端 ``OrchestrationPlan``（stages / +阶段目标 / 总控提示词 / finalSynthesis）整份存进 job 行，驱动后端 ``_run_dag`` +（stage 串行、stage 内并行）+ 前端大弹窗按 stage 分层画流程图 + resume 时重建计划。 +LONGTEXT（计划 JSON 可超 64 KB）。 + +Revision ID: 20260612_01 +Revises: 20260611_01 +Create Date: 2026-06-12 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op +from sqlalchemy import inspect +from sqlalchemy.dialects.mysql import LONGTEXT + + +revision = "20260612_01" +down_revision = "20260611_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_jobs") and not _has_column("roundtable_jobs", "orchestration_plan"): + col_type = LONGTEXT() if op.get_bind().dialect.name == "mysql" else sa.Text() + op.add_column("roundtable_jobs", sa.Column("orchestration_plan", col_type, nullable=True)) + + +def downgrade() -> None: + if _has_column("roundtable_jobs", "orchestration_plan"): + op.drop_column("roundtable_jobs", "orchestration_plan") diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_jobs/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_jobs/model.py index 1c4c2034..6555417a 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_jobs/model.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_jobs/model.py @@ -51,10 +51,14 @@ class RoundtableJobRow(Base): # ThreadIdMap: JSON object { "": "", ... }. thread_ids: Mapped[str | None] = mapped_column(PortableLongText, nullable=True) coordinator_name: Mapped[str | None] = mapped_column(String(128), nullable=True) - # "recommend"（总控自由派活）| "chain"（按业务链条顺序派活）. + # "recommend"（总控自由派活）| "chain"（按业务链条顺序派活）| "dag"（分层 DAG：stage 间串行、stage 内并行）. orchestration_mode: Mapped[str] = mapped_column(String(16), nullable=False, default="recommend") # chain 模式来源链条快照：JSON object { id, title }（recommend 模式为 null）. chain: Mapped[str | None] = mapped_column(PortableLongText, nullable=True) + # dag 模式分层编排计划：JSON object { mode, stages:[{id,agentIds,goal?}], finalSynthesis, + # coordinatorPrompt? }（对齐前端 OrchestrationPlan）。仅 dag 模式有值；驱动后端 _run_dag + # （stage 串行 / stage 内并行）+ 前端大弹窗按 stage 分层画流程图 + resume 时重建计划。 + orchestration_plan: Mapped[str | None] = mapped_column(PortableLongText, nullable=True) # 研讨对话记录：JSON array of { id, role('leader'|'seat'), agentId, name, content, cycle }。 # 让用户跑完 / 进行中都能看到真实研讨内容（leader 说明 + 各席位交付），而非只有流程图。 dialogues: Mapped[str | None] = mapped_column(PortableLongText, nullable=True) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_jobs/sql.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_jobs/sql.py index e9e9a92a..e7e0995b 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_jobs/sql.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/roundtable_jobs/sql.py @@ -55,7 +55,7 @@ class RoundtableJobRepository: for k in ("created_at", "updated_at"): if isinstance(d.get(k), datetime): d[k] = d[k].isoformat() - for k in ("dispatch_chain", "thread_ids", "chain", "dialogues", "step3"): + for k in ("dispatch_chain", "thread_ids", "chain", "orchestration_plan", "dialogues", "step3"): d[k] = _loads(d.get(k)) return d @@ -71,6 +71,7 @@ class RoundtableJobRepository: coordinator_name: str | None = None, thread_ids: dict[str, str] | None = None, chain: dict[str, Any] | None = None, + orchestration_plan: dict[str, Any] | None = None, dispatch_chain: list[dict[str, Any]] | None = None, status: str = "queued", phase: str = "initializing", @@ -92,6 +93,7 @@ class RoundtableJobRepository: coordinator_name=coordinator_name, orchestration_mode=orchestration_mode, chain=_dumps(chain), + orchestration_plan=_dumps(orchestration_plan), error=None, created_at=now, updated_at=now, diff --git a/offline-backend-20260512/backend/scripts/probe_roundtable_dag_job.py b/offline-backend-20260512/backend/scripts/probe_roundtable_dag_job.py new file mode 100644 index 00000000..4796febb --- /dev/null +++ b/offline-backend-20260512/backend/scripts/probe_roundtable_dag_job.py @@ -0,0 +1,110 @@ +"""后台挂起「分层 DAG」实测脚本：带一个并行 stage 的业务链。 + +POST 启动一个 dag 作业（stage1 两席位并行 → stage2 → stage3），轮询打印每帧的 +dispatchChain 节点状态，直观看到：①stage1 两席位**同时 active**（并行）；②stage2/3 +在前一 stage 全 done 后才 active（串行）；③orchestrationPlan 在响应里回得来（持久化生效）。 + +用法：PYTHONPATH=. .venv/Scripts/python scripts/probe_roundtable_dag_job.py +环境：GATEWAY_URL 默认 http://127.0.0.1:8001；AUTH_COOKIE 可选。 +""" + +from __future__ import annotations + +import asyncio +import os +import time + +import httpx + +GATEWAY_URL = os.getenv("GATEWAY_URL", "http://127.0.0.1:8001").rstrip("/") +AUTH_COOKIE = os.getenv("AUTH_COOKIE", "") +AUTH_BEARER = os.getenv("AUTH_BEARER", "") + +# 4 个已 seed 的圆桌席位。stage1 = 风险审查 ‖ 方案设计（并行），stage2 = 执行规划，stage3 = 总结。 +AGENTS = [ + {"agentId": "roundtable-risk-review", "name": "风险审查"}, + {"agentId": "roundtable-solution-design", "name": "方案设计"}, + {"agentId": "roundtable-execution-plan", "name": "执行规划"}, + {"agentId": "roundtable-summary", "name": "方案总结"}, +] +PLAN = { + "mode": "dag", + "stages": [ + {"id": "s0", "agentIds": ["roundtable-risk-review", "roundtable-solution-design"], "goal": "并行评估风险并产出初步方案"}, + {"id": "s1", "agentIds": ["roundtable-execution-plan"], "goal": "据风险与方案制定执行规划"}, + {"id": "s2", "agentIds": ["roundtable-summary"]}, + ], + "finalSynthesis": True, + "coordinatorPrompt": "先并行评估、再规划执行、最后总结，逐阶段承接。", +} +INTENT = "为一家中型制造企业设计一套数字化转型的整体方案" +TERMINAL = {"done", "error", "cancelled"} + + +def _headers() -> dict[str, str]: + h = {"Content-Type": "application/json"} + if AUTH_COOKIE: + h["Cookie"] = AUTH_COOKIE + if AUTH_BEARER: + h["Authorization"] = f"Bearer {AUTH_BEARER}" + return h + + +def _chain_str(nodes: list[dict]) -> str: + sym = {"pending": ".", "active": ">RUN", "done": "DONE"} + return " ".join(f"{n['name']}={sym.get(n['state'], '?')}" for n in nodes) + + +async def main() -> None: + async with httpx.AsyncClient(timeout=30.0, trust_env=False) as client: + print(f"[probe] POST {GATEWAY_URL}/api/roundtable-jobs mode=dag (stage1 并行)") + resp = await client.post( + f"{GATEWAY_URL}/api/roundtable-jobs", + headers=_headers(), + json={"intent": INTENT, "agents": AGENTS, "orchestrationMode": "dag", "orchestrationPlan": PLAN}, + ) + if resp.status_code != 201: + print(f"[probe] start failed: HTTP {resp.status_code}: {resp.text[:500]}") + return + job = resp.json() + job_id = job["id"] + # 验证 plan 在响应里回得来（持久化 + _to_response 生效） + plan_back = job.get("orchestrationPlan") + print(f"[probe] started {job_id} orchestrationMode={job.get('orchestrationMode')} " + f"orchestrationPlan stages={len((plan_back or {}).get('stages') or [])}") + + last = "" + t0 = time.perf_counter() + saw_parallel = False + while True: + await asyncio.sleep(1.0) + r = await client.get(f"{GATEWAY_URL}/api/roundtable-jobs/{job_id}", headers=_headers()) + if r.status_code != 200: + print(f"[probe] get failed: HTTP {r.status_code}: {r.text[:300]}") + return + j = r.json() + nodes = j.get("dispatchChain") or [] + active_seats = [n["name"] for n in nodes if n["state"] == "active"] + if len(active_seats) >= 2: + saw_parallel = True + line = f"status={j['status']} phase={j['phase']} consensus={j['consensusPercentage']}% [{_chain_str(nodes)}]" + if line != last: + print(f"[probe] +{time.perf_counter() - t0:5.1f}s {line}") + last = line + if j["status"] in TERMINAL: + print(f"\n[probe] terminal: {j['status']}") + if j["status"] == "error": + print(f"[probe] error detail: {j.get('error')}") + # 对话顺序证明 stage 串行 + seats = [d for d in (j.get("dialogues") or []) if d["role"] == "seat"] + print(f"[probe] 席位交付顺序: {[d['agentId'] for d in seats]}") + leaders = [d for d in (j.get("dialogues") or []) if d["role"] == "leader"] + print(f"[probe] 总控发言轮数: {len(leaders)}（3 stage 派活 + 1 收口 = 4）") + print(f"[probe] 是否观察到 stage 内并行（≥2 席位同时 active）: {'是 ✅' if saw_parallel else '否 ❌'}") + sample = next((d["content"] for d in seats if d.get("content")), "") + print(f"[probe] 席位交付样例: {sample[:80]}") + return + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/offline-backend-20260512/backend/scripts/sql/column_additions_mysql.sql b/offline-backend-20260512/backend/scripts/sql/column_additions_mysql.sql index 0a0daee2..330b6c41 100644 --- a/offline-backend-20260512/backend/scripts/sql/column_additions_mysql.sql +++ b/offline-backend-20260512/backend/scripts/sql/column_additions_mysql.sql @@ -77,6 +77,11 @@ CALL _add_column_if_missing('agents', 'square_id', "VARCHAR(64) NOT NULL DEFAULT -- String(128), nullable=True, index=True。 CALL _add_column_if_missing('roundtable_drafts', 'task_id', 'VARCHAR(128) NULL DEFAULT NULL'); +-- orchestration_plan: dag 模式分层编排计划 JSON(stages/阶段目标/总控提示词/finalSynthesis), +-- 对应 migration 20260612_01。驱动后端 _run_dag + 前端大弹窗分层流程图。 +-- ORM: RoundtableJobRow.orchestration_plan = PortableLongText(LONGTEXT), nullable。 +CALL _add_column_if_missing('roundtable_jobs', 'orchestration_plan', 'LONGTEXT NULL DEFAULT NULL'); + -- ---- 清理临时存储过程 -------------------------------------------------------- DROP PROCEDURE IF EXISTS _add_column_if_missing;