圆桌第三步流程图模式 + 后台 DAG 任务编排

- 方案总结报告:用户要求结构化输出/流程图时跳过 md 直出 flow-json 流程图,
  报告用文字展示;报告与流程图均可多轮修改、互相切换(detectFlowIntent 识别意图)
- 流程图画布放右侧沙箱位(sidePanel 插槽)不覆盖消息列表,可与 md 预览切换;
  节点随容器宽度自适应缩放并 fitView 居中
- 后台挂起支持真 DAG 编排(分层并行/阶段目标/总控提示词),新增
  orchestration_plan 持久化列 + 迁移;大弹窗流程图按 DAG 分层渲染

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
董宏凯andClaude Fable 5 committed 2026-06-12 13:52:17 +08:00
1 parent b33426416b
commit 086c280e5d
25 files changed
+880 -149

No files matched your search

@@ -65,6 +65,7 @@ function mapJob(raw: Record<string, unknown>): 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",
@@ -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
@@ -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<OrchestrationStage> | null;
}
const _NON_TERMINAL: ReadonlySet<JobStatus> = new Set<JobStatus>(["queued", "running", "awaiting_input"]);
@@ -105,6 +111,8 @@ export function DispatchChainModal({
onCancel,
error,
dialogues,
orchestrationMode,
stages,
}: DispatchChainModalProps) {
const [answer, setAnswer] = useState("");
const convoRef = useRef<HTMLDivElement>(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<string, ChainNodeState> = {};
const nodeByAgent = new Map<string, DispatchChainNode>();
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 (
<Dialog open={open} onOpenChange={onOpenChange}>
<DialogContent className="flex max-h-[90vh] w-[94vw] flex-col gap-0 sm:max-w-[1240px]">
@@ -167,51 +197,65 @@ export function DispatchChainModal({
<DialogDescription>{PHASE_LABEL[phase]}</DialogDescription>
</DialogHeader>
{/* 横向流程图(横向滚动) */}
<div className="custom-scrollbar shrink-0 overflow-x-auto overflow-y-hidden py-8">
<div className="flex min-w-max items-start gap-0 px-6">
{/* 总控节点 */}
<FlowNode
label="总控协调"
avatarType="human"
state={coordinatorState}
/>
{/* 流程图:dag 模式按 stage 分层(DAG);否则线性总控→席位→共识→报告 */}
{showDag ? (
// 分层 DAG 流程图:总控 → ┌Stage┐(并行席位纵向并列) → … → 🤝共识。
// 后台本期不绘报告(Step3 禁用) → 不显示「结果绘制」里程碑。
<DagFlowChart
stages={stages!}
resolveSeat={resolveDagSeat}
seatStates={dagSeatStates}
coordinatorState={coordinatorState}
consensusState={dagConsensusState}
showReportMilestone={false}
className="shrink-0"
/>
) : (
<div className="custom-scrollbar shrink-0 overflow-x-auto overflow-y-hidden py-8">
<div className="flex min-w-max items-start gap-0 px-6">
{/* 总控节点 */}
<FlowNode
label="总控协调"
avatarType="human"
state={coordinatorState}
/>
{/* 席位节点 */}
{nodes.map((node) => (
<Fragment key={node.agentId}>
<Connector
active={node.state === "active"}
done={node.state === "done"}
/>
<FlowNode
label={node.name}
avatarType={node.avatarType}
state={node.state}
order={node.order}
/>
</Fragment>
))}
{/* 席位节点 */}
{nodes.map((node) => (
<Fragment key={node.agentId}>
<Connector
active={node.state === "active"}
done={node.state === "done"}
/>
<FlowNode
label={node.name}
avatarType={node.avatarType}
state={node.state}
order={node.order}
/>
</Fragment>
))}
{/* 共识里程碑 */}
<Connector active={consensusActive} done={consensusDone} />
<MilestoneNode
label="共识达成"
emoji="🤝"
state={
consensusActive ? "active" : consensusDone ? "done" : "pending"
}
/>
{/* 共识里程碑 */}
<Connector active={consensusActive} done={consensusDone} />
<MilestoneNode
label="共识达成"
emoji="🤝"
state={
consensusActive ? "active" : consensusDone ? "done" : "pending"
}
/>
{/* 报告绘制里程碑 */}
<Connector active={reportActive} done={reportDone} />
<MilestoneNode
label="结果绘制"
emoji="📊"
state={reportActive ? "active" : reportDone ? "done" : "pending"}
/>
{/* 报告绘制里程碑 */}
<Connector active={reportActive} done={reportDone} />
<MilestoneNode
label="结果绘制"
emoji="📊"
state={reportActive ? "active" : reportDone ? "done" : "pending"}
/>
</div>
</div>
</div>
)}
{/* 研讨对话(leader 说明 + 各席位交付,真网关下即真模型真实对话) */}
{dialogues && dialogues.length > 0 && (
@@ -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 (
<div
style={{
background: meta.bg,
borderColor: meta.border,
color: meta.text,
width: FLOW_NODE_W,
width: scaledWidth,
}}
className="cursor-pointer rounded-xl border-2 px-3 py-2 shadow-sm transition hover:shadow-md"
title="点击查看详情"
@@ -78,11 +84,11 @@ function FlowCardNode({ data }: NodeProps) {
<Handle type="target" position={Position.Top} className="!h-1.5 !w-1.5 !bg-slate-400" />
<div className="mb-0.5 flex items-center gap-1.5">
<span style={{ background: meta.dot }} className="size-2 shrink-0 rounded-full" />
<span className="text-[9px] font-bold uppercase tracking-wider opacity-70">
<span className="text-[9px] font-bold uppercase tracking-wider opacity-70" style={{ fontSize: `${9 * scale}px` }}>
{meta.label}
</span>
</div>
<div className="break-words text-[11.5px] font-bold leading-snug">{node.label}</div>
<div className="break-words font-bold leading-snug" style={{ fontSize: `${11.5 * scale}px` }}>{node.label}</div>
<Handle type="source" position={Position.Bottom} className="!h-1.5 !w-1.5 !bg-slate-400" />
</div>
);
@@ -99,22 +105,48 @@ function FlowCanvasInner({
fullscreen,
onToggleFullscreen,
onClose,
onSwitchToReport,
}: FlowGraphPanelProps & { fullscreen: boolean; onToggleFullscreen: () => void }) {
const rf = useReactFlow();
const [selectedId, setSelectedId] = useState<string | null>(null);
const [scale, setScale] = useState(1);
const containerRef = useRef<HTMLDivElement>(null);
const fitTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
// 用户手动拖过的节点位置 —— 数据增量更新重新布局时保留,不被布局覆盖。
const manualPosRef = useRef<Map<string, { x: number; y: number }>>(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<Node[]>(derivedNodes);
@@ -180,6 +212,17 @@ function FlowCanvasInner({
)}
</div>
<div className="flex shrink-0 items-center gap-1">
{onSwitchToReport && (
<button
type="button"
title="切回 Markdown 报告预览(流程图可随时再打开)"
onClick={onSwitchToReport}
className="mr-1 inline-flex h-7 items-center gap-1 rounded-lg border border-slate-200/70 bg-white px-2 text-[10.5px] font-bold text-slate-600 transition hover:border-primary/40 hover:text-primary"
>
<FileText className="h-3.5 w-3.5" />
MD 报告
</button>
)}
<ToolButton title="放大" onClick={() => void rf.zoomIn({ duration: 150 })}>
<ZoomIn className="h-3.5 w-3.5" />
</ToolButton>
@@ -202,7 +245,7 @@ function FlowCanvasInner({
</div>
{/* 画布 */}
<div className="relative min-h-0 flex-1">
<div ref={containerRef} className="relative min-h-0 flex-1">
<ReactFlow
nodes={nodes}
edges={edges}
@@ -377,13 +420,14 @@ function ConnectionList({
// ── 外壳:覆盖右栏 / 全屏切换 ───────────────────────────────────────────────
export function FlowGraphPanel({ data, streaming, onClose }: FlowGraphPanelProps) {
export function FlowGraphPanel({ data, streaming, onClose, onSwitchToReport }: FlowGraphPanelProps) {
const [fullscreen, setFullscreen] = useState(false);
return (
<div
className={cn(
"flex flex-col bg-white",
fullscreen ? "fixed inset-0 z-[80]" : "absolute inset-0 z-20 rounded-2xl",
// 非全屏=填满所在面板(Step3 放在右侧沙箱位,不覆盖左侧消息列表);全屏才整页覆盖。
fullscreen ? "fixed inset-0 z-[80]" : "relative h-full min-h-0 w-full",
)}
>
<ReactFlowProvider>
@@ -393,6 +437,7 @@ export function FlowGraphPanel({ data, streaming, onClose }: FlowGraphPanelProps
fullscreen={fullscreen}
onToggleFullscreen={() => setFullscreen((f) => !f)}
onClose={onClose}
onSwitchToReport={onSwitchToReport}
/>
</ReactFlowProvider>
</div>
@@ -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";
@@ -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<GroupImperativeHandle>(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 (
<ThreadContext.Provider value={{ thread: threadStub, isMock: false }}>
<div className="h-full min-h-0 flex-1 bg-[#fffdfa]">{artifactPanelBody}</div>
<div className="h-full min-h-0 flex-1 bg-[#fffdfa]">{sidePanel ?? artifactPanelBody}</div>
</ThreadContext.Provider>
);
}
@@ -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",
)}
/>
<ResizablePanel
className={cn(
"bg-[#fffdfa] transition-all duration-300 ease-in-out",
!artifactsOpen && "opacity-0",
!panelOpen && "opacity-0",
)}
id="artifacts"
>
{artifactPanelBody}
{sidePanel ?? artifactPanelBody}
</ResizablePanel>
</ResizablePanelGroup>
</ThreadContext.Provider>
@@ -329,7 +329,7 @@ export function Step3ReportMode({
dialogues={dialogues}
error={error}
themeMismatch={themeMismatch}
onRetry={redraw}
onRetry={() => redraw()}
onRevisionSubmit={interject}
onArtifactStepClick={openArtifactFromStep}
onArtifactPreview={openArtifactPreview}
@@ -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({
)}
</aside>
{/* 右栏:智能体进度 + 沙箱(复用第 2 步)。流程图画布作为覆盖层叠在其上。 */}
{/* 右栏:智能体进度 + 沙箱(复用第 2 步)。流程图画布放在右侧沙箱位
(sidePanel 插槽),消息列表保持可见,可与 md 沙箱预览互相切换。 */}
<section className="relative flex-1 min-w-0 rounded-2xl border border-slate-200/60 bg-white overflow-hidden">
<Step2SandboxLayout
activeThreadId={activeArtifactThreadId ?? summaryThreadId}
isStreaming={isStreaming}
seatMessages={activeSeatMessages}
sidePanel={
flowGraph && flowView ? (
<FlowGraphPanel
data={flowGraph.graph}
streaming={flowGraph.streaming}
onClose={() => setFlowView(false)}
onSwitchToReport={
mdArtifact
? () => {
setFlowView(false);
openArtifactPreview(mdArtifact.threadId, mdArtifact.path);
}
: undefined
}
/>
) : undefined
}
>
<AgentProgressPanel
phase={phase}
dialogues={displayDialogues}
error={error}
busy={busy}
mdAvailable={!!mdArtifact}
onRetry={redraw}
onAsk={ask}
onAsk={handleAsk}
onPause={pause}
onArtifactStepClick={openArtifactFromStep}
onArtifactPreview={openArtifactPreview}
@@ -381,14 +435,6 @@ export function Step3SummaryMode({
</button>
)}
{/* 流程图画布(覆盖右栏;全屏在组件内部切换) */}
{flowGraph && flowView && (
<FlowGraphPanel
data={flowGraph.graph}
streaming={flowGraph.streaming}
onClose={() => setFlowView(false)}
/>
)}
</section>
</div>
@@ -498,6 +544,7 @@ function AgentProgressPanel({
dialogues,
error,
busy,
mdAvailable,
onRetry,
onAsk,
onPause,
@@ -512,6 +559,8 @@ function AgentProgressPanel({
dialogues: ReturnType<typeof useStep3Summary>["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 && (
<div className="inline-flex items-center gap-2 rounded-full border bg-white px-2.5 py-1 text-[11px] text-slate-500">
<span className="bg-primary size-1.5 animate-pulse rounded-full" />
报告正文实时输出,右侧沙箱同步预览 Markdown
{mdAvailable
? "报告正文实时输出,右侧沙箱同步预览 Markdown"
: "报告正文实时输出;输出结构化数据时流程图画布同步绘制"}
</div>
)}
@@ -638,7 +689,9 @@ function AgentProgressPanel({
{phase === "done" && (
<div className="flex items-center gap-2 text-xs font-bold text-emerald-600">
<CheckCircle className="w-4 h-4" />
已生成方案总结报告,右侧可预览 / 下载 .md;可在下方继续提修改意见或提问。
{mdAvailable
? "已生成方案总结报告,右侧可预览 / 下载 .md;可在下方继续修改报告 / 流程图或提问。"
: "已生成方案总结,流程图可在画布查看;可在下方继续修改总结 / 流程图或提问。"}
</div>
)}
</div>
@@ -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"
/>
@@ -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<void> => {
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<string, StubAiMessage[]> = {
[synthThreadId]: [
{
id: messageId,
type: "ai",
content: "",
tool_calls: [
const stub: Record<string, StubAiMessage[]> = 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],
);
@@ -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",
@@ -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)。
* 块未闭合(流式中)→「正在输出」措辞;已闭合 → 引导去看流程图画布。
@@ -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 真网关用;模拟网关忽略)。 */
@@ -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,
@@ -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,
@@ -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」之类的标签、徽章或水印
- ❌ 编造与原料明显矛盾的内容
- ❌ 忽视「业务链总控编排提示」的口径要求(它优先级最高)
@@ -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)
@@ -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",
]
@@ -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)
# ── 报告阶段 ─────────────────────────────────────────────────────────────────
@@ -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(
@@ -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}"
)
@@ -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")
@@ -51,10 +51,14 @@ class RoundtableJobRow(Base):
# ThreadIdMap: JSON object { "<agent_id>": "<thread_id>", ... }.
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)
@@ -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,
@@ -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())
@@ -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;